diff --git a/docs/guide/client-api.md b/docs/guide/client-api.md index 865dedbe..4f03d605 100644 --- a/docs/guide/client-api.md +++ b/docs/guide/client-api.md @@ -407,7 +407,7 @@ CustomAudio voice = media.customAudio("voice"); voice.push(pcm, PcmFormat.of(48000, 1)); ``` -`CustomAudio` takes PCM in pieces of any length and from any thread. It cuts the audio into the 10 ms chunks WebRTC works with and sends one every 10 ms from a thread of its own, so pushing a whole second at once is as fine as pushing every 10 ms. `queuedMillis()` tells how much is still waiting, and `clear()` drops it. +`CustomAudio` takes PCM in pieces of any length and from any thread. It cuts the audio into the 10 ms chunks WebRTC works with and sends one every 10 ms from a thread of its own, so pushing a whole second at once is as fine as pushing every 10 ms. `queuedMillis()` tells how much is still waiting, and `clear()` drops it. That thread does nothing else, so a slow push into WebRTC delays only the audio, not the timeouts of the runtime. Should it stall, as in a long garbage collection pause, it makes up for at most the last 50 ms instead of sending the whole pause at once; the queued audio then stays behind by the rest, until `clear()` drops it. ### Controlling Senders and Tracks diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudio.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudio.java index bbeaa2e9..d21e4fac 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudio.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudio.java @@ -23,6 +23,11 @@ * Audio may be pushed in pieces of any length, from any thread. The track * cuts it into 10 ms chunks and sends one every 10 ms from a thread of its * own, so pushing a whole second at once is as fine as pushing every 10 ms. + *

+ * Should that thread stall, such as in a long garbage collection pause, it + * makes up for at most the last 50 ms of the stall. The audio of the rest is + * not sent faster but later, so what is queued then stays that much behind, + * unless {@link #clear()} drops it. */ public interface CustomAudio extends LocalAudio { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java index a9d09633..295150a8 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/CustomAudioImpl.java @@ -28,14 +28,34 @@ /** * An audio track fed with audio the application pushes. Pushed audio is cut - * into 10 ms chunks, which the runtime's timer thread hands to the source one - * every 10 ms. That thread is the only one that pushes into the source, as - * {@link CustomAudioSource} requires. + * into 10 ms chunks, which the runtime's audio pacer hands to the source one + * every 10 ms. Its thread is the only one that pushes into the source, as + * {@link CustomAudioSource} requires, and it runs no timeouts, which a slow + * push would hold up. */ final class CustomAudioImpl extends LocalAudioImpl implements CustomAudio { + private static final long TICK_NANOS = TimeUnit.MILLISECONDS.toNanos(10); + + /** + * How late a tick may run and still send. The pacer runs at a fixed rate, + * so after a stall it runs the missed ticks back to back: those up to this + * late are made up for, so that a short stall keeps the pace, and the + * earlier ones are skipped, so that a long one does not burst out all the + * audio of the pause at once. Skipped ticks leave the queue as it is. + */ + private static final long MAX_LATE_NANOS = TimeUnit.MILLISECONDS.toNanos(50); + private final CustomAudioSource source; + // Confined to the pacer, once the ticker is scheduled. + + /** When the first tick was due. */ + private long start; + + /** The ticks run so far. */ + private long ticks; + // Guarded by queue. private final ArrayDeque queue = new ArrayDeque<>(); @@ -98,7 +118,8 @@ public void push(byte[] pcm, PcmFormat pcmFormat) { } if (ticker == null) { - ticker = rtc.timer.scheduleAtFixedRate(this::tick, 0, 10, TimeUnit.MILLISECONDS); + start = rtc.clock.getAsLong(); + ticker = rtc.audioPacer.scheduleAtFixedRate(this::tick, 0, TICK_NANOS, TimeUnit.NANOSECONDS); } } } @@ -144,6 +165,12 @@ private void flushPartial() { } private void tick() { + long late = rtc.clock.getAsLong() - (start + ticks++ * TICK_NANOS); + + if (late > MAX_LATE_NANOS) { + return; + } + Chunk chunk; synchronized (queue) { diff --git a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java index 1fb3e468..0c9f7860 100644 --- a/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java +++ b/webrtc/src/main/java/dev/onvoid/webrtc/client/WebRtc.java @@ -102,6 +102,15 @@ public final class WebRtc implements AutoCloseable { final ScheduledExecutorService timer; + /** + * The one thread that sends pushed audio, apart from the timer: a slow + * push must not delay the timeouts. + */ + final ScheduledExecutorService audioPacer; + + /** The time in nanoseconds, as {@link System#nanoTime()}. */ + final LongSupplier clock; + final Duration operationTimeout; /** How much audio an {@code onAudio} subscription buffers by default. */ @@ -166,6 +175,9 @@ private WebRtc(Builder builder) { this.framePool = pool(builder.frameThreads, "webrtc-client-frames-" + runtime + "-"); this.timer = builder.timer != null ? builder.timer : Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-timer-" + runtime + "-")); + this.audioPacer = builder.audioPacer != null ? builder.audioPacer + : Executors.newSingleThreadScheduledExecutor(threads("webrtc-client-audio-" + runtime + "-")); + this.clock = builder.clock; if (builder.eventExecutor != null) { this.eventExecutor = builder.eventExecutor; @@ -570,6 +582,7 @@ private void shutdownPools() { controlPool.shutdown(); framePool.shutdown(); timer.shutdownNow(); + audioPacer.shutdownNow(); if (ownsEventExecutor) { ((ExecutorService) eventExecutor).shutdown(); @@ -648,6 +661,8 @@ public static final class Builder { private ScheduledExecutorService timer; + private ScheduledExecutorService audioPacer; + private LongSupplier clock = System::nanoTime; @@ -850,8 +865,18 @@ Builder timer(ScheduledExecutorService timer) { } /** - * Replaces the clock of the load statistics and of the turns of the - * executors, so that tests decide how much time passes. + * Replaces the thread that sends pushed audio, so that tests decide + * when each chunk is due. The runtime shuts it down when it closes. + */ + Builder audioPacer(ScheduledExecutorService pacer) { + this.audioPacer = requireNonNull(pacer, "pacer"); + return this; + } + + /** + * Replaces the clock of the load statistics, of the turns of the + * executors and of the audio pacing, so that tests decide how much + * time passes. * * @param nanos the time in nanoseconds, as {@link System#nanoTime()}. */ diff --git a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java b/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java index d44de35d..6485cb3b 100644 --- a/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java +++ b/webrtc/src/test/java/dev/onvoid/webrtc/client/MediaTests.java @@ -39,6 +39,8 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; import dev.onvoid.webrtc.RTCStats; import dev.onvoid.webrtc.RTCStatsType; @@ -90,31 +92,17 @@ void pushedAudioArrives() throws Exception { void clearedAudioNotResent() throws Exception { // A format change sends the chunk begun before it, filled with // silence, not with the audio that clear() dropped from it. - ManualTimer timer = new ManualTimer(); + ManualTimer pacer = new ManualTimer(); WebRtc runtime = WebRtc.builder() .audio(AudioMode.HEADLESS) .errorHandler(errors::add) - .timer(timer) + .audioPacer(pacer) .build(); - LocalMediaImpl media = (LocalMediaImpl) runtime.media(); List sent = new ArrayList<>(); - CustomAudioImpl voice; - - try (LocalMediaImpl.Creation creation = media.begin()) { - CustomAudioSource source = new CustomAudioSource() { - - @Override - public void pushAudio(byte[] audioData, int bitsPerSample, int sampleRate, int channels, - int frameCount) { - sent.add(audioData.clone()); - } - }; - - voice = media.register(new CustomAudioImpl(media, runtime.factory.createAudioTrack("voice", source), - source)); - } try { + CustomAudioImpl voice = recordingAudio(runtime, sent::add); + // 160 bytes in 10 ms. PcmFormat format = PcmFormat.of(8000, 1); byte[] dropped = new byte[120]; @@ -127,7 +115,7 @@ public void pushAudio(byte[] audioData, int bitsPerSample, int sampleRate, int c voice.push(kept, format); voice.push(new byte[2], PcmFormat.of(16000, 1)); - timer.fire(Duration.ofMillis(10)); + pacer.fire(Duration.ofMillis(10)); assertEquals(1, sent.size()); assertEquals(160, sent.get(0).length); @@ -141,6 +129,83 @@ public void pushAudio(byte[] audioData, int bitsPerSample, int sampleRate, int c } } + @Test + void stalledAudioNotBurst() throws Exception { + // After a stall the pacer runs the missed ticks back to back. Those + // at most 50 ms late send, the earlier ones leave the audio queued. + ManualTimer pacer = new ManualTimer(); + AtomicLong now = new AtomicLong(); + WebRtc runtime = WebRtc.builder() + .audio(AudioMode.HEADLESS) + .errorHandler(errors::add) + .audioPacer(pacer) + .clock(now::get) + .build(); + List sent = new ArrayList<>(); + + try { + CustomAudioImpl voice = recordingAudio(runtime, sent::add); + Duration tick = Duration.ofMillis(10); + + // One second, 100 chunks of 160 bytes. + voice.push(new byte[16000], PcmFormat.of(8000, 1)); + pacer.fire(tick); + + assertEquals(1, sent.size()); + + // Stalled for a second: ticks 1 to 100 are due. + now.set(TimeUnit.SECONDS.toNanos(1)); + + for (int i = 0; i < 100; i++) { + pacer.fire(tick); + } + + // Ticks 95 to 100, due from 950 ms on. + assertEquals(7, sent.size()); + assertEquals(930, voice.queuedMillis()); + + // Back on pace. + now.addAndGet(tick.toNanos()); + pacer.fire(tick); + + assertEquals(8, sent.size()); + } + finally { + runtime.close(); + } + } + + @Test + void slowPushLeavesTimerFree() throws Exception { + // Pushed audio is sent on a thread of its own: a push that hangs in + // WebRTC holds up the audio, not the timeouts on the timer. + CountDownLatch pushing = new CountDownLatch(1); + CountDownLatch unblock = new CountDownLatch(1); + CustomAudioImpl voice = recordingAudio(rtc, data -> { + pushing.countDown(); + + try { + unblock.await(); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + + try { + voice.push(new byte[160], PcmFormat.of(8000, 1)); + assertTrue(pushing.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + CompletableFuture fired = new CompletableFuture<>(); + rtc.timer.schedule(() -> fired.complete(null), 1, TimeUnit.MILLISECONDS); + + await(fired); + } + finally { + unblock.countDown(); + } + } + @Test void oversizedFormatRejected() { // 10 ms of these hold 30720 samples, more than WebRTC takes at once. @@ -470,4 +535,23 @@ private static byte[] tone(int sampleRate, int channels, int millis) { return pcm; } + /** A pushed-audio track whose source hands each chunk to the consumer. */ + private static CustomAudioImpl recordingAudio(WebRtc runtime, Consumer chunks) { + LocalMediaImpl media = (LocalMediaImpl) runtime.media(); + + try (LocalMediaImpl.Creation creation = media.begin()) { + CustomAudioSource source = new CustomAudioSource() { + + @Override + public void pushAudio(byte[] audioData, int bitsPerSample, int sampleRate, int channels, + int frameCount) { + chunks.accept(audioData.clone()); + } + }; + + return media.register(new CustomAudioImpl(media, runtime.factory.createAudioTrack("voice", source), + source)); + } + } + }