implement alsa capture with timers

Add timer based alsa capture support
Improve format negotiation in alsa
Better track buffer owner in the links

pinos/modules/module-protocol-native.c

@@ -134,7 +134,7 @@ connection_data (SpaSource *source,
   void *message;
 
   if (mask & (SPA_IO_ERR | SPA_IO_HUP)) {
-    pinos_log_debug ("protocol-native %p: got connection error", client->impl);
+    pinos_log_error ("protocol-native %p: got connection error", client->impl);
     client_destroy (client);
     return;
   }
@@ -144,7 +144,7 @@ connection_data (SpaSource *source,
       PinosResource *resource;
       const PinosDemarshalFunc *demarshal;
 
-      pinos_log_debug ("protocol-native %p: got message %d from %u", client->impl, opcode, id);
+      pinos_log_trace ("protocol-native %p: got message %d from %u", client->impl, opcode, id);
 
       resource = pinos_map_lookup (&c->objects, id);
       if (resource == NULL) {

pinos/server/core.c

@@ -611,6 +611,10 @@ again:
     asprintf (error, "error node state");
     goto error;
   }
+  if (format == NULL) {
+    asprintf (error, "error get format");
+    goto error;
+  }
   return format;
 
 error:

pinos/server/link.c

@@ -47,7 +47,7 @@ typedef struct
   PinosListener output_port_destroy;
   PinosListener output_async_complete;
 
-  bool allocated;
+  void *buffer_owner;
   PinosMemblock buffer_mem;
   SpaBuffer **buffers;
   uint32_t n_buffers;
@@ -453,13 +453,20 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
         (out_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS))
       minsize = 0;
 
-    if (this->output->allocated) {
+    if (this->output->n_buffers) {
       out_flags = 0;
       in_flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
       impl->n_buffers = this->output->n_buffers;
       impl->buffers = this->output->buffers;
-      impl->allocated = false;
+      impl->buffer_owner = this->output;
       pinos_log_debug ("reusing %d output buffers %p", impl->n_buffers, impl->buffers);
+    } else if (this->input->n_buffers) {
+      out_flags = SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS;
+      in_flags = 0;
+      impl->n_buffers = this->input->n_buffers;
+      impl->buffers = this->input->buffers;
+      impl->buffer_owner = this->input;
+      pinos_log_debug ("reusing %d input buffers %p", impl->n_buffers, impl->buffers);
     } else {
       size_t data_sizes[1];
       ssize_t data_strides[1];
@@ -467,6 +474,7 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       data_sizes[0] = minsize;
       data_strides[0] = stride;
 
+      impl->buffer_owner = this;
       impl->n_buffers = max_buffers;
       impl->buffers = alloc_buffers (this,
                                      impl->n_buffers,
@@ -476,6 +484,8 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
                                      data_sizes,
                                      data_strides,
                                      &impl->buffer_mem);
+
+      pinos_log_debug ("allocating %d input buffers %p", impl->n_buffers, impl->buffers);
     }
 
     if (out_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) {
@@ -492,7 +502,7 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       this->output->n_buffers = impl->n_buffers;
       this->output->allocated = true;
       this->output->buffer_mem = impl->buffer_mem;
-      impl->allocated = false;
+      impl->buffer_owner = this->output;
       pinos_log_debug ("allocated %d buffers %p from output port", impl->n_buffers, impl->buffers);
     } else if (in_flags & SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS) {
       if ((res = spa_node_port_alloc_buffers (this->input->node->node,
@@ -508,7 +518,7 @@ do_allocation (PinosLink *this, SpaNodeState in_state, SpaNodeState out_state)
       this->input->n_buffers = impl->n_buffers;
       this->input->allocated = true;
       this->input->buffer_mem = impl->buffer_mem;
-      impl->allocated = false;
+      impl->buffer_owner = this->input;
       pinos_log_debug ("allocated %d buffers %p from input port", impl->n_buffers, impl->buffers);
     }
   }
@@ -682,7 +692,7 @@ on_port_destroy (PinosLink *this,
   } else
     return;
 
-  if (port->allocated) {
+  if (impl->buffer_owner == port) {
     impl->buffers = NULL;
     impl->n_buffers = 0;
 
@@ -743,7 +753,7 @@ pinos_link_free (PinosLink *link)
 
   pinos_work_queue_destroy (impl->work);
 
-  if (impl->allocated)
+  if (impl->buffer_owner == link)
     pinos_memblock_free (&impl->buffer_mem);
 
   free (impl);
@@ -876,7 +886,9 @@ pinos_link_new (PinosCore       *core,
 static void
 clear_port_buffers (PinosLink *link, PinosPort *port)
 {
-  if (!port->allocated) {
+  PinosLinkImpl *impl = SPA_CONTAINER_OF (link, PinosLinkImpl, this);
+
+  if (impl->buffer_owner != port) {
     pinos_log_debug ("link %p: clear buffers on port %p", link, port);
     spa_node_port_use_buffers (port->node->node,
                                port->direction,

pinos/server/node.c

@@ -323,6 +323,8 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
       if (po->buffer_id == SPA_ID_INVALID)
         continue;
 
+      pinos_log_trace ("node %p: have output %d", this, po->buffer_id);
+
       outport = this->output_port_map[i];
       spa_list_for_each (link, &outport->rt.links, rt.output_link) {
         if (link->rt.input == NULL || link->rt.output == NULL)
@@ -331,7 +333,6 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
         inport = link->rt.input;
         pi = &inport->node->transport->inputs[inport->port_id];
 
-        processed = true;
 
         pi->buffer_id = po->buffer_id;
 
@@ -344,6 +345,7 @@ on_node_event (SpaNode *node, SpaEvent *event, void *user_data)
                                              po->buffer_id)) < 0)
         pinos_log_warn ("node %p: error reuse buffer: %d", this, res);
 
+      processed = true;
       po->buffer_id = SPA_ID_INVALID;
     }
     if (processed) {

pinos/server/port.c

@@ -65,13 +65,6 @@ pinos_port_destroy (PinosPort *port)
 
   spa_list_remove (&port->link);
 
-  spa_node_port_use_buffers (port->node->node,
-                             port->direction,
-                             port->port_id,
-                             NULL, 0);
-  port->buffers = NULL;
-  port->n_buffers = 0;
-
   free (port);
 }
 

spa/plugins/alsa/alsa-sink.c

@@ -428,6 +428,8 @@ spa_alsa_sink_node_port_get_format (SpaNode          *node,
                                     const SpaFormat **format)
 {
   SpaALSASink *this;
+  SpaPODBuilder b = { NULL, };
+  SpaPODFrame f[2];
 
   if (node == NULL || format == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -440,7 +442,14 @@ spa_alsa_sink_node_port_get_format (SpaNode          *node,
   if (!this->have_format)
     return SPA_RESULT_NO_FORMAT;
 
-  *format = NULL;
+  spa_pod_builder_init (&b, this->format_buffer, sizeof (this->format_buffer));
+  spa_pod_builder_format (&b, &f[0], this->type.format,
+         this->type.media_type.audio, this->type.media_subtype.raw,
+         PROP (&f[1], this->type.format_audio.format,   SPA_POD_TYPE_ID,  this->current_format.info.raw.format),
+         PROP (&f[1], this->type.format_audio.rate,     SPA_POD_TYPE_INT, this->current_format.info.raw.rate),
+         PROP (&f[1], this->type.format_audio.channels, SPA_POD_TYPE_INT, this->current_format.info.raw.channels));
+
+  *format = SPA_POD_BUILDER_DEREF (&b, f[0].ref, SpaFormat);
 
   return SPA_RESULT_OK;
 }

spa/plugins/alsa/alsa-source.c

@@ -470,6 +470,8 @@ spa_alsa_source_node_port_get_format (SpaNode          *node,
                                       const SpaFormat **format)
 {
   SpaALSASource *this;
+  SpaPODBuilder b = { NULL, };
+  SpaPODFrame f[2];
 
   if (node == NULL || format == NULL)
     return SPA_RESULT_INVALID_ARGUMENTS;
@@ -482,7 +484,14 @@ spa_alsa_source_node_port_get_format (SpaNode          *node,
   if (!this->have_format)
     return SPA_RESULT_NO_FORMAT;
 
-  *format = NULL;
+  spa_pod_builder_init (&b, this->format_buffer, sizeof (this->format_buffer));
+  spa_pod_builder_format (&b, &f[0], this->type.format,
+         this->type.media_type.audio, this->type.media_subtype.raw,
+         PROP (&f[1], this->type.format_audio.format,   SPA_POD_TYPE_ID,  this->current_format.info.raw.format),
+         PROP (&f[1], this->type.format_audio.rate,     SPA_POD_TYPE_INT, this->current_format.info.raw.rate),
+         PROP (&f[1], this->type.format_audio.channels, SPA_POD_TYPE_INT, this->current_format.info.raw.channels));
+
+  *format = SPA_POD_BUILDER_DEREF (&b, f[0].ref, SpaFormat);
 
   return SPA_RESULT_OK;
 }
@@ -549,6 +558,7 @@ spa_alsa_source_node_port_use_buffers (SpaNode         *node,
     return SPA_RESULT_NO_FORMAT;
 
   if (this->n_buffers > 0) {
+    spa_alsa_pause (this, false);
     if ((res = spa_alsa_clear_buffers (this)) < 0)
       return res;
   }

spa/plugins/alsa/alsa-utils.c

@@ -14,8 +14,6 @@
 
 #define CHECK(s,msg) if ((err = (s)) < 0) { spa_log_error (state->log, msg ": %s", snd_strerror(err)); return err; }
 
-static void alsa_on_fd_events (SpaSource *source);
-
 static int
 spa_alsa_open (SpaALSAState *state)
 {
@@ -36,7 +34,6 @@ spa_alsa_open (SpaALSAState *state)
                        SND_PCM_NO_AUTO_CHANNELS |
                        SND_PCM_NO_AUTO_FORMAT), "open failed");
 
-
   state->timerfd = timerfd_create (CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
   state->opened = true;
 
@@ -171,22 +168,6 @@ spa_alsa_set_format (SpaALSAState *state, SpaAudioInfo *fmt, SpaPortFormatFlags
   state->rate = info->rate;
   state->frame_size = info->channels * 2;
 
-#if 0
-  period_size = props->period_size;
-  periods = props->periods;
-
-  spa_log_info (state->log, "trying period frames %zd and periods %u", period_size, periods);
-
-  dir = 0;
-  CHECK (snd_pcm_hw_params_set_period_size_near (hndl, params, &period_size, &dir), "set_period_size_near");
-  state->period_frames = period_size;
-
-  dir = 0;
-  CHECK (snd_pcm_hw_params_set_periods_near (hndl, params, &periods, &dir), "set_periods_near");
-  state->buffer_frames = periods * state->period_frames;
-
-  CHECK (snd_pcm_hw_params_set_buffer_size (hndl, params, state->buffer_frames), "set_buffer_size");
-#else
   CHECK (snd_pcm_hw_params_get_buffer_size_max (params, &state->buffer_frames), "get_buffer_size_max");
 
   CHECK (snd_pcm_hw_params_set_buffer_size_near (hndl, params, &state->buffer_frames), "set_buffer_size_near");
@@ -196,7 +177,6 @@ spa_alsa_set_format (SpaALSAState *state, SpaAudioInfo *fmt, SpaPortFormatFlags
   CHECK (snd_pcm_hw_params_set_period_size_near (hndl, params, &period_size, &dir), "set_period_size_near");
   state->period_frames = period_size;
   periods = state->buffer_frames / state->period_frames;
-#endif
 
   spa_log_info (state->log, "buffer frames %zd, period frames %zd, periods %u",
       state->buffer_frames, state->period_frames, periods);
@@ -226,67 +206,18 @@ set_swparams (SpaALSAState *state)
 
   /* start the transfer */
   CHECK (snd_pcm_sw_params_set_start_threshold (hndl, params, LONG_MAX), "set_start_threshold");
-#if 1
   CHECK (snd_pcm_sw_params_get_boundary (params, &boundary), "get_boundary");
 
   CHECK (snd_pcm_sw_params_set_stop_threshold (hndl, params, boundary), "set_stop_threshold");
-#else
-  CHECK (snd_pcm_sw_params_set_stop_threshold (hndl, params, -1), "set_stop_threshold");
-#endif
 
-//  CHECK (snd_pcm_sw_params_set_silence_threshold (hndl, params, 0U), "set_silence_threshold");
-
-  /* enable period events when requested */
-//  CHECK (snd_pcm_sw_params_set_period_event (hndl, params, props->period_event ? 1 : 0), "set_period_event");
   CHECK (snd_pcm_sw_params_set_period_event (hndl, params, 0), "set_period_event");
 
-#if 1
-  /* allow the transfer when at least period_size samples can be processed */
-  /* or disable this mechanism when period event is enabled (aka interrupt like style processing) */
-//  CHECK (snd_pcm_sw_params_set_avail_min (hndl, params, state->period_frames), "set_avail_min");
-#else
-  CHECK (snd_pcm_sw_params_set_avail_min (hndl, params, 0), "set_avail_min");
-#endif
-
   /* write the parameters to the playback device */
   CHECK (snd_pcm_sw_params (hndl, params), "sw_params");
 
   return 0;
 }
 
-/*
- *   Underrun and suspend recovery
- */
-static int
-xrun_recovery (SpaALSAState *state, snd_pcm_t *hndl, int err)
-{
-  snd_pcm_status_t *status;
-
-  snd_pcm_status_alloca (&status);
-
-  if ((err = snd_pcm_status (hndl, status)) < 0) {
-    spa_log_error (state->log, "snd_pcm_status error: %s", snd_strerror (err));
-  }
-
-  if (snd_pcm_status_get_state (status) == SND_PCM_STATE_SUSPENDED) {
-    spa_log_warn (state->log, "SUSPENDED, trying to resume");
-
-    if ((err = snd_pcm_prepare (hndl)) < 0) {
-      spa_log_error (state->log, "snd_pcm_prepare error: %s", snd_strerror (err));
-    }
-  }
-  if (snd_pcm_status_get_state (status) == SND_PCM_STATE_XRUN) {
-    spa_log_warn (state->log, "XRUN");
-  }
-
-  if (spa_alsa_pause (state, true) != SPA_RESULT_OK)
-    return -1;
-  if (spa_alsa_start (state, true) != SPA_RESULT_OK)
-    return -1;
-
-  return err;
-}
-
 static snd_pcm_uframes_t
 pull_frames_queue (SpaALSAState *state,
                    const snd_pcm_channel_area_t *my_areas,
@@ -338,7 +269,6 @@ pull_frames_queue (SpaALSAState *state,
     spa_log_warn (state->log, "underrun, want %zd frames", total_frames);
     snd_pcm_areas_silence (my_areas, offset, state->channels, total_frames, state->format);
   }
-  //spa_log_warn (state->log, "written %zd frames", total_frames);
   return total_frames;
 }
 
@@ -361,7 +291,7 @@ pull_frames_ringbuffer (SpaALSAState *state,
   avail = spa_ringbuffer_get_read_areas (&b->rb->ringbuffer, areas);
   size = SPA_MIN (avail, frames * state->frame_size);
 
-  spa_log_debug (state->log, "%u %u %u %u %zd %zd",
+  spa_log_trace (state->log, "%u %u %u %u %zd %zd",
       areas[0].offset, areas[0].len,
       areas[1].offset, areas[1].len, offset, size);
 
@@ -387,140 +317,44 @@ pull_frames_ringbuffer (SpaALSAState *state,
 
   return frames;
 }
-static int
+
-mmap_write (SpaALSAState *state)
+static snd_pcm_uframes_t
+push_frames_queue (SpaALSAState *state,
+                   const snd_pcm_channel_area_t *my_areas,
+                   snd_pcm_uframes_t offset,
+                   snd_pcm_uframes_t frames)
 {
-  snd_pcm_t *hndl = state->hndl;
+  snd_pcm_uframes_t total_frames = 0;
-  int err;
-  snd_pcm_sframes_t avail;
-  snd_pcm_uframes_t offset, frames, size;
-  const snd_pcm_channel_area_t *my_areas;
-
-#if 0
-  snd_pcm_status_t *status;
-
-  snd_pcm_status_alloca (&status);
-
-  if ((err = snd_pcm_status (hndl, status)) < 0) {
-    spa_log_error (state->log, "snd_pcm_status error: %s", snd_strerror (err));
-    return -1;
-  }
-
-  avail = snd_pcm_status_get_avail (status);
-#else
-  if ((avail = snd_pcm_avail_update (hndl)) < 0) {
-    spa_log_error (state->log, "snd_pcm_avail_update error: %s", snd_strerror (avail));
-    return -1;
-  }
-#endif
-
-  size = avail;
-  while (size > 0) {
-    frames = size;
-    if ((err = snd_pcm_mmap_begin (hndl, &my_areas, &offset, &frames)) < 0) {
-      spa_log_error (state->log, "snd_pcm_mmap_begin error: %s", snd_strerror(err));
-      return -1;
-    }
-    if (frames < state->period_frames)
-      break;
-    else
-      frames = state->period_frames;
-
-    if (state->ringbuffer)
-      frames = pull_frames_ringbuffer (state, my_areas, offset, frames);
-    else
-      frames = pull_frames_queue (state, my_areas, offset, frames);
-
-    if ((err = snd_pcm_mmap_commit (hndl, offset, frames)) < 0) {
-      spa_log_error (state->log, "snd_pcm_mmap_commit error: %s", snd_strerror(err));
-      if (err != -EPIPE && err != -ESTRPIPE)
-        return -1;
-    }
-    size -= frames;
-  }
-  return 0;
-}
-
-static int
-mmap_read (SpaALSAState *state)
-{
-  snd_pcm_t *hndl = state->hndl;
-  int err;
-  snd_pcm_sframes_t avail;
-  snd_pcm_uframes_t offset, frames, size;
-  snd_pcm_status_t *status;
-  const snd_pcm_channel_area_t *my_areas;
-  SpaALSABuffer *b;
-  snd_htimestamp_t htstamp = { 0, 0 };
-  int64_t now;
-  uint8_t *dest = NULL;
-  size_t destsize;
-
-  snd_pcm_status_alloca(&status);
-
-  if ((err = snd_pcm_status (hndl, status)) < 0) {
-    spa_log_error (state->log, "snd_pcm_status error: %s", snd_strerror(err));
-    return err;
-  }
-
-  avail = snd_pcm_status_get_avail (status);
-  snd_pcm_status_get_htstamp (status, &htstamp);
-  now = (int64_t)htstamp.tv_sec * SPA_NSEC_PER_SEC + (int64_t)htstamp.tv_nsec;
-
-  state->last_ticks = state->sample_count * SPA_USEC_PER_SEC / state->rate;
-  state->last_monotonic = now;
 
   if (spa_list_is_empty (&state->free)) {
-    b = NULL;
     spa_log_warn (state->log, "no more buffers");
-  } else {
+  }
+  else {
+    uint8_t *src;
+    size_t n_bytes;
+    SpaALSABuffer *b;
+    SpaData *d;
+    SpaPortOutput *output;
+
     b = spa_list_first (&state->free, SpaALSABuffer, link);
     spa_list_remove (&b->link);
 
-    dest = b->outbuf->datas[0].data;
-    destsize = b->outbuf->datas[0].maxsize;
-
     if (b->h) {
       b->h->seq = state->sample_count;
       b->h->pts = state->last_monotonic;
       b->h->dts_offset = 0;
     }
-    avail = SPA_MIN (avail, destsize / state->frame_size);
-  }
-
-  state->sample_count += avail;
-
-  size = avail;
-  while (size > 0) {
-    frames = size;
-    if ((err = snd_pcm_mmap_begin (hndl, &my_areas, &offset, &frames)) < 0) {
-      spa_log_error (state->log, "snd_pcm_mmap_begin error: %s", snd_strerror (err));
-      return -1;
-    }
-
-    if (b) {
-      size_t n_bytes = frames * state->frame_size;
-
-      memcpy (dest,
-              (uint8_t *)my_areas[0].addr + (offset * state->frame_size),
-              n_bytes);
-      dest += n_bytes;
-    }
-
-    if ((err = snd_pcm_mmap_commit (hndl, offset, frames)) < 0) {
-      spa_log_error (state->log, "snd_pcm_mmap_commit error: %s", snd_strerror(err));
-      return -1;
-    }
-    size -= frames;
-  }
-
-  if (b) {
-    SpaData *d;
-    SpaPortOutput *output;
 
     d = b->outbuf->datas;
+
+    total_frames = SPA_MIN (frames, d[0].maxsize / state->frame_size);
+    src = SPA_MEMBER (my_areas[0].addr, offset * state->frame_size, uint8_t);
+    n_bytes = total_frames * state->frame_size;
+
+    memcpy (d[0].data, src, n_bytes);
+
     d[0].chunk->offset = 0;
-    d[0].chunk->size = avail * state->frame_size;
+    d[0].chunk->size = n_bytes;
     d[0].chunk->stride = 0;
 
     if ((output = state->io)) {
@@ -533,77 +367,16 @@ mmap_read (SpaALSAState *state)
       state->event_cb (&state->node, &event, state->user_data);
     }
   }
-  return 0;
+  return total_frames;
 }
 
-static inline short
+static snd_pcm_uframes_t
-spa_io_to_poll (SpaIO mask)
+push_frames_ringbuffer (SpaALSAState *state,
+                        const snd_pcm_channel_area_t *my_areas,
+                        snd_pcm_uframes_t offset,
+                        snd_pcm_uframes_t frames)
 {
-  short events = 0;
+  return frames;
-
-  if (mask & SPA_IO_IN)
-    events |= POLLIN;
-  if (mask & SPA_IO_OUT)
-    events |= POLLOUT;
-  if (mask & SPA_IO_ERR)
-    events |= POLLERR;
-  if (mask & SPA_IO_HUP)
-    events |= POLLHUP;
-
-  return events;
-}
-
-static inline SpaIO
-spa_poll_to_io (short events)
-{
-  SpaIO mask = 0;
-
-  if (events & POLLIN)
-    mask |= SPA_IO_IN;
-  if (events & POLLOUT)
-    mask |= SPA_IO_OUT;
-  if (events & POLLERR)
-    mask |= SPA_IO_ERR;
-  if (events & POLLHUP)
-    mask |= SPA_IO_HUP;
-
-  return mask;
-}
-
-static void
-alsa_on_fd_events (SpaSource *source)
-{
-  SpaALSAState *state = source->data;
-  snd_pcm_t *hndl = state->hndl;
-  int err, i;
-  unsigned short revents = 0;
-
-  for (i = 0; i < state->n_fds; i++) {
-    state->fds[i].revents = spa_io_to_poll (state->sources[i].rmask);
-    state->sources[i].rmask = 0;
-  }
-
-  snd_pcm_poll_descriptors_revents (hndl,
-                                    state->fds,
-                                    state->n_fds,
-                                    &revents);
-  if (revents & POLLERR) {
-    if ((err = xrun_recovery (state, hndl, err)) < 0) {
-      spa_log_error (state->log, "error: %s", snd_strerror (err));
-    }
-  }
-
-  if (state->stream == SND_PCM_STREAM_CAPTURE) {
-    if (!(revents & POLLIN))
-      return;
-
-    mmap_read (state);
-  } else {
-    if (!(revents & POLLOUT))
-      return;
-
-    mmap_write (state);
-  }
 }
 
 static int
@@ -623,22 +396,21 @@ alsa_try_resume (SpaALSAState *state)
 }
 
 static inline void
-calc_timeout (size_t frames,
+calc_timeout (size_t target,
-              size_t cb_threshold,
+              size_t current,
               size_t rate,
               snd_htimestamp_t *now,
               struct timespec *ts)
 {
-  size_t to_play_usec;
+  size_t to_sleep_usec;
 
   ts->tv_sec = now->tv_sec;
-  /* adjust sleep time to target our callback threshold */
+  if (target > current)
-  if (frames > cb_threshold)
+    to_sleep_usec = (target - current) * 1000000 / rate;
-    to_play_usec = (frames - cb_threshold) * 1000000 / rate;
   else
-    to_play_usec = 0;
+    to_sleep_usec = 0;
 
-  ts->tv_nsec = to_play_usec * 1000 + now->tv_nsec;
+  ts->tv_nsec = to_sleep_usec * 1000 + now->tv_nsec;
 
   while (ts->tv_nsec > 1000000000L) {
     ts->tv_sec++;
@@ -647,7 +419,7 @@ calc_timeout (size_t frames,
 }
 
 static void
-alsa_on_timeout_event (SpaSource *source)
+alsa_on_playback_timeout_event (SpaSource *source)
 {
   uint64_t exp;
   int res;
@@ -680,7 +452,7 @@ alsa_on_timeout_event (SpaSource *source)
     avail = state->buffer_frames;
 
   filled = state->buffer_frames - avail;
-  if (filled > state->threshold + 10) {
+  if (filled > state->threshold + 16) {
     if (snd_pcm_state (hndl) == SND_PCM_STATE_SUSPENDED) {
       spa_log_error (state->log, "suspended: try resume");
       if ((res = alsa_try_resume (state)) < 0)
@@ -734,14 +506,89 @@ alsa_on_timeout_event (SpaSource *source)
   timerfd_settime (state->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);
 }
 
+
+static void
+alsa_on_capture_timeout_event (SpaSource *source)
+{
+  uint64_t exp;
+  int res;
+  SpaALSAState *state = source->data;
+  snd_pcm_t *hndl = state->hndl;
+  snd_pcm_sframes_t avail;
+  snd_pcm_uframes_t total_read = 0;
+  struct itimerspec ts;
+  const snd_pcm_channel_area_t *my_areas;
+  snd_pcm_status_t *status;
+  snd_htimestamp_t htstamp;
+
+  read (state->timerfd, &exp, sizeof (uint64_t));
+
+  snd_pcm_status_alloca(&status);
+
+  if ((res = snd_pcm_status (hndl, status)) < 0) {
+    spa_log_error (state->log, "snd_pcm_status error: %s", snd_strerror (res));
+    return;
+  }
+
+  avail = snd_pcm_status_get_avail (status);
+  snd_pcm_status_get_htstamp (status, &htstamp);
+
+  state->last_ticks = state->sample_count + avail;
+  state->last_monotonic = (int64_t)htstamp.tv_sec * SPA_NSEC_PER_SEC + (int64_t)htstamp.tv_nsec;
+
+  if (avail < state->threshold) {
+    if (snd_pcm_state (hndl) == SND_PCM_STATE_SUSPENDED) {
+      spa_log_error (state->log, "suspended: try resume");
+      if ((res = alsa_try_resume (state)) < 0)
+        return;
+    }
+  } else {
+    snd_pcm_uframes_t to_read = avail;
+
+    while (total_read < to_read) {
+      snd_pcm_uframes_t read, frames, offset;
+
+      frames = to_read - total_read;
+      if ((res = snd_pcm_mmap_begin (hndl, &my_areas, &offset, &frames)) < 0) {
+        spa_log_error (state->log, "snd_pcm_mmap_begin error: %s", snd_strerror(res));
+        return;
+      }
+
+      if (state->ringbuffer)
+        read = push_frames_ringbuffer (state, my_areas, offset, frames);
+      else
+        read = push_frames_queue (state, my_areas, offset, frames);
+
+      if (read < to_read)
+        to_read = 0;
+
+      if ((res = snd_pcm_mmap_commit (hndl, offset, read)) < 0) {
+        spa_log_error (state->log, "snd_pcm_mmap_commit error: %s", snd_strerror(res));
+        if (res != -EPIPE && res != -ESTRPIPE)
+          return;
+      }
+      total_read += read;
+    }
+    state->sample_count += total_read;
+  }
+  calc_timeout (state->threshold, avail - total_read, state->rate, &htstamp, &ts.it_value);
+
+//  printf ("timeout %ld %ld %ld %ld\n", total_read, avail, ts.it_value.tv_sec, ts.it_value.tv_nsec);
+  ts.it_interval.tv_sec = 0;
+  ts.it_interval.tv_nsec = 0;
+  timerfd_settime (state->timerfd, TFD_TIMER_ABSTIME, &ts, NULL);
+}
+
 SpaResult
 spa_alsa_start (SpaALSAState *state, bool xrun_recover)
 {
-  int err, i;
+  int err;
 
   if (state->started)
     return SPA_RESULT_OK;
 
+  spa_log_trace (state->log, "alsa %p: start", state);
+
   CHECK (set_swparams (state), "swparams");
   if (!xrun_recover)
     snd_pcm_dump (state->hndl, state->output);
@@ -751,52 +598,29 @@ spa_alsa_start (SpaALSAState *state, bool xrun_recover)
     return SPA_RESULT_ERROR;
   }
 
-  for (i = 0; i < state->n_fds; i++)
+  if (state->stream == SND_PCM_STREAM_PLAYBACK) {
-    spa_loop_remove_source (state->data_loop, &state->sources[i]);
+    state->source.func = alsa_on_playback_timeout_event;
-
+  } else {
-#if 0
+    state->source.func = alsa_on_capture_timeout_event;
-  if ((state->n_fds = snd_pcm_poll_descriptors_count (state->hndl)) <= 0) {
-    spa_log_error (state->log, "Invalid poll descriptors count %d", state->n_fds);
-    return SPA_RESULT_ERROR;
   }
+  state->source.data = state;
+  state->source.fd = state->timerfd;
+  state->source.mask = SPA_IO_IN;
+  state->source.rmask = 0;
+  spa_loop_add_source (state->data_loop, &state->source);
 
-  if ((err = snd_pcm_poll_descriptors (state->hndl, state->fds, state->n_fds)) < 0) {
+  state->threshold = 1024;
-    spa_log_error (state->log, "snd_pcm_poll_descriptors: %s", snd_strerror(err));
-    return SPA_RESULT_ERROR;
-  }
-
-  for (i = 0; i < state->n_fds; i++) {
-    state->sources[i].func = alsa_on_fd_events;
-    state->sources[i].data = state;
-    state->sources[i].fd = state->fds[i].fd;
-    state->sources[i].mask = spa_poll_to_io (state->fds[i].events);
-    state->sources[i].rmask = 0;
-    state->fds[i].revents  = 0;
-    spa_loop_add_source (state->data_loop, &state->sources[i]);
-  }
-#else
-  state->n_fds = 1;
-  state->sources[0].func = alsa_on_timeout_event;
-  state->sources[0].data = state;
-  state->sources[0].fd = state->timerfd;
-  state->sources[0].mask = SPA_IO_IN;
-  state->sources[0].rmask = 0;
-  spa_loop_add_source (state->data_loop, &state->sources[0]);
-#endif
-
-  state->threshold = 40;
-  state->alsa_started = false;
 
   if (state->stream == SND_PCM_STREAM_PLAYBACK) {
-    alsa_on_timeout_event (&state->sources[0]);
+    state->alsa_started = false;
+   } else {
+    if ((err = snd_pcm_start (state->hndl)) < 0) {
+      spa_log_error (state->log, "snd_pcm_start: %s", snd_strerror (err));
+      return SPA_RESULT_ERROR;
+    }
+    state->alsa_started = true;
   }
-
+  state->source.func (&state->source);
-#if 0
-  if ((err = snd_pcm_start (state->hndl)) < 0) {
-    spa_log_error (state->log, "snd_pcm_start: %s", snd_strerror (err));
-    return SPA_RESULT_ERROR;
-  }
-#endif
 
   state->started = true;
 
@@ -806,14 +630,14 @@ spa_alsa_start (SpaALSAState *state, bool xrun_recover)
 SpaResult
 spa_alsa_pause (SpaALSAState *state, bool xrun_recover)
 {
-  int err, i;
+  int err;
 
   if (!state->started)
     return SPA_RESULT_OK;
 
-  for (i = 0; i < state->n_fds; i++)
+  spa_log_trace (state->log, "alsa %p: pause", state);
-    spa_loop_remove_source (state->data_loop, &state->sources[i]);
+
-  state->n_fds = 0;
+  spa_loop_remove_source (state->data_loop, &state->source);
 
   if ((err = snd_pcm_drop (state->hndl)) < 0)
     spa_log_error (state->log, "snd_pcm_drop %s", snd_strerror (err));

spa/plugins/alsa/alsa-utils.h

@@ -158,9 +158,7 @@ struct _SpaALSAState {
   size_t ready_offset;
 
   bool started;
-  int n_fds;
+  SpaSource source;
-  struct pollfd fds[16];
-  SpaSource sources[16];
   int timerfd;
   bool alsa_started;
   int threshold;

spa/plugins/v4l2/v4l2-source.c

@@ -709,6 +709,7 @@ spa_v4l2_source_node_port_use_buffers (SpaNode         *node,
     return SPA_RESULT_NO_FORMAT;
 
   if (state->n_buffers) {
+    spa_v4l2_stream_off (this);
     if ((res = spa_v4l2_clear_buffers (this)) < 0)
       return res;
   }