Add support for trace logging in lockfree ringbuffer

Fix some crashes when the connection is dead.
Small cleanups in the audio mixer
Only propose alloc_buffer when we are using export_buf in v4l2

pinos/client/context.c

@@ -37,7 +37,7 @@ typedef struct {
 
   int fd;
   PinosConnection *connection;
-  SpaSource source;
+  SpaSource *source;
 
   bool disconnecting;
   PinosListener  need_flush;
@@ -388,7 +388,8 @@ do_flush_event (SpaLoopUtils *utils,
 {
   PinosContextImpl *impl = data;
   if (impl->connection)
-    pinos_connection_flush (impl->connection);
+    if (!pinos_connection_flush (impl->connection))
+      pinos_context_disconnect (&impl->this);
 }
 
 static void
@@ -633,12 +634,12 @@ pinos_context_connect_fd (PinosContext      *context,
 
   impl->fd = fd;
 
-  pinos_loop_add_io (context->loop,
+  impl->source = pinos_loop_add_io (context->loop,
-                     fd,
+                                    fd,
-                     SPA_IO_IN | SPA_IO_HUP | SPA_IO_ERR,
+                                    SPA_IO_IN | SPA_IO_HUP | SPA_IO_ERR,
-                     false,
+                                    false,
-                     on_context_data,
+                                    on_context_data,
-                     impl);
+                                    impl);
 
   context->core_proxy = pinos_proxy_new (context,
                                          0,
@@ -673,6 +674,7 @@ pinos_context_connect_fd (PinosContext      *context,
 no_registry:
   pinos_proxy_destroy (context->core_proxy);
 no_proxy:
+  pinos_loop_destroy_source (context->loop, impl->source);
   pinos_connection_destroy (impl->connection);
 error_close:
   close (fd);
@@ -694,6 +696,10 @@ pinos_context_disconnect (PinosContext *context)
 
   impl->disconnecting = true;
 
+  if (impl->source)
+    pinos_loop_destroy_source (context->loop, impl->source);
+  impl->source = NULL;
+
   if (context->registry_proxy)
     pinos_proxy_destroy (context->registry_proxy);
   context->registry_proxy = NULL;

pinos/client/log.c

@@ -18,6 +18,10 @@
  */
 
 #include <stdio.h>
+#include <unistd.h>
+#include <errno.h>
+
+#include <spa/ringbuffer.h>
 
 #include <pinos/client/log.h>
 
@@ -25,6 +29,15 @@
 
 SpaLogLevel pinos_log_level = DEFAULT_LOG_LEVEL;
 
+#define TRACE_BUFFER (16*1024)
+
+typedef struct {
+  SpaLog log;
+  SpaRingbuffer  trace_rb;
+  uint8_t        trace_data[TRACE_BUFFER];
+  SpaSource     *source;
+} DebugLog;
+
 static void
 do_logv (SpaLog        *log,
          SpaLogLevel    level,
@@ -34,22 +47,28 @@ do_logv (SpaLog        *log,
          const char    *fmt,
          va_list        args)
 {
-  char text[16*1024], location[128];
+  DebugLog *l = SPA_CONTAINER_OF (log, DebugLog, log);
-  static const char *levels[] = {
+  char text[1024], location[1024];
-    "-",
+  static const char *levels[] = { "-", "E", "W", "I", "D", "T", };
-    "E",
+  int size;
-    "W",
+
-    "I",
-    "D",
-    "T",
-  };
   vsnprintf (text, sizeof(text), fmt, args);
-  if (1) {
+  size = snprintf (location, sizeof(location), "[%s][%s:%i %s()] %s\n",
-    snprintf (location, sizeof(location), "%s:%i %s()", strrchr (file, '/')+1, line, func);
+      levels[level], strrchr (file, '/')+1, line, func, text);
-    fprintf(stderr, "[%s][%s] %s\n", levels[level], location, text);
+
-  } else {
+  if (SPA_UNLIKELY (level == SPA_LOG_LEVEL_TRACE && l->source)) {
-    fprintf(stderr, "[%s] %s\n", levels[level], text);
+    uint32_t index;
-  }
+    uint64_t count = 1;
+
+    spa_ringbuffer_get_write_index (&l->trace_rb, &index);
+    spa_ringbuffer_write_data (&l->trace_rb, l->trace_data,
+                                index & l->trace_rb.mask,
+                                location, size);
+    spa_ringbuffer_write_update (&l->trace_rb, index + size);
+
+    write (l->source->fd, &count, sizeof(uint64_t));
+  } else
+    fputs (location, stderr);
 }
 
 static void
@@ -66,25 +85,63 @@ do_log (SpaLog        *log,
   va_end (args);
 }
 
-static SpaLog log = {
+static DebugLog log = {
-  sizeof (SpaLog),
+  { sizeof (SpaLog),
-  NULL,
+    NULL,
-  DEFAULT_LOG_LEVEL,
+    DEFAULT_LOG_LEVEL,
-  do_log,
+    do_log,
-  do_logv,
+    do_logv,
+  },
+  {  0, 0, TRACE_BUFFER, TRACE_BUFFER - 1 },
 };
 
 SpaLog *
 pinos_log_get (void)
 {
-  return &log;
+  return &log.log;
 }
 
 void
 pinos_log_set_level (SpaLogLevel level)
 {
   pinos_log_level = level;
-  log.level = level;
+  log.log.level = level;
+}
+
+static void
+on_trace_event (SpaSource *source)
+{
+  int32_t avail;
+  uint32_t index;
+  uint64_t count;
+
+  if (read (source->fd, &count, sizeof (uint64_t)) != sizeof (uint64_t))
+    fprintf (stderr, "failed to read event fd: %s", strerror (errno));
+
+  while ((avail = spa_ringbuffer_get_read_index (&log.trace_rb, &index)) > 0) {
+    uint32_t offset, first, written;
+
+    if (avail > log.trace_rb.size) {
+      index += avail - log.trace_rb.size;
+      avail = log.trace_rb.size;
+    }
+    offset = index & log.trace_rb.mask;
+    first = SPA_MIN (avail, log.trace_rb.size - offset);
+
+    written = fprintf (stderr, "%*s", first, log.trace_data + offset);
+    if (SPA_UNLIKELY (avail > first)) {
+      written += fprintf (stderr, "%*s", avail - first, log.trace_data + first);
+    }
+    spa_ringbuffer_read_update (&log.trace_rb, index + written);
+  }
+}
+
+void
+pinos_log_set_trace_event (SpaSource *source)
+{
+  log.source = source;
+  log.source->func = on_trace_event;
+  log.source->data = &log;
 }
 
 void
@@ -97,7 +154,7 @@ pinos_log_log (SpaLogLevel  level,
   if (SPA_UNLIKELY (pinos_log_level_enabled (level))) {
     va_list args;
     va_start (args, fmt);
-    do_logv (&log, level, file, line, func, fmt, args);
+    do_logv (&log.log, level, file, line, func, fmt, args);
     va_end (args);
   }
 }
@@ -111,6 +168,6 @@ pinos_log_logv (SpaLogLevel  level,
                 va_list      args)
 {
   if (SPA_UNLIKELY (pinos_log_level_enabled (level))) {
-    do_logv (&log, level, file, line, func, fmt, args);
+    do_logv (&log.log, level, file, line, func, fmt, args);
   }
 }

pinos/client/log.h

@@ -21,6 +21,7 @@
 #define __PINOS_LOG_H__
 
 #include <spa/log.h>
+#include <spa/loop.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -31,6 +32,8 @@ extern SpaLogLevel pinos_log_level;
 SpaLog *      pinos_log_get       (void);
 
 void          pinos_log_set_level (SpaLogLevel level);
+void          pinos_log_set_trace_event (SpaSource *source);
+
 
 void          pinos_log_log       (SpaLogLevel  level,
                                    const char  *file,

pinos/client/protocol-native.c

@@ -79,6 +79,9 @@ core_marshal_client_update (void          *object,
   SpaPODFrame f;
   int i, n_items;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   n_items = props ? props->n_items : 0;
@@ -107,6 +110,9 @@ core_marshal_sync (void     *object,
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f,
@@ -124,6 +130,9 @@ core_marshal_get_registry (void     *object,
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f,
@@ -145,6 +154,9 @@ core_marshal_create_node (void          *object,
   SpaPODFrame f;
   uint32_t i, n_items;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   n_items = props ? props->n_items : 0;
@@ -180,6 +192,9 @@ core_marshal_create_client_node (void          *object,
   SpaPODFrame f;
   uint32_t i, n_items;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   n_items = props ? props->n_items : 0;
@@ -214,6 +229,9 @@ core_marshal_update_types (void          *object,
   SpaPODFrame f;
   uint32_t i;
 
+  if (connection == NULL)
+    return;
+
   spa_pod_builder_add (&b.b,
       SPA_POD_TYPE_STRUCT, &f,
         SPA_POD_TYPE_INT, first_id,
@@ -460,6 +478,9 @@ client_node_marshal_update (void           *object,
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f,
@@ -488,6 +509,9 @@ client_node_marshal_port_update (void              *object,
   SpaPODFrame f[2];
   int i, n_items;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_add (&b.b,
@@ -546,6 +570,9 @@ client_node_marshal_event (void     *object,
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f,
@@ -562,6 +589,9 @@ client_node_marshal_destroy (void    *object)
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f, 0);
@@ -990,6 +1020,9 @@ registry_marshal_bind (void          *object,
   Builder b = { { NULL, 0, 0, NULL, write_pod }, connection };
   SpaPODFrame f;
 
+  if (connection == NULL)
+    return;
+
   core_update_map (proxy->context);
 
   spa_pod_builder_struct (&b.b, &f,

pinos/daemon/main.c

@@ -42,6 +42,14 @@ main (int argc, char *argv[])
   }
 
   loop = pinos_main_loop_new ();
+#if 0
+  {
+    SpaSource *source;
+    source = pinos_loop_add_event (loop->loop, NULL, NULL);
+    pinos_log_set_trace_event (source);
+  }
+#endif
+
   core = pinos_core_new (loop, NULL);
 
   pinos_daemon_config_run_commands (config, core);

spa/lib/debug.c

@@ -18,9 +18,12 @@
  */
 
 #include <stdio.h>
+#include <unistd.h>
+#include <errno.h>
 
 #include <lib/mapper.h>
 #include <spa/format-utils.h>
+#include <spa/loop.h>
 #include "debug.h"
 
 SpaResult
@@ -442,6 +445,15 @@ spa_debug_dict (const SpaDict *dict)
 
 #define DEFAULT_LOG_LEVEL SPA_LOG_LEVEL_INFO
 
+#define TRACE_BUFFER 4096
+
+typedef struct {
+  SpaLog log;
+  SpaRingbuffer  trace_rb;
+  uint8_t        trace_data[TRACE_BUFFER];
+  SpaSource     *source;
+} DebugLog;
+
 static void
 do_logv (SpaLog        *log,
          SpaLogLevel    level,
@@ -451,22 +463,28 @@ do_logv (SpaLog        *log,
          const char    *fmt,
          va_list        args)
 {
-  char text[16*1024], location[128];
+  DebugLog *l = SPA_CONTAINER_OF (log, DebugLog, log);
-  static const char *levels[] = {
+  char text[512], location[1024];
-    "-",
+  static const char *levels[] = { "-", "E", "W", "I", "D", "T", };
-    "E",
+  int size;
-    "W",
+
-    "I",
-    "D",
-    "T",
-  };
   vsnprintf (text, sizeof(text), fmt, args);
-  if (1) {
+  size = snprintf (location, sizeof(location), "[%s][%s:%i %s()] %s\n",
-    snprintf (location, sizeof(location), "%s:%i %s()", strrchr (file, '/')+1, line, func);
+      levels[level], strrchr (file, '/')+1, line, func, text);
-    fprintf(stderr, "[%s][%s] %s\n", levels[level], location, text);
+
-  } else {
+  if (SPA_UNLIKELY (level == SPA_LOG_LEVEL_TRACE) && l->source) {
-    fprintf(stderr, "[%s] %s\n", levels[level], text);
+    uint32_t index;
-  }
+    uint64_t count = 1;
+
+    spa_ringbuffer_get_write_index (&l->trace_rb, &index);
+    spa_ringbuffer_write_data (&l->trace_rb, l->trace_data,
+                                index & l->trace_rb.mask,
+                                location, size);
+    spa_ringbuffer_write_update (&l->trace_rb, index + size);
+
+    write (l->source->fd, &count, sizeof(uint64_t));
+  } else
+    fputs (location, stderr);
 }
 
 static void
@@ -483,16 +501,54 @@ do_log (SpaLog        *log,
   va_end (args);
 }
 
-static SpaLog log = {
+static DebugLog log = {
-  sizeof (SpaLog),
+  { sizeof (SpaLog),
-  NULL,
+    NULL,
-  DEFAULT_LOG_LEVEL,
+    DEFAULT_LOG_LEVEL,
-  do_log,
+    do_log,
-  do_logv,
+    do_logv,
+  },
+  {  0, 0, TRACE_BUFFER, TRACE_BUFFER - 1 },
 };
 
 SpaLog *
 spa_log_get_default (void)
 {
-  return &log;
+  return &log.log;
+}
+
+static void
+on_trace_event (SpaSource *source)
+{
+  int32_t avail;
+  uint32_t index;
+  uint64_t count;
+
+  if (read (source->fd, &count, sizeof (uint64_t)) != sizeof (uint64_t))
+    fprintf (stderr, "failed to read event fd: %s", strerror (errno));
+
+  while ((avail = spa_ringbuffer_get_read_index (&log.trace_rb, &index)) > 0) {
+    uint32_t offset, first, written;
+
+    if (avail > log.trace_rb.size) {
+      index += avail - log.trace_rb.size;
+      avail = log.trace_rb.size;
+    }
+    offset = index & log.trace_rb.mask;
+    first = SPA_MIN (avail, log.trace_rb.size - offset);
+
+    written = fprintf (stderr, "%*s", first, log.trace_data + offset);
+    if (SPA_UNLIKELY (avail > first)) {
+      written += fprintf (stderr, "%*s", avail - first, log.trace_data + first);
+    }
+    spa_ringbuffer_read_update (&log.trace_rb, index + written);
+  }
+}
+
+void
+spa_log_default_set_trace_event (SpaSource *source)
+{
+  log.source = source;
+  log.source->func = on_trace_event;
+  log.source->data = &log;
 }

spa/lib/debug.h

@@ -42,6 +42,8 @@ SpaResult spa_debug_dump_mem  (const void *data, size_t size);
 SpaResult spa_debug_dict  (const SpaDict *dict);
 
 SpaLog * spa_log_get_default (void);
+void     spa_log_default_trace_purge (void);
+
 
 #ifdef __cplusplus
 }  /* extern "C" */

spa/plugins/audiomixer/audiomixer.c

@@ -575,22 +575,24 @@ spa_audiomixer_node_port_send_command (SpaNode        *node,
   return SPA_RESULT_NOT_IMPLEMENTED;
 }
 
-static void
+static inline void
 add_port_data (SpaAudioMixer *this, MixerBuffer *out, SpaAudioMixerPort *port, int layer)
 {
   int i;
   int16_t *op, *ip;
   size_t os, is, chunk;
   MixerBuffer *b;
+  SpaData *id, *od;
 
-  op = SPA_MEMBER (out->outbuf->datas[0].data, out->outbuf->datas[0].chunk->offset, void);
+  od = out->outbuf->datas;
-  os = out->outbuf->datas[0].chunk->size;
+  op = SPA_MEMBER (od[0].data, od[0].chunk->offset, void);
+  os = od[0].chunk->size;
 
   b = spa_list_first (&port->queue, MixerBuffer, link);
 
-  ip = SPA_MEMBER (b->outbuf->datas[0].data,
+  id = b->outbuf->datas;
-      port->queued_offset + b->outbuf->datas[0].chunk->offset, void);
+  ip = SPA_MEMBER (id[0].data, port->queued_offset + id[0].chunk->offset, void);
-  is = b->outbuf->datas[0].chunk->size - port->queued_offset;
+  is = id[0].chunk->size - port->queued_offset;
 
   chunk = SPA_MIN (os, is);
 
@@ -629,6 +631,7 @@ mix_output (SpaAudioMixer *this, size_t n_bytes)
   int i, layer;
   SpaAudioMixerPort *outport;
   SpaPortIO *output;
+  SpaData *od;
 
   outport = &this->out_ports[0];
   output = outport->io;
@@ -638,16 +641,16 @@ mix_output (SpaAudioMixer *this, size_t n_bytes)
 
   outbuf = spa_list_first (&outport->queue, MixerBuffer, link);
   spa_list_remove (&outbuf->link);
+  outbuf->outstanding = true;
 
-  n_bytes = SPA_MIN (n_bytes, outbuf->outbuf->datas[0].maxsize);
+  od = outbuf->outbuf->datas;
+  n_bytes = SPA_MIN (n_bytes, od[0].maxsize);
+  od[0].chunk->offset = 0;
+  od[0].chunk->size = n_bytes;
+  od[0].chunk->stride = 0;
 
   spa_log_trace (this->log, "audiomixer %p: dequeue output buffer %d %zd", this, outbuf->outbuf->id, n_bytes);
 
-  outbuf->outstanding = true;
-  outbuf->outbuf->datas[0].chunk->offset = 0;
-  outbuf->outbuf->datas[0].chunk->size = n_bytes;
-  outbuf->outbuf->datas[0].chunk->stride = 0;
-
   for (layer = 0, i = 0; i < MAX_PORTS; i++) {
     SpaAudioMixerPort *port = &this->in_ports[i];
 
@@ -662,11 +665,6 @@ mix_output (SpaAudioMixer *this, size_t n_bytes)
     }
     add_port_data (this, outbuf, port, layer++);
   }
-  if (layer == 0) {
-    this->state = STATE_IN;
-    return SPA_RESULT_NEED_INPUT;
-  }
-
   output->buffer_id = outbuf->outbuf->id;
   output->status = SPA_RESULT_HAVE_OUTPUT;
   this->state = STATE_OUT;
@@ -747,10 +745,8 @@ spa_audiomixer_node_process_output (SpaNode *node)
   this = SPA_CONTAINER_OF (node, SpaAudioMixer, node);
 
   port = &this->out_ports[0];
-  spa_return_val_if_fail (port->io != NULL, SPA_RESULT_ERROR);
   output = port->io;
-
+  spa_return_val_if_fail (output != NULL, SPA_RESULT_ERROR);
-  spa_return_val_if_fail (port->n_buffers > 0, SPA_RESULT_NO_BUFFERS);
 
   /* recycle */
   if (output->buffer_id != SPA_ID_INVALID) {

spa/plugins/v4l2/v4l2-utils.c

@@ -897,7 +897,7 @@ spa_v4l2_set_format (SpaV4l2Source *this, SpaVideoInfo *format, bool try_only)
   framerate->denom = streamparm.parm.capture.timeperframe.numerator;
 
   state->fmt = fmt;
-  state->info.flags = SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS |
+  state->info.flags = (state->export_buf ? SPA_PORT_INFO_FLAG_CAN_ALLOC_BUFFERS : 0) |
                       SPA_PORT_INFO_FLAG_CAN_USE_BUFFERS |
                       SPA_PORT_INFO_FLAG_LIVE;
   state->info.maxbuffering = -1;
@@ -1052,7 +1052,7 @@ spa_v4l2_use_buffers (SpaV4l2Source *this, SpaBuffer **buffers, uint32_t n_buffe
 
     if (buffers[i]->n_datas < 1) {
       spa_log_error (state->log, "v4l2: invalid memory on buffer %p", buffers[i]);
-      continue;
+      return SPA_RESULT_ERROR;
     }
     d = buffers[i]->datas;
 
@@ -1243,6 +1243,8 @@ spa_v4l2_stream_off (SpaV4l2Source *this)
   if (!state->started)
     return SPA_RESULT_OK;
 
+  spa_v4l2_port_set_enabled (this, false);
+
   type = V4L2_BUF_TYPE_VIDEO_CAPTURE;
   if (xioctl (state->fd, VIDIOC_STREAMOFF, &type) < 0) {
     spa_log_error (this->log, "VIDIOC_STREAMOFF: %s", strerror (errno));

spa/tests/test-mixer.c

@@ -123,8 +123,10 @@ typedef struct {
   unsigned int n_fds;
 } AppData;
 
-#define BUFFER_SIZE1     4092
+#define MIN_LATENCY     64
-#define BUFFER_SIZE2     4096
+
+#define BUFFER_SIZE1    MIN_LATENCY
+#define BUFFER_SIZE2    MIN_LATENCY - 4
 
 static void
 init_buffer (AppData *data, SpaBuffer **bufs, Buffer *ba, int n_buffers, size_t size)
@@ -297,7 +299,7 @@ do_invoke (SpaLoop       *loop,
 }
 
 static SpaResult
-make_nodes (AppData *data)
+make_nodes (AppData *data, const char *device)
 {
   SpaResult res;
   SpaProps *props;
@@ -315,8 +317,8 @@ make_nodes (AppData *data)
 
   spa_pod_builder_init (&b, buffer, sizeof (buffer));
   spa_pod_builder_props (&b, &f[0], data->type.props,
-      SPA_POD_PROP (&f[1], data->type.props_device, 0, SPA_POD_TYPE_STRING, 1, "hw:1"),
+      SPA_POD_PROP (&f[1], data->type.props_device, 0, SPA_POD_TYPE_STRING, 1, device ? device : "hw:0"),
-      SPA_POD_PROP (&f[1], data->type.props_min_latency, 0, SPA_POD_TYPE_INT, 1, 128));
+      SPA_POD_PROP (&f[1], data->type.props_min_latency, 0, SPA_POD_TYPE_INT, 1, MIN_LATENCY));
   props = SPA_POD_BUILDER_DEREF (&b, f[0].ref, SpaProps);
 
   if ((res = spa_node_set_props (data->sink, props)) < 0)
@@ -583,7 +585,7 @@ main (int argc, char *argv[])
 
   init_type (&data.type, data.map);
 
-  if ((res = make_nodes (&data)) < 0) {
+  if ((res = make_nodes (&data, argc > 1 ? argv[1] : NULL)) < 0) {
     printf ("can't make nodes: %d\n", res);
     return -1;
   }