hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Various fixes to live pushing.

Browse source
This commit is contained in:
Thulinma 2014-06-16 14:23:55 +02:00
parent 840cfaabed
commit 4d9d5aa5b5
3 changed files with 51 additions and 41 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
src/input/input_buffer.cpp
View file

@ -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<int, DTSC::Track>::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];

50
src/output/output.cpp
View file

@ -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<long unsigned int>::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<long unsigned int>::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();

1
src/output/output.h
View file

@ -76,6 +76,7 @@ namespace Mist {
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::map<unsigned long, unsigned long> lastKeyTime;///< Stores the time of the last keyframe, for preventing duplicates
bool sought;///<If a seek has been done, this is set to true. Used for seeking on prepareNext().
protected://these are to be messed with by child classes
unsigned int getKeyForTime(long unsigned int trackId, long long timeStamp);
IPC::sharedPage streamIndex;///< Shared memory used for metadata