From 4d9d5aa5b511deaeef392537604c1a6c59a71bdb Mon Sep 17 00:00:00 2001 From: Thulinma Date: Mon, 16 Jun 2014 14:23:55 +0200 Subject: [PATCH] Various fixes to live pushing. --- src/input/input_buffer.cpp | 41 +++++++++++++++++-------------- src/output/output.cpp | 50 ++++++++++++++++++++------------------ src/output/output.h | 1 + 3 files changed, 51 insertions(+), 41 deletions(-) diff --git a/src/input/input_buffer.cpp b/src/input/input_buffer.cpp index 16cde760..41d21b1b 100644 --- a/src/input/input_buffer.cpp +++ b/src/input/input_buffer.cpp @@ -151,8 +151,6 @@ namespace Mist { } if (givenTracks.count(value)){ givenTracks.erase(value); - indexPages.erase(value); - dataPages.erase(value); inputLoc.erase(value); } continue; @@ -166,17 +164,15 @@ namespace Mist { thisData[2] = (tmpTid >> 8) & 0xFF; thisData[3] = (tmpTid) & 0xFF; unsigned long tNum = ((long)(thisData[4]) << 8) | thisData[5]; - INFO_MSG("Assigning temporary ID %d to incoming track %lu for user %d", tmpTid, tNum, id); + DEBUG_MSG(DLVL_HIGH, "Assigning temporary ID %d to incoming track %lu for user %d", tmpTid, tNum, id); char tempMetaName[100]; sprintf(tempMetaName, "liveStream_%s%d", config->getString("streamname").c_str(), tmpTid); metaPages[tmpTid].init(tempMetaName, 8388608, true); } if (negotiateTracks.count(value)){ - INFO_MSG("Negotiating %lu", value); //Track is currently under negotiation, check whether the metadata has been submitted if (metaPages[value].mapped){ - INFO_MSG("Mapped %lu", value); unsigned int len = ntohl(((int *)metaPages[value].mapped)[1]); unsigned int i = 0; JSON::Value JSONMeta; @@ -187,42 +183,50 @@ namespace Mist { } std::string tempId = tmpMeta.tracks.begin()->second.getIdentifier(); - DEBUG_MSG(DLVL_DEVEL, "Attempting colision detection for track %s", tempId.c_str()); + DEBUG_MSG(DLVL_HIGH, "Attempting colision detection for track %s", tempId.c_str()); int finalMap = -1; - for (std::map::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) { - if (it->second.type == "video"){ - finalMap = 1; - } - if (it->second.type == "audio"){ - finalMap = 2; - } + if (tmpMeta.tracks.begin()->second.type == "video"){ + finalMap = 1; + } + if (tmpMeta.tracks.begin()->second.type == "audio"){ + finalMap = 2; } //Remove the "negotiate" status in either case negotiateTracks.erase(value); metaPages.erase(value); if (finalMap != -1 && givenTracks.count(finalMap)) { - DEBUG_MSG(DLVL_DEVEL, "Collision of new track %lu with track %d detected! Declining track", value, finalMap); + WARN_MSG("Collision of new track %lu with track %d detected! Declining track", value, finalMap); thisData[0] = 0xFF; thisData[1] = 0xFF; thisData[2] = 0xFF; thisData[3] = 0xFF; } else { if (finalMap == -1){ - DEBUG_MSG(DLVL_DEVEL, "Invalid track type detected, discarding"); + WARN_MSG("Invalid track type detected, declining."); + thisData[0] = 0xFF; + thisData[1] = 0xFF; + thisData[2] = 0xFF; + thisData[3] = 0xFF; continue; }else{ //Resume either if we have more than 1 keyframe on the replacement track (assume it was already pushing before the track "dissapeared" //or if the firstms of the replacement track is later than the lastms on the existing track - if (tmpMeta.tracks.begin()->second.keys.size() > 1|| tmpMeta.tracks.begin()->second.firstms >= myMeta.tracks[finalMap].lastms){ - DEBUG_MSG(DLVL_DEVEL, "Allowing negotiation track %lu, from user %u, to resume pushing final track number %d", value, id, finalMap); + if (tmpMeta.tracks.begin()->second.keys.size() > 1 || !myMeta.tracks.count(finalMap) || tmpMeta.tracks.begin()->second.firstms >= myMeta.tracks[finalMap].lastms){ + if (myMeta.tracks.count(finalMap) && myMeta.tracks[finalMap].lastms > 0){ + INFO_MSG("Allowing negotiation track %lu, from user %u, to resume pushing final track number %d", value, id, finalMap); + }else{ + INFO_MSG("Allowing negotiation track %lu, from user %u, to start pushing final track number %d", value, id, finalMap); + } }else{ //Otherwise replace existing track - DEBUG_MSG(DLVL_DEVEL, "Re-push initiated for track %lu, from user %u, will replace final track number %d", value, id, finalMap); + INFO_MSG("Re-push initiated for track %lu, from user %u, will replace final track number %d", value, id, finalMap); myMeta.tracks.erase(finalMap); + dataPages.erase(finalMap); } } givenTracks.insert(finalMap); if (!myMeta.tracks.count(finalMap)){ + DEBUG_MSG(DLVL_HIGH, "Inserting metadata for track number %d", finalMap); myMeta.tracks[finalMap] = tmpMeta.tracks.begin()->second; myMeta.tracks[finalMap].trackID = finalMap; } @@ -255,6 +259,7 @@ namespace Mist { //update current page int currentPage = dataPages[value].rbegin()->first; updateMetaFromPage(value, currentPage); + INFO_MSG("To go, track %lu: %lli", value, 8388608 - inputLoc[value][currentPage].curOffset); if (inputLoc[value][currentPage].curOffset > 8388608) { int nextPage = currentPage + inputLoc[value][currentPage].keyNum; char nextPageName[100]; diff --git a/src/output/output.cpp b/src/output/output.cpp index 70c517f2..004c1ddb 100644 --- a/src/output/output.cpp +++ b/src/output/output.cpp @@ -36,6 +36,7 @@ namespace Mist { firstTime = 0; parseData = false; wantRequest = true; + sought = false; isInitialized = false; isBlocking = false; lastStats = 0; @@ -119,7 +120,7 @@ namespace Mist { Util::sleep(100); newTid = ((long)(tmp[6 * bufConnOffset]) << 24) | ((long)(tmp[6 * bufConnOffset + 1]) << 16) | ((long)(tmp[6 * bufConnOffset + 2]) << 8) | tmp[6 * bufConnOffset + 3]; } - DEBUG_MSG(DLVL_DEVEL, "Track %d temporarily mapped to %d", tid, newTid); + DEBUG_MSG(DLVL_VERYHIGH, "Track %d temporarily mapped to %d", tid, newTid); char pageName[100]; sprintf(pageName, "liveStream_%s%d", streamName.c_str(), newTid); @@ -131,17 +132,17 @@ namespace Mist { JSON::Value tmpVal = tmpMeta.toJSON(); std::string tmpStr = tmpVal.toNetPacked(); memcpy(metaPages[newTid].mapped, tmpStr.data(), tmpStr.size()); - DEBUG_MSG(DLVL_DEVEL, "Temporary metadata written for incoming track %d, handling as track %d", tid, newTid); + DEBUG_MSG(DLVL_VERYHIGH, "Temporary metadata written for incoming track %d, handling as track %d", tid, newTid); unsigned short firstPage = 0xFFFF; unsigned int finalTid = newTid; while (firstPage == 0xFFFF){ - DEBUG_MSG(DLVL_DEVEL, "Re-checking at offset %d", bufConnOffset); + DEBUG_MSG(DLVL_VERYHIGH, "Re-checking at offset %d", bufConnOffset); Util::sleep(100); finalTid = ((long)(tmp[6 * bufConnOffset]) << 24) | ((long)(tmp[6 * bufConnOffset + 1]) << 16) | ((long)(tmp[6 * bufConnOffset + 2]) << 8) | tmp[6 * bufConnOffset + 3]; firstPage = ((long)(tmp[6 * bufConnOffset + 4]) << 8) | tmp[6 * bufConnOffset + 5]; if (finalTid == 0xFFFFFFFF){ - DEBUG_MSG(DLVL_DEVEL, "Buffer has declined incoming track %d", tid); + WARN_MSG("Buffer has declined incoming track %d", tid); return; } } @@ -149,19 +150,17 @@ namespace Mist { finalTid = ((long)(tmp[6 * bufConnOffset]) << 24) | ((long)(tmp[6 * bufConnOffset + 1]) << 16) | ((long)(tmp[6 * bufConnOffset + 2]) << 8) | tmp[6 * bufConnOffset + 3]; firstPage = ((long)(tmp[6 * bufConnOffset + 4]) << 8) | tmp[6 * bufConnOffset + 5]; if (finalTid == 0xFFFFFFFF){ - DEBUG_MSG(DLVL_DEVEL, "Buffer has declined incoming track %d", tid); + WARN_MSG("Buffer has declined incoming track %d", tid); memset(tmp + 6 * bufConnOffset, 0, 6); return; } - DEBUG_MSG(DLVL_DEVEL, "Buffer accepted incoming track %d, temporary mapping %d as final mapping %d", tid, newTid, finalTid); - DEBUG_MSG(DLVL_DEVEL, "Buffer has indicated that incoming track %d should start writing on track %d, page %d", tid, finalTid, firstPage); + INFO_MSG("Buffer has indicated that incoming track %d should start writing on track %d, page %d", tid, finalTid, firstPage); memset(pageName, 0, 100); sprintf(pageName, "%s%d_%d", streamName.c_str(), finalTid, firstPage); curPages[finalTid].init(pageName, 8 * 1024 * 1024); trackMap[tid] = finalTid; bookKeeping[finalTid] = DTSCPageData(); - DEBUG_MSG(DLVL_DEVEL, "Done negotiating for incoming track %d", tid); } @@ -205,6 +204,7 @@ namespace Mist { //open new page char nextPage[100]; sprintf(nextPage, "%s%llu_%d", streamName.c_str(), tNum, bookKeeping[tNum].pageNum + bookKeeping[tNum].keyNum); + INFO_MSG("Continuing track %llu on page %d", tNum, bookKeeping[tNum].pageNum + bookKeeping[tNum].keyNum); curPages[tNum].init(nextPage, 26 * 1024 * 1024); bookKeeping[tNum].pageNum += bookKeeping[tNum].keyNum; bookKeeping[tNum].keyNum = 0; @@ -342,21 +342,7 @@ namespace Mist { DEBUG_MSG(DLVL_MEDIUM, "Selected tracks: %s", selected.str().c_str()); #endif - unsigned int firstms = 0x0; - for (std::set::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){ - lastKeyTime[*it] = 0xFFFFFFFF; - if (myMeta.tracks[*it].firstms > firstms){ - firstms = myMeta.tracks[*it].firstms; - } - } - if (myMeta.live){ - if (firstms < 5000){ - firstms = 0; - } - seek(firstms); - }else{ - seek(0); - } + sought = false; capa.null(); } @@ -448,6 +434,7 @@ namespace Mist { /// Prepares all tracks from selectedTracks for seeking to the specified ms position. /// \todo Make this actually seek, instead of always loading position zero. void Output::seek(long long pos){ + sought = true; firstTime = Util::getMS() - pos; if (!isInitialized){ initialize(); @@ -542,6 +529,23 @@ namespace Mist { } void Output::prepareNext(){ + if (!sought){ + unsigned int firstms = 0x0; + for (std::set::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){ + lastKeyTime[*it] = 0xFFFFFFFF; + if (myMeta.tracks[*it].firstms > firstms){ + firstms = myMeta.tracks[*it].firstms; + } + } + if (myMeta.live){ + if (firstms < 5000){ + firstms = 0; + } + seek(firstms); + }else{ + seek(0); + } + } static unsigned int emptyCount = 0; if (!buffer.size()){ currentPacket.null(); diff --git a/src/output/output.h b/src/output/output.h index e2f0f45f..c76abf64 100644 --- a/src/output/output.h +++ b/src/output/output.h @@ -76,6 +76,7 @@ namespace Mist { std::map nxtKeyNum;///< Contains the number of the next key, for page seeking purposes. std::set buffer;///< A sorted list of next-to-be-loaded packets. std::map lastKeyTime;///< Stores the time of the last keyframe, for preventing duplicates + bool sought;///