npv/pipeline/local/code.frag.c

   1 STATIC void fatal(u8 *fmt, ...)
   2 {
   3         va_list ap;
   4
   5         npv_perr("pipeline:");
   6         va_start(ap, fmt);
   7         npv_vfatal(fmt, ap);
   8         va_end(ap); /* unreachable */
   9 }
  10 STATIC void pout(u8 *fmt, ...)
  11 {
  12         va_list ap;
  13
  14         npv_pout("pipeline:");
  15         va_start(ap, fmt);
  16         npv_vpout(fmt, ap);
  17         va_end(ap);
  18 }
  19 STATIC void wait(long ns)
  20 {
  21         struct timespec wanted;
  22         struct timespec rem;
  23
  24         memset(&wanted, 0, sizeof(wanted));
  25         memset(&rem, 0, sizeof(rem));
  26         wanted.tv_nsec = ns;
  27         loop {
  28                 int r;
  29
  30                 /* linux bug: cannot specify CLOCK_MONOTONIC_RAW */
  31                 r = clock_nanosleep(CLOCK_MONOTONIC, 0, &wanted, &rem);
  32                 if (r == 0)
  33                         break;
  34                 if (r != EINTR)
  35                         fatal("data wait timer failed:%d\n", r);
  36                 /* r == EINTR */
  37                 memcpy(&wanted, &rem, sizeof(wanted));
  38                 memset(&rem, 0, sizeof(rem));
  39         }
  40 }
  41 #define EOF_FMT 2
  42 STATIC void read(void) { loop /* infinite loop */
  43 {
  44         u8 r;
  45         /*
  46          * we do finer-grained locking in there, since we do not want to lock
  47          * the pkt qs during slow access
  48          */
  49         r = npv_fmt_pkts_read_and_q();
  50         if (r == EOF_FMT) {
  51                 npv_pkt_q_lock(npv_audio_pkt_q_p);
  52                 if (!npv_pkt_q_has_eof(npv_audio_pkt_q_p))
  53                         npv_pkt_q_enq(npv_audio_pkt_q_p, eof_pkt_l);
  54                 npv_pkt_q_unlock(npv_audio_pkt_q_p);
  55
  56                 npv_pkt_q_lock(npv_video_pkt_q_p);
  57                 if (!npv_pkt_q_has_eof(npv_video_pkt_q_p))
  58                         npv_pkt_q_enq(npv_video_pkt_q_p, eof_pkt_l);
  59                 npv_pkt_q_unlock(npv_video_pkt_q_p);
  60         }
  61         /* r == LIMITS_REACHED || EAGAIN */
  62         /* should be enough to avoid non kept-alive network connexion */
  63         wait(100000000); /* 100ms */
  64 }}
  65 #undef EOF_FMT
  66 STATIC void *read_thd_entry(void *arg)
  67 {
  68         int r;
  69         sigset_t sset;
  70
  71         r = sigfillset(&sset);
  72         if (r == -1)
  73                 fatal("read thread:unable to get a full signal mask\n");
  74         r = pthread_sigmask(SIG_SETMASK, &sset, 0);
  75         if (r != 0)
  76                 fatal("read thread:unable to \"block\" \"all\" signals\n");
  77         read();
  78         /* unreachable */
  79 }
  80 /*
  81  * our alsa audio buf is (rate/4)~0.25s (interactivity like vol processing).
  82  * then we would like to have ~2 times this buf (~double buf) of decoded audio
  83  * frs. namely we try to have ~(2 *0.25s) of decoded audio frames. we presume
  84  * the audio dec is fine grained enough in order to base our calculation on the
  85  * pts-es of 2 sets of audio frs.
  86  */
  87 STATIC bool have_enough_predecoded_audio_frs(void)
  88 {
  89         int64_t pts_min;
  90         int64_t pts_max;
  91         int64_t pts_delta;
  92         int64_t pts_delta_limit;
  93
  94         npv_audio_dec_sets_lock();
  95         if (npv_audio_dec_sets_p.eof_receive) {
  96                 npv_audio_dec_sets_unlock();
  97                 return true;
  98         } else if (npv_audio_dec_sets_p.n == 0) {
  99                 npv_audio_dec_sets_unlock();
 100                 return false;
 101         }
 102         /* from here we have at least 1 sets of audio frs */
 103         /* we presume the audio sets are in pts order */
 104         pts_min = npv_audio_dec_sets_p.a[0]->pts;
 105         pts_max = npv_audio_dec_sets_p.a[npv_audio_dec_sets_p.n - 1]->pts;
 106         npv_audio_dec_sets_unlock();
 107
 108         pts_delta = pts_max - pts_min;
 109         /* 2 * 0.25s = 500 ms */
 110         pts_delta_limit  = 500 * npv_audio_st_p.tb.den / npv_audio_st_p.tb.num
 111                                                                         / 1000;
 112         if (pts_delta >= pts_delta_limit)
 113                 return true;
 114         return false;
 115 }
 116 STATIC void audio(void) { loop /* infinite loop */
 117 {
 118         if (have_enough_predecoded_audio_frs()) {
 119                 wait(250000000); /* (rate/4)~0.25s */
 120                 continue;
 121         }
 122         /* can be long, finer-grained locking is done in there */
 123         npv_audio_pkts_send();
 124         npv_audio_dec_sets_receive_avail();
 125
 126 }}
 127 STATIC void *audio_thd_entry(void *arg)
 128 {
 129         int r;
 130         sigset_t sset;
 131
 132         r = sigfillset(&sset);
 133         if (r == -1)
 134                 fatal("send thread:unable to get a full signal mask\n");
 135         r = pthread_sigmask(SIG_SETMASK, &sset, 0);
 136         if (r != 0)
 137                 fatal("send thread:unable to \"block\" \"all\" signals\n");
 138         audio();
 139         /* unreachable */
 140 }
 141 /*
 142  * same heuristics than for audio. there is some sort of sync due to video fr
 143  * dropping based on "audio now" in the main thd
 144  */
 145 STATIC bool have_enough_predecoded_video_frs(void)
 146 {
 147         int64_t pts_min;
 148         int64_t pts_max;
 149         int64_t pts_delta;
 150         int64_t pts_delta_limit;
 151
 152         npv_video_dec_frs_lock();
 153         if (npv_video_dec_frs_p.eof_receive) {
 154                 npv_video_dec_frs_unlock();
 155                 return true;
 156         } else if (npv_video_dec_frs_p.n  == 0) {
 157                 npv_video_dec_frs_unlock();
 158                 return false;
 159         }
 160         /* from here we have at least 1 video fr */
 161         /* we presume the video frs are in pts order */
 162         pts_min = npv_video_dec_frs_p.a[0]->pts;
 163         pts_max = npv_video_dec_frs_p.a[npv_video_dec_frs_p.n - 1]->pts;
 164         npv_video_dec_frs_unlock();
 165
 166         pts_delta = pts_max - pts_min;
 167         /* 2 * 0.25s = 500 ms */
 168         pts_delta_limit  = 500 * npv_video_st_p.tb.den / npv_video_st_p.tb.num
 169                                                                         / 1000;
 170         if (pts_delta >= pts_delta_limit)
 171                 return true;
 172         return false;
 173 }
 174 STATIC void video(void) { loop /* infinite loop */
 175 {
 176         if (have_enough_predecoded_video_frs()) {
 177                 wait(250000000); /* (audio rate/4)~0.25s */
 178                 continue;
 179         }
 180         /* can be long, finer-grained locking is done in there */
 181         npv_video_pkts_send();
 182         npv_video_dec_frs_receive_avail();
 183 }}
 184 STATIC void *video_thd_entry(void *arg)
 185 {
 186         int r;
 187         sigset_t sset;
 188
 189         r = sigfillset(&sset);
 190         if (r == -1)
 191                 fatal("send thread:unable to get a full signal mask\n");
 192         r = pthread_sigmask(SIG_SETMASK, &sset, 0);
 193         if (r != 0)
 194                 fatal("send thread:unable to \"block\" \"all\" signals\n");
 195         video();
 196         /* unreachable */
 197 }