From b98205ce48fad74ad74913b393598116f3610c91 Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Wed, 21 Oct 2026 13:08:15 +0000 Subject: [PATCH 1/5] fix: udp:// slots read the socket on a thread of their own; no slot checks during a push (#251) An app cannot get a socket buffer above rmem_max (256 KiB on the Aceline box), about one 1080p key frame, so any stall of the receive loop lost datagrams of the burst. A high-priority thread now only reads the socket into a ring of 1024 datagrams; reordering, assembly and the decoder run as before on the session thread. Stream checks of other slots in the U view pause while the U cell receives a push. /stats.json gets a udp object per cell: rcvbuf, received, ring_dropped, rtp_lost and socket_drops where /proc/net/udp is readable. --- .../java/com/openipc/decoder/Decoder.java | 19 +- .../java/com/openipc/decoder/QuadCell.java | 4 + .../java/com/openipc/decoder/RtpListener.java | 171 ++++++++++++++++-- .../java/com/openipc/decoder/RtspSession.java | 13 ++ .../com/openipc/decoder/RtpListenerTest.java | 39 ++++ docs/debug.md | 3 +- docs/transports.md | 1 + 7 files changed, 229 insertions(+), 21 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/Decoder.java b/android/app/src/main/java/com/openipc/decoder/Decoder.java index 562afeb3..73e45768 100755 --- a/android/app/src/main/java/com/openipc/decoder/Decoder.java +++ b/android/app/src/main/java/com/openipc/decoder/Decoder.java @@ -1697,6 +1697,9 @@ private Map debugState() { if (jl != null) c.put("jieli", jl); Map hb = cell.heartbeatStats(); if (hb != null) c.put("heartbeat", hb); + // A udp:// push: a full socket buffer or ring, or loss on the way (#251) + Map udp = cell.udpListenStats(); + if (udp != null) c.put("udp", udp); if (cell.rotate() != 0) c.put("rotate", cell.rotate()); if (cell.mirror()) { c.put("mirror", true); @@ -1724,7 +1727,17 @@ private void updateProbes() { ? mHosts[i] : null; } streamProbes.update(urls, mTransports, udpViewShown ? null : mHost, - mResumed && (udpViewShown || !quadEnabled)); + mResumed && (udpViewShown ? !pushReceived() : !quadEnabled)); + } + + /** + * The U cell receives a push: no checks of other cameras then, their RTSP sessions made + * the box drop datagrams of the push (#251). + */ + private boolean pushReceived() { + QuadCell[] cells = quadCells; + QuadCell u = udpViewShown && cells != null ? cells[U_CELL] : null; + return u != null && u.isConnected(); } /** @@ -1754,7 +1767,9 @@ private String probeText(int i, long nowMs) { if (skip != null) return skip; StreamProbe.Result r = probeResult(i); if (r != null) return r.text(nowMs) + (streamProbes.isRunning(i) ? " \u00b7 checking" : ""); - return streamProbes.isRunning(i) ? "Checking stream\u2026" : "Waiting for a stream check"; + if (streamProbes.isRunning(i)) return "Checking stream\u2026"; + return pushReceived() ? "Not checked while the U cell receives a push" + : "Waiting for a stream check"; } /** diff --git a/android/app/src/main/java/com/openipc/decoder/QuadCell.java b/android/app/src/main/java/com/openipc/decoder/QuadCell.java index 2135ad6d..74030fea 100644 --- a/android/app/src/main/java/com/openipc/decoder/QuadCell.java +++ b/android/app/src/main/java/com/openipc/decoder/QuadCell.java @@ -438,6 +438,7 @@ static boolean isKeyFrameNal(int type, boolean h265) { /** The cooingdv heartbeat of an RTSP cell once it used one (#187), else null. */ java.util.Map heartbeatStats() { return session.heartbeatStats(); } + java.util.Map udpListenStats() { return session.udpListenStats(); } /** Degrees counter-clockwise the frames are turned (#200, #201), 0 for none. */ int rotate() { return session.rotate(); } @@ -520,6 +521,9 @@ private void checkUdpStall() { * Connected, not paused and a frame shown within the last {@code ms} milliseconds (#98); * on key frames within KEY_FRAMES_LIVE_MS, or the cell would blink between key frames. */ + /** A stream is coming in: "connected", paused or not. */ + boolean isConnected() { return "connected".equals(state); } + boolean showsVideo(long ms) { long window = keyFramesOnly ? Math.max(ms, KEY_FRAMES_LIVE_MS) : ms; return pausedReason == null && "connected".equals(state) && renderedWithin(window); diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java index e92ed1ff..d1db7633 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpListener.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -11,14 +11,20 @@ import android.net.wifi.WifiManager; import android.util.Log; +import java.io.BufferedReader; import java.io.Closeable; +import java.io.FileReader; import java.io.IOException; import java.net.DatagramPacket; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.MulticastSocket; import java.net.SocketTimeoutException; +import java.util.LinkedHashMap; import java.util.Locale; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.TimeUnit; /** * Receives the stream that Majestic sends to "outgoing" destinations (udp://host:port): @@ -26,10 +32,18 @@ * before every IDR, so it can be joined at any time without an SDP (#68). * A MulticastSocket is used for unicast too: joining a group from a DatagramChannel needs * API 24. + *

+ * An app cannot get a socket buffer above {@code rmem_max} (256 KiB on many boxes), which is + * about one 1080p key frame, so a thread of its own does nothing but read the socket into a + * ring; reordering, assembly and the decoder never hold up the receive (#251). */ class RtpListener implements Closeable { private static final String TAG = "OpenIPCDecoder"; private static final int RCVBUF = 2 * 1024 * 1024; + /** Datagrams held between the receive thread and the rest: a few key frames. */ + static final int RING = 1024; + /** Starting size of a ring slot; one grows for a longer datagram. */ + private static final int SLOT = 1600; interface Handler { void onPacket(byte[] buf, int offset, int length); @@ -152,9 +166,17 @@ boolean h265(int pt) { * numbers, which must not push video packets out as late ones. */ private final RtpReorder[] reorders = new RtpReorder[128]; + private final ArrayBlockingQueue free = new ArrayBlockingQueue<>(RING); + private final ArrayBlockingQueue filled = new ArrayBlockingQueue<>(RING); private volatile boolean closed; private boolean joined; private long rejected; + /** The buffer the kernel actually gave, as getReceiveBufferSize() says. */ + private final int rcvbuf; + private volatile long received; + /** Datagrams thrown away because the ring was full. */ + private volatile long ringDropped; + private volatile IOException readError; RtpListener(String tag, Url url) throws IOException { this.tag = tag; @@ -183,41 +205,100 @@ boolean h265(int pt) { throw e; } this.socket = s; + int got; + try { + got = s.getReceiveBufferSize(); + } catch (IOException e) { + got = -1; + } + rcvbuf = got; + Log.i(TAG, tag + ": UDP port " + url.port + ", receive buffer " + got + " bytes (asked " + + RCVBUF + ")"); + for (int i = 0; i < RING; i++) free.add(new byte[SLOT + 4]); if (joined) MulticastLock.acquire(); } - /** Receive until {@code keepGoing} turns false or {@link #close} is called. */ - void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { + /** + * Reads the socket until it is closed. A slot is a datagram with its length in the first + * four bytes, copied from one large buffer: datagrams are short, slots stay small. + */ + private void readLoop() { byte[] buf = new byte[65536]; DatagramPacket p = new DatagramPacket(buf, buf.length); - RtpReorder.Sink sink = handler::onPacket; InetAddress source = url.source; + try { + while (!closed) { + try { + p.setLength(buf.length); + socket.receive(p); + } catch (SocketTimeoutException e) { + continue; + } + if (source != null && !source.equals(p.getAddress())) { + if (rejected++ % 100 == 0) { + Log.w(TAG, tag + ": UDP from an unexpected sender ignored (" + + rejected + ")"); + } + continue; + } + int len = p.getLength(); + if (len < 12) continue; + received++; + byte[] slot = free.poll(); + if (slot == null) { + if (ringDropped++ % 100 == 0) { + Log.w(TAG, tag + ": UDP ring full, datagram dropped (" + ringDropped + ")"); + } + continue; + } + if (slot.length < len + 4) slot = new byte[len + 4]; + slot[0] = (byte) (len >>> 24); + slot[1] = (byte) (len >>> 16); + slot[2] = (byte) (len >>> 8); + slot[3] = (byte) len; + System.arraycopy(buf, p.getOffset(), slot, 4, len); + filled.offer(slot); + } + } catch (IOException e) { + if (!closed) readError = e; + } + } + + /** Receive until {@code keepGoing} turns false or {@link #close} is called. */ + void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { + RtpReorder.Sink sink = handler::onPacket; + Thread reader = new Thread(this::readLoop, tag + "-udp"); + reader.setDaemon(true); + // Above the decoder and the UI: a stalled receive is a lost packet + reader.setPriority(Thread.MAX_PRIORITY); + reader.start(); try { while (!closed && keepGoing.holds()) { + IOException e = readError; + if (e != null) throw e; long now = System.nanoTime(); long ms = 250; for (RtpReorder r : reorders) { long d = r != null ? r.msUntilDeadline(now) : -1; if (d >= 0) ms = Math.min(ms, d); } - socket.setSoTimeout((int) Math.max(1, ms)); + byte[] slot; try { - p.setLength(buf.length); - socket.receive(p); - if (source != null && !source.equals(p.getAddress())) { - if (rejected++ % 100 == 0) { - Log.w(TAG, tag + ": UDP from an unexpected sender ignored (" - + rejected + ")"); - } - continue; - } - if (p.getLength() < 12) continue; - int pt = buf[p.getOffset() + 1] & 0x7F; + slot = filled.poll(Math.max(1, ms), TimeUnit.MILLISECONDS); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + break; + } + // Everything waiting, then the reorder deadlines + while (slot != null) { + int len = ((slot[0] & 0xFF) << 24) | ((slot[1] & 0xFF) << 16) + | ((slot[2] & 0xFF) << 8) | (slot[3] & 0xFF); + int pt = slot[5] & 0x7F; RtpReorder r = reorders[pt]; if (r == null) r = reorders[pt] = new RtpReorder(64, UdpReceiver.REORDER_MS); - r.push(buf, p.getOffset(), p.getLength(), System.nanoTime(), sink); - } catch (SocketTimeoutException ignored) { - // Check the condition and the reorder deadlines + r.push(slot, 4, len, System.nanoTime(), sink); + free.offer(slot); + slot = filled.poll(); } now = System.nanoTime(); for (RtpReorder r : reorders) { @@ -238,6 +319,60 @@ void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { } } + /** + * For /stats.json (#251): the buffer the kernel gave, datagrams read, dropped by a full + * ring, missing from the sequence of the video payload type {@code videoPt} (-1: every + * type), and the kernel's drop count of the port when the app may read it. + */ + Map stats(int videoPt) { + Map m = new LinkedHashMap<>(); + m.put("rcvbuf", rcvbuf); + m.put("received", received); + m.put("ring_dropped", ringDropped); + long lost = 0; + for (int pt = 0; pt < reorders.length; pt++) { + RtpReorder r = reorders[pt]; + if (r != null && (videoPt < 0 || pt == videoPt)) lost += r.skipped(); + } + m.put("rtp_lost", lost); + long drops = socketDrops(url.port); + if (drops >= 0) m.put("socket_drops", drops); + return m; + } + + /** + * The kernel's drop count of the UDP sockets on {@code port}, from /proc/net/udp and udp6, + * or -1: apps of newer Android versions may not read them. + */ + static long socketDrops(int port) { + long total = -1; + for (String file : new String[] {"/proc/net/udp", "/proc/net/udp6"}) { + long d = socketDrops(file, port); + if (d >= 0) total = Math.max(0, total) + d; + } + return total; + } + + static long socketDrops(String file, int port) { + String hexPort = String.format(Locale.ROOT, ":%04X", port); + long total = -1; + try (BufferedReader in = new BufferedReader(new FileReader(file))) { + in.readLine(); // the header + String line; + while ((line = in.readLine()) != null) { + String[] f = line.trim().split("\\s+"); + // sl local_address rem_address st ... drops (the last column) + if (f.length < 13 || !f[1].endsWith(hexPort)) continue; + try { + total = Math.max(0, total) + Long.parseLong(f[f.length - 1]); + } catch (NumberFormatException ignored) {} + } + } catch (IOException | RuntimeException e) { + return -1; + } + return total; + } + /** From any thread; a blocked receive ends with the socket. */ @Override public synchronized void close() { diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index 1ae256c5..858dea12 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -104,6 +104,8 @@ default void onDetail(String text) {} private boolean udpPlaying; /** Listener of a udp:// slot, else null. */ private RtpListener listener; + /** Video payload type of the stream the listener receives, or -1 (#251). */ + private volatile int listenVideoPt = -1; /** A udp:// slot is listening: silence is reported, never "fixed" by a rebind. */ private volatile boolean listening; /** Client of a flowufo:// slot (#187), else null: it probes on its own when silent. */ @@ -732,6 +734,15 @@ Map heartbeatStats() { } } + /** Receive statistics for /stats.json while a udp:// slot listens (#251), else null. */ + Map udpListenStats() { + RtpListener l; + synchronized (lock) { + l = listener; + } + return l != null ? l.stats(listenVideoPt) : null; + } + /** FLOW-UFO statistics for /stats.json while a flowufo:// slot runs (#187), else null. */ Map flowUfoStats() { FlowUfo.Client c = ufo; @@ -936,6 +947,7 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { } listener = l; listening = true; + listenVideoPt = -1; // No RTSP here: nothing of an older connection may be sent by the watchdog keepalive = null; lastPacketMs = System.nanoTime() / 1_000_000; @@ -971,6 +983,7 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { sdp.videoPt = pt; sdp.videoH265 = u.h265(pt); current[0] = sdp; + listenVideoPt = pt; Log.i(TAG, tag + ": RTP payload type " + pt + (sdp.videoH265 ? ", H.265" : ", H.264")); callback.onSdp(sdp); diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java index c0f8b53b..2744bcb0 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -17,6 +17,8 @@ import org.junit.Before; import org.junit.Test; +import java.io.File; +import java.io.FileWriter; import java.io.IOException; import java.net.DatagramPacket; import java.net.DatagramSocket; @@ -25,6 +27,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -290,4 +293,40 @@ public void silenceIsReportedWithoutRebinding() throws Exception { } assertEquals(2, Collections.frequency(rec.statuses, "connected")); } + + @Test + public void udpStatsCountWhatArrivedAndWhatIsMissing() throws Exception { + int port = freePort(); + rec.latch = new CountDownLatch(6); + listen("udp://@:" + port); + send(port, 96, 3); + seq += 5; // five packets lost on the way + send(port, 96, 3); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + Map st = session.udpListenStats(); + assertTrue((Integer) st.get("rcvbuf") > 0); + assertEquals(6L, st.get("received")); + assertEquals(0L, st.get("ring_dropped")); + assertEquals(5L, st.get("rtp_lost")); + } + + @Test + public void socketDropsAreReadForThePort() throws Exception { + File f = File.createTempFile("udp", ".txt"); + try { + try (FileWriter w = new FileWriter(f)) { + w.write(" sl local_address rem_address st tx_queue rx_queue tr tm->when" + + " retrnsmt uid timeout inode ref pointer drops\n"); + w.write(" 12: 00000000:15E0 00000000:0000 07 00000000:00000000 00:00000000" + + " 00000000 10123 0 4711 2 0000000000000000 57\n"); + w.write(" 13: 00000000:15E1 00000000:0000 07 00000000:00000000 00:00000000" + + " 00000000 10123 0 4712 2 0000000000000000 9\n"); + } + assertEquals(57, RtpListener.socketDrops(f.getPath(), 5600)); + assertEquals(-1, RtpListener.socketDrops(f.getPath(), 5602)); + assertEquals(-1, RtpListener.socketDrops(f.getPath() + ".none", 5600)); + } finally { + f.delete(); + } + } } diff --git a/docs/debug.md b/docs/debug.md index 86cfc63b..b906bed5 100644 --- a/docs/debug.md +++ b/docs/debug.md @@ -151,10 +151,11 @@ One entry per grid cell; `slot` is 1–8 or `U`. An empty cell has only `slot`, | `load` | macroblocks/s the cell costs in its current mode, once size and rate are known | | `paused` | paused by the decode planner (`Paused: decoder overloaded`) | | `audio` | this cell's sound plays | +| `udp` | `udp://` slots and U: `rcvbuf` (socket buffer the kernel gave), `received`, `ring_dropped`, `rtp_lost`, `socket_drops` (if readable) | ### slots -Single view and U view: every cell of the grid as the debug page shows it (3×3 in the U view or when the active slot is 5–8). +Single view and U view: every cell of the grid as the debug page shows it (3×3 in the U view or when the active slot is 5–8). In the U view the other slots are not checked while the U cell receives a push (#251). | Field | Meaning | |-------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| diff --git a/docs/transports.md b/docs/transports.md index 48afc7ac..b7856539 100644 --- a/docs/transports.md +++ b/docs/transports.md @@ -77,6 +77,7 @@ udp://[source@][address]:port[?codec=h264|h265][&pt=N] - Addresses must be numeric (IPv4 or IPv6 in brackets); no name lookup. A port is required. - A codec change on the camera is followed while playing. Video only. - Two slots cannot listen on the same port. The [U cell](usage.md#the-u-cell) listens on 5600. +- A thread of its own reads the socket into a ring of 1024 datagrams: an app gets a socket buffer of at most `rmem_max` (often 256 KiB, about one 1080p key frame), so a burst must not wait for the decoder. Where datagrams went missing is in the cell's `udp` object in [`/stats.json`](debug.md#cells): `socket_drops` (socket buffer full, when Android lets the app read `/proc/net/udp`), `ring_dropped` (ring full) and `rtp_lost` (sequence gaps, lost anywhere on the way). A sender that paces its key frames avoids the bursts (#251). - On Wi‑Fi the app holds a multicast lock while it is in a group. - A pushed stream cannot fall back to TCP: when every key frame loses a piece, the cell decodes the damaged frames (`· damaged`). From 21d86e3bd9e07b973538aa0d6160e8b1f30e959e Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Wed, 21 Oct 2026 13:08:15 +0000 Subject: [PATCH 2/5] 2.18.3: udp:// receive thread, no slot checks during a push (#251) --- CHANGELOG.md | 9 +++++++++ android/app/build.gradle | 4 ++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 64615d9e..c52eb837 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,15 @@ Versions after 2.17 are `MAJOR.MINOR.PATCH` (2.17 is followed by 2.17.1 or 2.18. `docs/testing.md`, "Versions and releases"); the headings below keep the numbers as they were released. +## 2.18.3 — 2026-10-21 + +- `udp://` slots and the U cell: a thread of its own reads the socket into a ring of 1024 + datagrams, so key-frame bursts no longer overflow the small socket buffer an app gets + (`rmem_max`) while the decoder or the UI hold up the receive; the log shows the buffer the + kernel gave. While the U cell receives a push, the other slots get no stream checks, which + made the box drop datagrams. `/stats.json` adds `udp` to the cell: `rcvbuf`, `received`, + `ring_dropped`, `rtp_lost` and, where the app may read it, `socket_drops` (#251). + ## 2.18.2 — 2026-10-20 - Single view: the decoder is set up for the stream's size from its SPS, so streams above diff --git a/android/app/build.gradle b/android/app/build.gradle index ccfb652e..de0d418c 100755 --- a/android/app/build.gradle +++ b/android/app/build.gradle @@ -23,8 +23,8 @@ android { applicationId "com.openipc.decoder" minSdk 21 targetSdk 36 - versionCode 45 - versionName "2.18.2" + versionCode 46 + versionName "2.18.3" buildConfigField "String", "GIT_HASH", "\"${getGitHash()}\"" } From ebf57859845f8e5523daa96d40eab5aaa250be03 Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Wed, 21 Oct 2026 13:16:41 +0000 Subject: [PATCH 3/5] =?UTF-8?q?fix:=20udp://=20listener=20review=20fixes?= =?UTF-8?q?=20=E2=80=94=20batches,=20published=20counters,=20probe=20hold,?= =?UTF-8?q?=20bounded=20ring=20(#251)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Qodo on #252: the session thread handles at most 64 datagrams before it checks the stop condition and the reorder deadlines; rtp_lost and socket_drops are published by the session thread (/proc read at most once a second, never on the UI thread) and count only the known video payload type; a slot grown for a long datagram is not kept in the ring; stream checks are held from the one-second cell tick, not only on debug requests; isConnected() no longer splits showsVideo() from its javadoc. --- .../java/com/openipc/decoder/Decoder.java | 6 +- .../java/com/openipc/decoder/QuadCell.java | 6 +- .../java/com/openipc/decoder/RtpListener.java | 56 ++++++++++++++----- .../java/com/openipc/decoder/RtspSession.java | 7 +-- .../com/openipc/decoder/StreamProbes.java | 16 ++++++ .../com/openipc/decoder/RtpListenerTest.java | 7 +++ 6 files changed, 74 insertions(+), 24 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/Decoder.java b/android/app/src/main/java/com/openipc/decoder/Decoder.java index 73e45768..7fe300bf 100755 --- a/android/app/src/main/java/com/openipc/decoder/Decoder.java +++ b/android/app/src/main/java/com/openipc/decoder/Decoder.java @@ -1170,6 +1170,8 @@ public void run() { v.setVisibility(showStats ? View.VISIBLE : View.GONE); if (log) Log.i(TAG, "Quad-" + i + ": " + line); } + // Not only on debug requests: a push that starts ends the checks within a second + streamProbes.hold(pushReceived()); checkDecoderStuck(cells); planDecoding(cells, now); updateDebugUrls(); @@ -1726,8 +1728,10 @@ private void updateProbes() { urls[i] = mHosts[i] != null && !mHosts[i].isEmpty() && !mHosts[i].equals(DEFAULT_URL) ? mHosts[i] : null; } + boolean push = pushReceived(); + streamProbes.hold(push); streamProbes.update(urls, mTransports, udpViewShown ? null : mHost, - mResumed && (udpViewShown ? !pushReceived() : !quadEnabled)); + mResumed && (udpViewShown ? !push : !quadEnabled)); } /** diff --git a/android/app/src/main/java/com/openipc/decoder/QuadCell.java b/android/app/src/main/java/com/openipc/decoder/QuadCell.java index 74030fea..a3859597 100644 --- a/android/app/src/main/java/com/openipc/decoder/QuadCell.java +++ b/android/app/src/main/java/com/openipc/decoder/QuadCell.java @@ -517,13 +517,13 @@ private void checkUdpStall() { /** On key frames only, a frame comes once per GOP, which can be several seconds (#85). */ static final long KEY_FRAMES_LIVE_MS = 10_000; + /** A stream is coming in: "connected", paused or not. */ + boolean isConnected() { return "connected".equals(state); } + /** * Connected, not paused and a frame shown within the last {@code ms} milliseconds (#98); * on key frames within KEY_FRAMES_LIVE_MS, or the cell would blink between key frames. */ - /** A stream is coming in: "connected", paused or not. */ - boolean isConnected() { return "connected".equals(state); } - boolean showsVideo(long ms) { long window = keyFramesOnly ? Math.max(ms, KEY_FRAMES_LIVE_MS) : ms; return pausedReason == null && "connected".equals(state) && renderedWithin(window); diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java index d1db7633..2f1b6ab9 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpListener.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -42,8 +42,10 @@ class RtpListener implements Closeable { private static final int RCVBUF = 2 * 1024 * 1024; /** Datagrams held between the receive thread and the rest: a few key frames. */ static final int RING = 1024; - /** Starting size of a ring slot; one grows for a longer datagram. */ + /** Size of a ring slot; a longer datagram gets a slot of its own, not kept (Qodo on #252). */ private static final int SLOT = 1600; + /** Datagrams handled before the stop condition and the reorder deadlines again. */ + private static final int BATCH = 64; interface Handler { void onPacket(byte[] buf, int offset, int length); @@ -177,10 +179,20 @@ boolean h265(int pt) { /** Datagrams thrown away because the ring was full. */ private volatile long ringDropped; private volatile IOException readError; + /** + * Published by the session thread for /stats.json (Qodo on #252): the video payload type + * (-1 until known), the gaps in its sequence, the kernel's drop count (-1: unreadable). + */ + private volatile int videoPt; + private volatile long rtpLost; + private volatile long socketDrops = -1; + private boolean dropsReadable = true; + private long dropsReadNs; RtpListener(String tag, Url url) throws IOException { this.tag = tag; this.url = url; + this.videoPt = url.pt; MulticastSocket s = new MulticastSocket(null); try { if (url.multicast()) { @@ -289,21 +301,23 @@ void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { Thread.currentThread().interrupt(); break; } - // Everything waiting, then the reorder deadlines - while (slot != null) { + // A batch of what is waiting, then the condition and the reorder deadlines + for (int n = 0; slot != null; n++) { int len = ((slot[0] & 0xFF) << 24) | ((slot[1] & 0xFF) << 16) | ((slot[2] & 0xFF) << 8) | (slot[3] & 0xFF); int pt = slot[5] & 0x7F; RtpReorder r = reorders[pt]; if (r == null) r = reorders[pt] = new RtpReorder(64, UdpReceiver.REORDER_MS); r.push(slot, 4, len, System.nanoTime(), sink); - free.offer(slot); - slot = filled.poll(); + // A slot grown for a long datagram is not kept: the ring stays small + free.offer(slot.length > SLOT + 4 ? new byte[SLOT + 4] : slot); + slot = n + 1 < BATCH ? filled.poll() : null; } now = System.nanoTime(); for (RtpReorder r : reorders) { if (r != null) r.poll(now, sink); } + publishStats(now); } } catch (IOException e) { if (!closed) throw e; @@ -321,25 +335,37 @@ void run(UdpReceiver.Condition keepGoing, Handler handler) throws IOException { /** * For /stats.json (#251): the buffer the kernel gave, datagrams read, dropped by a full - * ring, missing from the sequence of the video payload type {@code videoPt} (-1: every - * type), and the kernel's drop count of the port when the app may read it. + * ring, missing from the sequence of the video payload type (0 until it is known), and + * the kernel's drop count of the port when the app may read it. */ - Map stats(int videoPt) { + Map stats() { Map m = new LinkedHashMap<>(); m.put("rcvbuf", rcvbuf); m.put("received", received); m.put("ring_dropped", ringDropped); - long lost = 0; - for (int pt = 0; pt < reorders.length; pt++) { - RtpReorder r = reorders[pt]; - if (r != null && (videoPt < 0 || pt == videoPt)) lost += r.skipped(); - } - m.put("rtp_lost", lost); - long drops = socketDrops(url.port); + m.put("rtp_lost", rtpLost); + long drops = socketDrops; if (drops >= 0) m.put("socket_drops", drops); return m; } + /** The video payload type, once the session knows it; its gaps are {@code rtp_lost}. */ + void setVideoPt(int pt) { videoPt = pt; } + + /** On the session thread: the loss counters for stats(), /proc at most once a second. */ + private void publishStats(long now) { + int pt = videoPt; + RtpReorder r = pt >= 0 ? reorders[pt] : null; + rtpLost = r != null ? r.skipped() : 0; + if (dropsReadable && now - dropsReadNs >= 1_000_000_000L) { + dropsReadNs = now; + long d = socketDrops(url.port); + // Not readable now is not readable later: no more tries + if (d < 0) dropsReadable = false; + socketDrops = d; + } + } + /** * The kernel's drop count of the UDP sockets on {@code port}, from /proc/net/udp and udp6, * or -1: apps of newer Android versions may not read them. diff --git a/android/app/src/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index 858dea12..ac87a5fa 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -104,8 +104,6 @@ default void onDetail(String text) {} private boolean udpPlaying; /** Listener of a udp:// slot, else null. */ private RtpListener listener; - /** Video payload type of the stream the listener receives, or -1 (#251). */ - private volatile int listenVideoPt = -1; /** A udp:// slot is listening: silence is reported, never "fixed" by a rebind. */ private volatile boolean listening; /** Client of a flowufo:// slot (#187), else null: it probes on its own when silent. */ @@ -740,7 +738,7 @@ Map udpListenStats() { synchronized (lock) { l = listener; } - return l != null ? l.stats(listenVideoPt) : null; + return l != null ? l.stats() : null; } /** FLOW-UFO statistics for /stats.json while a flowufo:// slot runs (#187), else null. */ @@ -947,7 +945,6 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { } listener = l; listening = true; - listenVideoPt = -1; // No RTSP here: nothing of an older connection may be sent by the watchdog keepalive = null; lastPacketMs = System.nanoTime() / 1_000_000; @@ -983,7 +980,7 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { sdp.videoPt = pt; sdp.videoH265 = u.h265(pt); current[0] = sdp; - listenVideoPt = pt; + l.setVideoPt(pt); Log.i(TAG, tag + ": RTP payload type " + pt + (sdp.videoH265 ? ", H.265" : ", H.264")); callback.onSdp(sdp); diff --git a/android/app/src/main/java/com/openipc/decoder/StreamProbes.java b/android/app/src/main/java/com/openipc/decoder/StreamProbes.java index 7a3c08a6..536b39c8 100644 --- a/android/app/src/main/java/com/openipc/decoder/StreamProbes.java +++ b/android/app/src/main/java/com/openipc/decoder/StreamProbes.java @@ -165,6 +165,21 @@ synchronized void cancel() { } } + /** The U cell receives a push: no probes start, running ones end (#251). */ + private boolean held; + + /** From the UI thread about once a second, and with every update. */ + void hold(boolean on) { + synchronized (lock) { + if (held == on) return; + held = on; + if (!on) return; + for (Probe p : probes) { + if (p != null && !p.cancelled()) p.cancel(); + } + } + } + StreamProbes(String userAgent) { this.userAgent = userAgent; ((ThreadPoolExecutor) workers).allowCoreThreadTimeOut(true); @@ -266,6 +281,7 @@ private void tick() { ticker = null; return; } + if (held) return; int count = 0; for (boolean r : running) if (r) count++; if (count >= MAX_RUNNING) return; diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java index 2744bcb0..63ac9dcc 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -303,7 +303,14 @@ public void udpStatsCountWhatArrivedAndWhatIsMissing() throws Exception { seq += 5; // five packets lost on the way send(port, 96, 3); assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + // Published by the session thread after the batch Map st = session.udpListenStats(); + long deadline = System.currentTimeMillis() + 2000; + while (!Long.valueOf(5).equals(st.get("rtp_lost")) + && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + st = session.udpListenStats(); + } assertTrue((Integer) st.get("rcvbuf") > 0); assertEquals(6L, st.get("received")); assertEquals(0L, st.get("ring_dropped")); From 8882ce603147d59eb46a1285a4dde5350980107a Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Wed, 21 Oct 2026 13:27:56 +0000 Subject: [PATCH 4/5] fix: rtp_lost keeps losses of a long jump and of earlier video types (#251) Qodo on #252: a forward jump of 128 or more packets now adds the missing ones to RtpReorder.skipped (not a jump of 1000 or more: a restarted stream), and rtp_lost sums every payload type that was video in the listener, so a codec change does not lower it. --- .../java/com/openipc/decoder/RtpListener.java | 10 +++++++-- .../java/com/openipc/decoder/RtpReorder.java | 3 +++ .../com/openipc/decoder/RtpListenerTest.java | 22 +++++++++++++++++++ .../com/openipc/decoder/RtpReorderTest.java | 10 +++++++++ 4 files changed, 43 insertions(+), 2 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java index 2f1b6ab9..17528738 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpListener.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -184,6 +184,8 @@ boolean h265(int pt) { * (-1 until known), the gaps in its sequence, the kernel's drop count (-1: unreadable). */ private volatile int videoPt; + /** Every payload type that was video in this listener: losses before a codec change stay. */ + private final boolean[] videoTypes = new boolean[128]; private volatile long rtpLost; private volatile long socketDrops = -1; private boolean dropsReadable = true; @@ -355,8 +357,12 @@ Map stats() { /** On the session thread: the loss counters for stats(), /proc at most once a second. */ private void publishStats(long now) { int pt = videoPt; - RtpReorder r = pt >= 0 ? reorders[pt] : null; - rtpLost = r != null ? r.skipped() : 0; + if (pt >= 0) videoTypes[pt] = true; + long lost = 0; + for (int i = 0; i < reorders.length; i++) { + if (videoTypes[i] && reorders[i] != null) lost += reorders[i].skipped(); + } + rtpLost = lost; if (dropsReadable && now - dropsReadNs >= 1_000_000_000L) { dropsReadNs = now; long d = socketDrops(url.port); diff --git a/android/app/src/main/java/com/openipc/decoder/RtpReorder.java b/android/app/src/main/java/com/openipc/decoder/RtpReorder.java index 6397a5de..14dff303 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpReorder.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpReorder.java @@ -77,6 +77,9 @@ void push(byte[] buf, int offset, int len, long nowNs, Sink sink) { } else if (d >= SLOTS) { // Too far ahead to wait for: hand over what is held, then continue from here flushAll(sink); + // The rest of the jump is lost too, unless it is so far that the stream restarted + int gap = (short) (seq - expected); + if (gap > 0 && gap < RESTART_BEHIND) skipped += gap; restart(seq, buf, offset, len, sink); } else { int i = seq & (SLOTS - 1); diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java index 63ac9dcc..8c0ca560 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -336,4 +336,26 @@ public void socketDropsAreReadForThePort() throws Exception { f.delete(); } } + + @Test + public void lossBeforeACodecChangeIsKept() throws Exception { + int port = freePort(); + rec.latch = new CountDownLatch(4); + listen("udp://@:" + port); + send(port, 96, 2); + seq += 3; // three H.264 packets lost + send(port, 96, 2); + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + rec.latch = new CountDownLatch(2); + send(port, 97, 2); // the camera switched to H.265 + assertTrue(rec.latch.await(3, TimeUnit.SECONDS)); + long deadline = System.currentTimeMillis() + 2000; + Map st = session.udpListenStats(); + while (!Long.valueOf(3).equals(st.get("rtp_lost")) + && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + st = session.udpListenStats(); + } + assertEquals(3L, st.get("rtp_lost")); + } } diff --git a/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java b/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java index 19d2216b..3689bf56 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpReorderTest.java @@ -112,4 +112,14 @@ public void streamRestartFarBehindIsFollowed() { push(r, 0, 30000, 30001, 7, 8); assertEquals(Arrays.asList(30000, 30001, 7, 8), out); } + + @Test + public void aLongJumpCountsAsLostUnlessTheStreamRestarted() { + RtpReorder r = new RtpReorder(64, 40); + push(r, 0, 10, 11, 212); // 200 packets lost in one burst (Qodo on #252) + assertEquals(Arrays.asList(10, 11, 212), out); + assertEquals(200, r.skipped()); + push(r, 0, 20213); // a new stream, not a loss + assertEquals(200, r.skipped()); + } } From 926ecd974b0e2ced8923a42ace9554fc1f9602df Mon Sep 17 00:00:00 2001 From: Compot Developer Date: Wed, 21 Oct 2026 15:32:33 +0000 Subject: [PATCH 5/5] fix: socket_drops counts this listener's socket only; held probes say not checked (#251) Qodo on #252: /proc/net/udp{,6} rows are matched by the bound address and port (the group of a multicast slot, the wildcard otherwise), so another group on the same port is not counted; a slot not checked because the U cell receives a push has probe state "not checked", as its text says. --- .../java/com/openipc/decoder/Decoder.java | 5 +- .../java/com/openipc/decoder/RtpListener.java | 54 +++++++++++++++---- .../com/openipc/decoder/RtpListenerTest.java | 15 ++++-- 3 files changed, 60 insertions(+), 14 deletions(-) diff --git a/android/app/src/main/java/com/openipc/decoder/Decoder.java b/android/app/src/main/java/com/openipc/decoder/Decoder.java index 7fe300bf..c785280a 100755 --- a/android/app/src/main/java/com/openipc/decoder/Decoder.java +++ b/android/app/src/main/java/com/openipc/decoder/Decoder.java @@ -1837,8 +1837,9 @@ private List probeSlots() { StreamProbe.Result r = probeResult(i); Map p = new LinkedHashMap<>(); p.put("state", r != null ? (r.ok ? "ok" : "error") - : streamProbes.isRunning(i) ? "checking" : u || StreamProbes.skipReason( - mHosts[i], DEFAULT_URL, udpViewShown ? null : mHost) != null + : streamProbes.isRunning(i) ? "checking" : u || pushReceived() + || StreamProbes.skipReason(mHosts[i], DEFAULT_URL, + udpViewShown ? null : mHost) != null ? "not checked" : "waiting"); if (r != null) { p.put("checked_s_ago", (now - r.atMs) / 1000); diff --git a/android/app/src/main/java/com/openipc/decoder/RtpListener.java b/android/app/src/main/java/com/openipc/decoder/RtpListener.java index 17528738..df1a597d 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtpListener.java +++ b/android/app/src/main/java/com/openipc/decoder/RtpListener.java @@ -365,7 +365,7 @@ private void publishStats(long now) { rtpLost = lost; if (dropsReadable && now - dropsReadNs >= 1_000_000_000L) { dropsReadNs = now; - long d = socketDrops(url.port); + long d = socketDrops(url.address, url.port); // Not readable now is not readable later: no more tries if (d < 0) dropsReadable = false; socketDrops = d; @@ -373,20 +373,56 @@ private void publishStats(long now) { } /** - * The kernel's drop count of the UDP sockets on {@code port}, from /proc/net/udp and udp6, - * or -1: apps of newer Android versions may not read them. + * The kernel's drop count of this listener's socket: the port and its bound address (the + * group of a multicast slot, or the wildcard), from /proc/net/udp and udp6; -1 when apps + * may not read them (Qodo on #252: other groups on the port are not counted). */ - static long socketDrops(int port) { + static long socketDrops(InetAddress bound, int port) { long total = -1; - for (String file : new String[] {"/proc/net/udp", "/proc/net/udp6"}) { - long d = socketDrops(file, port); + String[][] files = { + {"/proc/net/udp", procAddress(bound, false)}, + {"/proc/net/udp6", procAddress(bound, true)}, + }; + for (String[] f : files) { + if (f[1] == null) continue; + long d = socketDrops(f[0], f[1] + String.format(Locale.ROOT, ":%04X", port)); if (d >= 0) total = Math.max(0, total) + d; } return total; } - static long socketDrops(String file, int port) { - String hexPort = String.format(Locale.ROOT, ":%04X", port); + /** + * An address as /proc/net/udp{,6} prints it: 32-bit words in host (little-endian) order; + * an IPv4 address in udp6 is IPv4-mapped. Null when it cannot be in that file. + */ + static String procAddress(InetAddress a, boolean v6File) { + byte[] b; + if (a == null) { + b = new byte[v6File ? 16 : 4]; + } else if (a.getAddress().length == 4) { + if (!v6File) { + b = a.getAddress(); + } else { + b = new byte[16]; + b[10] = (byte) 0xFF; + b[11] = (byte) 0xFF; + System.arraycopy(a.getAddress(), 0, b, 12, 4); + } + } else { + if (!v6File) return null; + b = a.getAddress(); + } + StringBuilder sb = new StringBuilder(); + for (int w = 0; w < b.length; w += 4) { + for (int i = 3; i >= 0; i--) { + sb.append(String.format(Locale.ROOT, "%02X", b[w + i] & 0xFF)); + } + } + return sb.toString(); + } + + /** Sum of the drops column of the rows whose local address is {@code local}, or -1. */ + static long socketDrops(String file, String local) { long total = -1; try (BufferedReader in = new BufferedReader(new FileReader(file))) { in.readLine(); // the header @@ -394,7 +430,7 @@ static long socketDrops(String file, int port) { while ((line = in.readLine()) != null) { String[] f = line.trim().split("\\s+"); // sl local_address rem_address st ... drops (the last column) - if (f.length < 13 || !f[1].endsWith(hexPort)) continue; + if (f.length < 13 || !f[1].equalsIgnoreCase(local)) continue; try { total = Math.max(0, total) + Long.parseLong(f[f.length - 1]); } catch (NumberFormatException ignored) {} diff --git a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java index 8c0ca560..bcf8cfbe 100644 --- a/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java +++ b/android/app/src/test/java/com/openipc/decoder/RtpListenerTest.java @@ -326,12 +326,21 @@ public void socketDropsAreReadForThePort() throws Exception { + " retrnsmt uid timeout inode ref pointer drops\n"); w.write(" 12: 00000000:15E0 00000000:0000 07 00000000:00000000 00:00000000" + " 00000000 10123 0 4711 2 0000000000000000 57\n"); + w.write(" 14: 030201EF:15E0 00000000:0000 07 00000000:00000000 00:00000000" + + " 00000000 10123 0 4713 2 0000000000000000 5\n"); w.write(" 13: 00000000:15E1 00000000:0000 07 00000000:00000000 00:00000000" + " 00000000 10123 0 4712 2 0000000000000000 9\n"); } - assertEquals(57, RtpListener.socketDrops(f.getPath(), 5600)); - assertEquals(-1, RtpListener.socketDrops(f.getPath(), 5602)); - assertEquals(-1, RtpListener.socketDrops(f.getPath() + ".none", 5600)); + // The wildcard socket of the port, not the group bound on the same port + assertEquals(57, RtpListener.socketDrops(f.getPath(), "00000000:15E0")); + assertEquals(5, RtpListener.socketDrops(f.getPath(), "030201EF:15E0")); + assertEquals(-1, RtpListener.socketDrops(f.getPath(), "00000000:15E2")); + assertEquals(-1, RtpListener.socketDrops(f.getPath() + ".none", "00000000:15E0")); + assertEquals("030201EF", + RtpListener.procAddress(InetAddress.getByName("239.1.2.3"), false)); + assertEquals("0000000000000000FFFF0000030201EF", + RtpListener.procAddress(InetAddress.getByName("239.1.2.3"), true)); + assertEquals("00000000000000000000000000000000", RtpListener.procAddress(null, true)); } finally { f.delete(); }