fix(packet_pool): reducing latency for single-packet PES by using a heuristic to determine when a single-packet PES is received and to send eventually

Author: cobeq

data.go

@@ -182,3 +182,26 @@ func isPSIComplete(ps []*Packet) bool {
 
 	return i.Len() >= i.Offset()
 }
+
+// isPESComplete checks whether payload fully contains PES packet
+func isPESComplete(payload []byte) bool {
+	i := astikit.NewBytesIterator(payload)
+
+	i.Seek(4)
+
+	// Get next bytes
+	var bs []byte
+	var err error
+	if bs, err = i.NextBytesNoCopy(2); err != nil {
+		return false
+	}
+
+	pesLength := uint16(bs[0])<<8 | uint16(bs[1])
+
+	if pesLength == 0 {
+		// any length
+		return false
+	}
+
+	return int(pesLength)+pesHeaderLength <= len(payload)
+}

packet_pool.go

@@ -23,6 +23,7 @@ func newPacketAccumulator(pid uint16, programMap *programMap) *packetAccumulator
 func (b *packetAccumulator) add(p *Packet) (ps []*Packet) {
 	mps := b.q
 
+	var needSkipBufferMPS bool
 	// Empty buffer if we detect a discontinuity
 	if hasDiscontinuity(mps, p) {
 		// Reset current slice or make new
@@ -31,6 +32,8 @@ func (b *packetAccumulator) add(p *Packet) (ps []*Packet) {
 		} else {
 			mps = make([]*Packet, 0, 10)
 		}
+	} else {
+		needSkipBufferMPS = len(mps) == 0 || isPacketsAlreadySent(mps)
 	}
 
 	// Throw away packet if it's the same as the previous one
@@ -40,7 +43,9 @@ func (b *packetAccumulator) add(p *Packet) (ps []*Packet) {
 
 	// Flush buffer if new payload starts here
 	if p.Header.PayloadUnitStartIndicator {
-		ps = mps
+		if !needSkipBufferMPS {
+			ps = mps
+		}
 		mps = make([]*Packet, 0, cap(mps))
 	}
 
@@ -52,6 +57,8 @@ func (b *packetAccumulator) add(p *Packet) (ps []*Packet) {
 		isPSIComplete(mps) {
 		ps = mps
 		mps = nil
+	} else if needSkipBufferMPS && isPayloadCompletePES(p) {
+		ps = mps
 	}
 
 	b.q = mps
@@ -110,6 +117,10 @@ func (b *packetPool) dumpUnlocked() (ps []*Packet) {
 		ps = b.b[uint32(k)].q
 		delete(b.b, uint32(k))
 		if len(ps) > 0 {
+			if isPacketsAlreadySent(ps) {
+				ps = nil
+				continue
+			}
 			return
 		}
 	}
@@ -150,3 +161,11 @@ func isSameAsPrevious(ps []*Packet, p *Packet) bool {
 	l := len(ps)
 	return l > 0 && p.Header.HasPayload && p.Header.ContinuityCounter == ps[l-1].Header.ContinuityCounter
 }
+
+func isPayloadCompletePES(packet *Packet) bool {
+	return packet.Header.PayloadUnitStartIndicator && isPESPayload(packet.Payload) && isPESComplete(packet.Payload)
+}
+
+func isPacketsAlreadySent(ps []*Packet) bool {
+	return len(ps) == 1 && isPayloadCompletePES(ps[0])
+}

packet_pool_test.go

@@ -49,3 +49,38 @@ func TestPacketPool(t *testing.T) {
 	ps = b.dumpUnlocked()
 	assert.Len(t, ps, 0)
 }
+
+func TestPacketPoolWithRarePackets(t *testing.T) {
+	payloadDVBTeletext := hexToBytes(`000001bd00b2848024293c972af5ffffffffffffffffffffffffffffffff
+	ffffffffffffffffffffffffffffff10032cf5e4a8a80b0ba80ba80b2692
+	040404040404040404040404040404040404040404040404040404040404
+	0404032cd5e4a8a85757a8a8a80b26a80404040404040404040404040404
+	040404040404040404040404040404040404ff2cffffffffffffffffffff
+	ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff
+	ffffffff`)
+	b := newPacketPool(nil)
+	ps := b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 0, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 1, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 2, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 3, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 3, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 0)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 4, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 6, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 0)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 7, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 7, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 0)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 9, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 0)
+	ps = b.addUnlocked(&Packet{Header: PacketHeader{ContinuityCounter: 10, HasPayload: true, PayloadUnitStartIndicator: true, PID: 1004}, Payload: payloadDVBTeletext})
+	assert.Len(t, ps, 1)
+	ps = b.dumpUnlocked()
+	assert.Len(t, ps, 0)
+}