diff options
Diffstat (limited to 'app/src/main')
4 files changed, 154 insertions, 136 deletions
diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java b/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java index da0bb0a..eaf0db1 100644 --- a/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java +++ b/app/src/main/java/invalid/lena/scrcpy/AudioFrames.java @@ -4,6 +4,7 @@ package invalid.lena.scrcpy; // implementation; tests use a recording stub. Kept android-free. // // start(fourcc) tells the sink which wire codec it should expect. +// ptsUs is the wire presentation timestamp with protocol flags removed. // For raw payloads are interleaved s16le PCM. For opus the first // frame has isConfig=true and payload is the OpusHead (the server // pre-strips the AOPUSHDR/AOPUSDLY/AOPUSPRL container); subsequent @@ -12,7 +13,7 @@ public interface AudioFrames { void start(int fourcc); - void feed(byte[] data, int off, int len, boolean isConfig); + void feed(byte[] data, int off, int len, long ptsUs, boolean isConfig); void release(); } diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioSink.java b/app/src/main/java/invalid/lena/scrcpy/AudioSink.java index 5507c5c..f14f85c 100644 --- a/app/src/main/java/invalid/lena/scrcpy/AudioSink.java +++ b/app/src/main/java/invalid/lena/scrcpy/AudioSink.java @@ -14,6 +14,8 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.util.ArrayDeque; import java.util.Deque; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; // Audio output: AudioTrack writing 48 kHz stereo 16-bit PCM. The @@ -25,6 +27,7 @@ public final class AudioSink implements AudioFrames { private static final int CHANNEL_OUT = AudioFormat.CHANNEL_OUT_STEREO; private static final int ENCODING = AudioFormat.ENCODING_PCM_16BIT; private static final int MAX_PENDING_OPUS = 16; + private static final int MAX_PENDING_PCM = 16; private static final int MAX_PACKET_BYTES = 256 * 1024; // Defaults documented at <https://developer.android.com/reference/android/media/MediaCodec#CSD>. @@ -34,24 +37,29 @@ public final class AudioSink implements AudioFrames { private volatile MediaCodec opusCodec; private HandlerThread opusThread; private Handler opusHandler; + private Thread opusOutputThread; private volatile boolean opusConfigured; private int fourcc; private final Object lock = new Object(); - private final Runnable onFatalError; - private final AtomicBoolean fatalReported = new AtomicBoolean(); + private final AtomicBoolean failed = new AtomicBoolean(); private final AtomicBoolean playbackReported = new AtomicBoolean(); private final Deque<Integer> freeOpusInputs = new ArrayDeque<>(); - private final Deque<byte[]> pendingOpus = new ArrayDeque<>(MAX_PENDING_OPUS); + private final Deque<Packet> pendingOpus = new ArrayDeque<>(MAX_PENDING_OPUS); + private final ArrayBlockingQueue<byte[]> pendingPcm = + new ArrayBlockingQueue<>(MAX_PENDING_PCM); - private volatile boolean released; - private long droppedBytes; - private long lastDropLogMs; + private static final class Packet { + final byte[] data; + final long ptsUs; - public AudioSink(Runnable onFatalError) { - this.onFatalError = onFatalError; + Packet(byte[] data, long ptsUs) { + this.data = data; + this.ptsUs = ptsUs; + } } + private volatile boolean released; @Override public void start(int fourcc) { this.fourcc = fourcc; @@ -96,25 +104,31 @@ public final class AudioSink implements AudioFrames { } catch (IOException e) { throw new IllegalStateException("audio sink: no opus decoder", e); } - opusCodec.setCallback(new MediaCodec.Callback() { - @Override public void onInputBufferAvailable(MediaCodec mc, int idx) { + opusOutputThread = new Thread(this::playOpus, "audio-out"); + opusOutputThread.start(); + opusCodec.setCallback(new MediaCodec.Callback() { + @Override public void onInputBufferAvailable(MediaCodec mc, int idx) { synchronized (lock) { - if (released || mc != opusCodec) return; - byte[] packet = pendingOpus.pollFirst(); - if (packet == null) freeOpusInputs.offerLast(idx); - else submitOpus(mc, idx, packet); + if (released || failed.get() || mc != opusCodec) return; + Packet packet = pendingOpus.pollFirst(); + if (packet == null) { + freeOpusInputs.offerLast(idx); + } else { + lock.notifyAll(); + submitOpus(mc, idx, packet); + } } } @Override public void onOutputBufferAvailable(MediaCodec mc, int idx, MediaCodec.BufferInfo info) { + byte[] pcm = null; try { ByteBuffer out = mc.getOutputBuffer(idx); - if (out != null && info.size > 0 && !released) { - byte[] pcm = new byte[info.size]; + if (out != null && info.size > 0 && !released && !failed.get()) { + pcm = new byte[info.size]; out.position(info.offset); out.limit(info.offset + info.size); out.get(pcm); - writePcm(pcm, 0, pcm.length); } } catch (IllegalStateException e) { reportFatal(e, "audio sink: opus output"); @@ -122,6 +136,7 @@ public final class AudioSink implements AudioFrames { try { mc.releaseOutputBuffer(idx, false); } catch (IllegalStateException ignored) {} } + if (pcm != null) queuePcm(pcm); } @Override public void onError(MediaCodec mc, MediaCodec.CodecException e) { reportFatal(e, "audio sink: opus codec error"); @@ -134,7 +149,7 @@ public final class AudioSink implements AudioFrames { } @Override - public void feed(byte[] data, int off, int len, boolean isConfig) { + public void feed(byte[] data, int off, int len, long ptsUs, boolean isConfig) { if (data == null || off < 0 || len < 0 || off > data.length - len) { throw new IndexOutOfBoundsException("invalid audio frame range"); } @@ -142,16 +157,16 @@ public final class AudioSink implements AudioFrames { reportFatal(null, "audio sink: packet too large (" + len + " bytes)"); return; } - if (released || len == 0) return; + if (released || failed.get() || len == 0) return; if (fourcc == Wire.CODEC_OPUS) { - feedOpus(data, off, len, isConfig); + feedOpus(data, off, len, ptsUs, isConfig); } else { // Raw PCM: passthrough. writePcm(data, off, len); } } - private void feedOpus(byte[] data, int off, int len, boolean isConfig) { + private void feedOpus(byte[] data, int off, int len, long ptsUs, boolean isConfig) { if (isConfig) { if (opusConfigured) return; // already configured // OpusHead is 19 bytes; pre_skip lives at bytes [10..11]. Reject @@ -183,31 +198,45 @@ public final class AudioSink implements AudioFrames { if (!opusConfigured) return; byte[] packet = new byte[len]; System.arraycopy(data, off, packet, 0, len); + Packet p = new Packet(packet, ptsUs); synchronized (lock) { if (released || opusCodec == null) return; + // Apply socket backpressure during decoder bursts. Dropping a + // compressed packet corrupts playback; growing the queue only + // postpones the same failure. + while (!released && !failed.get() && freeOpusInputs.isEmpty() + && pendingOpus.size() == MAX_PENDING_OPUS) { + try { + lock.wait(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return; + } + } + if (released || failed.get() || opusCodec == null) return; Integer idx = freeOpusInputs.pollFirst(); if (idx != null) { - submitOpus(opusCodec, idx, packet); + submitOpus(opusCodec, idx, p); return; } - if (pendingOpus.size() == MAX_PENDING_OPUS) pendingOpus.pollFirst(); - pendingOpus.offerLast(packet); + pendingOpus.offerLast(p); } } // Must be called with lock held. Async MediaCodec input indices belong // to the codec instance that delivered them. - private void submitOpus(MediaCodec codec, int idx, byte[] packet) { + private void submitOpus(MediaCodec codec, int idx, Packet packet) { try { ByteBuffer in = codec.getInputBuffer(idx); - if (in == null || packet.length > in.capacity()) { + if (in == null || packet.data.length > in.capacity()) { reportFatal(null, - "audio sink: opus packet exceeds codec input (" + packet.length + " bytes)"); + "audio sink: opus packet exceeds codec input (" + + packet.data.length + " bytes)"); return; } in.clear(); - in.put(packet); - codec.queueInputBuffer(idx, 0, packet.length, 0, 0); + in.put(packet.data); + codec.queueInputBuffer(idx, 0, packet.data.length, packet.ptsUs, 0); } catch (IllegalStateException e) { reportFatal(e, "audio sink: opus queueInputBuffer"); } @@ -220,32 +249,60 @@ public final class AudioSink implements AudioFrames { return b; } - private void writePcm(byte[] data, int off, int len) { - AudioTrack t = track; - if (released || t == null || len <= 0) return; - int written = t.write(data, off, len, AudioTrack.WRITE_NON_BLOCKING); - if (written < 0) { - reportFatal(null, "audio sink: AudioTrack write rc=" + written); - return; + private void queuePcm(byte[] pcm) { + try { + while (!released && !failed.get() + && !pendingPcm.offer(pcm, 100, TimeUnit.MILLISECONDS)) { + // Wait for the blocking AudioTrack writer to make room. + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - if (written > 0 && playbackReported.compareAndSet(false, true)) { - Log.i("audio sink: playback started"); + } + + private void playOpus() { + try { + while (!released && !failed.get()) { + byte[] pcm = pendingPcm.take(); + writePcm(pcm, 0, pcm.length); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } catch (RuntimeException e) { + reportFatal(e, "audio sink: AudioTrack write"); } - if (written < len) { - droppedBytes += (len - written); - long now = System.currentTimeMillis(); - if (now - lastDropLogMs > 1000L) { - lastDropLogMs = now; - Log.w("audio sink: ring full, dropped %d bytes so far", droppedBytes); + } + + private void writePcm(byte[] data, int off, int len) { + AudioTrack t = track; + if (released || failed.get() || t == null || len <= 0) return; + int end = off + len; + while (!released && !failed.get() && off < end) { + int written = t.write(data, off, end - off, AudioTrack.WRITE_BLOCKING); + if (written <= 0) { + if (!released) { + reportFatal(null, "audio sink: AudioTrack write rc=" + written); + } + return; + } + off += written; + if (playbackReported.compareAndSet(false, true)) { + Log.i("audio sink: playback started"); } } } private void reportFatal(Exception error, String message) { - if (!fatalReported.compareAndSet(false, true)) return; + if (!failed.compareAndSet(false, true)) return; if (error == null) Log.e("%s", message); else Log.e(error, "%s", message); - if (onFatalError != null) onFatalError.run(); + synchronized (lock) { + pendingOpus.clear(); + lock.notifyAll(); + } + pendingPcm.clear(); + Thread t = opusOutputThread; + if (t != null) t.interrupt(); } @Override @@ -257,20 +314,35 @@ public final class AudioSink implements AudioFrames { opusCodec = null; freeOpusInputs.clear(); pendingOpus.clear(); + lock.notifyAll(); } + AudioTrack t = track; + track = null; + // Stop playback first so the blocking writer can return before + // AudioTrack is released. + if (t != null) { + try { t.pause(); t.flush(); t.stop(); } catch (Exception ignored) {} + } + HandlerThread ht = opusThread; opusThread = null; opusHandler = null; + Thread out = opusOutputThread; + opusOutputThread = null; + if (out != null) out.interrupt(); if (c != null) { try { c.stop(); } catch (Exception ignored) {} try { c.release(); } catch (Exception ignored) {} } if (ht != null) ht.quitSafely(); - - AudioTrack t = track; - track = null; - if (t == null) return; - try { t.pause(); t.flush(); t.stop(); } catch (Exception ignored) {} - try { t.release(); } catch (Exception ignored) {} + if (out != null && out != Thread.currentThread()) { + try { out.join(1_000); } + catch (InterruptedException e) { Thread.currentThread().interrupt(); } + if (out.isAlive()) Log.e("audio sink: output thread did not stop"); + } + pendingPcm.clear(); + if (t != null) { + try { t.release(); } catch (Exception ignored) {} + } } } diff --git a/app/src/main/java/invalid/lena/scrcpy/AudioStream.java b/app/src/main/java/invalid/lena/scrcpy/AudioStream.java index 13b39f5..45ea69c 100644 --- a/app/src/main/java/invalid/lena/scrcpy/AudioStream.java +++ b/app/src/main/java/invalid/lena/scrcpy/AudioStream.java @@ -2,7 +2,6 @@ package invalid.lena.scrcpy; import java.io.IOException; import java.io.InputStream; -import java.util.concurrent.TimeUnit; // Reads the scrcpy audio socket and drives an AudioFrames sink. // @@ -19,7 +18,10 @@ import java.util.concurrent.TimeUnit; // Takes a plain InputStream; caller owns stream lifecycle. public final class AudioStream { - private static final long FLAG_CONFIG = 1L << 62; + private static final long FLAG_SESSION = 1L << 63; + private static final long FLAG_CONFIG = 1L << 62; + private static final long FLAG_KEYFRAME = 1L << 61; + private static final long PTS_MASK = ~(FLAG_SESSION | FLAG_CONFIG | FLAG_KEYFRAME); // Generous upper bound for one audio packet (a raw PCM block or an // opus packet is a few KB). A corrupt or hostile length field must @@ -28,30 +30,12 @@ public final class AudioStream { private final InputStream source; private final AudioFrames sink; - private final Runnable onFatalError; private Thread thread; private volatile boolean stop; - // Liveness signal for Session's stall watchdog. The audio socket is - // the only one that is reliably continuous: the server captures PCM - // at 48 kHz whether or not anything on the target's screen moves, so - // silence here means the wire is gone. Video is useless for this - a - // static screen queues nothing to the encoder and legitimately - // produces no frames for minutes. - private volatile boolean started; - private volatile long lastPacketAtMs; - - public boolean isStarted() { return started; } - public long lastPacketAtMs() { return lastPacketAtMs; } - public AudioStream(InputStream source, AudioFrames sink) { - this(source, sink, null); - } - - public AudioStream(InputStream source, AudioFrames sink, Runnable onFatalError) { this.source = source; this.sink = sink; - this.onFatalError = onFatalError; } public void start() { @@ -85,9 +69,13 @@ public final class AudioStream { throw new IOException("audio: unexpected codec " + Wire.fourccName(fourcc)); } Log.i("audio meta codec=%s", Wire.fourccName(fourcc)); - sink.start(fourcc); - lastPacketAtMs = monotonicMs(); - started = true; + boolean playback = true; + try { + sink.start(fourcc); + } catch (RuntimeException e) { + Log.e(e, "audio output disabled"); + playback = false; + } byte[] hdr = new byte[12]; byte[] payload = new byte[16 * 1024]; @@ -97,33 +85,28 @@ public final class AudioStream { long ptsAndFlags = Wire.readBe64(hdr, 0); int size = Wire.readBe32(hdr, 8); boolean cfg = (ptsAndFlags & FLAG_CONFIG) != 0; + long ptsUs = ptsAndFlags & PTS_MASK; if (size <= 0 || size > MAX_FRAME_SIZE) { throw new IOException("audio frame size out of range: " + size); } if (size > payload.length) payload = new byte[size]; Wire.readFully(source, payload, 0, size); - lastPacketAtMs = monotonicMs(); - sink.feed(payload, 0, size, cfg); + if (playback) { + try { + sink.feed(payload, 0, size, ptsUs, cfg); + } catch (RuntimeException e) { + Log.e(e, "audio output disabled"); + playback = false; + } + } if (++frames == 1) Log.i("audio frame n=1 size=%d cfg=%s", size, cfg); } } catch (IOException e) { - if (!stop) { - Log.e(e, "audio reader"); - reportFatal(); - } + if (!stop) Log.e(e, "audio reader"); } catch (Exception e) { Log.e(e, "audio reader unexpected"); - if (!stop) reportFatal(); } finally { Log.i("audio reader: end"); } } - - private void reportFatal() { - if (onFatalError != null) onFatalError.run(); - } - - private static long monotonicMs() { - return TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); - } } diff --git a/app/src/main/java/invalid/lena/scrcpy/Session.java b/app/src/main/java/invalid/lena/scrcpy/Session.java index 294b27d..39ef38a 100644 --- a/app/src/main/java/invalid/lena/scrcpy/Session.java +++ b/app/src/main/java/invalid/lena/scrcpy/Session.java @@ -67,22 +67,8 @@ public final class Session { // deterministic failure walks the ladder and then gives up. private static final long[] RECONNECT_BACKOFF_MS = {1_000L, 2_000L, 5_000L, 10_000L, 15_000L}; - // Comfortably longer than STALL_MS + STALL_POLL_MS. If it were not, - // a stall-detected drop would always look "healthy" (a stall is only - // declared after STALL_MS of silence, so the connection is at least - // that old by then), the budget would reset on every stall and the - // ladder would never be reached for the one failure it exists for. private static final long HEALTHY_MS = 120_000L; - // Stall watchdog. adb streams have no read timeout - AdbStream.read() - // waits on its queue until data arrives or the stream is closed - so a - // target that disappears without closing the socket (Wi-Fi dropping - // mid-frame, NAT rebinding, a carrier idle timeout on a VPN link) - // leaves the readers parked and the session frozen forever with no - // reconnect. Poll for silence on the audio socket instead; see - // AudioStream.isStarted() for why audio and not video. - private static final long STALL_MS = 30_000L; - private static final long STALL_POLL_MS = 5_000L; private static final long BRING_UP_DEADLINE_MS = 90_000L; private static final long STOP_JOIN_MS = 10_000L; @@ -118,9 +104,8 @@ public final class Session { } } - // Counted down by the video reader when its loop exits; the session - // thread parks on it for the live duration of a connection. Swapped - // for a fresh latch on each reconnect cycle. + // Counted down when video or control ends; audio is optional and never + // ends a session. Swapped for a fresh latch on each reconnect cycle. private volatile CountDownLatch endSignal = new CountDownLatch(1); public Session(Context ctx, Adb adb, Devices.Device target, Surface surface, Listener listener) { @@ -242,14 +227,12 @@ public final class Session { if (!connect()) return; // gave up: onError + onStopped fired long upAt = monotonicMs(); CountDownLatch latch; - AudioStream as; synchronized (this) { if (stopped) return; latch = endSignal; - as = audioStream; } try { - awaitEndOrStall(latch, as); + latch.await(); } catch (InterruptedException ie) { return; } @@ -286,29 +269,6 @@ public final class Session { } } - // Park until the pipeline dies, stop() fires, or the audio socket has - // been silent long enough that the link must be gone. Returning - // without the latch firing leaves the readers parked; the caller's - // tearDownInstalled() closes their streams, which unblocks them. - // - // The watchdog only arms once audio is actually flowing. If the - // target cannot capture audio the server reports the stream disabled, - // AudioStream returns immediately, and there is no reliable idle - // signal left - in that case fall back to waiting indefinitely rather - // than inventing one from the video socket, which is legitimately - // silent whenever the target's screen is static. - private void awaitEndOrStall(CountDownLatch latch, AudioStream as) - throws InterruptedException { - while (!latch.await(STALL_POLL_MS, TimeUnit.MILLISECONDS)) { - if (stopped || as == null || !as.isStarted()) continue; - long idle = monotonicMs() - as.lastPacketAtMs(); - if (idle >= STALL_MS) { - Log.w("session: no audio for %d ms - link presumed dead", idle); - return; - } - } - } - // Run the bring-up retry ladder once. Returns true when a session is // live (read threads started); false if the budget was exhausted, in // which case onError() + onStopped() have already fired. @@ -429,8 +389,10 @@ public final class Session { ctrl = new Controller(ctx, cs::send); cs.setInboundSink(ctrl); - ak = new AudioSink(ended); - as = new AudioStream(s.audioIn, ak, ended); + // Audio is optional. Failure to capture or play it must not tear + // down video and control. + ak = new AudioSink(); + as = new AudioStream(s.audioIn, ak); vk = new VideoSink(surface, ended); Controller ctrlRef = ctrl; |