mirrors/neutrino
1
0
Fork 0
mirror of https://github.com/tuxbox-neutrino/neutrino.git synced 2025-08-27 15:32:59 +02:00
Code Issues Packages Projects Releases Wiki Activity

eitd/sectionsd.cpp: testing abstract classes and common thread code

Browse Source
This commit is contained in:
[CST] Focus
2012-02-29 19:48:38 +04:00
parent 4c6199eef5
commit 55dbf7f63e
2 changed files with 323 additions and 391 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
src/eitd/eitd.h
View File

@@ -44,6 +44,21 @@ typedef SIevent * SIeventPtr;
typedef SIservice * SIservicePtr; typedef SIservice * SIservicePtr;
#endif #endif
/* period to restart EIT reading */
#define TIME_EIT_SCHEDULED_PAUSE 60 * 60
/* force EIT thread to change filter after, seconds */
#define TIME_EIT_SKIPPING 120 // 90 <- Canal diditaal 19.2e -> ~100 seconds for 0x5x
/* a little more time for freesat epg */
#define TIME_FSEIT_SKIPPING 240
/* Timeout in ms for reading from dmx in EIT threads. Dont make this too long
since we are holding the start_stop lock during this read! */
#define EIT_READ_TIMEOUT 100
/* Number of DMX read timeouts, after which we check if there is an EIT at all
for EIT and PPT threads... */
#define CHECK_RESTART_DMX_AFTER_TIMEOUTS (2000 / EIT_READ_TIMEOUT) // 2 seconds
struct OrderServiceUniqueKeyFirstStartTimeEventUniqueKey struct OrderServiceUniqueKeyFirstStartTimeEventUniqueKey
{ {
bool operator()(const SIeventPtr &p1, const SIeventPtr &p2) bool operator()(const SIeventPtr &p1, const SIeventPtr &p2)
@@ -83,40 +98,64 @@ typedef std::map<t_channel_id, SIservicePtr, std::less<t_channel_id> > MySIservi
#define MAX_SECTION_LENGTH (0x0fff + 3) #define MAX_SECTION_LENGTH (0x0fff + 3)
/* abstract section reading class */
class CSectionThread : public OpenThreads::Thread, public DMX class CSectionThread : public OpenThreads::Thread, public DMX
{ {
protected: protected:
uint8_t *static_buf; uint8_t *static_buf;
int timeoutsDMX; int timeoutsDMX;
unsigned timeoutInMSeconds;
bool running; bool running;
//std::string name;
int event_count; // debug int event_count; // debug
/* section read timeout, msec */
unsigned timeoutInMSeconds;
/* read timeouts count to switch to next filter */
int skipTimeouts;
/* time to switch to next filter */
int skipTime;
bool sendToSleepNow;
virtual void run() {}; /* should thread wait for time set before process loop */
int Sleep(unsigned int timeout) bool wait_for_time;
{ /* time in seconds to sleep when requested, wait forever if 0 */
struct timespec abs_wait; int sleep_time;
struct timeval now;
gettimeofday(&now, NULL);
TIMEVAL_TO_TIMESPEC(&now, &abs_wait);
abs_wait.tv_sec += timeout;
dprintf("%s: going to sleep for %d seconds...\n", name.c_str(), timeout);
pthread_mutex_lock(&start_stop_mutex);
int rs = pthread_cond_timedwait( &change_cond, &start_stop_mutex, &abs_wait );
pthread_mutex_unlock( &start_stop_mutex );
return rs;
}
bool addEvents();
/* thread hooks */
/* add filters when thread started */
virtual void addFilters() {};
/* check if thread should go sleep */
virtual bool shouldSleep() { return false; };
/* check if thread should continue to sleep after wakeup or timeout */
virtual bool checkSleep() { return false; };
/* called before sleep loop */
virtual void beforeSleep() {};
/* called after sleep loop */
virtual void afterSleep() {};
/* called inside sleep loop, after lock, before wait */
virtual void beforeWait() {};
/* called inside sleep loop, before lock, after wait */
virtual void afterWait() {};
/* process section after getSection */
virtual void processSection(int rc) { if(rc < 0) return; };
/* cleanup before exit */
virtual void cleanup() {};
/* sleep for sleep_time, forever if sleep_time = 0 */
int Sleep();
/* main thread function */
void run();
public: public:
CSectionThread() CSectionThread()
{ {
static_buf = new uint8_t[MAX_SECTION_LENGTH]; static_buf = new uint8_t[MAX_SECTION_LENGTH];
timeoutsDMX = 0; timeoutsDMX = 0;
running = false; running = false;
event_count = 0; event_count = 0;
sendToSleepNow = 0;
} }
~CSectionThread() ~CSectionThread()
@@ -151,23 +190,69 @@ printf("%s::Stop: to close\n", name.c_str());
} }
}; };
class CEitThread : public CSectionThread /* abstract eit events reading class */
class CEventsThread : public CSectionThread
{ {
private: protected:
void run(); /* default hooks */
bool shouldSleep();
bool checkSleep();
void processSection(int rc);
/* private */
bool addEvents();
public:
CEventsThread(std::string tname, unsigned short pid = 0x12)
{
name = tname;
pID = pid;
/* defaults for events threads, redefined if needed in derived */
timeoutInMSeconds = EIT_READ_TIMEOUT;
skipTimeouts = CHECK_RESTART_DMX_AFTER_TIMEOUTS;
skipTime = TIME_EIT_SKIPPING;
sleep_time = TIME_EIT_SCHEDULED_PAUSE;
wait_for_time = true;
};
}; };
class CCNThread : public CSectionThread class CEitThread : public CEventsThread
{ {
private: private:
/* overloaded hooks */
void addFilters();
void beforeSleep();
public:
CEitThread();
};
class CFreeSatThread : public CEventsThread
{
private:
/* overloaded hooks */
void addFilters();
public:
CFreeSatThread();
};
class CCNThread : public CEventsThread
{
private:
/* private */
bool updating; bool updating;
cDemux * eitDmx; cDemux * eitDmx;
void sendCNEvent(); void sendCNEvent();
bool startUpdateFilter();
bool stopUpdateFilter(); /* overloaded hooks */
void run(); void beforeWait();
void afterWait();
void addFilters();
void beforeSleep();
void processSection(int rc);
void cleanup();
public: public:
CCNThread();
bool checkUpdate(); bool checkUpdate();
}; };

579
src/eitd/sectionsd.cpp
View File

@@ -67,13 +67,6 @@
#define DEBUG_EIT_THREAD #define DEBUG_EIT_THREAD
#define DEBUG_CN_THREAD #define DEBUG_CN_THREAD
/* period to restart EIT reading */
#define TIME_EIT_SCHEDULED_PAUSE 60 * 60
/* force EIT thread to change filter after, seconds */
#define TIME_EIT_SKIPPING 120 // 90 <- Canal diditaal 19.2e -> ~100 seconds for 0x5x
// a little more time for freesat epg
#define TIME_FSEIT_SKIPPING 240
static bool sectionsd_ready = false; static bool sectionsd_ready = false;
/*static*/ bool reader_ready = true; /*static*/ bool reader_ready = true;
static unsigned int max_events; static unsigned int max_events;
@@ -88,13 +81,6 @@ static unsigned int max_events;
#define READ_TIMEOUT_IN_SECONDS 2 #define READ_TIMEOUT_IN_SECONDS 2
#define WRITE_TIMEOUT_IN_SECONDS 2 #define WRITE_TIMEOUT_IN_SECONDS 2
// Timeout in ms for reading from dmx in EIT threads. Dont make this too long
// since we are holding the start_stop lock during this read!
#define EIT_READ_TIMEOUT 100
// Number of DMX read timeouts, after which we check if there is an EIT at all
// for EIT and PPT threads...
#define CHECK_RESTART_DMX_AFTER_TIMEOUTS (2000 / EIT_READ_TIMEOUT) // 2 seconds
// Time in seconds we are waiting for an EIT version number // Time in seconds we are waiting for an EIT version number
//#define TIME_EIT_VERSION_WAIT 3 // old //#define TIME_EIT_VERSION_WAIT 3 // old
#define TIME_EIT_VERSION_WAIT 10 #define TIME_EIT_VERSION_WAIT 10
@@ -152,14 +138,11 @@ static pthread_rwlock_t messagingLock = PTHREAD_RWLOCK_INITIALIZER;
static pthread_cond_t timeThreadSleepCond = PTHREAD_COND_INITIALIZER; static pthread_cond_t timeThreadSleepCond = PTHREAD_COND_INITIALIZER;
static pthread_mutex_t timeThreadSleepMutex = PTHREAD_MUTEX_INITIALIZER; static pthread_mutex_t timeThreadSleepMutex = PTHREAD_MUTEX_INITIALIZER;
//static DMX dmxEIT(0x12, 3000 /*320*/);
static CEitThread threadEIT; static CEitThread threadEIT;
//static DMX dmxCN(0x12, 512, false, 1);
static CCNThread threadCN; static CCNThread threadCN;
#ifdef ENABLE_FREESATEPG #ifdef ENABLE_FREESATEPG
static DMX dmxFSEIT(3842, 320); static CFreeSatThread threadFSEIT;
#endif #endif
#ifdef ENABLE_SDT #ifdef ENABLE_SDT
@@ -1454,152 +1437,121 @@ xprintf("timeThread: going to sleep for %d sec\n\n", seconds);
pthread_exit(NULL); pthread_exit(NULL);
} }
/********************************************************************************/
#ifdef ENABLE_FREESATEPG /* abstract CSectionThread functions */
//--------------------------------------------------------------------- /********************************************************************************/
// Freesat EIT-thread /* sleep for sleep_time seconds, forever if sleep_time = 0 */
// reads Freesat EPG-data int CSectionThread::Sleep()
//---------------------------------------------------------------------
static void *fseitThread(void *)
{ {
int rs;
struct timespec abs_wait;
struct timeval now;
/* we are holding the start_stop lock during this timeout, so don't if(sleep_time) {
make it too long... */ gettimeofday(&now, NULL);
unsigned timeoutInMSeconds = EIT_READ_TIMEOUT; TIMEVAL_TO_TIMESPEC(&now, &abs_wait);
bool sendToSleepNow = false; abs_wait.tv_sec += sleep_time;
pthread_setcanceltype (PTHREAD_CANCEL_ASYNCHRONOUS, 0); }
xprintf("%s: going to sleep for %d seconds...\n", name.c_str(), sleep_time);
pthread_mutex_lock(&start_stop_mutex);
dmxFSEIT.addfilter(0x60, 0xfe); //other TS, scheduled, freesat epg is only broadcast using table_ids 0x60 (scheduled) and 0x61 (scheduled later) beforeWait();
if(sleep_time)
rs = pthread_cond_timedwait( &change_cond, &start_stop_mutex, &abs_wait );
else
rs = pthread_cond_wait(&change_cond, &start_stop_mutex);
afterWait();
pthread_mutex_unlock( &start_stop_mutex );
return rs;
}
/* common thread main function */
void CSectionThread::run()
{
xprintf("%s::run:: starting, pid %d (%lu)\n", name.c_str(), getpid(), pthread_self());
if (sections_debug) if (sections_debug)
dump_sched_info("freesatEitThread"); dump_sched_info(name.c_str());
dprintf("[%sThread] pid %d (%lu) start\n", "fseit", getpid(), pthread_self()); addFilters();
int timeoutsDMX = 0;
uint8_t *static_buf = new uint8_t[MAX_SECTION_LENGTH];
int rc;
dmxFSEIT.start(); // -> unlock if (wait_for_time) {
if (!scanning) waitForTimeset();
dmxFSEIT.request_pause(); xprintf("%s::run:: time set.\n", name.c_str());
}
waitForTimeset(); DMX::start();
dmxFSEIT.lastChanged = time_monotonic();
while(!sectionsd_stop) { while (running) {
while (!scanning) { if (shouldSleep()) {
if(sectionsd_stop) #ifdef DEBUG_EIT_THREAD
xprintf("%s: going to sleep %d seconds, running %d scanning %d blacklisted %d events %d\n",
name.c_str(), sleep_time, running, scanning, channel_is_blacklisted, event_count);
#endif
event_count = 0;
beforeSleep();
int rs = 0;
do {
real_pause();
rs = Sleep();
#ifdef DEBUG_EIT_THREAD
xprintf("%s: wakeup, running %d scanning %d blacklisted %d reason %d\n\n",
name.c_str(), running, scanning, channel_is_blacklisted, rs);
#endif
} while (checkSleep());
if(!running)
break; break;
sleep(1);
}
if(sectionsd_stop)
break;
time_t zeit = time_monotonic();
rc = dmxFSEIT.getSection(static_buf, timeoutInMSeconds, timeoutsDMX); afterSleep();
if (rc < 0)
continue;
if (timeoutsDMX < 0)
{
if ( dmxFSEIT.filter_index + 1 < (signed) dmxFSEIT.filters.size() )
{
if (timeoutsDMX == -1)
dprintf("[freesatEitThread] skipping to next filter(%d) (> DMX_HAS_ALL_SECTIONS_SKIPPING)\n", dmxFSEIT.filter_index+1 );
if (timeoutsDMX == -2)
dprintf("[freesatEitThread] skipping to next filter(%d) (> DMX_HAS_ALL_CURRENT_SECTIONS_SKIPPING)\n", dmxFSEIT.filter_index+1 );
timeoutsDMX = 0;
dmxFSEIT.change(dmxFSEIT.filter_index + 1);
}
else {
sendToSleepNow = true;
timeoutsDMX = 0;
}
}
if (timeoutsDMX >= CHECK_RESTART_DMX_AFTER_TIMEOUTS && scanning)
{
if ( dmxFSEIT.filter_index + 1 < (signed) dmxFSEIT.filters.size() )
{
dprintf("[freesatEitThread] skipping to next filter(%d) (> DMX_TIMEOUT_SKIPPING)\n", dmxFSEIT.filter_index+1 );
dmxFSEIT.change(dmxFSEIT.filter_index + 1);
}
else
sendToSleepNow = true;
timeoutsDMX = 0;
}
if (sendToSleepNow)
{
sendToSleepNow = false;
if(sectionsd_stop)
break;
dmxFSEIT.real_pause();
pthread_mutex_lock( &dmxFSEIT.start_stop_mutex );
writeLockMessaging();
messaging_zap_detected = false;
unlockMessaging();
struct timespec abs_wait;
struct timeval now;
gettimeofday(&now, NULL);
TIMEVAL_TO_TIMESPEC(&now, &abs_wait);
abs_wait.tv_sec += TIME_EIT_SCHEDULED_PAUSE;
dprintf("dmxFSEIT: going to sleep for %d seconds...\n", TIME_EIT_SCHEDULED_PAUSE);
int rs = pthread_cond_timedwait( &dmxFSEIT.change_cond, &dmxFSEIT.start_stop_mutex, &abs_wait );
pthread_mutex_unlock( &dmxFSEIT.start_stop_mutex );
if (rs == ETIMEDOUT) if (rs == ETIMEDOUT)
{ change(0); // -> restart, FIXME
dprintf("dmxFSEIT: waking up again - timed out\n");
// must call dmxFSEIT.change after! unpause otherwise dev is not open, sendToSleepNow = false;
// dmxFSEIT.lastChanged will not be set, and filter is advanced the next iteration
// maybe .change should imply .real_unpause()? -- seife
dprintf("New Filterindex: %d (ges. %d)\n", 2, (signed) dmxFSEIT.filters.size() );
dmxFSEIT.change(1); // -> restart
}
else if (rs == 0)
{
dprintf("dmxFSEIT: waking up again - requested from .change()\n");
}
else
{
dprintf("dmxFSEIT: waking up again - unknown reason %d\n",rs);
}
// update zeit after sleep
zeit = time_monotonic();
} }
else if (zeit > dmxFSEIT.lastChanged + TIME_FSEIT_SKIPPING )
{ int rc = getSection(static_buf, timeoutInMSeconds, timeoutsDMX);
processSection(rc);
time_t zeit = time_monotonic();
bool need_change = false;
if(timeoutsDMX < 0 || timeoutsDMX >= skipTimeouts) {
#ifdef DEBUG_EIT_THREAD
xprintf("%s: skipping to next filter %d from %d (timeouts %d)\n",
name.c_str(), filter_index+1, filters.size(), timeoutsDMX);
#endif
timeoutsDMX = 0;
need_change = true;
}
if (zeit > lastChanged + skipTime) {
#ifdef DEBUG_EIT_THREAD
xprintf("%s: skipping to next filter %d from %d (seconds %d)\n",
name.c_str(), filter_index+1, filters.size(), (int) (zeit - lastChanged));
#endif
need_change = true;
}
if(running && need_change && scanning) {
readLockMessaging(); readLockMessaging();
if (!next_filter())
if ( dmxFSEIT.filter_index + 1 < (signed) dmxFSEIT.filters.size() )
{
dprintf("[freesatEitThread] skipping to next filter(%d) (> TIME_FSEIT_SKIPPING)\n", dmxFSEIT.filter_index+1 );
dmxFSEIT.change(dmxFSEIT.filter_index + 1);
}
else
sendToSleepNow = true; sendToSleepNow = true;
unlockMessaging(); unlockMessaging();
} }
} // while running
addEvents();
//dprintf("[eitThread] added %d events (end)\n", eit.events().size());
} // for
delete[] static_buf; delete[] static_buf;
dputs("[freesatEitThread] end"); cleanup();
printf("[sectionsd] %s ended\n", name.c_str());
pthread_exit(NULL); pthread_exit(NULL);
} }
#endif
bool CSectionThread::addEvents() /********************************************************************************/
/* abstract CEventsThread functions */
/********************************************************************************/
bool CEventsThread::addEvents()
{ {
SIsectionEIT eit(static_buf); SIsectionEIT eit(static_buf);
@@ -1638,21 +1590,37 @@ bool CSectionThread::addEvents()
return true; return true;
} }
//--------------------------------------------------------------------- /* default check if thread should go to sleep */
// EIT-thread bool CEventsThread::shouldSleep()
// reads EPG-datas
//---------------------------------------------------------------------
void CEitThread::run()
{ {
name = "eitThread"; return (sendToSleepNow || !scanning || channel_is_blacklisted);
xprintf("%s::run:: starting, pid %d (%lu)\n", name.c_str(), getpid(), pthread_self()); }
pID = 0x12; /* default check if thread should continue to sleep */
timeoutInMSeconds = EIT_READ_TIMEOUT; bool CEventsThread::checkSleep()
{
return (running && (!scanning || channel_is_blacklisted));
}
bool sendToSleepNow = false; /* default section process */
void CEventsThread::processSection(int rc)
{
if(rc <= 0)
return;
addEvents();
}
/********************************************************************************/
/* EIT thread to read other TS CN + all scheduled events */
/********************************************************************************/
CEitThread::CEitThread()
: CEventsThread("eitThread")
{
}
/* EIT thread hooks */
void CEitThread::addFilters()
{
/* These filters are a bit tricky (index numbers): /* These filters are a bit tricky (index numbers):
- 0 Dummy filter, to make this thread sleep for some seconds - 0 Dummy filter, to make this thread sleep for some seconds
- 1 then get other TS's current/next (this TS's cur/next are - 1 then get other TS's current/next (this TS's cur/next are
@@ -1670,89 +1638,51 @@ void CEitThread::run()
#else #else
addfilter(0x60, 0xf0); //3 other TS, scheduled addfilter(0x60, 0xf0); //3 other TS, scheduled
#endif #endif
if (sections_debug)
dump_sched_info("eitThread");
waitForTimeset();
xprintf("%s::run:: time set.\n", name.c_str());
DMX::start(); // -> unlock
while (running) {
if(sendToSleepNow || !scanning || channel_is_blacklisted) {
#ifdef DEBUG_EIT_THREAD
xprintf("%s: going to sleep %d seconds, running %d scanning %d blacklisted %d events %d\n",
name.c_str(), TIME_EIT_SCHEDULED_PAUSE, running, scanning, channel_is_blacklisted, event_count);
#endif
event_count = 0;
writeLockMessaging();
messaging_zap_detected = false;
unlockMessaging();
int rs = 0;
do {
real_pause();
rs = Sleep(TIME_EIT_SCHEDULED_PAUSE);
#ifdef DEBUG_EIT_THREAD
xprintf("%s: wakeup, running %d scanning %d blacklisted %d reason %d\n\n", name.c_str(), running, scanning, channel_is_blacklisted, rs);
#endif
} while(running && (!scanning || channel_is_blacklisted));
if(!running)
break;
if (rs == ETIMEDOUT)
change(1); // -> restart
sendToSleepNow = false;
}
int rc = getSection(static_buf, timeoutInMSeconds, timeoutsDMX);
if (rc > 0)
addEvents();
time_t zeit = time_monotonic();
bool need_change = false;
if(timeoutsDMX < 0 || timeoutsDMX >= CHECK_RESTART_DMX_AFTER_TIMEOUTS) {
#ifdef DEBUG_EIT_THREAD
xprintf("%s: skipping to next filter %d from %d (timeouts %d)\n", name.c_str(), filter_index+1, filters.size(), timeoutsDMX);
#endif
timeoutsDMX = 0;
need_change = true;
}
if (zeit > lastChanged + TIME_EIT_SKIPPING) {
#ifdef DEBUG_EIT_THREAD
xprintf("%s: skipping to next filter %d from %d (TIME_EIT_SKIPPING)\n", name.c_str(), filter_index+1, filters.size());
#endif
need_change = true;
}
/* FIXME without dummy filter, there is race between service change and need_change */
if(running && need_change && scanning && !channel_is_blacklisted) {
readLockMessaging();
if (!next_filter())
sendToSleepNow = true;
unlockMessaging();
}
} // while running
delete[] static_buf;
printf("[sectionsd] %s ended\n", name.c_str());
pthread_exit(NULL);
} }
//--------------------------------------------------------------------- void CEitThread::beforeSleep()
// CN-thread: eit thread, but only current/next {
//--------------------------------------------------------------------- xprintf("%s: CScheduledThread::beforeSleep()\n", name.c_str());
bool CCNThread::startUpdateFilter() writeLockMessaging();
messaging_zap_detected = false;
unlockMessaging();
}
/********************************************************************************/
/* CN thread to read current TS CN events */
/********************************************************************************/
CCNThread::CCNThread()
: CEventsThread("cnThread")
{
sleep_time = 0;
cache = false;
skipTimeouts = 5000 / timeoutInMSeconds; // 5 seconds
skipTime = TIME_EIT_VERSION_WAIT;
updating = false;
eitDmx = new cDemux(0);
eitDmx->Open(DMX_PSI_CHANNEL);
}
/* CN thread hooks */
void CCNThread::cleanup()
{
delete eitDmx;
}
void CCNThread::addFilters()
{
addfilter(0x4e, 0xff); //0 current TS, current/next
}
void CCNThread::beforeWait()
{ {
xprintf("%s: set eit update filter, service = 0x%016llx, current version 0x%x got events %d (%s)\n", xprintf("%s: set eit update filter, service = 0x%016llx, current version 0x%x got events %d (%s)\n",
name.c_str(), messaging_current_servicekey, eit_version, messaging_have_CN, name.c_str(), messaging_current_servicekey, eit_version, messaging_have_CN,
updating ? "active" : "not active"); updating ? "active" : "not active");
if (updating || eit_version == 0xff) if (updating || eit_version == 0xff)
return false; return;
updating = true; updating = true;
@@ -1775,21 +1705,51 @@ bool CCNThread::startUpdateFilter()
mask[3] = (0x1F << 1) | 0x01; mask[3] = (0x1F << 1) | 0x01;
mode[3] = 0x1F << 1; mode[3] = 0x1F << 1;
eitDmx->sectionFilter(0x12, filter, mask, 4, 0 /*timeout*/, mode); eitDmx->sectionFilter(0x12, filter, mask, 4, 0 /*timeout*/, mode);
return true;
} }
bool CCNThread::stopUpdateFilter() void CCNThread::afterWait()
{ {
xprintf("%s: stop eit update filter (%s)\n", name.c_str(), updating ? "active" : "not active"); xprintf("%s: stop eit update filter (%s)\n", name.c_str(), updating ? "active" : "not active");
if(updating) { if(updating) {
updating = false; updating = false;
eitDmx->Stop(); eitDmx->Stop();
return true;
} }
return false;
} }
void CCNThread::beforeSleep()
{
if (sendToSleepNow && messaging_have_CN == 0x00) {
/* send a "no epg" event anyway before going to sleep */
sendCNEvent();
}
}
void CCNThread::processSection(int rc)
{
if(rc <= 0)
return;
addEvents();
readLockMessaging();
if (messaging_got_CN != messaging_have_CN) {
unlockMessaging();
writeLockMessaging();
messaging_have_CN = messaging_got_CN;
unlockMessaging();
#ifdef DEBUG_CN_THREAD
xprintf("%s: have CN: timeoutsDMX %d messaging_have_CN %x messaging_got_CN %x\n\n",
name.c_str(), timeoutsDMX, messaging_have_CN, messaging_got_CN);
#endif
dprintf("[cnThread] got current_next (0x%x) - sending event!\n", messaging_have_CN);
sendCNEvent();
lastChanged = time_monotonic();
}
else
unlockMessaging();
}
/* CN private functions */
bool CCNThread::checkUpdate() bool CCNThread::checkUpdate()
{ {
if(!updating) if(!updating)
@@ -1801,7 +1761,8 @@ bool CCNThread::checkUpdate()
if (ret > 0) { if (ret > 0) {
LongSection section(buf); LongSection section(buf);
printdate_ms(stdout); printdate_ms(stdout);
xprintf("%s: eit update filter: ### new version 0x%02x ###, Activate thread\n", name.c_str(), section.getVersionNumber()); xprintf("%s: eit update filter: ### new version 0x%02x ###, Activate thread\n",
name.c_str(), section.getVersionNumber());
writeLockMessaging(); writeLockMessaging();
messaging_have_CN = 0x00; messaging_have_CN = 0x00;
@@ -1820,110 +1781,23 @@ void CCNThread::sendCNEvent()
sizeof(messaging_current_servicekey)); sizeof(messaging_current_servicekey));
} }
void CCNThread::run() #ifdef ENABLE_FREESATEPG
/********************************************************************************/
/* Freesat EIT thread */
/********************************************************************************/
CFreeSatThread()
: CEventsThread("freeSatThread", 3842)
{ {
name = "cnThread"; skipTime = TIME_FSEIT_SKIPPING;
xprintf("%s::run:: starting, pid %d (%lu)\n", name.c_str(), getpid(), pthread_self()); };
pID = 0x12;
dmx_num = 0;
cache = false;
timeoutInMSeconds = EIT_READ_TIMEOUT; /* Freesat hooks */
bool sendToSleepNow = false; void CFreeSatThread::addFilters()
{
updating = false; //other TS, scheduled, freesat epg is only broadcast using table_ids 0x60 (scheduled) and 0x61 (scheduled later)
eitDmx = new cDemux(0); addfilter(0x60, 0xfe);
eitDmx->Open(DMX_PSI_CHANNEL);
// -- set EIT filter 0x4e
addfilter(0x4e, 0xff); //0 current TS, current/next
waitForTimeset();
#ifdef DEBUG_CN_THREAD
xprintf("CCNThread::run:: time set..\n");
#endif
DMX::start(); // -> unlock
while(running)
{
if(sendToSleepNow || !scanning || channel_is_blacklisted) {
#ifdef DEBUG_CN_THREAD
xprintf("%s: going to sleep, running %d scanning %d blacklisted %d events %d\n", name.c_str(), running, scanning, channel_is_blacklisted, event_count);
#endif
int rs = 0;
do {
real_pause();
pthread_mutex_lock( &start_stop_mutex );
if (!channel_is_blacklisted)
startUpdateFilter();
rs = pthread_cond_wait(&change_cond, &start_stop_mutex);
stopUpdateFilter();
pthread_mutex_unlock(&start_stop_mutex);
#ifdef DEBUG_CN_THREAD
xprintf("%s: wakeup, running %d scanning %d blacklisted %d reason %d\n\n", name.c_str(), running, scanning, channel_is_blacklisted, rs);
#endif
} while(running && (!scanning || channel_is_blacklisted));
if(!running)
break;
sendToSleepNow = false;
#if HAVE_IPBOX_HARDWARE
if (rs == 0)
change(0);
#endif
}
int rc = getSection(static_buf, timeoutInMSeconds, timeoutsDMX);
if (rc > 0)
addEvents();
time_t zeit = time_monotonic();
readLockMessaging();
if (messaging_got_CN != messaging_have_CN)
{
unlockMessaging();
writeLockMessaging();
messaging_have_CN = messaging_got_CN;
unlockMessaging();
#ifdef DEBUG_CN_THREAD
xprintf("%s: have CN: timeoutsDMX %d messaging_have_CN %x messaging_got_CN %x\n\n",
name.c_str(), timeoutsDMX, messaging_have_CN, messaging_got_CN);
#endif
dprintf("[cnThread] got current_next (0x%x) - sending event!\n", messaging_have_CN);
sendCNEvent();
lastChanged = zeit; /* this is ugly - needs somehting better */
}
else
unlockMessaging();
if(timeoutsDMX < 0) {
#ifdef DEBUG_CN_THREAD
xprintf("%s: timeoutsDMX %d messaging_got_CN %x messaging_have_CN %x sid %016llx\n\n",
name.c_str(), timeoutsDMX, messaging_got_CN, messaging_have_CN, messaging_current_servicekey);
#endif
timeoutsDMX = 0;
sendToSleepNow = true;
}
if (zeit > lastChanged + TIME_EIT_VERSION_WAIT) {
#ifdef DEBUG_CN_THREAD
xprintf("%s: zeit > lastChanged + TIME_EIT_VERSION_WAIT\n", name.c_str());
#endif
sendToSleepNow = true;
}
if (sendToSleepNow && messaging_have_CN == 0x00) {
/* send a "no epg" event anyway before going to sleep */
sendCNEvent();
}
} // for
delete[] static_buf;
stopUpdateFilter();
delete eitDmx;
printf("[sectionsd] %s ended\n", name.c_str());
pthread_exit(NULL);
} }
#endif
#ifdef ENABLE_SDT #ifdef ENABLE_SDT
static bool addService(const SIservice &s, const int is_actual) static bool addService(const SIservice &s, const int is_actual)
@@ -1954,16 +1828,6 @@ static bool addService(const SIservice &s, const int is_actual)
SIservicePtr sptr(sp); SIservicePtr sptr(sp);
#if 0
#define MAX_SIZE_SERVICENAME 50
char servicename[MAX_SIZE_SERVICENAME];
if (sptr->serviceName.empty()) {
sprintf(servicename, "%04x", sptr->service_id);
servicename[sizeof(servicename) - 1] = 0;
sptr->serviceName = servicename;
}
#endif
sptr->is_actual = is_actual; sptr->is_actual = is_actual;
writeLockServices(); writeLockServices();
@@ -2176,9 +2040,6 @@ extern cDemux * dmxUTC;
void sectionsd_main_thread(void * /*data*/) void sectionsd_main_thread(void * /*data*/)
{ {
pthread_t threadTOT, threadHouseKeeping; pthread_t threadTOT, threadHouseKeeping;
#ifdef ENABLE_FREESATEPG
pthread_t threadFSEIT;
#endif
int rc; int rc;
printf("$Id: sectionsd.cpp,v 1.305 2009/07/30 12:41:39 seife Exp $\n"); printf("$Id: sectionsd.cpp,v 1.305 2009/07/30 12:41:39 seife Exp $\n");
@@ -2249,13 +2110,7 @@ printf("SIevent size: %d\n", sizeof(SIevent));
threadCN.Start(); threadCN.Start();
#ifdef ENABLE_FREESATEPG #ifdef ENABLE_FREESATEPG
// EIT-Thread3 starten threadFSEIT.Start();
rc = pthread_create(&threadFSEIT, 0, fseitThread, 0);
if (rc) {
fprintf(stderr, "[sectionsd] failed to create fseit-thread (rc=%d)\n", rc);
return;
}
#endif #endif
#ifdef ENABLE_SDT #ifdef ENABLE_SDT
threadSDT.Start(); threadSDT.Start();
@@ -2301,16 +2156,7 @@ printf("SIevent size: %d\n", sizeof(SIevent));
pthread_mutex_unlock(&timeThreadSleepMutex); pthread_mutex_unlock(&timeThreadSleepMutex);
printf("pausing...\n"); printf("pausing...\n");
#if 0
dmxEIT.request_pause();
dmxCN.request_pause();
#endif
#ifdef ENABLE_FREESATEPG
dmxFSEIT.request_pause();
#endif
#ifdef ENABLE_SDT
//dmxSDT.request_pause();
#endif
pthread_cancel(threadHouseKeeping); pthread_cancel(threadHouseKeeping);
if(dmxUTC) dmxUTC->Stop(); if(dmxUTC) dmxUTC->Stop();
@@ -2331,7 +2177,8 @@ printf("SIevent size: %d\n", sizeof(SIevent));
threadSDT.Stop(); threadSDT.Stop();
#endif #endif
#ifdef ENABLE_FREESATEPG #ifdef ENABLE_FREESATEPG
dmxFSEIT.close(); printf("join FSEIT\n");
threadFSEIT.Stop();
#endif #endif
printf("[sectionsd] ended\n"); printf("[sectionsd] ended\n");
} }