[input] Adapt input_buffer so it can handle dynamic quality (sample rates etc)

Still WIP, player and outputs cannot handle this yet

src/input.c

@@ -40,6 +40,7 @@
 #include "input.h"
 
 // Disallow further writes to the buffer when its size is larger than this threshold
+// TODO untie from 44100
 #define INPUT_BUFFER_THRESHOLD STOB(88200)
 // How long (in sec) to wait for player read before looping in playback thread
 #define INPUT_LOOP_TIMEOUT 1
@@ -64,21 +65,36 @@ static struct input_definition *inputs[] = {
     NULL
 };
 
+struct marker
+{
+  uint64_t pos; // Position of marker measured in bytes
+  struct input_quality quality;
+  enum input_flags flags;
+
+  // Reverse linked list, yay!
+  struct marker *prev;
+};
+
 struct input_buffer
 {
   // Raw pcm stream data
   struct evbuffer *evbuf;
 
-  // If non-zero, remaining length of buffer until EOF
+  // If an input makes a write with a flag or a changed sample rate etc, we add
-  size_t eof;
+  // a marker to head, and when we read we check from the tail to see if there
-  // If non-zero, remaining length of buffer until read error occurred
+  // are updates to the player.
-  size_t error;
+  struct marker *marker_tail;
-  // If non-zero, remaining length of buffer until (possible) new metadata
-  size_t metadata;
 
   // Optional callback to player if buffer is full
   input_cb full_cb;
 
+  // Quality of write/read data
+  struct input_quality cur_write_quality;
+  struct input_quality cur_read_quality;
+
+  size_t bytes_written;
+  size_t bytes_read;
+
   // Locks for sharing the buffer between input and player thread
   pthread_mutex_t mutex;
   pthread_cond_t cond;
@@ -101,47 +117,6 @@ static size_t debug_elapsed;
 
 /* ------------------------------ MISC HELPERS ---------------------------- */
 
-static short
-flags_set(size_t len)
-{
-  short flags = 0;
-
-  if (input_buffer.error)
-    {
-      if (len >= input_buffer.error)
-	{
-	  flags |= INPUT_FLAG_ERROR;
-	  input_buffer.error = 0;
-	}
-      else
-	input_buffer.error -= len;
-    }
-
-  if (input_buffer.eof)
-    {
-      if (len >= input_buffer.eof)
-	{
-	  flags |= INPUT_FLAG_EOF;
-	  input_buffer.eof = 0;
-	}
-      else
-	input_buffer.eof -= len;
-    }
-
-  if (input_buffer.metadata)
-    {
-      if (len >= input_buffer.metadata)
-	{
-	  flags |= INPUT_FLAG_METADATA;
-	  input_buffer.metadata = 0;
-	}
-      else
-	input_buffer.metadata -= len;
-    }
-
-  return flags;
-}
-
 static int
 map_data_kind(int data_kind)
 {
@@ -166,6 +141,27 @@ map_data_kind(int data_kind)
     }
 }
 
+static void
+marker_add(short flags)
+{
+  struct marker *head;
+  struct marker *marker;
+
+  CHECK_NULL(L_PLAYER, marker = calloc(1, sizeof(struct marker)));
+
+  marker->pos = input_buffer.bytes_written;
+  marker->quality = input_buffer.cur_write_quality;
+  marker->flags = flags;
+
+  for (head = input_buffer.marker_tail; head && head->prev; head = head->prev)
+    ; // Fast forward to the head
+
+  if (!head)
+    input_buffer.marker_tail = marker;
+  else
+    head->prev = marker;
+}
+
 static int
 source_check_and_map(struct player_source *ps, const char *action, char check_setup)
 {
@@ -215,7 +211,7 @@ playback(void *arg)
   // Loops until input_loop_break is set or no more input, e.g. EOF
   ret = inputs[type]->start(ps);
   if (ret < 0)
-    input_write(NULL, 0, 0, INPUT_FLAG_ERROR);
+    input_write(NULL, NULL, INPUT_FLAG_ERROR);
 
 #ifdef DEBUG
   DPRINTF(E_DBG, L_PLAYER, "Playback loop stopped (break is %d, ret %d)\n", input_loop_break, ret);
@@ -240,7 +236,7 @@ input_wait(void)
 
 // Called by input modules from within the playback loop
 int
-input_write(struct evbuffer *evbuf, int sample_rate, int bits_per_sample, short flags)
+input_write(struct evbuffer *evbuf, struct input_quality *quality, short flags)
 {
   struct timespec ts;
   int ret;
@@ -271,20 +267,31 @@ input_write(struct evbuffer *evbuf, int sample_rate, int bits_per_sample, short
       return 0;
     }
 
-  if (evbuf)
+  // Change of quality. Note, the marker is placed at the last position of the
-    ret = evbuffer_add_buffer(input_buffer.evbuf, evbuf);
+  // last byte we wrote, even though that of course doesn't have the new quality
-  else
+  // yet. Not intuitive, but input_read() will understand.
+  if (quality && memcmp(quality, &input_buffer.cur_write_quality, sizeof(struct input_quality)) != 0)
+    {
+      input_buffer.cur_write_quality = *quality;
+      marker_add(INPUT_FLAG_QUALITY);
+    }
+
   ret = 0;
-
+  if (evbuf)
+    {
+      input_buffer.bytes_written += evbuffer_get_length(evbuf);
+      ret = evbuffer_add_buffer(input_buffer.evbuf, evbuf);
       if (ret < 0)
+	{
 	  DPRINTF(E_LOG, L_PLAYER, "Error adding stream data to input buffer\n");
+	  flags |= INPUT_FLAG_ERROR;
+	}
+    }
 
-  if (!input_buffer.error && (flags & INPUT_FLAG_ERROR))
+  // Note this marker is added at the post-write position, since EOF and ERROR
-    input_buffer.error = evbuffer_get_length(input_buffer.evbuf);
+  // belong there. We never want to add a marker for the NONBLOCK flag.
-  if (!input_buffer.eof && (flags & INPUT_FLAG_EOF))
+  if (flags & ~INPUT_FLAG_NONBLOCK)
-    input_buffer.eof = evbuffer_get_length(input_buffer.evbuf);
+    marker_add(flags);
-  if (!input_buffer.metadata && (flags & INPUT_FLAG_METADATA))
-    input_buffer.metadata = evbuffer_get_length(input_buffer.evbuf);
 
   pthread_mutex_unlock(&input_buffer.mutex);
 
@@ -298,6 +305,7 @@ input_write(struct evbuffer *evbuf, int sample_rate, int bits_per_sample, short
 int
 input_read(void *data, size_t size, short *flags)
 {
+  struct marker *marker;
   int len;
 
   *flags = 0;
@@ -310,23 +318,50 @@ input_read(void *data, size_t size, short *flags)
 
   pthread_mutex_lock(&input_buffer.mutex);
 
-#ifdef DEBUG
+  // First we check if there is a marker in the requested samples. If there is,
-  debug_elapsed += size;
+  // we only return data up until that marker. That way we don't have to deal
-  if (debug_elapsed > STOB(441000)) // 10 sec
+  // with multiple markers, and we don't return data that contains mixed sample
+  // rates, bits per sample or an EOF in the middle.
+  marker = input_buffer.marker_tail;
+  if (marker && marker->pos < input_buffer.bytes_read + size)
     {
-      DPRINTF(E_DBG, L_PLAYER, "Input buffer has %zu bytes\n", evbuffer_get_length(input_buffer.evbuf));
+      *flags = marker->flags;
-      debug_elapsed = 0;
+      if (*flags & INPUT_FLAG_QUALITY)
+	input_buffer.cur_read_quality = marker->quality;
+
+      size = marker->pos - input_buffer.bytes_read;
+      input_buffer.marker_tail = marker->prev;
+      free(marker);
     }
-#endif
 
   len = evbuffer_remove(input_buffer.evbuf, data, size);
   if (len < 0)
     {
       DPRINTF(E_LOG, L_PLAYER, "Error reading stream data from input buffer\n");
+      *flags |= INPUT_FLAG_ERROR;
       goto out_unlock;
     }
 
-  *flags = flags_set(len);
+  input_buffer.bytes_read += len;
+
+#ifdef DEBUG
+  // Logs if flags present or each 10 seconds
+  debug_elapsed += len;
+  if (*flags || (debug_elapsed / STOB(input_buffer.cur_read_quality.sample_rate) > 10))
+    {
+      debug_elapsed = 0;
+      DPRINTF(E_SPAM, L_PLAYER, "READ %zu bytes (%d/%d), WROTE %zu bytes (%d/%d), SIZE %zu (=%zu), FLAGS %04x\n",
+        input_buffer.bytes_read,
+        input_buffer.cur_read_quality.sample_rate,
+        input_buffer.cur_read_quality.bits_per_sample,
+        input_buffer.bytes_written,
+        input_buffer.cur_write_quality.sample_rate,
+        input_buffer.cur_write_quality.bits_per_sample,
+        evbuffer_get_length(input_buffer.evbuf),
+        input_buffer.bytes_written - input_buffer.bytes_read,
+        *flags);
+    }
+#endif
 
  out_unlock:
   pthread_cond_signal(&input_buffer.cond);
@@ -465,19 +500,30 @@ input_seek(struct player_source *ps, int seek_ms)
 void
 input_flush(short *flags)
 {
+  struct marker *marker;
   size_t len;
 
   pthread_mutex_lock(&input_buffer.mutex);
 
+  // We will return an OR of all the unread marker flags
+  *flags = 0;
+  for (marker = input_buffer.marker_tail; marker; marker = input_buffer.marker_tail)
+    {
+      *flags |= marker->flags;
+      input_buffer.marker_tail = marker->prev;
+      free(marker);
+    }
+
   len = evbuffer_get_length(input_buffer.evbuf);
 
   evbuffer_drain(input_buffer.evbuf, len);
 
-  *flags = flags_set(len);
+  memset(&input_buffer.cur_read_quality, 0, sizeof(struct input_quality));
+  memset(&input_buffer.cur_write_quality, 0, sizeof(struct input_quality));
+
+  input_buffer.bytes_read = 0;
+  input_buffer.bytes_written = 0;
 
-  input_buffer.error = 0;
-  input_buffer.eof = 0;
-  input_buffer.metadata = 0;
   input_buffer.full_cb = NULL;
 
   pthread_mutex_unlock(&input_buffer.mutex);
@@ -487,6 +533,14 @@ input_flush(short *flags)
 #endif
 }
 
+int
+input_quality_get(struct input_quality *quality)
+{
+  // No mutex, other threads should not be able to affect cur_read_quality
+  *quality = input_buffer.cur_read_quality;
+  return 0;
+}
+
 int
 input_metadata_get(struct input_metadata *metadata, struct player_source *ps, int startup, uint64_t rtptime)
 {

src/input.h

@@ -29,6 +29,8 @@ enum input_flags
   INPUT_FLAG_ERROR    = (1 << 2),
   // Flags possible new stream metadata
   INPUT_FLAG_METADATA = (1 << 3),
+  // Flags new stream quality
+  INPUT_FLAG_QUALITY  = (1 << 4),
 };
 
 struct player_source
@@ -81,6 +83,13 @@ struct player_source
 
 typedef int (*input_cb)(void);
 
+struct input_quality
+{
+  int sample_rate;
+  int bits_per_sample;
+  // Maybe some day also add channels here
+};
+
 struct input_metadata
 {
   uint32_t item_id;
@@ -146,14 +155,13 @@ int input_loop_break;
  * until the write can be made (unless INPUT_FILE_NONBLOCK is set).
  *
  * @in  evbuf    Raw PCM_LE audio data to write
- * @in  evbuf    Sample rate of the data
+ * @in  evbuf    Quality of the PCM (sample rate etc.)
- * @in  evbuf    Bits per sample (typically 16 or 24)
  * @in  flags    One or more INPUT_FLAG_*
  * @return       0 on success, EAGAIN if buffer was full (and _NONBLOCK is set),
  *               -1 on error
  */
 int
-input_write(struct evbuffer *evbuf, int sample_rate, int bits_per_sample, short flags);
+input_write(struct evbuffer *evbuf, struct input_quality *quality, short flags);
 
 /*
  * Input modules can use this to wait in the playback loop (like input_write()
@@ -224,6 +232,12 @@ input_seek(struct player_source *ps, int seek_ms);
 void
 input_flush(short *flags);
 
+/*
+ * Returns the current quality of data returned by intput_read().
+ */
+int
+input_quality_get(struct input_quality *quality);
+
 /*
  * Gets metadata from the input, returns 0 if metadata is set, otherwise -1
  */

src/inputs/file_http.c

@@ -59,17 +59,16 @@ static int
 start(struct player_source *ps)
 {
   struct transcode_ctx *ctx = ps->input_ctx;
+  struct input_quality quality = { 0 };
   struct evbuffer *evbuf;
   short flags;
-  int sample_rate;
-  int bps;
   int ret;
   int icy_timer;
 
   evbuf = evbuffer_new();
 
-  sample_rate = transcode_encode_query(ctx->encode_ctx, "sample_rate");
+  quality.sample_rate = transcode_encode_query(ctx->encode_ctx, "sample_rate");
-  bps = transcode_encode_query(ctx->encode_ctx, "bits_per_sample");
+  quality.bits_per_sample = transcode_encode_query(ctx->encode_ctx, "bits_per_sample");
 
   ret = -1;
   flags = 0;
@@ -84,7 +83,7 @@ start(struct player_source *ps)
       flags = ((ret == 0) ? INPUT_FLAG_EOF : 0) |
                (icy_timer ? INPUT_FLAG_METADATA : 0);
 
-      ret = input_write(evbuf, sample_rate, bps, flags);
+      ret = input_write(evbuf, &quality, flags);
       if (ret < 0)
 	break;
     }

src/inputs/pipe.c

@@ -831,6 +831,7 @@ static int
 start(struct player_source *ps)
 {
   struct pipe *pipe = ps->input_ctx;
+  struct input_quality quality = { 0 };
   struct evbuffer *evbuf;
   short flags;
   int ret;
@@ -842,13 +843,16 @@ start(struct player_source *ps)
       return -1;
     }
 
+  quality.sample_rate = pipe_sample_rate;
+  quality.bits_per_sample = pipe_bits_per_sample;
+
   ret = -1;
   while (!input_loop_break)
     {
       ret = evbuffer_read(evbuf, pipe->fd, PIPE_READ_MAX);
       if ((ret == 0) && (pipe->is_autostarted))
 	{
-	  input_write(evbuf, pipe_sample_rate, pipe_bits_per_sample, INPUT_FLAG_EOF); // Autostop
+	  input_write(evbuf, NULL, INPUT_FLAG_EOF); // Autostop
 	  break;
 	}
       else if ((ret == 0) || ((ret < 0) && (errno == EAGAIN)))
@@ -865,7 +869,7 @@ start(struct player_source *ps)
       flags = (pipe_metadata_is_new ? INPUT_FLAG_METADATA : 0);
       pipe_metadata_is_new = 0;
 
-      ret = input_write(evbuf, pipe_sample_rate, pipe_bits_per_sample, flags);
+      ret = input_write(evbuf, &quality, flags);
       if (ret < 0)
 	break;
     }
@@ -949,14 +953,14 @@ init(void)
     }
 
   pipe_sample_rate = cfg_getint(cfg_getsec(cfg, "library"), "pipe_sample_rate");
-  if (pipe_sample_rate != 44100 || pipe_sample_rate != 48000 || pipe_sample_rate != 96000)
+  if (pipe_sample_rate != 44100 && pipe_sample_rate != 48000 && pipe_sample_rate != 96000)
     {
       DPRINTF(E_FATAL, L_PLAYER, "The configuration of pipe_sample_rate is invalid: %d\n", pipe_sample_rate);
       return -1;
     }
 
   pipe_bits_per_sample = cfg_getint(cfg_getsec(cfg, "library"), "pipe_bits_per_sample");
-  if (pipe_bits_per_sample != 16 || pipe_bits_per_sample != 24)
+  if (pipe_bits_per_sample != 16 && pipe_bits_per_sample != 24)
     {
       DPRINTF(E_FATAL, L_PLAYER, "The configuration of pipe_bits_per_sample is invalid: %d\n", pipe_bits_per_sample);
       return -1;

src/spotify.c

@@ -719,7 +719,7 @@ playback_eot(void *arg, int *retval)
   g_state = SPOTIFY_STATE_STOPPING;
 
   // TODO 1) This will block for a while, but perhaps ok?
-  input_write(spotify_audio_buffer, 0, 0, INPUT_FLAG_EOF);
+  input_write(spotify_audio_buffer, NULL, INPUT_FLAG_EOF);
 
   *retval = 0;
   return COMMAND_END;
@@ -1007,6 +1007,7 @@ logged_out(sp_session *sess)
 static int music_delivery(sp_session *sess, const sp_audioformat *format,
                           const void *frames, int num_frames)
 {
+  struct input_quality quality = { 0 };
   size_t size;
   int ret;
 
@@ -1018,6 +1019,9 @@ static int music_delivery(sp_session *sess, const sp_audioformat *format,
       return num_frames;
     }
 
+  quality.sample_rate = format->sample_rate;
+  quality.bits_per_sample = 16;
+
   // Audio discontinuity, e.g. seek
   if (num_frames == 0)
     {
@@ -1037,7 +1041,7 @@ static int music_delivery(sp_session *sess, const sp_audioformat *format,
   // The input buffer only accepts writing when it is approaching depletion, and
   // because we use NONBLOCK it will just return if this is not the case. So in
   // most cases no actual write is made and spotify_audio_buffer will just grow.
-  input_write(spotify_audio_buffer, format->sample_rate, 16, INPUT_FLAG_NONBLOCK);
+  input_write(spotify_audio_buffer, &quality, INPUT_FLAG_NONBLOCK);
 
   return num_frames;
 }