diff --git a/lib/procs.cpp b/lib/procs.cpp index f136b92a..3fb7212c 100644 --- a/lib/procs.cpp +++ b/lib/procs.cpp @@ -444,6 +444,18 @@ int Util::Procs::Count() { /// Returns true if a process with this PID is currently active. bool Util::Procs::isActive(pid_t name) { tthread::lock_guard guard(plistMutex); - return (plist.count(name) == 1) && (kill(name, 0) == 0); + return (kill(name, 0) == 0); +} + +/// Forget about the given PID, keeping it running on shutdown. +void Util::Procs::forget(pid_t pid) { + tthread::lock_guard guard(plistMutex); + plist.erase(pid); +} + +/// Remember the given PID, killing it on shutdown. +void Util::Procs::remember(pid_t pid) { + tthread::lock_guard guard(plistMutex); + plist.insert(pid); } diff --git a/lib/procs.h b/lib/procs.h index 2c07cda8..667f570b 100644 --- a/lib/procs.h +++ b/lib/procs.h @@ -38,6 +38,8 @@ namespace Util { static int Count(); static bool isActive(pid_t name); static bool isRunning(pid_t pid); + static void forget(pid_t pid); + static void remember(pid_t pid); static std::set socketList; ///< Holds sockets that should be closed before forking }; } diff --git a/src/input/input.cpp b/src/input/input.cpp index 0222dbdd..607f44ee 100644 --- a/src/input/input.cpp +++ b/src/input/input.cpp @@ -142,6 +142,17 @@ namespace Mist { return 0; } + /// Default crash handler, cleans up Pull semaphore on crashes + void Input::onCrash(){ + if (streamName.size() && !needsLock()) { + //we have a Pull semaphore to clean up, do it + IPC::semaphore pullLock; + pullLock.open(std::string("/MstPull_" + streamName).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1); + pullLock.close(); + pullLock.unlink(); + } + } + void Input::convert() { //check filename for no - if (config->getString("output") != "-") { @@ -262,12 +273,18 @@ namespace Mist { } /// Main loop for stream-style inputs. - /// This loop will start the buffer without resume support, and then repeatedly call ..... followed by .... + /// This loop will do the following, in order: + /// - exit if another stream() input is already open for this streamname + /// - start a buffer in push mode + /// - connect to it + /// - run parseStreamHeader + /// - if there are tracks, register as a non-viewer on the user page of the buffer + /// - call getNext() in a loop, buffering packets void Input::stream(){ IPC::semaphore pullLock; pullLock.open(std::string("/MstPull_" + streamName).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1); if (!pullLock.tryWait()){ - DEBUG_MSG(DLVL_DEVEL, "A pull process for stream %s is already running", streamName.c_str()); + WARN_MSG("A pull process for stream %s is already running", streamName.c_str()); pullLock.close(); return; } diff --git a/src/input/input.h b/src/input/input.h index 66580bb1..f80ac6c7 100644 --- a/src/input/input.h +++ b/src/input/input.h @@ -20,7 +20,7 @@ namespace Mist { public: Input(Util::Config * cfg); virtual int run(); - virtual void onCrash(){} + virtual void onCrash(); virtual void argumentsParsed(){} virtual ~Input() {}; diff --git a/src/output/output.cpp b/src/output/output.cpp index 064270dc..56b07757 100644 --- a/src/output/output.cpp +++ b/src/output/output.cpp @@ -547,7 +547,10 @@ namespace Mist{ } return highest; } - + + /// Loads the page for the given trackId and keyNum into memory. + /// Overwrites any existing page for the same trackId. + /// Automatically calls thisPacket.null() if necessary. void Output::loadPageForKey(long unsigned int trackId, long long int keyNum){ if (!myMeta.tracks.count(trackId) || !myMeta.tracks[trackId].keys.size()){ WARN_MSG("Load for track %lu key %lld aborted - track is empty", trackId, keyNum); @@ -598,6 +601,10 @@ namespace Mist{ if (currKeyOpen.count(trackId) && currKeyOpen[trackId] == (unsigned int)pageNum){ return; } + //If we're loading the track thisPacket is on, null it to prevent accesses. + if (thisPacket && thisPacket.getTrackId() == trackId){ + thisPacket.null(); + } char id[NAME_BUFFER_SIZE]; snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_DATA, streamName.c_str(), trackId, pageNum); nProxy.curPage[trackId].init(id, DEFAULT_DATA_PAGE_SIZE); @@ -951,7 +958,7 @@ namespace Mist{ } ///Attempts to prepare a new packet for output. - ///If thisPacket evaluates to false, playback has completed. + ///If it returns true and 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. @@ -1009,7 +1016,9 @@ namespace Mist{ //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){ - nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime()); + if (thisPacket){ + nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime()); + } loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]); nxt.offset = 0; if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){ @@ -1022,7 +1031,6 @@ namespace Mist{ buffer.insert(nxt); } }else{ - thisPacket.null(); dropTrack(nxt.tid, "page load failure", true); } return false; @@ -1063,7 +1071,6 @@ namespace Mist{ //The next key showed up on another page! //We've simply reached the end of the page. Load the next key = next page. loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]); - thisPacket.null(); nxt.offset = 0; if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){ unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset); diff --git a/src/output/output.h b/src/output/output.h index a45bba37..405b248f 100644 --- a/src/output/output.h +++ b/src/output/output.h @@ -61,6 +61,8 @@ namespace Mist { uint32_t currTrackCount() const; virtual bool isReadyForPlay(); //virtuals. The optional virtuals have default implementations that do as little as possible. + /// This function is called whenever a packet is ready for sending. + /// Inside it, thisPacket is guaranteed to contain a valid packet. virtual void sendNext() {}//REQUIRED! Others are optional. bool prepareNext(); virtual void dropTrack(uint32_t trackId, std::string reason, bool probablyBad = true); diff --git a/src/output/output_rtmp.cpp b/src/output/output_rtmp.cpp index a69d0be5..f5179fe5 100644 --- a/src/output/output_rtmp.cpp +++ b/src/output/output_rtmp.cpp @@ -257,7 +257,7 @@ namespace Mist { // erase & pos = nextpos + 1; } - if (trackSwitch){ + if (trackSwitch && thisPacket){ seek(thisPacket.getTime()); } }