|
| 1 | +#include "rtp_reorder.h" |
| 2 | +#include "connection.h" |
| 3 | +#include "rtp.h" |
| 4 | +#include "snapshot.h" |
| 5 | +#include "stream.h" |
| 6 | +#include "utils.h" |
| 7 | +#include <string.h> |
| 8 | + |
| 9 | +void rtp_reorder_init(rtp_reorder_t *r) { memset(r, 0, sizeof(*r)); } |
| 10 | + |
| 11 | +void rtp_reorder_cleanup(rtp_reorder_t *r) { |
| 12 | + int pending = 0; |
| 13 | + for (int i = 0; i < RTP_REORDER_WINDOW_SIZE; i++) { |
| 14 | + if (r->slots[i]) { |
| 15 | + buffer_ref_put(r->slots[i]); |
| 16 | + r->slots[i] = NULL; |
| 17 | + pending++; |
| 18 | + } |
| 19 | + } |
| 20 | + if (pending > 0) { |
| 21 | + logger(LOG_DEBUG, |
| 22 | + "RTP reorder: Cleanup discarded %d pending packets (base_seq=%u)", |
| 23 | + pending, r->base_seq); |
| 24 | + } |
| 25 | + r->count = 0; |
| 26 | + r->initialized = 0; |
| 27 | +} |
| 28 | + |
| 29 | +/* Deliver single packet */ |
| 30 | +static int deliver_packet(buffer_ref_t *buf, connection_t *conn, |
| 31 | + int is_snapshot) { |
| 32 | + if (is_snapshot) { |
| 33 | + return snapshot_process_packet(&conn->stream.snapshot, buf->data_size, |
| 34 | + (uint8_t *)buf->data + buf->data_offset, |
| 35 | + conn); |
| 36 | + } |
| 37 | + return rtp_queue_buf_direct(conn, buf); |
| 38 | +} |
| 39 | + |
| 40 | +/* Flush consecutive packets, stop at hole |
| 41 | + * log_recovery: if true, log "Recovered" message (for Phase 2 reordering) */ |
| 42 | +static int flush_consecutive(rtp_reorder_t *r, connection_t *conn, |
| 43 | + int is_snapshot, int log_recovery) { |
| 44 | + int total_bytes = 0; |
| 45 | + int flushed = 0; |
| 46 | + uint16_t start_seq = r->base_seq; |
| 47 | + |
| 48 | + while (r->count > 0) { |
| 49 | + int slot = r->base_seq & RTP_REORDER_WINDOW_MASK; |
| 50 | + buffer_ref_t *buf = r->slots[slot]; |
| 51 | + |
| 52 | + if (!buf) |
| 53 | + break; /* Hole, stop */ |
| 54 | + |
| 55 | + int bytes = deliver_packet(buf, conn, is_snapshot); |
| 56 | + if (bytes > 0) |
| 57 | + total_bytes += bytes; |
| 58 | + |
| 59 | + buffer_ref_put(buf); |
| 60 | + r->slots[slot] = NULL; |
| 61 | + r->base_seq++; |
| 62 | + r->count--; |
| 63 | + flushed++; |
| 64 | + } |
| 65 | + |
| 66 | + if (log_recovery && flushed > 0) { |
| 67 | + logger(LOG_DEBUG, |
| 68 | + "RTP reorder: Recovered %d out-of-order packets (seq %u-%u)", flushed, |
| 69 | + start_seq, (uint16_t)(r->base_seq - 1)); |
| 70 | + } |
| 71 | + |
| 72 | + return total_bytes; |
| 73 | +} |
| 74 | + |
| 75 | +/* Force flush to make room */ |
| 76 | +static int force_flush_until(rtp_reorder_t *r, uint16_t target_seq, |
| 77 | + connection_t *conn, int is_snapshot) { |
| 78 | + int total_bytes = 0; |
| 79 | + int lost_count = 0; |
| 80 | + uint16_t start_seq = r->base_seq; |
| 81 | + |
| 82 | + while ((int16_t)(target_seq - r->base_seq) >= RTP_REORDER_WINDOW_SIZE) { |
| 83 | + int slot = r->base_seq & RTP_REORDER_WINDOW_MASK; |
| 84 | + buffer_ref_t *buf = r->slots[slot]; |
| 85 | + |
| 86 | + if (buf) { |
| 87 | + int bytes = deliver_packet(buf, conn, is_snapshot); |
| 88 | + if (bytes > 0) |
| 89 | + total_bytes += bytes; |
| 90 | + buffer_ref_put(buf); |
| 91 | + r->slots[slot] = NULL; |
| 92 | + r->count--; |
| 93 | + } else { |
| 94 | + lost_count++; |
| 95 | + } |
| 96 | + |
| 97 | + r->base_seq++; |
| 98 | + } |
| 99 | + |
| 100 | + if (lost_count > 0) { |
| 101 | + logger(LOG_DEBUG, "RTP reorder: Packet loss at seq %u (target=%u)", |
| 102 | + start_seq, target_seq); |
| 103 | + } |
| 104 | + |
| 105 | + return total_bytes; |
| 106 | +} |
| 107 | + |
| 108 | +int rtp_reorder_insert(rtp_reorder_t *r, buffer_ref_t *buf_ref, uint16_t seqn, |
| 109 | + connection_t *conn, int is_snapshot) { |
| 110 | + int total_bytes = 0; |
| 111 | + |
| 112 | + /* Phase 0: First packet - start collecting */ |
| 113 | + if (unlikely(r->initialized == 0)) { |
| 114 | + r->base_seq = seqn; /* Remember first seq as reference */ |
| 115 | + r->initialized = 1; /* Enter collecting phase */ |
| 116 | + |
| 117 | + /* Store first packet */ |
| 118 | + int slot = seqn & RTP_REORDER_WINDOW_MASK; |
| 119 | + buffer_ref_get(buf_ref); |
| 120 | + r->slots[slot] = buf_ref; |
| 121 | + r->count = 1; |
| 122 | + return 0; |
| 123 | + } |
| 124 | + |
| 125 | + /* Phase 1: Collecting initial packets |
| 126 | + * base_seq dynamically tracks the minimum sequence seen so far */ |
| 127 | + if (unlikely(r->initialized == 1)) { |
| 128 | + int slot = seqn & RTP_REORDER_WINDOW_MASK; |
| 129 | + |
| 130 | + if (!r->slots[slot]) { |
| 131 | + buffer_ref_get(buf_ref); |
| 132 | + r->slots[slot] = buf_ref; |
| 133 | + r->count++; |
| 134 | + |
| 135 | + /* Update min_seq: if this packet is earlier than current base_seq */ |
| 136 | + if ((int16_t)(seqn - r->base_seq) < 0) { |
| 137 | + r->base_seq = seqn; |
| 138 | + } |
| 139 | + } |
| 140 | + |
| 141 | + /* Collected enough? Start delivering from base_seq */ |
| 142 | + if (r->count >= RTP_REORDER_INIT_COLLECT) { |
| 143 | + r->initialized = 2; /* Enter active phase */ |
| 144 | + |
| 145 | + logger(LOG_DEBUG, |
| 146 | + "RTP reorder: Init complete, base_seq=%u (%d packets collected)", |
| 147 | + r->base_seq, r->count); |
| 148 | + |
| 149 | + /* Flush consecutive from base_seq (already the minimum) |
| 150 | + * Don't log "Recovered" - this is normal init, not reordering */ |
| 151 | + total_bytes += flush_consecutive(r, conn, is_snapshot, 0); |
| 152 | + } |
| 153 | + return total_bytes; |
| 154 | + } |
| 155 | + |
| 156 | + /* Phase 2: Active reordering (initialized == 2) */ |
| 157 | + int16_t seq_diff = (int16_t)(seqn - r->base_seq); |
| 158 | + |
| 159 | + /* Case 1: Expected sequence -> immediate delivery */ |
| 160 | + if (likely(seq_diff == 0)) { |
| 161 | + int bytes = deliver_packet(buf_ref, conn, is_snapshot); |
| 162 | + if (bytes > 0) |
| 163 | + total_bytes = bytes; |
| 164 | + r->base_seq++; |
| 165 | + |
| 166 | + total_bytes += flush_consecutive(r, conn, is_snapshot, 1); |
| 167 | + return total_bytes; |
| 168 | + } |
| 169 | + |
| 170 | + /* Case 2: Late/duplicate packet -> drop */ |
| 171 | + if (seq_diff < 0) { |
| 172 | + logger(LOG_DEBUG, "RTP reorder: Late packet dropped (seq=%u, base=%u)", |
| 173 | + seqn, r->base_seq); |
| 174 | + return 0; |
| 175 | + } |
| 176 | + |
| 177 | + /* Case 3: Beyond window -> force flush */ |
| 178 | + if (seq_diff >= RTP_REORDER_WINDOW_SIZE) { |
| 179 | + total_bytes += force_flush_until(r, seqn, conn, is_snapshot); |
| 180 | + } |
| 181 | + |
| 182 | + /* Store in slot */ |
| 183 | + int slot = seqn & RTP_REORDER_WINDOW_MASK; |
| 184 | + if (r->slots[slot]) { |
| 185 | + /* Slot occupied - duplicate packet, silently drop. |
| 186 | + * This is normal in some network environments where upstream devices |
| 187 | + * send redundant packets for reliability. */ |
| 188 | + return total_bytes; |
| 189 | + } |
| 190 | + |
| 191 | + buffer_ref_get(buf_ref); |
| 192 | + r->slots[slot] = buf_ref; |
| 193 | + r->count++; |
| 194 | + |
| 195 | + return total_bytes; |
| 196 | +} |
0 commit comments