Quelle  alsa-drain-contract.patch   Sprache: unbekannt

diff --git a/src/cubeb_alsa.c b/src/cubeb_alsa.c
index be9faa49..d5cb5ecb 100644
--- a/src/cubeb_alsa.c
+++ b/src/cubeb_alsa.c
@@ -318,8 +318,18 @@ alsa_set_stream_state(cubeb_stream * stm, enum stream_state state)
   poll_wake(ctx);
 }
 
+static void
+alsa_set_draining(cubeb_stream * stm)
+{
+  alsa_set_stream_state(stm, DRAINING);
+  if (stm->stream_type == SND_PCM_STREAM_PLAYBACK && stm->other_stream &&
+      stm->other_stream->state == RUNNING) {
+    alsa_set_stream_state(stm->other_stream, DRAINING);
+  }
+}
+
 static enum stream_state
-alsa_process_stream(cubeb_stream * stm)
+alsa_process_stream(cubeb_stream * stm, int other_stream_running)
 {
   unsigned short revents;
   snd_pcm_sframes_t avail;
@@ -376,6 +386,7 @@ alsa_process_stream(cubeb_stream * stm)
       (!stm->other_stream ||
        stm->other_stream->bufframes < stm->other_stream->buffer_size)) {
     snd_pcm_sframes_t wrote = stm->bufframes;
+    snd_pcm_sframes_t requested;
     struct cubeb_stream * mainstm = stm->other_stream ? stm->other_stream : stm;
     void * other_buffer = stm->other_stream ? stm->other_stream->buffer +
                                                   stm->other_stream->bufframes
@@ -388,6 +399,7 @@ alsa_process_stream(cubeb_stream * stm)
       wrote = stm->other_stream->buffer_size - stm->other_stream->bufframes;
     }
 
+    requested = wrote;
     pthread_mutex_unlock(&stm->mutex);
     wrote = stm->data_callback(mainstm, stm->user_ptr, stm->buffer,
                                other_buffer, wrote);
@@ -401,17 +413,27 @@ alsa_process_stream(cubeb_stream * stm)
       if (stm->other_stream) {
         stm->other_stream->bufframes += wrote;
       }
+      if (wrote < requested) {
+        if (!stm->other_stream) {
+          pthread_mutex_unlock(&stm->mutex);
+          return INACTIVE;
+        }
+        draining = 1;
+        set_timeout(&stm->drain_timeout, 0);
+      }
     }
   }
 
   /* Playback: Don't have enough data? Let's ask for more. */
   if (stm->stream_type == SND_PCM_STREAM_PLAYBACK &&
       avail > (snd_pcm_sframes_t)stm->bufframes &&
-      (!stm->other_stream || stm->other_stream->bufframes > 0)) {
+      (!stm->other_stream ||
+       (other_stream_running && stm->other_stream->bufframes > 0))) {
     long got = avail - stm->bufframes;
     void * other_buffer = stm->other_stream ? stm->other_stream->buffer : NULL;
     char * buftail =
         stm->buffer + WRAP(snd_pcm_frames_to_bytes)(stm->pcm, stm->bufframes);
+    long requested;
 
     /* Correct read size to the other stream available frames */
     if (stm->other_stream &&
@@ -419,6 +441,7 @@ alsa_process_stream(cubeb_stream * stm)
       got = stm->other_stream->bufframes;
     }
 
+    requested = got;
     pthread_mutex_unlock(&stm->mutex);
     got = stm->data_callback(stm, stm->user_ptr, other_buffer, buftail, got);
     pthread_mutex_lock(&stm->mutex);
@@ -431,6 +454,9 @@ alsa_process_stream(cubeb_stream * stm)
       if (stm->other_stream) {
         stream_buffer_decrement(stm->other_stream, got);
       }
+      if (got < requested) {
+        draining = 1;
+      }
     }
   }
 
@@ -446,7 +472,7 @@ alsa_process_stream(cubeb_stream * stm)
     stm->bufframes = avail;
 
     /* Mark as draining, unless we're waiting for capture */
-    if (!stm->other_stream || stm->other_stream->bufframes > 0) {
+    if (!stm->other_stream || draining || stm->other_stream->bufframes > 0) {
       set_timeout(&stm->drain_timeout, drain_time * 1000);
 
       draining = 1;
@@ -524,7 +550,10 @@ alsa_run(cubeb * ctx)
     stm = ctx->streams[i];
     if (stm && stm->state == DRAINING) {
       r = ms_until(&stm->drain_timeout);
-      if (r >= 0 && timeout > r) {
+      if (r < 0) {
+        r = 0;
+      }
+      if (timeout > r) {
         timeout = r;
       }
     }
@@ -552,11 +581,20 @@ alsa_run(cubeb * ctx)
          https://github.com/kinetiknz/cubeb/issues/135. */
       if (stm && stm->state == RUNNING && stm->fds &&
           any_revents(stm->fds, stm->nfds)) {
+        int other_stream_running =
+            !stm->other_stream || stm->other_stream->state == RUNNING;
         alsa_set_stream_state(stm, PROCESSING);
         pthread_mutex_unlock(&ctx->mutex);
-        state = alsa_process_stream(stm);
+        state = alsa_process_stream(stm, other_stream_running);
         pthread_mutex_lock(&ctx->mutex);
-        alsa_set_stream_state(stm, state);
+        if (state == DRAINING) {
+          alsa_set_draining(stm);
+        } else if (state == INACTIVE) {
+          alsa_set_stream_state(stm, state);
+          stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_STOPPED);
+        } else {
+          alsa_set_stream_state(stm, state);
+        }
       }
     }
   } else if (r == 0) {
@@ -565,7 +603,10 @@ alsa_run(cubeb * ctx)
       if (stm) {
         if (stm->state == DRAINING && ms_since(&stm->drain_timeout) >= 0) {
           alsa_set_stream_state(stm, INACTIVE);
-          stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_DRAINED);
+          if (!stm->other_stream ||
+              stm->stream_type == SND_PCM_STREAM_PLAYBACK) {
+            stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_DRAINED);
+          }
         } else if (stm->state == RUNNING &&
                    ms_since(&stm->last_activity) > CUBEB_WATCHDOG_MS) {
           alsa_set_stream_state(stm, ERROR);
diff --git a/test/test_callback_ret.cpp b/test/test_callback_ret.cpp
index 0737b60a..d2c1e353 100644
--- a/test/test_callback_ret.cpp
+++ b/test/test_callback_ret.cpp
@@ -31,6 +31,8 @@ struct user_state_callback_ret {
   std::atomic<int> cb_count{0};
   std::atomic<int> expected_cb_count{0};
   std::atomic<int> error_state{0};
+  std::atomic<int> stopped_state{0};
+  std::atomic<int> drained_state{0};
 };
 
 // Data callback that always returns 0
@@ -134,9 +136,11 @@ state_cb_ret(cubeb_stream * stream, void * user, cubeb_state state)
     break;
   case CUBEB_STATE_STOPPED:
     fprintf(stderr, "stream stopped\n");
+    u->stopped_state.fetch_add(1);
     break;
   case CUBEB_STATE_DRAINED:
     fprintf(stderr, "stream drained\n");
+    u->drained_state.fetch_add(1);
     break;
   case CUBEB_STATE_ERROR:
     fprintf(stderr, "stream error\n");
@@ -147,6 +151,46 @@ state_cb_ret(cubeb_stream * stream, void * user, cubeb_state state)
   }
 }
 
+enum expected_terminal_state {
+  NO_TERMINAL_STATE,
+  DRAINED_STATE,
+  SHORT_INPUT_CALLBACK
+};
+
+expected_terminal_state
+expected_terminal_state_for_callback(test_direction direction,
+                                     cubeb_data_callback data_cb)
+{
+  if (data_cb == data_cb_ret_zero || data_cb == data_cb_ret_nframes_minus_one) {
+    return direction == INPUT_ONLY ? SHORT_INPUT_CALLBACK : DRAINED_STATE;
+  }
+  return NO_TERMINAL_STATE;
+}
+
+bool
+wait_for_drained_state(user_state_callback_ret & user_state)
+{
+  for (int i = 0; i < 40; i++) {
+    if (user_state.drained_state > 0) {
+      return true;
+    }
+    delay(50);
+  }
+  return false;
+}
+
+bool
+wait_for_callback(user_state_callback_ret & user_state)
+{
+  for (int i = 0; i < 40; i++) {
+    if (user_state.cb_count > 0) {
+      return true;
+    }
+    delay(50);
+  }
+  return false;
+}
+
 void
 run_test_callback(test_direction direction, cubeb_data_callback data_cb,
                   const std::string & test_desc)
@@ -158,6 +202,10 @@ run_test_callback(test_direction direction, cubeb_data_callback data_cb,
   int r;
   user_state_callback_ret user_state;
   uint32_t latency_frames = 0;
+  expected_terminal_state expected_state =
+      expected_terminal_state_for_callback(direction, data_cb);
+  int stopped_before_stop = 0;
+  int drained_before_stop = 0;
 
   r = common_init(&ctx, "Cubeb callback return value example");
   ASSERT_EQ(r, CUBEB_OK) << "Error initializing cubeb library";
@@ -211,7 +259,15 @@ run_test_callback(test_direction direction, cubeb_data_callback data_cb,
       cleanup_stream_at_exit(stream, cubeb_stream_destroy);
 
   cubeb_stream_start(stream);
-  delay(100);
+  if (expected_state == NO_TERMINAL_STATE) {
+    delay(100);
+  } else if (expected_state == SHORT_INPUT_CALLBACK) {
+    wait_for_callback(user_state);
+  } else {
+    wait_for_drained_state(user_state);
+  }
+  stopped_before_stop = user_state.stopped_state;
+  drained_before_stop = user_state.drained_state;
   cubeb_stream_stop(stream);
 
   ASSERT_EQ(user_state.expected_cb_count, user_state.cb_count)
@@ -220,6 +276,13 @@ run_test_callback(test_direction direction, cubeb_data_callback data_cb,
   if (data_cb == data_cb_ret_error && user_state.cb_count != 0) {
     ASSERT_EQ(user_state.error_state, 1) << "Callback expected error state";
   }
+  if (expected_state == DRAINED_STATE && user_state.cb_count != 0) {
+    ASSERT_EQ(drained_before_stop, 1)
+        << "Output-capable short callback must drain for " << test_desc << "!";
+    ASSERT_EQ(stopped_before_stop, 0)
+        << "Output-capable short callback must not stop for " << test_desc
+        << "!";
+  }
 }
 
 TEST(cubeb, test_input_callback)