diff --git a/CHANGELOG.md b/CHANGELOG.md index 64615d9..c52eb83 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 ccfb652..de0d418 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()}\"" } 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 562afeb..c785280 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(); @@ -1697,6 +1699,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); @@ -1723,8 +1728,20 @@ 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 || !quadEnabled)); + mResumed && (udpViewShown ? !push : !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 +1771,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"; } /** @@ -1818,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/QuadCell.java b/android/app/src/main/java/com/openipc/decoder/QuadCell.java index 2135ad6..a385959 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(); } @@ -516,6 +517,9 @@ 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. 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 e92ed1f..df1a597 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,20 @@ * 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; + /** 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); @@ -152,13 +168,33 @@ 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; + /** + * 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; + /** 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; + 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()) { @@ -183,46 +219,107 @@ 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; + } + // 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(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); + // 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; @@ -238,6 +335,112 @@ 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 (0 until it is known), and + * the kernel's drop count of the port when the app may read it. + */ + Map stats() { + Map m = new LinkedHashMap<>(); + m.put("rcvbuf", rcvbuf); + m.put("received", received); + m.put("ring_dropped", ringDropped); + 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; + 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.address, 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 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(InetAddress bound, int port) { + long total = -1; + 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; + } + + /** + * 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 + 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].equalsIgnoreCase(local)) 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/RtpReorder.java b/android/app/src/main/java/com/openipc/decoder/RtpReorder.java index 6397a5d..14dff30 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/main/java/com/openipc/decoder/RtspSession.java b/android/app/src/main/java/com/openipc/decoder/RtspSession.java index 1ae256c..ac87a5f 100644 --- a/android/app/src/main/java/com/openipc/decoder/RtspSession.java +++ b/android/app/src/main/java/com/openipc/decoder/RtspSession.java @@ -732,6 +732,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() : null; + } + /** FLOW-UFO statistics for /stats.json while a flowufo:// slot runs (#187), else null. */ Map flowUfoStats() { FlowUfo.Client c = ufo; @@ -971,6 +980,7 @@ private boolean listen(int gen, RtpListener.Url u) throws IOException { sdp.videoPt = pt; sdp.videoH265 = u.h265(pt); current[0] = sdp; + 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 7a3c08a..536b39c 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 c0f8b53..bcf8cfb 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,78 @@ 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)); + // 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")); + 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(" 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"); + } + // 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(); + } + } + + @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 19d2216..3689bf5 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()); + } } diff --git a/docs/debug.md b/docs/debug.md index 86cfc63..b906bed 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 48afc7a..b785653 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`).