chiark / gitweb /
move audio translation to the right place
[disorder] / server / speaker.c
index 98ed297a7896467fe0b9de699a76af19e6b9e77e..72c161862917156fb682eaf39f8541a52a24ab05 100644 (file)
  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
  * USA
  */
-
-/* This program deliberately does not use the garbage collector even though it
- * might be convenient to do so.  This is for two reasons.  Firstly some libao
- * drivers are implemented using threads and we do not want to have to deal
- * with potential interactions between threading and garbage collection.
- * Secondly this process needs to be able to respond quickly and this is not
- * compatible with the collector hanging the program even relatively
- * briefly. */
+/** @file server/speaker.c
+ * @brief Speaker processs
+ *
+ * This program is responsible for transmitting a single coherent audio stream
+ * to its destination (over the network, to some sound API, to some
+ * subprocess).  It receives connections from decoders via file descriptor
+ * passing from the main server and plays them in the right order.
+ *
+ * For the <a href="http://www.alsa-project.org/">ALSA</a> API, 8- and 16- bit
+ * stereo and mono are supported, with any sample rate (within the limits that
+ * ALSA can deal with.)
+ *
+ * When communicating with a subprocess, <a
+ * href="http://sox.sourceforge.net/">sox</a> is invoked to convert the inbound
+ * data to a single consistent format.  The same applies for network (RTP)
+ * play, though in that case currently only 44.1KHz 16-bit stereo is supported.
+ *
+ * The inbound data starts with a structure defining the data format.  Note
+ * that this is NOT portable between different platforms or even necessarily
+ * between versions; the speaker is assumed to be built from the same source
+ * and run on the same host as the main server.
+ *
+ * This program deliberately does not use the garbage collector even though it
+ * might be convenient to do so.  This is for two reasons.  Firstly some sound
+ * APIs use thread threads and we do not want to have to deal with potential
+ * interactions between threading and garbage collection.  Secondly this
+ * process needs to be able to respond quickly and this is not compatible with
+ * the collector hanging the program even relatively briefly.
+ */
 
 #include <config.h>
 #include "types.h"
 # define MACHINE_AO_FMT AO_FMT_LITTLE
 #endif
 
-#define BUFFER_SECONDS 5                /* How many seconds of input to
-                                         * buffer. */
+/** @brief How many seconds of input to buffer
+ *
+ * While any given connection has this much audio buffered, no more reads will
+ * be issued for that connection.  The decoder will have to wait.
+ */
+#define BUFFER_SECONDS 5
 
 #define FRAMES 4096                     /* Frame batch size */
 
-#define NETWORK_BYTES 1024              /* Bytes to send per network packet */
-/* (don't make this too big or arithmetic will start to overflow) */
+/** @brief Bytes to send per network packet
+ *
+ * Don't make this too big or arithmetic will start to overflow.
+ */
+#define NETWORK_BYTES 1024
 
-#define RTP_AHEAD 2                     /* Max RTP playahead (seconds) */
+/** @brief Maximum RTP playahead (seconds) */
+#define RTP_AHEAD 2
 
-#define NFDS 256                        /* Max FDs to poll for */
+/** @brief Maximum number of FDs to poll for */
+#define NFDS 256
 
-/* Known tracks are kept in a linked list.  We don't normally to have
- * more than two - maybe three at the outside. */
+/** @brief Track structure
+ *
+ * Known tracks are kept in a linked list.  Usually there will be at most two
+ * of these but rearranging the queue can cause there to be more.
+ */
 static struct track {
   struct track *next;                   /* next track */
   int fd;                               /* input FD */
@@ -152,12 +185,12 @@ static void version(void) {
   exit(0);
 }
 
-/* Return the number of bytes per frame in FORMAT. */
+/** @brief Return the number of bytes per frame in @p format */
 static size_t bytes_per_frame(const ao_sample_format *format) {
   return format->channels * format->bits / 8;
 }
 
-/* Find track ID, maybe creating it if not found. */
+/** @brief Find track @p id, maybe creating it if not found */
 static struct track *findtrack(const char *id, int create) {
   struct track *t;
 
@@ -177,7 +210,7 @@ static struct track *findtrack(const char *id, int create) {
   return t;
 }
 
-/* Remove track ID (but do not destroy it). */
+/** @brief Remove track @p id (but do not destroy it) */
 static struct track *removetrack(const char *id) {
   struct track *t, **tt;
 
@@ -189,7 +222,7 @@ static struct track *removetrack(const char *id) {
   return t;
 }
 
-/* Destroy a track. */
+/** @brief Destroy a track */
 static void destroy(struct track *t) {
   D(("destroy %s", t->id));
   if(t->fd != -1) xclose(t->fd);
@@ -197,7 +230,7 @@ static void destroy(struct track *t) {
   free(t);
 }
 
-/* Notice a new FD. */
+/** @brief Notice a new connection */
 static void acquire(struct track *t, int fd) {
   D(("acquire %s %d", t->id, fd));
   if(t->fd != -1)
@@ -206,7 +239,111 @@ static void acquire(struct track *t, int fd) {
   nonblock(fd);
 }
 
-/* Read data into a sample buffer.  Return 0 on success, -1 on EOF. */
+/** @brief Return true if A and B denote identical libao formats, else false */
+static int formats_equal(const ao_sample_format *a,
+                         const ao_sample_format *b) {
+  return (a->bits == b->bits
+          && a->rate == b->rate
+          && a->channels == b->channels
+          && a->byte_format == b->byte_format);
+}
+
+/** @brief Compute arguments to sox */
+static void soxargs(const char ***pp, char **qq, ao_sample_format *ao) {
+  int n;
+
+  *(*pp)++ = "-t.raw";
+  *(*pp)++ = "-s";
+  *(*pp)++ = *qq; n = sprintf(*qq, "-r%d", ao->rate); *qq += n + 1;
+  *(*pp)++ = *qq; n = sprintf(*qq, "-c%d", ao->channels); *qq += n + 1;
+  /* sox 12.17.9 insists on -b etc; CVS sox insists on -<n> etc; both are
+   * deployed! */
+  switch(config->sox_generation) {
+  case 0:
+    if(ao->bits != 8
+       && ao->byte_format != AO_FMT_NATIVE
+       && ao->byte_format != MACHINE_AO_FMT) {
+      *(*pp)++ = "-x";
+    }
+    switch(ao->bits) {
+    case 8: *(*pp)++ = "-b"; break;
+    case 16: *(*pp)++ = "-w"; break;
+    case 32: *(*pp)++ = "-l"; break;
+    case 64: *(*pp)++ = "-d"; break;
+    default: fatal(0, "cannot handle sample size %d", (int)ao->bits);
+    }
+    break;
+  case 1:
+    switch(ao->byte_format) {
+    case AO_FMT_NATIVE: break;
+    case AO_FMT_BIG: *(*pp)++ = "-B"; break;
+    case AO_FMT_LITTLE: *(*pp)++ = "-L"; break;
+    }
+    *(*pp)++ = *qq; n = sprintf(*qq, "-%d", ao->bits/8); *qq += n + 1;
+    break;
+  }
+}
+
+/** @brief Enable format translation
+ *
+ * If necessary, replaces a tracks inbound file descriptor with one connected
+ * to a sox invocation, which performs the required translation.
+ */
+static void enable_translation(struct track *t) {
+  switch(config->speaker_backend) {
+  case BACKEND_COMMAND:
+  case BACKEND_NETWORK:
+    /* These backends need a specific sample format */
+    break;
+  case BACKEND_ALSA:
+    /* ALSA can cope */
+    return;
+  }
+  if(!formats_equal(&t->format, &config->sample_format)) {
+    char argbuf[1024], *q = argbuf;
+    const char *av[18], **pp = av;
+    int soxpipe[2];
+    pid_t soxkid;
+
+    *pp++ = "sox";
+    soxargs(&pp, &q, &t->format);
+    *pp++ = "-";
+    soxargs(&pp, &q, &config->sample_format);
+    *pp++ = "-";
+    *pp++ = 0;
+    if(debugging) {
+      for(pp = av; *pp; pp++)
+        D(("sox arg[%d] = %s", pp - av, *pp));
+      D(("end args"));
+    }
+    xpipe(soxpipe);
+    soxkid = xfork();
+    if(soxkid == 0) {
+      signal(SIGPIPE, SIG_DFL);
+      xdup2(t->fd, 0);
+      xdup2(soxpipe[1], 1);
+      fcntl(0, F_SETFL, fcntl(0, F_GETFL) & ~O_NONBLOCK);
+      close(soxpipe[0]);
+      close(soxpipe[1]);
+      close(t->fd);
+      execvp("sox", (char **)av);
+      _exit(1);
+    }
+    D(("forking sox for format conversion (kid = %d)", soxkid));
+    close(t->fd);
+    close(soxpipe[1]);
+    t->fd = soxpipe[0];
+    t->format = config->sample_format;
+    ready = 0;
+  }
+}
+
+/** @brief Read data into a sample buffer
+ * @param t Pointer to track
+ * @return 0 on success, -1 on EOF
+ *
+ * This is effectively the read callback on @c t->fd.
+ */
 static int fill(struct track *t) {
   size_t where, left;
   int n;
@@ -242,6 +379,8 @@ static int fill(struct track *t) {
       /* Check that our assumptions are met. */
       if(t->format.bits & 7)
         fatal(0, "bits per sample not a multiple of 8");
+      /* If the input format is unsuitable, arrange to translate it */
+      enable_translation(t);
       /* Make a new buffer for audio data. */
       t->size = bytes_per_frame(&t->format) * t->format.rate * BUFFER_SECONDS;
       t->buffer = xmalloc(t->size);
@@ -253,16 +392,7 @@ static int fill(struct track *t) {
   return 0;
 }
 
-/* Return true if A and B denote identical libao formats, else false. */
-static int formats_equal(const ao_sample_format *a,
-                         const ao_sample_format *b) {
-  return (a->bits == b->bits
-          && a->rate == b->rate
-          && a->channels == b->channels
-          && a->byte_format == b->byte_format);
-}
-
-/* Close the sound device. */
+/** @brief Close the sound device */
 static void idle(void) {
   D(("idle"));
 #if API_ALSA
@@ -284,7 +414,7 @@ static void idle(void) {
   ready = 0;
 }
 
-/* Abandon the current track */
+/** @brief Abandon the current track */
 static void abandon(void) {
   struct speaker_message sm;
 
@@ -300,6 +430,7 @@ static void abandon(void) {
 }
 
 #if API_ALSA
+/** @brief Log ALSA parameters */
 static void log_params(snd_pcm_hw_params_t *hwparams,
                        snd_pcm_sw_params_t *swparams) {
   snd_pcm_uframes_t f;
@@ -326,43 +457,11 @@ static void log_params(snd_pcm_hw_params_t *hwparams,
 }
 #endif
 
-static void soxargs(const char ***pp, char **qq, ao_sample_format *ao) {
-  int n;
-
-  *(*pp)++ = "-t.raw";
-  *(*pp)++ = "-s";
-  *(*pp)++ = *qq; n = sprintf(*qq, "-r%d", ao->rate); *qq += n + 1;
-  *(*pp)++ = *qq; n = sprintf(*qq, "-c%d", ao->channels); *qq += n + 1;
-  /* sox 12.17.9 insists on -b etc; CVS sox insists on -<n> etc; both are
-   * deployed! */
-  switch(config->sox_generation) {
-  case 0:
-    if(ao->bits != 8
-       && ao->byte_format != AO_FMT_NATIVE
-       && ao->byte_format != MACHINE_AO_FMT) {
-      *(*pp)++ = "-x";
-    }
-    switch(ao->bits) {
-    case 8: *(*pp)++ = "-b"; break;
-    case 16: *(*pp)++ = "-w"; break;
-    case 32: *(*pp)++ = "-l"; break;
-    case 64: *(*pp)++ = "-d"; break;
-    default: fatal(0, "cannot handle sample size %d", (int)ao->bits);
-    }
-    break;
-  case 1:
-    switch(ao->byte_format) {
-    case AO_FMT_NATIVE: break;
-    case AO_FMT_BIG: *(*pp)++ = "-B"; break;
-    case AO_FMT_LITTLE: *(*pp)++ = "-L"; break;
-    }
-    *(*pp)++ = *qq; n = sprintf(*qq, "-%d", ao->bits/8); *qq += n + 1;
-    break;
-  }
-}
-
-/* Make sure the sound device is open and has the right sample format.  Return
- * 0 on success and -1 on error. */
+/** @brief Enable sound output
+ *
+ * Makes sure the sound device is open and has the right sample format.  Return
+ * 0 on success and -1 on error.
+ */
 static int activate(void) {
   /* If we don't know the format yet we cannot start. */
   if(!playing->got_format) {
@@ -372,43 +471,6 @@ static int activate(void) {
   switch(config->speaker_backend) {
   case BACKEND_COMMAND:
   case BACKEND_NETWORK:
-    /* If we pass audio on to some other agent then we enforce the configured
-     * sample format on the *inbound* audio data. */
-    if(!formats_equal(&playing->format, &config->sample_format)) {
-      char argbuf[1024], *q = argbuf;
-      const char *av[18], **pp = av;
-      int soxpipe[2];
-      pid_t soxkid;
-      *pp++ = "sox";
-      soxargs(&pp, &q, &playing->format);
-      *pp++ = "-";
-      soxargs(&pp, &q, &config->sample_format);
-      *pp++ = "-";
-      *pp++ = 0;
-      if(debugging) {
-        for(pp = av; *pp; pp++)
-          D(("sox arg[%d] = %s", pp - av, *pp));
-        D(("end args"));
-      }
-      xpipe(soxpipe);
-      soxkid = xfork();
-      if(soxkid == 0) {
-        xdup2(playing->fd, 0);
-        xdup2(soxpipe[1], 1);
-        fcntl(0, F_SETFL, fcntl(0, F_GETFL) & ~O_NONBLOCK);
-        close(soxpipe[0]);
-        close(soxpipe[1]);
-        close(playing->fd);
-        execvp("sox", (char **)av);
-        _exit(1);
-      }
-      D(("forking sox for format conversion (kid = %d)", soxkid));
-      close(playing->fd);
-      close(soxpipe[1]);
-      playing->fd = soxpipe[0];
-      playing->format = config->sample_format;
-      ready = 0;
-    }
     if(!ready) {
       pcm_format = config->sample_format;
       bufsize = 3 * FRAMES;
@@ -541,6 +603,7 @@ static void fork_cmd(void) {
   xpipe(pfd);
   cmdpid = xfork();
   if(!cmdpid) {
+    signal(SIGPIPE, SIG_DFL);
     xdup2(pfd[0], 0);
     close(pfd[0]);
     close(pfd[1]);
@@ -553,9 +616,9 @@ static void fork_cmd(void) {
 }
 
 static void play(size_t frames) {
-  size_t avail_bytes, written_frames;
+  size_t avail_bytes, write_bytes, written_frames;
   ssize_t written_bytes;
-  struct rtp header;
+  struct rtp_header header;
   struct iovec vec[2];
 
   if(activate()) {
@@ -586,7 +649,7 @@ static void play(size_t frames) {
 
   switch(config->speaker_backend) {
 #if API_ALSA
-  case BACKEND_ALSA:
+  case BACKEND_ALSA: {
     snd_pcm_sframes_t pcm_written_frames;
     size_t avail_frames;
     int err;
@@ -618,6 +681,7 @@ static void play(size_t frames) {
     written_frames = pcm_written_frames;
     written_bytes = written_frames * bpf;
     break;
+  }
 #endif
   case BACKEND_COMMAND:
     if(avail_bytes > frames * bpf)
@@ -647,9 +711,14 @@ static void play(size_t frames) {
       struct timeval now;
       xgettimeofday(&now, 0);
       /* There's been a gap.  Fix up the RTP time accordingly. */
-      rtp_time += (((now.tv_sec + now.tv_usec /1000000.0)
-                    - (rtp_time_real.tv_sec + rtp_time_real.tv_usec / 1000000.0)) 
-                   * playing->format.rate * playing->format.channels);
+      const long offset =  (((now.tv_sec + now.tv_usec /1000000.0)
+                             - (rtp_time_real.tv_sec + rtp_time_real.tv_usec / 1000000.0)) 
+                            * playing->format.rate * playing->format.channels);
+      if(offset >= 0) {
+        info("offset RTP timestamp by %ld", offset);
+        rtp_time += offset;
+      }
+      rtp_time_real = now;
     }
     header.vpxcc = 2 << 6;              /* V=2, P=0, X=0, CC=0 */
     header.seq = htons(rtp_seq++);
@@ -673,33 +742,40 @@ static void play(size_t frames) {
      * generated per second is then the sampling rate times the channel
      * count.)"
      */
-    vec[0].iov_base = (void *)&header;
-    vec[0].iov_len = sizeof header;
-    vec[1].iov_base = playing->buffer + playing->start;
-    vec[1].iov_len = avail_bytes;
+    write_bytes = avail_bytes;
 #if 0
-    {
-      char buffer[3 * sizeof header + 1];
-      size_t n;
-      const uint8_t *ptr = (void *)&header;
-      
-      for(n = 0; n < sizeof header; ++n)
-        sprintf(&buffer[3 * n], "%02x ", *ptr++);
-      info(buffer);
-    }
+    while(write_bytes > 0 && (uint32_t)(playing->buffer + playing->start + write_bytes - 4) == 0)
+      write_bytes -= 4;
 #endif
-    do {
-      written_bytes = writev(bfd,
-                             vec,
-                             2);
-    } while(written_bytes < 0 && errno == EINTR);
-    if(written_bytes < 0) {
-      error(errno, "error transmitting audio data");
-      ++audio_errors;
-      if(audio_errors == 10)
-        fatal(0, "too many audio errors");
+    if(write_bytes) {
+      vec[0].iov_base = (void *)&header;
+      vec[0].iov_len = sizeof header;
+      vec[1].iov_base = playing->buffer + playing->start;
+      vec[1].iov_len = avail_bytes;
+#if 0
+      {
+        char buffer[3 * sizeof header + 1];
+        size_t n;
+        const uint8_t *ptr = (void *)&header;
+        
+        for(n = 0; n < sizeof header; ++n)
+          sprintf(&buffer[3 * n], "%02x ", *ptr++);
+        info(buffer);
+      }
+#endif
+      do {
+        written_bytes = writev(bfd,
+                               vec,
+                               2);
+      } while(written_bytes < 0 && errno == EINTR);
+      if(written_bytes < 0) {
+        error(errno, "error transmitting audio data");
+        ++audio_errors;
+        if(audio_errors == 10)
+          fatal(0, "too many audio errors");
       return;
-    }
+      }
+    } else
     audio_errors /= 2;
     written_bytes = avail_bytes;
     written_frames = written_bytes / bpf;
@@ -712,6 +788,7 @@ static void play(size_t frames) {
       ++rtp_time_real.tv_sec;
       rtp_time_real.tv_usec -= 1000000;
     }
+    assert(rtp_time_real.tv_usec < 1000000);
     break;
   default:
     assert(!"reached");
@@ -903,7 +980,7 @@ int main(int argc, char **argv) {
         }
         break;
 #if API_ALSA
-      case BACKEND_ALSA:
+      case BACKEND_ALSA: {
         /* We send sample data to ALSA as fast as it can accept it, relying on
          * the fact that it has a relatively small buffer to minimize pause
          * latency. */
@@ -925,6 +1002,7 @@ int main(int argc, char **argv) {
         if(alsa_nslots >= 0)
           fdno += alsa_nslots;
         break;
+      }
 #endif
       default:
         assert(!"unknown backend");