Skip to content

Commit b07f979

Browse files
kinetiknzkinetiknz
committed
alsa: fix drain contract
1 parent cdb54bb commit b07f979

2 files changed

Lines changed: 112 additions & 8 deletions

File tree

‎src/cubeb_alsa.c‎

Lines changed: 48 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -318,8 +318,18 @@ alsa_set_stream_state(cubeb_stream * stm, enum stream_state state)
318318
poll_wake(ctx);
319319
}
320320

321+
static void
322+
alsa_set_draining(cubeb_stream * stm)
323+
{
324+
alsa_set_stream_state(stm, DRAINING);
325+
if (stm->stream_type == SND_PCM_STREAM_PLAYBACK && stm->other_stream &&
326+
stm->other_stream->state == RUNNING) {
327+
alsa_set_stream_state(stm->other_stream, DRAINING);
328+
}
329+
}
330+
321331
static enum stream_state
322-
alsa_process_stream(cubeb_stream * stm)
332+
alsa_process_stream(cubeb_stream * stm, int other_stream_running)
323333
{
324334
unsigned short revents;
325335
snd_pcm_sframes_t avail;
@@ -376,6 +386,7 @@ alsa_process_stream(cubeb_stream * stm)
376386
(!stm->other_stream ||
377387
stm->other_stream->bufframes < stm->other_stream->buffer_size)) {
378388
snd_pcm_sframes_t wrote = stm->bufframes;
389+
snd_pcm_sframes_t requested;
379390
struct cubeb_stream * mainstm = stm->other_stream ? stm->other_stream : stm;
380391
void * other_buffer = stm->other_stream ? stm->other_stream->buffer +
381392
stm->other_stream->bufframes
@@ -388,6 +399,7 @@ alsa_process_stream(cubeb_stream * stm)
388399
wrote = stm->other_stream->buffer_size - stm->other_stream->bufframes;
389400
}
390401

402+
requested = wrote;
391403
pthread_mutex_unlock(&stm->mutex);
392404
wrote = stm->data_callback(mainstm, stm->user_ptr, stm->buffer,
393405
other_buffer, wrote);
@@ -401,24 +413,35 @@ alsa_process_stream(cubeb_stream * stm)
401413
if (stm->other_stream) {
402414
stm->other_stream->bufframes += wrote;
403415
}
416+
if (wrote < requested) {
417+
if (!stm->other_stream) {
418+
pthread_mutex_unlock(&stm->mutex);
419+
return INACTIVE;
420+
}
421+
draining = 1;
422+
set_timeout(&stm->drain_timeout, 0);
423+
}
404424
}
405425
}
406426

407427
/* Playback: Don't have enough data? Let's ask for more. */
408428
if (stm->stream_type == SND_PCM_STREAM_PLAYBACK &&
409429
avail > (snd_pcm_sframes_t)stm->bufframes &&
410-
(!stm->other_stream || stm->other_stream->bufframes > 0)) {
430+
(!stm->other_stream ||
431+
(other_stream_running && stm->other_stream->bufframes > 0))) {
411432
long got = avail - stm->bufframes;
412433
void * other_buffer = stm->other_stream ? stm->other_stream->buffer : NULL;
413434
char * buftail =
414435
stm->buffer + WRAP(snd_pcm_frames_to_bytes)(stm->pcm, stm->bufframes);
436+
long requested;
415437

416438
/* Correct read size to the other stream available frames */
417439
if (stm->other_stream &&
418440
got > (snd_pcm_sframes_t)stm->other_stream->bufframes) {
419441
got = stm->other_stream->bufframes;
420442
}
421443

444+
requested = got;
422445
pthread_mutex_unlock(&stm->mutex);
423446
got = stm->data_callback(stm, stm->user_ptr, other_buffer, buftail, got);
424447
pthread_mutex_lock(&stm->mutex);
@@ -431,6 +454,9 @@ alsa_process_stream(cubeb_stream * stm)
431454
if (stm->other_stream) {
432455
stream_buffer_decrement(stm->other_stream, got);
433456
}
457+
if (got < requested) {
458+
draining = 1;
459+
}
434460
}
435461
}
436462

@@ -446,7 +472,7 @@ alsa_process_stream(cubeb_stream * stm)
446472
stm->bufframes = avail;
447473

448474
/* Mark as draining, unless we're waiting for capture */
449-
if (!stm->other_stream || stm->other_stream->bufframes > 0) {
475+
if (!stm->other_stream || draining || stm->other_stream->bufframes > 0) {
450476
set_timeout(&stm->drain_timeout, drain_time * 1000);
451477

452478
draining = 1;
@@ -524,7 +550,10 @@ alsa_run(cubeb * ctx)
524550
stm = ctx->streams[i];
525551
if (stm && stm->state == DRAINING) {
526552
r = ms_until(&stm->drain_timeout);
527-
if (r >= 0 && timeout > r) {
553+
if (r < 0) {
554+
r = 0;
555+
}
556+
if (timeout > r) {
528557
timeout = r;
529558
}
530559
}
@@ -552,11 +581,20 @@ alsa_run(cubeb * ctx)
552581
https://github.com/kinetiknz/cubeb/issues/135. */
553582
if (stm && stm->state == RUNNING && stm->fds &&
554583
any_revents(stm->fds, stm->nfds)) {
584+
int other_stream_running =
585+
!stm->other_stream || stm->other_stream->state == RUNNING;
555586
alsa_set_stream_state(stm, PROCESSING);
556587
pthread_mutex_unlock(&ctx->mutex);
557-
state = alsa_process_stream(stm);
588+
state = alsa_process_stream(stm, other_stream_running);
558589
pthread_mutex_lock(&ctx->mutex);
559-
alsa_set_stream_state(stm, state);
590+
if (state == DRAINING) {
591+
alsa_set_draining(stm);
592+
} else if (state == INACTIVE) {
593+
alsa_set_stream_state(stm, state);
594+
stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_STOPPED);
595+
} else {
596+
alsa_set_stream_state(stm, state);
597+
}
560598
}
561599
}
562600
} else if (r == 0) {
@@ -565,7 +603,10 @@ alsa_run(cubeb * ctx)
565603
if (stm) {
566604
if (stm->state == DRAINING && ms_since(&stm->drain_timeout) >= 0) {
567605
alsa_set_stream_state(stm, INACTIVE);
568-
stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_DRAINED);
606+
if (!stm->other_stream ||
607+
stm->stream_type == SND_PCM_STREAM_PLAYBACK) {
608+
stm->state_callback(stm, stm->user_ptr, CUBEB_STATE_DRAINED);
609+
}
569610
} else if (stm->state == RUNNING &&
570611
ms_since(&stm->last_activity) > CUBEB_WATCHDOG_MS) {
571612
alsa_set_stream_state(stm, ERROR);

‎test/test_callback_ret.cpp‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@ struct user_state_callback_ret {
3131
std::atomic<int> cb_count{0};
3232
std::atomic<int> expected_cb_count{0};
3333
std::atomic<int> error_state{0};
34+
std::atomic<int> stopped_state{0};
35+
std::atomic<int> drained_state{0};
3436
};
3537

3638
// Data callback that always returns 0
@@ -134,9 +136,11 @@ state_cb_ret(cubeb_stream * stream, void * user, cubeb_state state)
134136
break;
135137
case CUBEB_STATE_STOPPED:
136138
fprintf(stderr, "stream stopped\n");
139+
u->stopped_state.fetch_add(1);
137140
break;
138141
case CUBEB_STATE_DRAINED:
139142
fprintf(stderr, "stream drained\n");
143+
u->drained_state.fetch_add(1);
140144
break;
141145
case CUBEB_STATE_ERROR:
142146
fprintf(stderr, "stream error\n");
@@ -147,6 +151,46 @@ state_cb_ret(cubeb_stream * stream, void * user, cubeb_state state)
147151
}
148152
}
149153

154+
enum expected_terminal_state {
155+
NO_TERMINAL_STATE,
156+
DRAINED_STATE,
157+
SHORT_INPUT_CALLBACK
158+
};
159+
160+
expected_terminal_state
161+
expected_terminal_state_for_callback(test_direction direction,
162+
cubeb_data_callback data_cb)
163+
{
164+
if (data_cb == data_cb_ret_zero || data_cb == data_cb_ret_nframes_minus_one) {
165+
return direction == INPUT_ONLY ? SHORT_INPUT_CALLBACK : DRAINED_STATE;
166+
}
167+
return NO_TERMINAL_STATE;
168+
}
169+
170+
bool
171+
wait_for_drained_state(user_state_callback_ret & user_state)
172+
{
173+
for (int i = 0; i < 40; i++) {
174+
if (user_state.drained_state > 0) {
175+
return true;
176+
}
177+
delay(50);
178+
}
179+
return false;
180+
}
181+
182+
bool
183+
wait_for_callback(user_state_callback_ret & user_state)
184+
{
185+
for (int i = 0; i < 40; i++) {
186+
if (user_state.cb_count > 0) {
187+
return true;
188+
}
189+
delay(50);
190+
}
191+
return false;
192+
}
193+
150194
void
151195
run_test_callback(test_direction direction, cubeb_data_callback data_cb,
152196
const std::string & test_desc)
@@ -158,6 +202,10 @@ run_test_callback(test_direction direction, cubeb_data_callback data_cb,
158202
int r;
159203
user_state_callback_ret user_state;
160204
uint32_t latency_frames = 0;
205+
expected_terminal_state expected_state =
206+
expected_terminal_state_for_callback(direction, data_cb);
207+
int stopped_before_stop = 0;
208+
int drained_before_stop = 0;
161209

162210
r = common_init(&ctx, "Cubeb callback return value example");
163211
ASSERT_EQ(r, CUBEB_OK) << "Error initializing cubeb library";
@@ -211,7 +259,15 @@ run_test_callback(test_direction direction, cubeb_data_callback data_cb,
211259
cleanup_stream_at_exit(stream, cubeb_stream_destroy);
212260

213261
cubeb_stream_start(stream);
214-
delay(100);
262+
if (expected_state == NO_TERMINAL_STATE) {
263+
delay(100);
264+
} else if (expected_state == SHORT_INPUT_CALLBACK) {
265+
wait_for_callback(user_state);
266+
} else {
267+
wait_for_drained_state(user_state);
268+
}
269+
stopped_before_stop = user_state.stopped_state;
270+
drained_before_stop = user_state.drained_state;
215271
cubeb_stream_stop(stream);
216272

217273
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,
220276
if (data_cb == data_cb_ret_error && user_state.cb_count != 0) {
221277
ASSERT_EQ(user_state.error_state, 1) << "Callback expected error state";
222278
}
279+
if (expected_state == DRAINED_STATE && user_state.cb_count != 0) {
280+
ASSERT_EQ(drained_before_stop, 1)
281+
<< "Output-capable short callback must drain for " << test_desc << "!";
282+
ASSERT_EQ(stopped_before_stop, 0)
283+
<< "Output-capable short callback must not stop for " << test_desc
284+
<< "!";
285+
}
223286
}
224287

225288
TEST(cubeb, test_input_callback)

0 commit comments

Comments
 (0)