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

Recording, HLS Push, UDP (Multicast) Input, Threaded TS

Browse source
This commit is contained in:
Erik Zandvliet 2016-01-28 15:00:25 +01:00 • committed by Thulinma
parent 1c3e143709
commit c25a533729
29 changed files with 1809 additions and 815 deletions
Download patch file Download diff file Expand all files Collapse all files

99
src/input/input.cpp
View file

@ -29,11 +29,7 @@ namespace Mist {
Input::Input(Util::Config * cfg) : InOutBase() {
config = cfg;
#ifdef INPUT_LIVE
standAlone = false;
#else
standAlone = true;
#endif
JSON::Value option;
option["long"] = "json";
@ -90,7 +86,7 @@ namespace Mist {
}
void Input::checkHeaderTimes(std::string streamFile) {
if (streamFile == "-") {
if (streamFile == "-" || streamFile == "push://") {
return;
}
std::string headerFile = streamFile + ".dtsh";
@ -124,6 +120,7 @@ namespace Mist {
config->getOption("streamname") = streamName;
}
streamName = config->getString("streamname");
nProxy.streamName = streamName;
if (config->getBool("json")) {
std::cout << capa.toString() << std::endl;
return 0;
@ -132,26 +129,19 @@ namespace Mist {
std::cerr << config->getString("cmd") << " setup failed." << std::endl;
return 0;
}
//Do not read the header if this is a live stream
#ifndef INPUT_LIVE
checkHeaderTimes(config->getString("input"));
if (!readHeader()) {
std::cerr << "Reading header for " << config->getString("input") << " failed." << std::endl;
return 0;
}
parseHeader();
#endif
//Live inputs only have a serve() mode
#ifndef INPUT_LIVE
if (!config->getString("streamname").size()) {
if (!streamName.size()) {
convert();
} else {
#endif
serve();
#ifndef INPUT_LIVE
}
#endif
return 0;
}
@ -203,44 +193,6 @@ namespace Mist {
void Input::serve(){
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
#ifdef INPUT_LIVE
unsigned int giveUpCounter = 0;
while (!Util::startInput(streamName) && config->is_active && ++giveUpCounter < 20) {
Util::sleep(500);
}
if (giveUpCounter >= 20){
FAIL_MSG("Could not start buffer for stream '%s', aborting stream input!", streamName.c_str());
config->is_active = false;
}
userClient = IPC::sharedClient(userPageName, 30, true);
getNext();
while (thisPacket || config->is_active) {
unsigned long tid = thisPacket.getTrackId();
//Check for eligibility of track
IPC::userConnection userConn(userClient.getData());
if (trackOffset.count(tid) && !userConn.getTrackId(trackOffset[tid])) {
trackOffset.erase(tid);
trackState.erase(tid);
trackMap.erase(tid);
trackBuffer.erase(tid);
pagesByTrack.erase(tid);
metaPages.erase(tid);
curPageNum.erase(tid);
curPage.erase(tid);
INFO_MSG("Erasing track %d", tid);
continue;
}
if (thisPacket) {
continueNegotiate(thisPacket.getTrackId());
bufferLivePacket(thisPacket);
} else {
Util::sleep(100);
}
getNext();
userClient.keepAlive();
}
userClient.finish();
#else
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_READY", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
@ -283,7 +235,6 @@ namespace Mist {
Util::sleep(1000);
}
}
#endif
finish();
DEBUG_MSG(DLVL_DEVEL, "Input for stream %s closing clean", streamName.c_str());
//end player functionality
@ -297,7 +248,7 @@ namespace Mist {
}
removeUnused();
if (standAlone) {
for (std::map<unsigned long, IPC::sharedPage>::iterator it = metaPages.begin(); it != metaPages.end(); it++) {
for (std::map<unsigned long, IPC::sharedPage>::iterator it = nProxy.metaPages.begin(); it != nProxy.metaPages.end(); it++) {
it->second.master = true;
}
}
@ -316,9 +267,9 @@ namespace Mist {
bufferRemove(it->first, it2->first);
pageCounter[it->first].erase(it2->first);
for (int i = 0; i < 8192; i += 8) {
unsigned int thisKeyNum = ntohl(((((long long int *)(metaPages[it->first].mapped + i))[0]) >> 32) & 0xFFFFFFFF);
unsigned int thisKeyNum = ntohl(((((long long int *)(nProxy.metaPages[it->first].mapped + i))[0]) >> 32) & 0xFFFFFFFF);
if (thisKeyNum == it2->first) {
(((long long int *)(metaPages[it->first].mapped + i))[0]) = 0;
(((long long int *)(nProxy.metaPages[it->first].mapped + i))[0]) = 0;
}
}
change = true;
@ -359,13 +310,13 @@ namespace Mist {
for (int i = 0; i < it->second.keys.size(); i++) {
if (newData) {
//i+1 because keys are 1-indexed
pagesByTrack[it->first][i + 1].firstTime = it->second.keys[i].getTime();
nProxy.pagesByTrack[it->first][i + 1].firstTime = it->second.keys[i].getTime();
newData = false;
}
pagesByTrack[it->first].rbegin()->second.keyNum++;
pagesByTrack[it->first].rbegin()->second.partNum += it->second.keys[i].getParts();
pagesByTrack[it->first].rbegin()->second.dataSize += it->second.keySizes[i];
if (pagesByTrack[it->first].rbegin()->second.dataSize > FLIP_DATA_PAGE_SIZE) {
nProxy.pagesByTrack[it->first].rbegin()->second.keyNum++;
nProxy.pagesByTrack[it->first].rbegin()->second.partNum += it->second.keys[i].getParts();
nProxy.pagesByTrack[it->first].rbegin()->second.dataSize += it->second.keySizes[i];
if (nProxy.pagesByTrack[it->first].rbegin()->second.dataSize > FLIP_DATA_PAGE_SIZE) {
newData = true;
}
}
@ -398,7 +349,7 @@ namespace Mist {
}
if (myMeta.tracks[tid].keys[bookKeeping[tid].curKey].getParts() + 1 == curData[tid].partNum) {
if (curData[tid].dataSize > FLIP_DATA_PAGE_SIZE) {
pagesByTrack[tid][bookKeeping[tid].first] = curData[tid];
nProxy.pagesByTrack[tid][bookKeeping[tid].first] = curData[tid];
bookKeeping[tid].first += curData[tid].keyNum;
curData[tid].keyNum = 0;
curData[tid].dataSize = 0;
@ -415,17 +366,17 @@ namespace Mist {
getNext(false);
}
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
if (curData.count(it->first) && !pagesByTrack[it->first].count(bookKeeping[it->first].first)) {
pagesByTrack[it->first][bookKeeping[it->first].first] = curData[it->first];
if (curData.count(it->first) && !nProxy.pagesByTrack[it->first].count(bookKeeping[it->first].first)) {
nProxy.pagesByTrack[it->first][bookKeeping[it->first].first] = curData[it->first];
}
}
}
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
if (!pagesByTrack.count(it->first)) {
if (!nProxy.pagesByTrack.count(it->first)) {
DEBUG_MSG(DLVL_WARN, "No pages for track %d found", it->first);
} else {
DEBUG_MSG(DLVL_MEDIUM, "Track %d (%s) split into %lu pages", it->first, myMeta.tracks[it->first].codec.c_str(), pagesByTrack[it->first].size());
for (std::map<unsigned long, DTSCPageData>::iterator it2 = pagesByTrack[it->first].begin(); it2 != pagesByTrack[it->first].end(); it2++) {
DEBUG_MSG(DLVL_MEDIUM, "Track %d (%s) split into %lu pages", it->first, myMeta.tracks[it->first].codec.c_str(), nProxy.pagesByTrack[it->first].size());
for (std::map<unsigned long, DTSCPageData>::iterator it2 = nProxy.pagesByTrack[it->first].begin(); it2 != nProxy.pagesByTrack[it->first].end(); it2++) {
DEBUG_MSG(DLVL_VERYHIGH, "Page %lu-%lu, (%llu bytes)", it2->first, it2->first + it2->second.keyNum - 1, it2->second.dataSize);
}
}
@ -443,10 +394,10 @@ namespace Mist {
if (keyNum < 1) {
keyNum = 1;
}
if (isBuffered(track, keyNum)) {
if (nProxy.isBuffered(track, keyNum)) {
//get corresponding page number
int pageNumber = 0;
for (std::map<unsigned long, DTSCPageData>::iterator it = pagesByTrack[track].begin(); it != pagesByTrack[track].end(); it++) {
for (std::map<unsigned long, DTSCPageData>::iterator it = nProxy.pagesByTrack[track].begin(); it != nProxy.pagesByTrack[track].end(); it++) {
if (it->first <= keyNum) {
pageNumber = it->first;
} else {
@ -457,13 +408,13 @@ namespace Mist {
VERYHIGH_MSG("Track %u, key %u is already buffered in page %d. Cancelling bufferFrame", track, keyNum, pageNumber);
return true;
}
if (!pagesByTrack.count(track)) {
if (!nProxy.pagesByTrack.count(track)) {
WARN_MSG("No pages for track %u found! Cancelling bufferFrame", track);
return false;
}
//Update keynum to point to the corresponding page
INFO_MSG("Loading key %u from page %lu", keyNum, (--(pagesByTrack[track].upper_bound(keyNum)))->first);
keyNum = (--(pagesByTrack[track].upper_bound(keyNum)))->first;
INFO_MSG("Loading key %u from page %lu", keyNum, (--(nProxy.pagesByTrack[track].upper_bound(keyNum)))->first);
keyNum = (--(nProxy.pagesByTrack[track].upper_bound(keyNum)))->first;
if (!bufferStart(track, keyNum)) {
WARN_MSG("bufferStart failed! Cancelling bufferFrame");
return false;
@ -474,8 +425,8 @@ namespace Mist {
trackSelect(trackSpec.str());
seek(myMeta.tracks[track].keys[keyNum - 1].getTime());
long long unsigned int stopTime = myMeta.tracks[track].lastms + 1;
if ((int)myMeta.tracks[track].keys.size() > keyNum - 1 + pagesByTrack[track][keyNum].keyNum) {
stopTime = myMeta.tracks[track].keys[keyNum - 1 + pagesByTrack[track][keyNum].keyNum].getTime();
if ((int)myMeta.tracks[track].keys.size() > keyNum - 1 + nProxy.pagesByTrack[track][keyNum].keyNum) {
stopTime = myMeta.tracks[track].keys[keyNum - 1 + nProxy.pagesByTrack[track][keyNum].keyNum].getTime();
}
DEBUG_MSG(DLVL_HIGH, "Playing from %llu to %llu", myMeta.tracks[track].keys[keyNum - 1].getTime(), stopTime);
getNext();

129
src/input/input_buffer.cpp
View file

@ -89,17 +89,34 @@ namespace Mist {
capa["optional"]["segmentsize"]["type"] = "uint";
capa["optional"]["segmentsize"]["default"] = 5000LL;
option.null();
option["arg"] = "integer";
option["arg"] = "string";
option["long"] = "udp-port";
option["short"] = "U";
option["help"] = "The UDP port on which to listen for TS Packets";
option["value"].append(0LL);
option["value"].append("");
config->addOption("udpport", option);
capa["optional"]["udpport"]["name"] = "TS/UDP port";
capa["optional"]["udpport"]["help"] = "The UDP port on which to listen for TS Packets, or 0 for disabling TS Input";
capa["optional"]["udpport"]["help"] = "The UDP port on which to listen for TS Packets, or 0 for disabling TS Input, optionally prefixed with the interface IP to listen on.";
capa["optional"]["udpport"]["option"] = "--udp-port";
capa["optional"]["udpport"]["type"] = "uint";
capa["optional"]["udpport"]["default"] = 0LL;
capa["optional"]["udpport"]["type"] = "str";
capa["optional"]["udpport"]["default"] = "";
option.null();
option["arg"] = "string";
option["long"] = "multicast-interface";
option["short"] = "M";
option["help"] = "The interface(s)s on which to listen for UDP Multicast packets, space separated.";
option["value"].append("");
config->addOption("multicastinterface", option);
capa["optional"]["multicastinterface"]["name"] = "TS Multicast interface";
capa["optional"]["multicastinterface"]["help"] = "The interface(s) on which to listen for UDP Multicast packets, comma separated.";
capa["optional"]["multicastinterface"]["option"] = "--multicast-interface";
capa["optional"]["multicastinterface"]["type"] = "str";
capa["optional"]["multicastinterface"]["default"] = "";
option.null();
/*LTS-end*/
capa["source_match"] = "push://*";
capa["priority"] = 9ll;
@ -130,12 +147,12 @@ namespace Mist {
DEBUG_MSG(DLVL_DEVEL, "Cleaning up, removing last keyframes");
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
std::map<unsigned long, DTSCPageData> & locations = bufferLocations[it->first];
if (!metaPages.count(it->first) || !metaPages[it->first].mapped){
if (!nProxy.metaPages.count(it->first) || !nProxy.metaPages[it->first].mapped){
continue;
}
//First detect all entries on metaPage
for (int i = 0; i < 8192; i += 8){
int * tmpOffset = (int *)(metaPages[it->first].mapped + i);
int * tmpOffset = (int *)(nProxy.metaPages[it->first].mapped + i);
if (tmpOffset[0] == 0 && tmpOffset[1] == 0){
continue;
}
@ -206,14 +223,14 @@ namespace Mist {
snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, streamName.c_str());
IPC::semaphore liveMeta(liveSemName, O_CREAT | O_RDWR, ACCESSPERMS, 1);
liveMeta.wait();
if (!metaPages.count(0) || !metaPages[0].mapped){
if (!nProxy.metaPages.count(0) || !nProxy.metaPages[0].mapped){
char pageName[NAME_BUFFER_SIZE];
snprintf(pageName, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
metaPages[0].init(pageName, DEFAULT_META_PAGE_SIZE, true);
metaPages[0].master = false;
nProxy.metaPages[0].init(pageName, DEFAULT_META_PAGE_SIZE, true);
nProxy.metaPages[0].master = false;
}
myMeta.writeTo(metaPages[0].mapped);
memset(metaPages[0].mapped + myMeta.getSendLen(), 0, (metaPages[0].len > myMeta.getSendLen() ? std::min(metaPages[0].len - myMeta.getSendLen(), 4ll) : 0));
myMeta.writeTo(nProxy.metaPages[0].mapped);
memset(nProxy.metaPages[0].mapped + myMeta.getSendLen(), 0, (nProxy.metaPages[0].len > myMeta.getSendLen() ? std::min(nProxy.metaPages[0].len - myMeta.getSendLen(), 4ll) : 0));
liveMeta.post();
}
@ -296,15 +313,15 @@ namespace Mist {
if (myMeta.tracks[tid].keys[0].getNumber() >= (++(bufferLocations[tid].begin()))->first || !config->is_active){
//Find page in indexpage and null it
for (int i = 0; i < 8192; i += 8){
unsigned int thisKeyNum = ((((long long int *)(metaPages[tid].mapped + i))[0]) >> 32) & 0xFFFFFFFF;
if (thisKeyNum == htonl(bufferLocations[tid].begin()->first) && ((((long long int *)(metaPages[tid].mapped + i))[0]) != 0)){
(((long long int *)(metaPages[tid].mapped + i))[0]) = 0;
unsigned int thisKeyNum = ((((long long int *)(nProxy.metaPages[tid].mapped + i))[0]) >> 32) & 0xFFFFFFFF;
if (thisKeyNum == htonl(bufferLocations[tid].begin()->first) && ((((long long int *)(nProxy.metaPages[tid].mapped + i))[0]) != 0)){
(((long long int *)(nProxy.metaPages[tid].mapped + i))[0]) = 0;
}
}
DEBUG_MSG(DLVL_DEVEL, "Erasing track %d, keys %lu-%lu from buffer", tid, bufferLocations[tid].begin()->first, bufferLocations[tid].begin()->first + bufferLocations[tid].begin()->second.keyNum - 1);
bufferRemove(tid, bufferLocations[tid].begin()->first);
for (int i = 0; i < 1024; i++){
int * tmpOffset = (int *)(metaPages[tid].mapped + (i * 8));
int * tmpOffset = (int *)(nProxy.metaPages[tid].mapped + (i * 8));
int tmpNum = ntohl(tmpOffset[0]);
if (tmpNum == bufferLocations[tid].begin()->first){
tmpOffset[0] = 0;
@ -312,12 +329,12 @@ namespace Mist {
}
}
curPageNum.erase(tid);
nProxy.curPageNum.erase(tid);
char thisPageName[NAME_BUFFER_SIZE];
snprintf(thisPageName, NAME_BUFFER_SIZE, SHM_TRACK_DATA, config->getString("streamname").c_str(), (unsigned long)tid, bufferLocations[tid].begin()->first);
curPage[tid].init(thisPageName, 20971520);
curPage[tid].master = true;
curPage.erase(tid);
nProxy.curPage[tid].init(thisPageName, 20971520);
nProxy.curPage[tid].master = true;
nProxy.curPage.erase(tid);
bufferLocations[tid].erase(bufferLocations[tid].begin());
} else {
@ -334,13 +351,13 @@ namespace Mist {
for (std::map<unsigned long, DTSCPageData>::iterator it = bufferLocations[tid].begin(); it != bufferLocations[tid].end(); it++){
char thisPageName[NAME_BUFFER_SIZE];
snprintf(thisPageName, NAME_BUFFER_SIZE, SHM_TRACK_DATA, config->getString("streamname").c_str(), tid, it->first);
curPage[tid].init(thisPageName, 20971520, false, false);
curPage[tid].master = true;
curPage.erase(tid);
nProxy.curPage[tid].init(thisPageName, 20971520, false, false);
nProxy.curPage[tid].master = true;
nProxy.curPage.erase(tid);
}
bufferLocations.erase(tid);
metaPages[tid].master = true;
metaPages.erase(tid);
nProxy.metaPages[tid].master = true;
nProxy.metaPages.erase(tid);
}
void inputBuffer::finish() {
@ -410,9 +427,9 @@ namespace Mist {
while (bufferLocations[tid].size()){
char thisPageName[NAME_BUFFER_SIZE];
snprintf(thisPageName, NAME_BUFFER_SIZE, SHM_TRACK_DATA, config->getString("streamname").c_str(), (unsigned long)tid, bufferLocations[tid].begin()->first);
curPage[tid].init(thisPageName, 20971520);
curPage[tid].master = true;
curPage.erase(tid);
nProxy.curPage[tid].init(thisPageName, 20971520);
nProxy.curPage[tid].master = true;
nProxy.curPage.erase(tid);
bufferLocations[tid].erase(bufferLocations[tid].begin());
}
if (pushLocation.count(it->first)){
@ -427,9 +444,9 @@ namespace Mist {
}
pushLocation.erase(it->first);
}
curPageNum.erase(it->first);
metaPages[it->first].master = true;
metaPages.erase(it->first);
nProxy.curPageNum.erase(it->first);
nProxy.metaPages[it->first].master = true;
nProxy.metaPages.erase(it->first);
activeTracks.erase(it->first);
myMeta.tracks.erase(it->first);
changed = true;
@ -522,8 +539,8 @@ namespace Mist {
activeTracks.erase(value);
bufferLocations.erase(value);
}
metaPages[value].master = true;
metaPages.erase(value);
nProxy.metaPages[value].master = true;
nProxy.metaPages.erase(value);
continue;
}
}
@ -546,13 +563,13 @@ namespace Mist {
//The track id is set to the value of a track that we are currently negotiating about
if (negotiatingTracks.count(value)){
//If the metadata page for this track is not yet registered, initialize it
if (!metaPages.count(value) || !metaPages[value].mapped){
if (!nProxy.metaPages.count(value) || !nProxy.metaPages[value].mapped){
char tempMetaName[NAME_BUFFER_SIZE];
snprintf(tempMetaName, NAME_BUFFER_SIZE, SHM_TRACK_META, config->getString("streamname").c_str(), value);
metaPages[value].init(tempMetaName, 8388608, false, false);
nProxy.metaPages[value].init(tempMetaName, 8388608, false, false);
}
//If this tracks metdata page is not initialize, skip the entire element for now. It will be instantiated later
if (!metaPages[value].mapped) {
if (!nProxy.metaPages[value].mapped) {
//remove the negotiation if it has timed out
if (++negotiationTimeout[value] >= 1000){
negotiatingTracks.erase(value);
@ -564,13 +581,13 @@ namespace Mist {
//The page exist, now we try to read in the metadata of the track
//Store the size of the dtsc packet to read.
unsigned int len = ntohl(((int *)metaPages[value].mapped)[1]);
unsigned int len = ntohl(((int *)nProxy.metaPages[value].mapped)[1]);
//Temporary variable, won't be used again
unsigned int tempForReadingMeta = 0;
//Read in the metadata through a temporary JSON object
///\todo Optimize this part. Find a way to not have to store the metadata in JSON first, but read it from the page immediately
JSON::Value tempJSONForMeta;
JSON::fromDTMI((const unsigned char *)metaPages[value].mapped + 8, len, tempForReadingMeta, tempJSONForMeta);
JSON::fromDTMI((const unsigned char *)nProxy.metaPages[value].mapped + 8, len, tempForReadingMeta, tempJSONForMeta);
//Construct a metadata object for the current track
DTSC::Meta trackMeta(tempJSONForMeta);
//If the track metadata does not contain the negotiated track, assume the metadata is currently being written, and skip the element for now. It will be instantiated in the next call.
@ -579,8 +596,8 @@ namespace Mist {
if (++negotiationTimeout[value] >= 1000){
negotiatingTracks.erase(value);
//Set master to true before erasing the page, because we are responsible for cleaning up unused pages
metaPages[value].master = true;
metaPages.erase(value);
nProxy.metaPages[value].master = true;
nProxy.metaPages.erase(value);
negotiationTimeout.erase(value);
}
continue;
@ -603,8 +620,8 @@ namespace Mist {
//Remove the "negotiate" status in either case
negotiatingTracks.erase(value);
//Set master to true before erasing the page, because we are responsible for cleaning up unused pages
metaPages[value].master = true;
metaPages.erase(value);
nProxy.metaPages[value].master = true;
nProxy.metaPages.erase(value);
//Check if the track collides, and whether the track it collides with is active.
if (collidesWith != -1 && activeTracks.count(collidesWith)){/*LTS*/
@ -639,8 +656,8 @@ namespace Mist {
//Set master to true before erasing the page, because we are responsible for cleaning up unused pages
updateMeta();
eraseTrackDataPages(value);
metaPages[finalMap].master = true;
metaPages.erase(finalMap);
nProxy.metaPages[finalMap].master = true;
nProxy.metaPages.erase(finalMap);
bufferLocations.erase(finalMap);
}
@ -666,12 +683,12 @@ namespace Mist {
//If the track is active, and this is the element responsible for pushing it
if (activeTracks.count(value) && pushLocation[value] == data){
//Open the track index page if we dont have it open yet
if (!metaPages.count(value) || !metaPages[value].mapped){
if (!nProxy.metaPages.count(value) || !nProxy.metaPages[value].mapped){
char firstPage[NAME_BUFFER_SIZE];
snprintf(firstPage, NAME_BUFFER_SIZE, SHM_TRACK_INDEX, config->getString("streamname").c_str(), value);
metaPages[value].init(firstPage, 8192, false, false);
nProxy.metaPages[value].init(firstPage, 8192, false, false);
}
if (metaPages[value].mapped){
if (nProxy.metaPages[value].mapped){
//Update the metadata for this track
updateTrackMeta(value);
hasPush = true;
@ -684,7 +701,7 @@ namespace Mist {
VERYHIGH_MSG("Updating meta for track %d", tNum);
//Store a reference for easier access
std::map<unsigned long, DTSCPageData> & locations = bufferLocations[tNum];
char * mappedPointer = metaPages[tNum].mapped;
char * mappedPointer = nProxy.metaPages[tNum].mapped;
//First detect all entries on metaPage
for (int i = 0; i < 8192; i += 8) {
@ -725,27 +742,27 @@ namespace Mist {
//Otherwise open and parse the page
//Open the page if it is not yet open
if (!curPageNum.count(tNum) || curPageNum[tNum] != pageNum || !curPage[tNum].mapped){
if (!nProxy.curPageNum.count(tNum) || nProxy.curPageNum[tNum] != pageNum || !nProxy.curPage[tNum].mapped){
//DO NOT ERASE THE PAGE HERE, master is not set to true
curPageNum.erase(tNum);
nProxy.curPageNum.erase(tNum);
char nextPageName[NAME_BUFFER_SIZE];
snprintf(nextPageName, NAME_BUFFER_SIZE, SHM_TRACK_DATA, config->getString("streamname").c_str(), tNum, pageNum);
curPage[tNum].init(nextPageName, 20971520);
nProxy.curPage[tNum].init(nextPageName, 20971520);
//If the page can not be opened, stop here
if (!curPage[tNum].mapped){
if (!nProxy.curPage[tNum].mapped){
WARN_MSG("Could not open page: %s", nextPageName);
return;
}
curPageNum[tNum] = pageNum;
nProxy.curPageNum[tNum] = pageNum;
}
DTSC::Packet tmpPack;
if (!curPage[tNum].mapped[pageData.curOffset]){
if (!nProxy.curPage[tNum].mapped[pageData.curOffset]){
VERYHIGH_MSG("No packet on page %lu for track %lu, waiting...", pageNum, tNum);
return;
}
tmpPack.reInit(curPage[tNum].mapped + pageData.curOffset, 0);
tmpPack.reInit(nProxy.curPage[tNum].mapped + pageData.curOffset, 0);
//No new data has been written on the page since last update
if (!tmpPack){
return;
@ -761,7 +778,7 @@ namespace Mist {
//Update the offset on the page with the size of the current packet
pageData.curOffset += tmpPack.getDataLen();
//Attempt to read in the next packet
tmpPack.reInit(curPage[tNum].mapped + pageData.curOffset, 0);
tmpPack.reInit(nProxy.curPage[tNum].mapped + pageData.curOffset, 0);
}
}

310
src/input/input_ts.cpp
View file

@ -10,9 +10,16 @@
#include <mist/flv_tag.h>
#include <mist/defines.h>
#include <mist/ts_packet.h>
#include <mist/timing.h>
#include <mist/mp4_generic.h>
#include "input_ts.h"
#include <mist/tinythread.h>
#include <sys/stat.h>
/// \todo Implement this trigger equivalent...
/*
if(Triggers::shouldTrigger("STREAM_PUSH", smp)){
@ -27,10 +34,82 @@ if(Triggers::shouldTrigger("STREAM_PUSH", smp)){
}
*/
#ifdef TSLIVE_INPUT
std::string globalStreamName;
TS::Stream liveStream(true);
Util::Config * cfgPointer = NULL;
#define THREAD_TIMEOUT 15
std::map<unsigned long long, unsigned long long> threadTimer;
std::set<unsigned long> claimableThreads;
void parseThread(void * ignored) {
std::string semName = "MstInTSStreamClaim" + globalStreamName;
IPC::semaphore lock(semName.c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
int tid = -1;
lock.wait();
if (claimableThreads.size()) {
tid = *claimableThreads.begin();
claimableThreads.erase(claimableThreads.begin());
}
lock.post();
if (tid == -1) {
return;
}
if (liveStream.isDataTrack(tid)){
if (!Util::startInput(globalStreamName)) {
return;
}
}
Mist::negotiationProxy myProxy;
myProxy.streamName = globalStreamName;
DTSC::Meta myMeta;
if (liveStream.isDataTrack(tid)){
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, globalStreamName.c_str());
myProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
}
threadTimer[tid] = Util::bootSecs();
while (Util::bootSecs() - threadTimer[tid] < THREAD_TIMEOUT && cfgPointer->is_active) {
liveStream.parse(tid);
if (liveStream.hasPacket(tid)){
liveStream.initializeMetadata(myMeta, tid);
DTSC::Packet pack;
liveStream.getPacket(tid, pack);
if (pack && myMeta.tracks.count(tid)){
myProxy.bufferLivePacket(pack, myMeta);
}
lock.wait();
threadTimer[tid] = Util::bootSecs();
lock.post();
}
if (liveStream.isDataTrack(tid)){
myProxy.userClient.keepAlive();
}
if (!liveStream.hasPacket(tid)){
Util::sleep(100);
}
}
lock.wait();
threadTimer.erase(tid);
lock.post();
liveStream.eraseTrack(tid);
myProxy.userClient.finish();
}
#endif
namespace Mist {
/// Constructor of TS Input
/// \arg cfg Util::Config that contains all current configurations.
inputTS::inputTS(Util::Config * cfg) : Input(cfg) {
@ -44,44 +123,80 @@ namespace Mist {
capa["codecs"][0u][1u].append("AC3");
capa["optional"]["port"]["name"] = "UDP Port";
capa["optional"]["port"]["help"] = "The udp port on which to listen for incoming UDP Packets";
capa["optional"]["port"]["type"] = "uint";
capa["optional"]["port"]["default"] = 9876;
capa["optional"]["port"]["help"] = "The UDP port on which to listen for incoming UDP Packets, optionally prefixed by the interface IP.";
capa["optional"]["port"]["type"] = "string";
capa["optional"]["port"]["default"] = "9876";
capa["optional"]["port"]["option"] = "--port";
cfg->addOption("port",
JSON::fromString("{\"arg\":\"integer\",\"value\":9876,\"short\":\"p\",\"long\":\"port\",\"help\":\"The udp port on which to listen for incoming UDP Packets.\"}"));
JSON::fromString("{\"arg\":\"string\",\"value\":9876,\"short\":\"p\",\"long\":\"port\",\"help\":\"The UDP port on which to listen for incoming UDP Packets, optionally prefixed by the interface IP.\"}"));
capa["optional"]["multicastinterface"]["name"] = "TS Multicast interface";
capa["optional"]["multicastinterface"]["help"] = "The interface(s) on which to listen for UDP Multicast packets, comma separated.";
capa["optional"]["multicastinterface"]["option"] = "--multicast-interface";
capa["optional"]["multicastinterface"]["type"] = "str";
capa["optional"]["multicastinterface"]["default"] = "";
cfg->addOption("multicastinterface",
JSON::fromString("{\"arg\":\"string\",\"value\":\"\",\"short\":\"M\",\"long\":\"multicast-interface\",\"help\":\"The interfaces on which to listen for UDP Multicast packets, space separatered.\"}"));
pushing = false;
inFile = NULL;
#ifdef TSLIVE_INPUT
standAlone = false;
#endif
}
inputTS::~inputTS() {
if (inFile){
if (inFile) {
fclose(inFile);
}
#ifdef TSLIVE_INPUT
std::string semName = "MstInTSStreamClaim" + globalStreamName;
IPC::semaphore lock(semName.c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
lock.wait();
threadTimer.clear();
claimableThreads.clear();
lock.post();
#endif
}
///Setup of TS Input
#ifdef TSLIVE_INPUT
///Live Setup of TS Input
bool inputTS::setup() {
#ifdef INPUT_LIVE
INFO_MSG("Setup start");
if (config->getString("input") == "-") {
inFile = stdin;
}else{
} else {
pushing = true;
udpCon.setBlocking(false);
udpCon.bind(config->getInteger("port"));
std::string ipPort = config->getString("port");
size_t colon = ipPort.rfind(':');
if (colon != std::string::npos) {
udpCon.bind(JSON::Value(ipPort.substr(colon + 1)).asInt(), ipPort.substr(0, colon), config->getString("multicastinterface"));
} else {
udpCon.bind(JSON::Value(ipPort).asInt(), "", config->getString("multicastinterface"));
}
}
INFO_MSG("Setup complete");
return true;
}
#else
if (config->getString("input") != "-"){
///Setup of TS Input
bool inputTS::setup() {
if (config->getString("input") != "-") {
inFile = fopen(config->getString("input").c_str(), "r");
}
if (!inFile) {
return false;
}
#endif
return true;
}
#endif
///Track selector of TS Input
///\arg trackSpec specifies which tracks are to be selected
///\todo test whether selecting a subset of tracks work
@ -99,33 +214,40 @@ namespace Mist {
}
}
#ifdef TSLIVE_INPUT
//This implementation in used in the live version of TS input, where no header is available in advance.
//Reading the header returns true in this case, to continue parsing the actual stream.
bool inputTS::readHeader() {
return true;
}
#else
///Reads headers from a TS stream, and saves them into metadata
///It works by going through the entire TS stream, and every time
///It encounters a new PES start, it writes the currently found PES data
///for a specific track to metadata. After the entire stream has been read,
///for a specific track to metadata. After the entire stream has been read,
///it writes the remaining metadata.
///\todo Find errors, perhaps parts can be made more modular
bool inputTS::readHeader(){
bool inputTS::readHeader() {
if (!inFile) {
return false;
}
DTSC::File tmp(config->getString("input") + ".dtsh");
if (tmp){
if (tmp) {
myMeta = tmp.getMeta();
return true;
}
}
TS::Packet packet;//to analyse and extract data
fseek(inFile, 0, SEEK_SET);//seek to beginning
bool first = true;
long long int lastBpos = 0;
while (packet.FromFile(inFile) && !feof(inFile)){
while (packet.FromFile(inFile) && !feof(inFile)) {
tsStream.parse(packet, lastBpos);
lastBpos = ftell(inFile);
while(tsStream.hasPacketOnEachTrack()){
if (first){
while (tsStream.hasPacketOnEachTrack()) {
if (first) {
tsStream.initializeMetadata(myMeta);
first = false;
}
@ -142,106 +264,176 @@ namespace Mist {
oFile.close();
return true;
}
#endif
///Gets the next packet that is to be sent
///At the moment, the logic of sending the last packet that was finished has been implemented,
///Gets the next packet that is to be sent
///At the moment, the logic of sending the last packet that was finished has been implemented,
///but the seeking and finding data is not yet ready.
///\todo Finish the implementation
void inputTS::getNext(bool smart){
void inputTS::getNext(bool smart) {
INSANE_MSG("Getting next");
thisPacket.null();
bool hasPacket = (selectedTracks.size() == 1 ? tsStream.hasPacket(*selectedTracks.begin()) : tsStream.hasPacketOnEachTrack());
while (!hasPacket && (pushing || !feof(inFile)) && config->is_active){
while (!hasPacket && (pushing || !feof(inFile)) && config->is_active) {
if (!pushing) {
unsigned int bPos = ftell(inFile);
tsBuf.FromFile(inFile);
if (selectedTracks.count(tsBuf.getPID())){
if (selectedTracks.count(tsBuf.getPID())) {
tsStream.parse(tsBuf, bPos);
}
}else{
while (udpCon.Receive()){
} else {
while (udpCon.Receive()) {
udpDataBuffer.append(udpCon.data, udpCon.data_len);
while (udpDataBuffer.size() > 188 && (udpDataBuffer[0] != 0x47 || udpDataBuffer[188] != 0x47)){
while (udpDataBuffer.size() > 188 && (udpDataBuffer[0] != 0x47 || udpDataBuffer[188] != 0x47)) {
size_t syncPos = udpDataBuffer.find("\107", 1);
udpDataBuffer.erase(0, syncPos);
}
while (udpDataBuffer.size() >= 188){
while (udpDataBuffer.size() >= 188) {
tsBuf.FromPointer(udpDataBuffer.data());
tsStream.parse(tsBuf, 0);
udpDataBuffer.erase(0,188);
udpDataBuffer.erase(0, 188);
}
}
if (userClient.getData()){
userClient.keepAlive();
}
Util::sleep(500);
}
if (userClient.getData()){
userClient.keepAlive();
}
hasPacket = (selectedTracks.size() == 1 ? tsStream.hasPacket(*selectedTracks.begin()) : tsStream.hasPacketOnEachTrack());
}
if (!hasPacket){
if(inFile && !feof(inFile)){
if (!hasPacket) {
if (inFile && !feof(inFile)) {
getNext();
}
if (pushing){
if (pushing) {
sleep(500);
}
return;
}
if (selectedTracks.size() == 1){
if (selectedTracks.size() == 1) {
tsStream.getPacket(*selectedTracks.begin(), thisPacket);
}else{
} else {
tsStream.getEarliestPacket(thisPacket);
}
tsStream.initializeMetadata(myMeta);
if (!myMeta.tracks.count(thisPacket.getTrackId())){
if (!myMeta.tracks.count(thisPacket.getTrackId())) {
getNext();
}
}
void inputTS::readPMT(){
void inputTS::readPMT() {
//save current file position
int bpos = ftell(inFile);
if (fseek(inFile, 0, SEEK_SET)){
if (fseek(inFile, 0, SEEK_SET)) {
FAIL_MSG("Seek to 0 failed");
return;
}
TS::Packet tsBuffer;
while (!tsStream.hasPacketOnEachTrack() && tsBuffer.FromFile(inFile)){
while (!tsStream.hasPacketOnEachTrack() && tsBuffer.FromFile(inFile)) {
tsStream.parse(tsBuffer, 0);
}
//Clear leaves the PMT in place
tsStream.clear();
//Restore original file position
if (fseek(inFile, bpos, SEEK_SET)){
if (fseek(inFile, bpos, SEEK_SET)) {
return;
}
}
///Seeks to a specific time
void inputTS::seek(int seekTime){
void inputTS::seek(int seekTime) {
tsStream.clear();
readPMT();
unsigned long seekPos = 0xFFFFFFFFull;
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++) {
unsigned long thisBPos = 0;
for (std::deque<DTSC::Key>::iterator keyIt = myMeta.tracks[*it].keys.begin(); keyIt != myMeta.tracks[*it].keys.end(); keyIt++){
if (keyIt->getTime() > seekTime){
for (std::deque<DTSC::Key>::iterator keyIt = myMeta.tracks[*it].keys.begin(); keyIt != myMeta.tracks[*it].keys.end(); keyIt++) {
if (keyIt->getTime() > seekTime) {
break;
}
thisBPos = keyIt->getBpos();
}
if (thisBPos < seekPos){
if (thisBPos < seekPos) {
seekPos = thisBPos;
}
}
fseek(inFile, seekPos, SEEK_SET);//seek to the correct position
}
#ifdef TSLIVE_INPUT
void inputTS::serve() {
cfgPointer = config;
globalStreamName = streamName;
unsigned long long threadCheckTimer = Util::bootSecs();
while (config->is_active) {
if (!pushing) {
unsigned int bPos = ftell(inFile);
int ctr = 0;
while (ctr < 20 && tsBuf.FromFile(inFile)){
liveStream.add(tsBuf);
ctr++;
}
} else {
while (udpCon.Receive()) {
int offset = 0;
//Try to read full TS Packets
//Assumption here is made that one UDP Datagram consists of complete TS Packets.
//Assumption made because of possible packet loss issues
while ((udpCon.data_len - offset) >= 188) {
//Watch out! We push here to a global, in order for threads to be able to access it.
liveStream.add(udpCon.data + offset);
offset += 188;
}
if (offset < udpCon.data_len) {
WARN_MSG("%d bytes left in datagram", udpCon.data_len - offset);
}
}
}
//Check for and spawn threads here.
if (Util::bootSecs() - threadCheckTimer > 2) {
std::set<unsigned long> activeTracks = liveStream.getActiveTracks();
std::string semName = "MstInTSStreamClaim" + globalStreamName;
IPC::semaphore lock(semName.c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
lock.wait();
for (std::set<unsigned long>::iterator it = activeTracks.begin(); it != activeTracks.end(); it++) {
if (threadTimer.count(*it) && ((Util::bootSecs() - threadTimer[*it]) > (2 * THREAD_TIMEOUT))) {
WARN_MSG("Thread for track %d timed out %d seconds ago without a clean shutdown.", *it, Util::bootSecs() - threadTimer[*it]);
threadTimer.erase(*it);
}
if (!threadTimer.count(*it)) {
//Add to list of unclaimed threads
claimableThreads.insert(*it);
//Spawn thread here.
tthread::thread thisThread(parseThread, 0);
thisThread.detach();
}
}
lock.post();
threadCheckTimer = Util::bootSecs();
}
Util::sleep(100);
}
finish();
INFO_MSG("Input for stream %s closing clean", streamName.c_str());
}
void inputTS::finish() {
std::string semName = "MstInTSStreamClaim" + globalStreamName;
IPC::semaphore lock(semName.c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
int threadCount = 0;
do {
lock.wait();
threadCount = threadTimer.size();
lock.post();
} while (threadCount);
}
#endif
}

11
src/input/input_ts.h
View file

@ -22,6 +22,17 @@ namespace Mist {
void trackSelect(std::string trackSpec);
void readPMT();
#ifdef TSLIVE_INPUT
//Live tsinput does not have a header, so parseheader should do nothing
void parseHeader() { }
//In case of live TS Input, we override the default serve function
void serve();
void finish();
#endif
FILE * inFile;///<The input file with ts data
TS::Stream tsStream;///<Used for parsing the incoming ts stream