Improved generic prepareNext() and stats() handling
This commit is contained in:
parent
5ad6679783
commit
0d5719f76c
2 changed files with 239 additions and 234 deletions
|
@ -57,7 +57,6 @@ namespace Mist {
|
||||||
maxSkipAhead = 7500;
|
maxSkipAhead = 7500;
|
||||||
minSkipAhead = 5000;
|
minSkipAhead = 5000;
|
||||||
realTime = 1000;
|
realTime = 1000;
|
||||||
completeKeyReadyTimeOut = false;
|
|
||||||
if (myConn){
|
if (myConn){
|
||||||
setBlocking(true);
|
setBlocking(true);
|
||||||
}else{
|
}else{
|
||||||
|
@ -172,7 +171,7 @@ namespace Mist {
|
||||||
//wait max 10 seconds for sync
|
//wait max 10 seconds for sync
|
||||||
while ((!tmpEx.getSync() || tmpEx.getSync() == 2) && i++ < 100){
|
while ((!tmpEx.getSync() || tmpEx.getSync() == 2) && i++ < 100){
|
||||||
Util::wait(100);
|
Util::wait(100);
|
||||||
stats();
|
stats(true);
|
||||||
tmpEx = IPC::statExchange(statsPage.getData());
|
tmpEx = IPC::statExchange(statsPage.getData());
|
||||||
}
|
}
|
||||||
HIGH_MSG("USER_NEW sync achieved: %u", (unsigned int)tmpEx.getSync());
|
HIGH_MSG("USER_NEW sync achieved: %u", (unsigned int)tmpEx.getSync());
|
||||||
|
@ -252,7 +251,7 @@ namespace Mist {
|
||||||
onFail();
|
onFail();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
stats();
|
stats(true);
|
||||||
updateMeta();
|
updateMeta();
|
||||||
if (myMeta.live && !isReadyForPlay()){
|
if (myMeta.live && !isReadyForPlay()){
|
||||||
unsigned long long waitUntil = Util::epoch() + 15;
|
unsigned long long waitUntil = Util::epoch() + 15;
|
||||||
|
@ -506,7 +505,7 @@ namespace Mist {
|
||||||
}else{
|
}else{
|
||||||
nxtKeyNum[trackId] = 0;
|
nxtKeyNum[trackId] = 0;
|
||||||
}
|
}
|
||||||
stats();
|
stats(true);
|
||||||
Util::wait(100);
|
Util::wait(100);
|
||||||
pageNum = pageNumForKey(trackId, keyNum);
|
pageNum = pageNumForKey(trackId, keyNum);
|
||||||
}
|
}
|
||||||
|
@ -516,8 +515,7 @@ namespace Mist {
|
||||||
}else{
|
}else{
|
||||||
nxtKeyNum[trackId] = 0;
|
nxtKeyNum[trackId] = 0;
|
||||||
}
|
}
|
||||||
stats();
|
stats(true);
|
||||||
nxtKeyNum[trackId] = pageNum;
|
|
||||||
|
|
||||||
if (currKeyOpen.count(trackId) && currKeyOpen[trackId] == (unsigned int)pageNum){
|
if (currKeyOpen.count(trackId) && currKeyOpen[trackId] == (unsigned int)pageNum){
|
||||||
return;
|
return;
|
||||||
|
@ -526,7 +524,7 @@ namespace Mist {
|
||||||
snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_DATA, streamName.c_str(), trackId, pageNum);
|
snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_DATA, streamName.c_str(), trackId, pageNum);
|
||||||
nProxy.curPage[trackId].init(id, DEFAULT_DATA_PAGE_SIZE);
|
nProxy.curPage[trackId].init(id, DEFAULT_DATA_PAGE_SIZE);
|
||||||
if (!(nProxy.curPage[trackId].mapped)){
|
if (!(nProxy.curPage[trackId].mapped)){
|
||||||
DEBUG_MSG(DLVL_FAIL, "Initializing page %s failed", nProxy.curPage[trackId].name.c_str());
|
FAIL_MSG("Initializing page %s failed", nProxy.curPage[trackId].name.c_str());
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
currKeyOpen[trackId] = pageNum;
|
currKeyOpen[trackId] = pageNum;
|
||||||
|
@ -909,8 +907,8 @@ namespace Mist {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
/*LTS-END*/
|
/*LTS-END*/
|
||||||
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler started");
|
DONTEVEN_MSG("MistOut client handler started");
|
||||||
while (config->is_active && myConn.connected() && (wantRequest || parseData)){
|
while (config->is_active && myConn && (wantRequest || parseData)){
|
||||||
if (wantRequest){
|
if (wantRequest){
|
||||||
requestHandler();
|
requestHandler();
|
||||||
}
|
}
|
||||||
|
@ -919,21 +917,84 @@ namespace Mist {
|
||||||
initialize();
|
initialize();
|
||||||
}
|
}
|
||||||
if ( !sentHeader){
|
if ( !sentHeader){
|
||||||
DEBUG_MSG(DLVL_DONTEVEN, "sendHeader");
|
DONTEVEN_MSG("sendHeader");
|
||||||
sendHeader();
|
sendHeader();
|
||||||
}
|
}
|
||||||
prepareNext();
|
if (!sought){
|
||||||
|
if (myMeta.live){
|
||||||
|
long unsigned int mainTrack = getMainSelectedTrack();
|
||||||
|
//cancel if there are no keys in the main track
|
||||||
|
if (!myMeta.tracks.count(mainTrack) || !myMeta.tracks[mainTrack].keys.size()){break;}
|
||||||
|
//seek to the newest keyframe, unless that is <5s, then seek to the oldest keyframe
|
||||||
|
unsigned long long seekPos = myMeta.tracks[mainTrack].keys.rbegin()->getTime();
|
||||||
|
if (seekPos < 5000){
|
||||||
|
seekPos = myMeta.tracks[mainTrack].keys.begin()->getTime();
|
||||||
|
}
|
||||||
|
seek(seekPos);
|
||||||
|
}else{
|
||||||
|
seek(0);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (prepareNext()){
|
||||||
if (thisPacket){
|
if (thisPacket){
|
||||||
|
|
||||||
|
|
||||||
|
//slow down processing, if real time speed is wanted
|
||||||
|
if (realTime){
|
||||||
|
while (thisPacket.getTime() > (((Util::getMS() - firstTime)*1000)+maxSkipAhead)/realTime) {
|
||||||
|
Util::sleep(std::min(thisPacket.getTime() - (((Util::getMS() - firstTime)*1000)+minSkipAhead)/realTime, 1000llu));
|
||||||
|
stats();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
//delay the stream until its current keyframe is complete, if only complete keys wanted
|
||||||
|
if (completeKeysOnly){
|
||||||
|
bool completeKeyReady = false;
|
||||||
|
int timeoutTries = 40;//wait default 250ms*40=10 seconds
|
||||||
|
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
|
||||||
|
if (it->second.keys.size() >1){
|
||||||
|
int thisTimeoutTries = ((it->second.lastms - it->second.firstms) / (it->second.keys.size()-1)) / 125;
|
||||||
|
if (thisTimeoutTries > timeoutTries) timeoutTries = thisTimeoutTries;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
while(!completeKeyReady && timeoutTries>0){
|
||||||
|
completeKeyReady = true;
|
||||||
|
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
|
||||||
|
if (!myMeta.tracks[*it].keys.size() || myMeta.tracks[*it].keys.rbegin()->getTime() + myMeta.tracks[*it].keys.rbegin()->getLength() <= thisPacket.getTime() ){
|
||||||
|
completeKeyReady = false;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!completeKeyReady){
|
||||||
|
timeoutTries--;//we count down
|
||||||
|
stats();
|
||||||
|
Util::wait(250);
|
||||||
|
updateMeta();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (timeoutTries<=0){
|
||||||
|
WARN_MSG("Waiting for key frame timed out");
|
||||||
|
completeKeysOnly = false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
sendNext();
|
sendNext();
|
||||||
}else{
|
}else{
|
||||||
|
/*LTS-START*/
|
||||||
|
if(Triggers::shouldTrigger("CONN_STOP", streamName)){
|
||||||
|
std::string payload = streamName+"\n" + getConnectedHost() +"\n"+capa["name"].asStringRef()+"\n";
|
||||||
|
Triggers::doTrigger("CONN_STOP", payload, streamName);
|
||||||
|
}
|
||||||
|
/*LTS-END*/
|
||||||
if (!onFinish()){
|
if (!onFinish()){
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
stats();
|
stats();
|
||||||
}
|
}
|
||||||
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler shutting down: %s, %s, %s", myConn.connected() ? "conn_active" : "conn_closed", wantRequest ? "want_request" : "no_want_request", parseData ? "parsing_data" : "not_parsing_data");
|
MEDIUM_MSG("MistOut client handler shutting down: %s, %s, %s", myConn.connected() ? "conn_active" : "conn_closed", wantRequest ? "want_request" : "no_want_request", parseData ? "parsing_data" : "not_parsing_data");
|
||||||
|
|
||||||
/*LTS-START*/
|
/*LTS-START*/
|
||||||
if(Triggers::shouldTrigger("CONN_CLOSE", streamName)){
|
if(Triggers::shouldTrigger("CONN_CLOSE", streamName)){
|
||||||
|
@ -942,7 +1003,7 @@ namespace Mist {
|
||||||
}
|
}
|
||||||
/*LTS-END*/
|
/*LTS-END*/
|
||||||
|
|
||||||
stats();
|
stats(true);
|
||||||
nProxy.userClient.finish();
|
nProxy.userClient.finish();
|
||||||
statsPage.finish();
|
statsPage.finish();
|
||||||
myConn.close();
|
myConn.close();
|
||||||
|
@ -963,97 +1024,78 @@ namespace Mist {
|
||||||
return *(selectedTracks.begin());
|
return *(selectedTracks.begin());
|
||||||
}
|
}
|
||||||
|
|
||||||
void Output::prepareNext(){
|
void Output::dropTrack(uint32_t trackId, std::string reason, bool probablyBad){
|
||||||
if (!myConn){return;}//stop if connection was closed
|
//depending on whether this is probably bad and the current debug level, print a message
|
||||||
|
unsigned int printLevel = DLVL_INFO;
|
||||||
|
if (probablyBad){
|
||||||
|
printLevel = DLVL_WARN;
|
||||||
|
}
|
||||||
|
DEBUG_MSG(printLevel, "Dropping %s (%s) track %lu@k%lu (nextP=%d, lastP=%d): %s", streamName.c_str(), myMeta.tracks[trackId].codec.c_str(), (long unsigned)trackId, nxtKeyNum[trackId]+1, pageNumForKey(trackId, nxtKeyNum[trackId]+1), pageNumMax(trackId), reason.c_str());
|
||||||
|
//now actually drop the track from the buffer
|
||||||
|
for (std::set<sortedPageInfo>::iterator it = buffer.begin(); it != buffer.end(); ++it){
|
||||||
|
if (it->tid == trackId){
|
||||||
|
buffer.erase(it);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
///Attempts to prepare a new packet for output.
|
||||||
|
///If thisPacket evaluates to false, playback has completed.
|
||||||
|
///Could be called repeatedly in a loop if you really really want a new packet.
|
||||||
|
/// \returns true if thisPacket was filled with the next packet.
|
||||||
|
/// \returns false if we could not reliably determine the next packet yet.
|
||||||
|
bool Output::prepareNext(){
|
||||||
static bool atLivePoint = false;
|
static bool atLivePoint = false;
|
||||||
static int nonVideoCount = 0;
|
static int nonVideoCount = 0;
|
||||||
if (!sought){
|
|
||||||
if (myMeta.live){
|
|
||||||
long unsigned int mainTrack = getMainSelectedTrack();
|
|
||||||
if (!myMeta.tracks.count(mainTrack)){
|
|
||||||
myConn.close();
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
if (myMeta.tracks[mainTrack].keys.size() < 2){
|
|
||||||
if (!myMeta.tracks[mainTrack].keys.size()){
|
|
||||||
myConn.close();
|
|
||||||
return;
|
|
||||||
}else{
|
|
||||||
seek(myMeta.tracks[mainTrack].keys.begin()->getTime());
|
|
||||||
prepareNext();
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
unsigned long long seekPos = myMeta.tracks[mainTrack].keys.rbegin()->getTime();
|
|
||||||
if (seekPos < 5000){
|
|
||||||
seekPos = 0;
|
|
||||||
}
|
|
||||||
seek(seekPos);
|
|
||||||
}else{
|
|
||||||
seek(0);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
static unsigned int emptyCount = 0;
|
static unsigned int emptyCount = 0;
|
||||||
if (!buffer.size()){
|
if (!buffer.size()){
|
||||||
thisPacket.null();
|
thisPacket.null();
|
||||||
DEBUG_MSG(DLVL_DEVEL, "Buffer completely played out");
|
INFO_MSG("Buffer completely played out");
|
||||||
onFinish();
|
return true;
|
||||||
/*LTS-START*/
|
|
||||||
if(Triggers::shouldTrigger("CONN_STOP", streamName)){
|
|
||||||
std::string payload = streamName+"\n" + getConnectedHost() +"\n"+capa["name"].asStringRef()+"\n";
|
|
||||||
Triggers::doTrigger("CONN_STOP", payload, streamName);
|
|
||||||
}
|
|
||||||
/*LTS-END*/
|
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
sortedPageInfo nxt = *(buffer.begin());
|
sortedPageInfo nxt = *(buffer.begin());
|
||||||
buffer.erase(buffer.begin());
|
|
||||||
|
|
||||||
if (!myMeta.tracks.count(nxt.tid)){
|
if (!myMeta.tracks.count(nxt.tid)){
|
||||||
MEDIUM_MSG("Track %s@%u disappeared from stream - dropping.");
|
dropTrack(nxt.tid, "disappeared from metadata", true);
|
||||||
prepareNext();
|
return false;
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
DEBUG_MSG(DLVL_DONTEVEN, "Loading track %u (next=%lu), %llu ms", nxt.tid, nxtKeyNum[nxt.tid], nxt.time);
|
DONTEVEN_MSG("Loading track %u (next=%lu), %llu ms", nxt.tid, nxtKeyNum[nxt.tid], nxt.time);
|
||||||
|
|
||||||
|
//if we're going to read past the end of the data page, load the next page
|
||||||
|
//this only happens for VoD
|
||||||
if (nxt.offset >= nProxy.curPage[nxt.tid].len){
|
if (nxt.offset >= nProxy.curPage[nxt.tid].len){
|
||||||
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
||||||
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
|
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
|
||||||
nxt.offset = 0;
|
nxt.offset = 0;
|
||||||
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
|
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
|
||||||
if (getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset) < nxt.time){
|
if (getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset) < nxt.time){
|
||||||
ERROR_MSG("Time going backwards in track %u - dropping track.", nxt.tid);
|
dropTrack(nxt.tid, "time going backwards");
|
||||||
}else{
|
}else{
|
||||||
nxt.time = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
nxt.time = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
||||||
|
//swap out the next object in the buffer with a new one
|
||||||
|
buffer.erase(buffer.begin());
|
||||||
buffer.insert(nxt);
|
buffer.insert(nxt);
|
||||||
}
|
}
|
||||||
prepareNext();
|
}else{
|
||||||
return;
|
dropTrack(nxt.tid, "page load failure", true);
|
||||||
}
|
}
|
||||||
}
|
return false;
|
||||||
|
|
||||||
if (!nProxy.curPage.count(nxt.tid) || !nProxy.curPage[nxt.tid].mapped){
|
|
||||||
//mapping failure? Drop this track and go to next.
|
|
||||||
//not an error - usually means end of stream.
|
|
||||||
WARN_MSG("Track %u no page - dropping track.", nxt.tid);
|
|
||||||
prepareNext();
|
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
//have we arrived at the end of the memory page? (4 zeroes mark the end)
|
//have we arrived at the end of the memory page? (4 zeroes mark the end)
|
||||||
if (!memcmp(nProxy.curPage[nxt.tid].mapped + nxt.offset, "\000\000\000\000", 4)){
|
if (!memcmp(nProxy.curPage[nxt.tid].mapped + nxt.offset, "\000\000\000\000", 4)){
|
||||||
//if we don't currently know where we are, we're lost. We should drop the track.
|
//if we don't currently know where we are, we're lost. We should drop the track.
|
||||||
if (!nxt.time){
|
if (!nxt.time){
|
||||||
WARN_MSG("Timeless empty packet on track %u - dropping track.", nxt.tid);
|
dropTrack(nxt.tid, "timeless empty packet");
|
||||||
prepareNext();
|
return false;
|
||||||
return;
|
|
||||||
}
|
}
|
||||||
//if this is a live stream, we might have just reached the live point.
|
//if this is a live stream, we might have just reached the live point.
|
||||||
//check where the next key is
|
//check where the next key is
|
||||||
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
||||||
int nextPage = pageNumForKey(nxt.tid, nxtKeyNum[nxt.tid]+1);
|
int nextPage = pageNumForKey(nxt.tid, nxtKeyNum[nxt.tid]+1);
|
||||||
//are we live, and the next key hasn't shown up on another page? then we're waiting.
|
//are we live, and the next key hasn't shown up on another page, then we're waiting.
|
||||||
if (myMeta.live && currKeyOpen.count(nxt.tid) && (currKeyOpen[nxt.tid] == (unsigned int)nextPage || nextPage == -1)){
|
if (myMeta.live && currKeyOpen.count(nxt.tid) && (currKeyOpen[nxt.tid] == (unsigned int)nextPage || nextPage == -1)){
|
||||||
if (++emptyCount < 100){
|
if (++emptyCount < 100){
|
||||||
Util::wait(250);
|
Util::wait(250);
|
||||||
|
@ -1062,60 +1104,69 @@ namespace Mist {
|
||||||
reconnect();//reconnect every 2 seconds
|
reconnect();//reconnect every 2 seconds
|
||||||
}else{
|
}else{
|
||||||
if (emptyCount % 4 == 0){
|
if (emptyCount % 4 == 0){
|
||||||
stats();
|
|
||||||
updateMeta();
|
updateMeta();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
buffer.insert(nxt);
|
|
||||||
}else{
|
}else{
|
||||||
//after ~25 seconds, give up and drop the track.
|
//after ~25 seconds, give up and drop the track.
|
||||||
WARN_MSG("Empty packet on %s track %u (%s) @ key %lu (nPage=%d, lPage=%d) - could not reload, dropping track.", streamName.c_str(), nxt.tid, myMeta.tracks[nxt.tid].type.c_str(), nxtKeyNum[nxt.tid]+1, nextPage, pageNumMax(nxt.tid));
|
dropTrack(nxt.tid, "could not reload empty packet");
|
||||||
}
|
}
|
||||||
}else{
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
//We've simply reached the end of the page. Load the next key = next page.
|
//We've simply reached the end of the page. Load the next key = next page.
|
||||||
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
|
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
|
||||||
nxt.offset = 0;
|
nxt.offset = 0;
|
||||||
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
|
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
|
||||||
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
||||||
if (nextTime && nextTime < nxt.time){
|
if (nextTime && nextTime < nxt.time){
|
||||||
WARN_MSG("Time going backwards in track %u - dropping track.", nxt.tid);
|
dropTrack(nxt.tid, "time going backwards");
|
||||||
}else{
|
}else{
|
||||||
if (nextTime){
|
if (nextTime){
|
||||||
nxt.time = nextTime;
|
nxt.time = nextTime;
|
||||||
}
|
}
|
||||||
|
//swap out the next object in the buffer with a new one
|
||||||
|
buffer.erase(buffer.begin());
|
||||||
buffer.insert(nxt);
|
buffer.insert(nxt);
|
||||||
MEDIUM_MSG("Next page for track %u starts at %llu.", nxt.tid, nxt.time);
|
MEDIUM_MSG("Next page for track %u starts at %llu.", nxt.tid, nxt.time);
|
||||||
}
|
}
|
||||||
}else{
|
}else{
|
||||||
WARN_MSG("Could not load next memory page for track %u - dropping track.", nxt.tid);
|
dropTrack(nxt.tid, "page load failure");
|
||||||
}
|
}
|
||||||
|
return false;
|
||||||
}
|
}
|
||||||
prepareNext();
|
|
||||||
return;
|
//we've handled all special cases - at this point the packet should exist
|
||||||
}
|
//let's load it
|
||||||
thisPacket.reInit(nProxy.curPage[nxt.tid].mapped + nxt.offset, 0, true);
|
thisPacket.reInit(nProxy.curPage[nxt.tid].mapped + nxt.offset, 0, true);
|
||||||
if (thisPacket){
|
//if it failed, drop the track and continue
|
||||||
|
if (!thisPacket){
|
||||||
|
dropTrack(nxt.tid, "packet load failure");
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
emptyCount = 0;//valid packet - reset empty counter
|
||||||
|
|
||||||
|
//if there's a timestamp mismatch, print this.
|
||||||
|
//except for live, where we never know the time in advance
|
||||||
if (thisPacket.getTime() != nxt.time && nxt.time && !atLivePoint){
|
if (thisPacket.getTime() != nxt.time && nxt.time && !atLivePoint){
|
||||||
static int warned = 0;
|
static int warned = 0;
|
||||||
if (warned < 10){
|
if (warned < 5){
|
||||||
WARN_MSG("Loaded track %ld@%llu instead of %u@%llu for %s", thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, streamName.c_str());
|
WARN_MSG("Loaded %s track %ld@%llu instead of %u@%llu (%dms) for %s", streamName.c_str(), thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, (int)((long long)thisPacket.getTime() - (long long)nxt.time), streamName.c_str());
|
||||||
if (++warned == 10){
|
if (++warned == 5){
|
||||||
WARN_MSG("Further warnings about time mismatches printed on HIGH level.");
|
WARN_MSG("Further warnings about time mismatches printed on HIGH level.");
|
||||||
}
|
}
|
||||||
}else{
|
}else{
|
||||||
HIGH_MSG("Loaded track %ld@%llu instead of %u@%llu for %s", thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, streamName.c_str());
|
HIGH_MSG("Loaded %s track %ld@%llu instead of %u@%llu (%dms) for %s", streamName.c_str(), thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, (int)((long long)thisPacket.getTime() - (long long)nxt.time), streamName.c_str());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
atLivePoint = false;
|
|
||||||
bool isVideoTrack = (myMeta.tracks[nxt.tid].type == "video");
|
//when live, every keyframe, check correctness of the keyframe number
|
||||||
if ((isVideoTrack && thisPacket.getFlag("keyframe")) || (!isVideoTrack && (++nonVideoCount % 30 == 0))){
|
if (myMeta.live && thisPacket.getFlag("keyframe")){
|
||||||
if (myMeta.live){
|
//Check whether returned keyframe is correct. If not, wait for approximately 10 seconds while checking.
|
||||||
if (myMeta.tracks[nxt.tid].type == "video"){
|
|
||||||
//Check whether returned keyframe is correct. If not, wait for approximately 5 seconds while checking.
|
|
||||||
//Failure here will cause tracks to drop due to inconsistent internal state.
|
//Failure here will cause tracks to drop due to inconsistent internal state.
|
||||||
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
||||||
int counter = 0;
|
int counter = 0;
|
||||||
while(counter < 100 && myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
|
while(counter < 40 && myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
|
||||||
if (counter++){
|
if (counter++){
|
||||||
//Only sleep 250ms if this is not the first updatemeta try
|
//Only sleep 250ms if this is not the first updatemeta try
|
||||||
Util::wait(250);
|
Util::wait(250);
|
||||||
|
@ -1126,76 +1177,29 @@ namespace Mist {
|
||||||
if (myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
|
if (myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
|
||||||
WARN_MSG("Keyframe value is not correct - state will now be inconsistent.");
|
WARN_MSG("Keyframe value is not correct - state will now be inconsistent.");
|
||||||
}
|
}
|
||||||
}else{
|
|
||||||
//On non-video tracks, just update metadata and assume everything else is correct
|
|
||||||
updateMeta();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
|
|
||||||
EXTREME_MSG("Track %u @ %llums = key %lu", nxt.tid, thisPacket.getTime(), nxtKeyNum[nxt.tid]);
|
EXTREME_MSG("Track %u @ %llums = key %lu", nxt.tid, thisPacket.getTime(), nxtKeyNum[nxt.tid]);
|
||||||
}
|
}
|
||||||
emptyCount = 0;
|
|
||||||
}
|
//always assume we're not at the live point
|
||||||
|
atLivePoint = false;
|
||||||
|
//we assume the next packet is the next on this same page
|
||||||
nxt.offset += thisPacket.getDataLen();
|
nxt.offset += thisPacket.getDataLen();
|
||||||
if (realTime){
|
|
||||||
while (nxt.time > (((Util::getMS() - firstTime)*1000)+maxSkipAhead)/realTime) {
|
|
||||||
Util::sleep(nxt.time - (((Util::getMS() - firstTime)*1000)+minSkipAhead)/realTime);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
//delay the stream until its current keyframe is complete
|
|
||||||
if (completeKeysOnly){
|
|
||||||
bool completeKeyReady = false;
|
|
||||||
int timeoutTries = 40;//attempts to updateMeta before timeOut and moving on; default is approximately 10 seconds
|
|
||||||
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
|
|
||||||
if (it->second.keys.size() >1){
|
|
||||||
int thisTimeoutTries = ((it->second.lastms - it->second.firstms) / (it->second.keys.size()-1)) / 125;
|
|
||||||
if (thisTimeoutTries > timeoutTries) timeoutTries = thisTimeoutTries;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
while(!completeKeyReady && timeoutTries>0){
|
|
||||||
completeKeyReady = true;
|
|
||||||
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
|
|
||||||
if (!myMeta.tracks[*it].keys.size() || myMeta.tracks[*it].keys.rbegin()->getTime() + myMeta.tracks[*it].keys.rbegin()->getLength() <= nxt.time ){
|
|
||||||
completeKeyReady = false;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (!completeKeyReady){
|
|
||||||
if (completeKeyReadyTimeOut){
|
|
||||||
INSANE_MSG("Complete Key not ready and time-out is being skipped");
|
|
||||||
timeoutTries = 0;
|
|
||||||
}else{
|
|
||||||
INSANE_MSG("Timeout: %d",timeoutTries);
|
|
||||||
timeoutTries--;//we count down
|
|
||||||
stats();
|
|
||||||
Util::wait(250);
|
|
||||||
updateMeta();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (timeoutTries<=0){
|
|
||||||
if (!completeKeyReadyTimeOut){
|
|
||||||
WARN_MSG("Wait for keyframe Timeout triggered! Ended to avoid endless loops");
|
|
||||||
}
|
|
||||||
completeKeyReadyTimeOut = true;
|
|
||||||
}else{
|
|
||||||
//untimeout handling
|
|
||||||
completeKeyReadyTimeOut = false;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (nProxy.curPage[nxt.tid]){
|
|
||||||
if (nxt.offset < nProxy.curPage[nxt.tid].len){
|
if (nxt.offset < nProxy.curPage[nxt.tid].len){
|
||||||
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
|
||||||
if (nextTime){
|
if (nextTime){
|
||||||
nxt.time = nextTime;
|
nxt.time = nextTime;
|
||||||
}else{
|
}else{
|
||||||
++nxt.time;
|
++nxt.time;
|
||||||
|
//no packet -> we are at the live point
|
||||||
atLivePoint = true;
|
atLivePoint = true;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
//exchange the current packet in the buffer for the next one
|
||||||
|
buffer.erase(buffer.begin());
|
||||||
buffer.insert(nxt);
|
buffer.insert(nxt);
|
||||||
}
|
|
||||||
stats();
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns the name as it should be used in statistics.
|
/// Returns the name as it should be used in statistics.
|
||||||
|
@ -1209,21 +1213,23 @@ namespace Mist {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void Output::stats(){
|
void Output::stats(bool force){
|
||||||
if (!isInitialized){
|
//cancel stats update if not initialized
|
||||||
return;
|
if (!isInitialized){return;}
|
||||||
}
|
//also cancel if it has been less than a second since the last update
|
||||||
|
//unless force is set to true
|
||||||
|
unsigned long long int now = Util::epoch();
|
||||||
|
if (now == lastStats && !force){return;}
|
||||||
|
lastStats = now;
|
||||||
|
|
||||||
EXTREME_MSG("Writing stats: %s, %s, %lu", getConnectedHost().c_str(), streamName.c_str(), crc & 0xFFFFFFFFu);
|
EXTREME_MSG("Writing stats: %s, %s, %lu", getConnectedHost().c_str(), streamName.c_str(), crc & 0xFFFFFFFFu);
|
||||||
if (statsPage.getData()){
|
if (statsPage.getData()){
|
||||||
unsigned long long int now = Util::epoch();
|
|
||||||
if (now != lastStats){
|
|
||||||
/*LTS-START*/
|
/*LTS-START*/
|
||||||
if (!statsPage.isAlive()){
|
if (!statsPage.isAlive()){
|
||||||
myConn.close();
|
myConn.close();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
/*LTS-END*/
|
/*LTS-END*/
|
||||||
lastStats = now;
|
|
||||||
IPC::statExchange tmpEx(statsPage.getData());
|
IPC::statExchange tmpEx(statsPage.getData());
|
||||||
tmpEx.now(now);
|
tmpEx.now(now);
|
||||||
if (tmpEx.host() == std::string("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000", 16)){
|
if (tmpEx.host() == std::string("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000", 16)){
|
||||||
|
@ -1242,7 +1248,6 @@ namespace Mist {
|
||||||
}
|
}
|
||||||
statsPage.keepAlive();
|
statsPage.keepAlive();
|
||||||
}
|
}
|
||||||
}
|
|
||||||
doSync();
|
doSync();
|
||||||
int tNum = 0;
|
int tNum = 0;
|
||||||
if (!nProxy.userClient.getData()){
|
if (!nProxy.userClient.getData()){
|
||||||
|
@ -1250,7 +1255,7 @@ namespace Mist {
|
||||||
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
|
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
|
||||||
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
|
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
|
||||||
if (!nProxy.userClient.getData()){
|
if (!nProxy.userClient.getData()){
|
||||||
DEBUG_MSG(DLVL_WARN, "Player connection failure - aborting output");
|
WARN_MSG("Player connection failure - aborting output");
|
||||||
myConn.close();
|
myConn.close();
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
@ -1269,7 +1274,7 @@ namespace Mist {
|
||||||
}
|
}
|
||||||
nProxy.userClient.keepAlive();
|
nProxy.userClient.keepAlive();
|
||||||
if (tNum > SIMUL_TRACKS){
|
if (tNum > SIMUL_TRACKS){
|
||||||
DEBUG_MSG(DLVL_WARN, "Too many tracks selected, using only first %d", SIMUL_TRACKS);
|
WARN_MSG("Too many tracks selected, using only first %d", SIMUL_TRACKS);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -54,7 +54,7 @@ namespace Mist {
|
||||||
/*LTS-END*/
|
/*LTS-END*/
|
||||||
//non-virtual generic functions
|
//non-virtual generic functions
|
||||||
int run();
|
int run();
|
||||||
void stats();
|
void stats(bool force = false);
|
||||||
void seek(unsigned long long pos);
|
void seek(unsigned long long pos);
|
||||||
bool seek(unsigned int tid, unsigned long long pos, bool getNextKey = false);
|
bool seek(unsigned int tid, unsigned long long pos, bool getNextKey = false);
|
||||||
void stop();
|
void stop();
|
||||||
|
@ -67,7 +67,8 @@ namespace Mist {
|
||||||
virtual bool isReadyForPlay();
|
virtual bool isReadyForPlay();
|
||||||
//virtuals. The optional virtuals have default implementations that do as little as possible.
|
//virtuals. The optional virtuals have default implementations that do as little as possible.
|
||||||
virtual void sendNext() {}//REQUIRED! Others are optional.
|
virtual void sendNext() {}//REQUIRED! Others are optional.
|
||||||
virtual void prepareNext();
|
bool prepareNext();
|
||||||
|
virtual void dropTrack(uint32_t trackId, std::string reason, bool probablyBad = true);
|
||||||
virtual void onRequest();
|
virtual void onRequest();
|
||||||
virtual bool onFinish() {
|
virtual bool onFinish() {
|
||||||
return false;
|
return false;
|
||||||
|
@ -105,7 +106,6 @@ namespace Mist {
|
||||||
std::map<unsigned long, unsigned long> nxtKeyNum;///< Contains the number of the next key, for page seeking purposes.
|
std::map<unsigned long, unsigned long> nxtKeyNum;///< Contains the number of the next key, for page seeking purposes.
|
||||||
std::set<sortedPageInfo> buffer;///< A sorted list of next-to-be-loaded packets.
|
std::set<sortedPageInfo> buffer;///< A sorted list of next-to-be-loaded packets.
|
||||||
bool sought;///<If a seek has been done, this is set to true. Used for seeking on prepareNext().
|
bool sought;///<If a seek has been done, this is set to true. Used for seeking on prepareNext().
|
||||||
bool completeKeyReadyTimeOut;//a bool to see if there has been a keyframe TimeOut for complete keys in Live
|
|
||||||
protected://these are to be messed with by child classes
|
protected://these are to be messed with by child classes
|
||||||
virtual std::string getConnectedHost();
|
virtual std::string getConnectedHost();
|
||||||
virtual std::string getConnectedBinHost();
|
virtual std::string getConnectedBinHost();
|
||||||
|
|
Loading…
Add table
Reference in a new issue