开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 多媒体 > DVD > 查看源码
sync.cc
资源名称：dvd-munitions.tar.gz　[点击查看]
上传用户：aoeyumen
上传日期：2007-01-06
资源大小：3329k
文件大小：8k
源码类别：

DVD

开发平台：

Unix_Linux

sync.cc：源码内容
  1. /*
  2.    File: sync.cc
  4.    By: Alex Theo de Jong
  5.    Created: February 1996
  7.    Description:
  8.    Synchronization object for multiple streams
  9. */
  11. #ifdef __GNUG__
  12. #pragma implementation
  13. #endif
  15. #include "athread.hh"
  17. #include <stdio.h>
  18. #include <iostream.h>
  19. #include <sys/time.h>
  20. #ifdef LINUX
  21. #include <sys/sem.h>
  22. #endif
  24. #include "error.hh"
  25. #include "debug.hh"
  26. #include "util.hh"
  27. #include "sync.hh"
  29. /*
  30.  *
  31.  * StampQueue
  32.  *
  33.  */
  35. StampQueue::StampQueue(int max) :
  36.   in(max), out(0),
  37.   index_in(0), index_out(0),
  38.   resync(0),
  39.   newtime(0.0),maximum(max), total(0)
  40. {
  41.   if (!(stamps=new double[maximum]))
  42.     error("could not allocate memory");
  43.   if (!(bytes=new int[maximum]))
  44.     error("could not allocate memory");  
  45. }
  47. StampQueue::~StampQueue(){
  48.   delete[] stamps;
  49.   delete[] bytes;
  50. }
  52. int StampQueue::get(double& stamp){
  53. #ifdef TRACE
  54.   if (total==0) warning("stamp queue underflow (stamp)");
  55. #endif
  56.   wait_out();
  57.   lock();
  58.   stamp=stamps[index_out];
  59.   TRACER("StampQueue::get " << dtoa(stamp));
  60.   if (++index_out>=maximum) index_out=0; 
  61.   total--;
  62.   unlock();
  63.   post_in();
  64.   return 1;
  65. }
  67. int StampQueue::get(double& stamp, int& b){
  68. #ifdef TRACE
  69.   if (total==0) warning("stamp queue underflow (stamp and bytes)");
  70. #endif
  71.   wait_out();
  72.   lock();
  73.   stamp=stamps[index_out];
  74.   b+=bytes[index_out];
  75.   if (++index_out>=maximum) index_out=0; 
  76.   TRACER("StampQueue::get" << dtoa(stamp) << " " << itoa(bytes[index_out]));
  77.   total--;
  78.   unlock();
  79.   post_in();
  80.   return 1;
  81. }
  83. int StampQueue::put(const double stamp, const int b){
  84. #ifdef TRACE
  85.   TRACER("StampQueue::put " << dtoa(stamp) << " " << itoa(b));
  86.   if (total==maximum){
  87.     warning("stamp queue overflow");
  88.   }
  89. #endif
  90.   wait_in();
  91.   lock();
  92.   stamps[index_in]=stamp;
  93.   bytes[index_in]=b;
  94.   if (++index_in>=maximum) index_in=0; 
  95.   total++;
  96.   unlock();
  97.   post_out();
  98.   return 1;
  99. }
  101. /*
  102.  *
  103.  * SyncTimer
  104.  *
  105.  */
  107. SyncTimer::SyncTimer(int max) : StampQueue(max), time(-1.0) {
  108. }
  110. int SyncTimer::update(){
  111.   get(newtime);
  112.   lock();
  113.   // == continuous re-sync if time-stamps are not used
  114.   //  < only re-sync if next time stamp drops (ie. jumping back into stream)
  115.   resync=(newtime<=time) ? 1 : 0;    
  116. #ifdef TRACE
  117.   if (resync==1) TRACER("re_sync time");
  118. #endif
  119.   time=newtime;
  120. TRACER("Timeupdate " << dtoa(time));
  121.   time_cond.broadcast();
  122.   unlock();
  123.   return resync;
  124. }
  126. int SyncTimer::done(){
  127.   TRACER("int SyncTimer::done()");
  128.   lock();
  129.   time=0x0fffffff; // maximum time
  130.   time_cond.broadcast();
  131.   unlock();
  132.   return 1;
  133. }
  135. /*
  136.  *
  137.  * SyncData
  138.  *
  139.  */
  141. SyncData::SyncData(int max, SyncTimer* t) : 
  142.   StampQueue(max),
  143.   terminated(0),
  144.   time(0.0),
  145.   bytes(0)
  146. {
  147.   timer=t;
  148. }
  151. int SyncData::wait(){
  152. #ifdef UPTIGHT
  153.   bytelock();  // bytes
  154. #endif
  155.   DEBUGGER("W" << itoa(bytes));
  157.   if (bytes<=0){
  158.     get(newtime, bytes);
  159. //message("wait for time " << dtoa(newtime) << " " << dtoa(time));
  160.     if (newtime<=time){
  161.       resync=0; // found new sync time; stop re-sync
  162.       TRACER("reset data resync");
  163.     }
  164. //msg("9");
  165.     time_lock.lock();
  166. //msg("8");
  167.     time=newtime;
  168.     time_cond.signal();
  169. //msg("7");
  170.     time_lock.unlock();
  171. //msg("6");
  173.     timer->lock();
  174. //message("wait for time " << dtoa(timer->time) << " it is " << dtoa(time) << " resync " << itoa(resync));
  175.     while (resync==0 && (time>timer->time) && (terminated==0)){
  176. //message("inwhile");
  178.       timer->time_cond.wait(&timer->time_lock);
  179. //message("wait for time " << dtoa(timer->time) << "it is " << dtoa(time));
  180.     }
  181.     TRACER("Twait " << dtoa(time) << " =< " << dtoa(timer->time));
  182.     timer->unlock();
  183. #ifdef UPTIGHT
  184.     byteunlock();
  185. #endif
  187.     return 1;
  188.   }
  189. #ifdef UPTIGHT
  190.   byteunlock();
  191. #endif
  192.   return 0;
  193. }
  195. int SyncData::skip(){
  196.   TRACER("int SyncData::skip()");
  198. #ifdef UPTIGHT
  199.   bytelock();
  200. #endif
  201.   if (bytes<=0){
  202.     get(newtime, bytes);
  203.     if (newtime<time)
  204.       resync=0;     // found new sync time; stop re-sync
  205.     time_lock.lock();
  206.     time=newtime;
  207.     time_cond.signal();
  208.     time_lock.unlock();
  209. #ifdef UPTIGHT
  210.     byteunlock();
  211. #endif
  212.     return 1;
  213.   }
  214. #ifdef UPTIGHT
  215.   byteunlock();
  216. #endif
  217.   return 0;
  218. }
  220. int SyncData::update(){
  221.   time_lock.lock();
  222.   if (timer->resync==1){
  223.     resync=1;  // re_sync data
  224.     TRACER("resync data");
  225.   }
  226.   else {
  227.     while ((time<=timer->time) && (terminated==0)){
  228.       time_cond.wait(&time_lock);
  229.     }
  230.   }
  231.   DEBUGGER("T" << dtoa(time) << "<" << dtoa(timer->time));
  232.   time_lock.unlock();
  233.   return resync;
  234. }
  236. int SyncData::done(int term=0){ return (term) ? terminated=term : terminated; }
  239. /*
  240.  *
  241.  * Synchronization
  242.  *
  243.  */
  245. /*
  246.   type  sync id     notes
  247.    0     1, 2       video + audio
  248.    1     1          video
  249.    2     2          audio
  250.    n     1,2,... n  video + audio + others
  251. */
  253. Synchronization::Synchronization(int type, int t_qsize, int f_qsize) : 
  254.   timer(t_qsize),
  255.   terminate(0), 
  256.   terminated(0) 
  257. {
  258.   int i;
  259.   int error=0;
  260.   sched_param param;
  261.   int policy;
  262.   for (i=0; i<Max_Sync_Process; i++) syncs[i]=0; // set all to 0
  263.   if (type>2){
  264.     for (i=0; i<Max_Sync_Process; i++)
  265.       syncs[i]=new SyncData(f_qsize, &timer);
  266.   }
  267.   else if (type==0){
  268.     for (i=0; i<2 && i<Max_Sync_Process; i++)
  269.       syncs[i]=new SyncData(f_qsize, &timer);  // id 1 & 2
  270.   }
  271.   else if (type==1){
  272.     syncs[0]=new SyncData(f_qsize, &timer);    // only id=1
  273.   }
  274.   else if (type==2){
  275.     syncs[0]=0;
  276.     syncs[1]=new SyncData(f_qsize, &timer);    // only id=2
  277.   }
  279.   if ((error=athr_create((void*(*)(void*)) Synchronization::init, this, &id))<0){
  280.     error("failed to create thread");
  281.     athr_exit(0);
  282.   }
  283.   if ((error=athr_getschedparam(id, &policy, &param))<0){
  284.     warning("could not get thread prio - ignored");
  285.   }
  286.   else {
  287. #ifdef LINUX
  288.     param.sched_priority+=1;
  289. //    policy = SCHED_RR;
  290.     TRACER("TIMERPRIORITY=" << param.sched_priority << "(" << param.sched_priority-1 << ")");
  291. #else
  292.     param.prio+=1;
  293.     TRACER("TIMERPRIORITY=" << param.prio << "(" << param.prio-1 << ")");
  294. #endif
  295.     if ((error=athr_setschedparam(id, policy, &param))<0){
  296.       warning("could not set thread prio - ignored");
  297.     }
  298.   }
  299. }
  301. Synchronization::~Synchronization(){
  302.   if (!terminated){
  303.     TRACER("waiting for synchronization thread to terminate ..."); 
  304.     athr_join(id);
  305.   }
  306.   TRACER("synchronization thread terminate!"); 
  307. }
  309. void* Synchronization::init(Synchronization* s){
  310.   TRACER("void* Synchronization::init(Synchronization* s)");
  311.   int i;
  312.   int resync=0;
  314.   while (!s->terminate){
  315.     resync=s->timer.update();
  317. #ifdef TRACE
  318.     msg("!");
  319.     if (s->syncs[0] && s->syncs[1]){
  320.       if (s->syncs[0]->total==0 && s->syncs[1]->total==s->syncs[1]->maximum){
  321.         TRACER("video stamps underflow, audio stamps overflow");
  322.       }
  323.       if (s->syncs[0]->total==s->syncs[0]->maximum && s->syncs[1]->total==0){
  324.         TRACER("audio stamps underflow, video stamps overflow");
  325.       }      
  326.     }
  327. #endif
  328.     for (i=0; ((s->terminate==0) && (i<Max_Sync_Process)); i++){
  329.       if ((s->syncs[i]!=0) && ((s->syncs[i]->total!=0) || resync)){
  330. #ifdef TRACE
  331.         msg(itoa(i));
  332. #endif
  333.         s->syncs[i]->update();
  334.       }
  335.     }
  336.   }
  338.   // finish up
  339.   while (s->timer.total) s->timer.update();
  340.   s->timer.done();
  341.   s->terminated=1;
  342.   athr_exit(0);
  343.   return 0;
  344. }
  347. int Synchronization::stop(){
  348.   TRACER("int Synchronization::stop()");
  349.   terminate=1;
  350.   double finaltime=0.0;
  351.   for (int i=0; i<Max_Sync_Process; i++){
  352.     if (syncs[i]){
  353.       syncs[i]->terminated=1;
  354.       syncs[i]->put(finaltime); // put last dummy time stamp
  355.       syncs[i]->time_cond.signal();
  356.     }
  357.   }
  358.   finaltime=0x0fffffff;
  359.   timer.put(finaltime);
  360.   return terminated;
  361. }
  363. int Synchronization::usedbytes(int ID, int b){ return (ID!=0) ? syncs[ID-1]->usedbytes(b) : -1; }
  364. int Synchronization::wait(int ID){ return (ID!=0) ? syncs[ID-1]->wait() : -1; }
  365. int Synchronization::skip(int ID){ return (ID!=0) ? syncs[ID-1]->skip() : -1; }
  366. int Synchronization::put(const double stamp){ return timer.put(stamp,0); }
  367. int Synchronization::put(int ID, const double s, const int b){ return (ID!=0) ? syncs[ID-1]->put(s, b) : -1; }
  368. int Synchronization::pause(){ return athr_suspend(id); }   // stop timer
  369. int Synchronization::resume(){ return athr_continue(id); } // continu timer
  370. int Synchronization::done(int ID){ return (ID!=0) ? syncs[ID-1]->done() : -1; }
  373. #ifdef MAIN
  375. main(int argc, char** argv){
  376.   // Just a test to compile and link
  377.   Synchronization sync();
  379.   while (1){
  380.   }
  381. }
  382.       // set higher priority for reading process
  383.       int prio=0;
  384.       thr_getprio(thr_self(), &prio);
  385.       prio++;
  386.       thr_setprio(thr_self(), prio);
  389. #endif