fix: send pushed audio from a thread apart from the timeouts

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
 

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.
+ * <p>
+ * 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 {
 

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<Chunk> 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) {

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()}.
 		 */

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<byte[]> 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<byte[]> 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<Void> 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<byte[]> 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));
+		}
+	}
+
 }