Removed 5s timeout from streamAlive call.

This commit is contained in:
Thulinma 2016-05-05 14:28:39 +02:00
parent 6386060c10
commit fa41e02047
8 changed files with 36 additions and 27 deletions

View file

@ -95,13 +95,13 @@ namespace IPC {
///\param oflag The flags with which to open the semaphore ///\param oflag The flags with which to open the semaphore
///\param mode The mode in which to create the semaphore, if O_CREAT is given in oflag, ignored otherwise ///\param mode The mode in which to create the semaphore, if O_CREAT is given in oflag, ignored otherwise
///\param value The initial value of the semaphore if O_CREAT is given in oflag, ignored otherwise ///\param value The initial value of the semaphore if O_CREAT is given in oflag, ignored otherwise
semaphore::semaphore(const char * name, int oflag, mode_t mode, unsigned int value) { semaphore::semaphore(const char * name, int oflag, mode_t mode, unsigned int value, bool noWait) {
#if defined(__CYGWIN__) || defined(_WIN32) #if defined(__CYGWIN__) || defined(_WIN32)
mySem = 0; mySem = 0;
#else #else
mySem = SEM_FAILED; mySem = SEM_FAILED;
#endif #endif
open(name, oflag, mode, value); open(name, oflag, mode, value, noWait);
} }
///\brief The deconstructor ///\brief The deconstructor
@ -126,7 +126,7 @@ namespace IPC {
///\param oflag The flags with which to open the semaphore ///\param oflag The flags with which to open the semaphore
///\param mode The mode in which to create the semaphore, if O_CREAT is given in oflag, ignored otherwise ///\param mode The mode in which to create the semaphore, if O_CREAT is given in oflag, ignored otherwise
///\param value The initial value of the semaphore if O_CREAT is given in oflag, ignored otherwise ///\param value The initial value of the semaphore if O_CREAT is given in oflag, ignored otherwise
void semaphore::open(const char * name, int oflag, mode_t mode, unsigned int value) { void semaphore::open(const char * name, int oflag, mode_t mode, unsigned int value, bool noWait) {
close(); close();
int timer = 0; int timer = 0;
while (!(*this) && timer++ < 10) { while (!(*this) && timer++ < 10) {
@ -165,7 +165,7 @@ namespace IPC {
mySem = sem_open(name, oflag); mySem = sem_open(name, oflag);
} }
if (!(*this)) { if (!(*this)) {
if (errno == ENOENT) { if (errno == ENOENT && !noWait) {
Util::wait(500); Util::wait(500);
} else { } else {
break; break;

View file

@ -60,10 +60,10 @@ namespace IPC {
class semaphore { class semaphore {
public: public:
semaphore(); semaphore();
semaphore(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0); semaphore(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0, bool noWait = false);
~semaphore(); ~semaphore();
operator bool() const; operator bool() const;
void open(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0); void open(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0, bool noWait = false);
int getVal() const; int getVal() const;
void post(); void post();
void wait(); void wait();

View file

@ -109,7 +109,7 @@ JSON::Value Util::getStreamConfig(std::string streamname){
bool Util::streamAlive(std::string & streamname){ bool Util::streamAlive(std::string & streamname){
char semName[NAME_BUFFER_SIZE]; char semName[NAME_BUFFER_SIZE];
snprintf(semName, NAME_BUFFER_SIZE, SEM_INPUT, streamname.c_str()); snprintf(semName, NAME_BUFFER_SIZE, SEM_INPUT, streamname.c_str());
IPC::semaphore playerLock(semName, O_RDWR, ACCESSPERMS, 1); IPC::semaphore playerLock(semName, O_RDWR, ACCESSPERMS, 1, true);
if (!playerLock){return false;} if (!playerLock){return false;}
if (!playerLock.tryWait()) { if (!playerLock.tryWait()) {
playerLock.close(); playerLock.close();
@ -310,7 +310,13 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
FAIL_MSG("Starting process %s for stream %s failed: %s", argv[0], streamname.c_str(), strerror(errno)); FAIL_MSG("Starting process %s for stream %s failed: %s", argv[0], streamname.c_str(), strerror(errno));
_exit(42); _exit(42);
} }
return true;
unsigned int waiting = 0;
while (!streamAlive(streamname) && ++waiting < 40){
Util::sleep(250);
}
return streamAlive(streamname);
} }
/* roxlu-begin */ /* roxlu-begin */

View file

@ -783,7 +783,9 @@ void Controller::fillActive(JSON::Value & req, JSON::Value & rep, bool onlyNow){
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, it->c_str()); snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, it->c_str());
streamIndex.init(pageId, DEFAULT_META_PAGE_SIZE, false, false); streamIndex.init(pageId, DEFAULT_META_PAGE_SIZE, false, false);
if (streamIndex.mapped){ if (streamIndex.mapped){
IPC::semaphore metaLocker(std::string("liveMeta@" + (*it)).c_str(), O_CREAT | O_RDWR, (S_IRWXU|S_IRWXG|S_IRWXO), 1); static char liveSemName[NAME_BUFFER_SIZE];
snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, it->c_str());
IPC::semaphore metaLocker(liveSemName, O_CREAT | O_RDWR, (S_IRWXU|S_IRWXG|S_IRWXO), 1);
metaLocker.wait(); metaLocker.wait();
DTSC::Scan strm = DTSC::Packet(streamIndex.mapped, streamIndex.len, true).getScan(); DTSC::Scan strm = DTSC::Packet(streamIndex.mapped, streamIndex.len, true).getScan();
long long lms = 0; long long lms = 0;

View file

@ -195,8 +195,14 @@ namespace Mist {
/// ~~~~~~~~~~~~~~~ /// ~~~~~~~~~~~~~~~
// //
void Input::serve(){ void Input::serve(){
if (!isBuffer) {
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
bufferFrame(it->first, 1);
}
}
char userPageName[NAME_BUFFER_SIZE]; char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str()); snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
userPage.init(userPageName, PLAY_EX_SIZE, true);
/*LTS-START*/ /*LTS-START*/
if(Triggers::shouldTrigger("STREAM_READY", config->getString("streamname"))){ if(Triggers::shouldTrigger("STREAM_READY", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n"; std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
@ -205,12 +211,6 @@ namespace Mist {
} }
} }
/*LTS-END*/ /*LTS-END*/
userPage.init(userPageName, PLAY_EX_SIZE, true);
if (!isBuffer) {
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
bufferFrame(it->first, 1);
}
}
DEBUG_MSG(DLVL_DEVEL, "Input for stream %s started", streamName.c_str()); DEBUG_MSG(DLVL_DEVEL, "Input for stream %s started", streamName.c_str());

View file

@ -278,15 +278,6 @@ namespace Mist {
onFail(); onFail();
return; return;
} }
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
nProxy.metaPages.clear();
nProxy.metaPages[0].init(pageId, DEFAULT_META_PAGE_SIZE);
if (!nProxy.metaPages[0].mapped){
FAIL_MSG("Could not connect to server for %s", streamName.c_str());
onFail();
return;
}
if (statsPage.getData()){ if (statsPage.getData()){
statsPage.finish(); statsPage.finish();
} }
@ -297,6 +288,15 @@ namespace Mist {
char userPageName[NAME_BUFFER_SIZE]; char userPageName[NAME_BUFFER_SIZE];
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);
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
nProxy.metaPages.clear();
nProxy.metaPages[0].init(pageId, DEFAULT_META_PAGE_SIZE);
if (!nProxy.metaPages[0].mapped){
FAIL_MSG("Could not connect to server for %s", streamName.c_str());
onFail();
return;
}
updateMeta(); updateMeta();
} }

View file

@ -88,8 +88,7 @@ namespace Mist {
} }
} }
}else{ }else{
fastAsPossibleTime = 50000;//50 seconds realTime = 1000;
realTime = 0;
} }
} }

View file

@ -348,7 +348,9 @@ namespace Mist {
json_resp["on_error"] = config->getString("nostreamtext"); json_resp["on_error"] = config->getString("nostreamtext");
} }
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1); IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore metaLocker(std::string("liveMeta@" + streamName).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1); static char liveSemName[NAME_BUFFER_SIZE];
snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, streamName.c_str());
IPC::semaphore metaLocker(liveSemName, O_CREAT | O_RDWR, ACCESSPERMS, 1);
bool metaLock = false; bool metaLock = false;
configLock.wait(); configLock.wait();
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE); IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);