Merged in improve-ts-encoder-performance (pull request #83)

Improve ts encoder performance

Approved-by: kortschak <dan@kortschak.io>

stream/mts/encoder.go

@@ -122,8 +122,7 @@ var (
 )
 
 const (
-	psiPacketSize = 184
-	psiSendCount  = 7
+	psiSndCnt = 7
 )
 
 // timeLocation holds time and location data
@@ -197,6 +196,8 @@ type Encoder struct {
 	clock         time.Duration
 	frameInterval time.Duration
 	ptsOffset     time.Duration
+	tsSpace       [PacketSize]byte
+	pesSpace      [pes.MaxPesSize]byte
 
 	psiCount int
 
@@ -240,7 +241,7 @@ func (e *Encoder) Encode(nalu []byte) error {
 		Data:         nalu,
 		HeaderLength: 5,
 	}
-	buf := pesPkt.Bytes()
+	buf := pesPkt.Bytes(e.pesSpace[:pes.MaxPesSize])
 
 	pusi := true
 	for len(buf) != 0 {
@@ -268,7 +269,7 @@ func (e *Encoder) Encode(nalu []byte) error {
 			}
 		}
 		e.psiCount--
-		_, err := e.dst.Write(pkt.Bytes())
+		_, err := e.dst.Write(pkt.Bytes(e.tsSpace[:PacketSize]))
 		if err != nil {
 			return err
 		}
@@ -288,9 +289,9 @@ func (e *Encoder) writePSI() error {
 		PID:     patPid,
 		CC:      e.ccFor(patPid),
 		AFC:     hasPayload,
-		Payload: addPadding(patTable),
+		Payload: patTable,
 	}
-	_, err := e.dst.Write(patPkt.Bytes())
+	_, err := e.dst.Write(patPkt.Bytes(e.tsSpace[:PacketSize]))
 	if err != nil {
 		return err
 	}
@@ -311,25 +312,16 @@ func (e *Encoder) writePSI() error {
 		PID:     pmtPid,
 		CC:      e.ccFor(pmtPid),
 		AFC:     hasPayload,
-		Payload: addPadding(pmtTable),
+		Payload: pmtTable,
 	}
-	_, err = e.dst.Write(pmtPkt.Bytes())
+	_, err = e.dst.Write(pmtPkt.Bytes(e.tsSpace[:PacketSize]))
 	if err != nil {
 		return err
 	}
-	e.psiCount = psiSendCount
+	e.psiCount = psiSndCnt
 	return nil
 }
 
-// addPadding adds an appropriate amount of padding to a pat or pmt table for
-// addition to an mpegts packet.
-func addPadding(d []byte) []byte {
-	for len(d) < psiPacketSize {
-		d = append(d, 0xff)
-	}
-	return d
-}
-
 // tick advances the clock one frame interval.
 func (e *Encoder) tick() {
 	e.clock += e.frameInterval

stream/mts/mpegts.go

@@ -33,8 +33,8 @@ import (
 )
 
 const (
-	mpegTsSize        = 188
-	mpegtsPayloadSize = 176
+	PacketSize  = 188
+	PayloadSize = 176
 )
 
 /*
@@ -130,13 +130,13 @@ type Packet struct {
 // FindPMT will take a clip of mpegts and try to find a PMT table - if one
 // is found, then it is returned, otherwise nil and an error is returned.
 func FindPMT(d []byte) (p []byte, err error) {
-	if len(d) < mpegTsSize {
+	if len(d) < PacketSize {
 		return nil, errors.New("Mmpegts data not of valid length")
 	}
-	for i := 0; i < len(d); i += mpegTsSize {
+	for i := 0; i < len(d); i += PacketSize {
 		pid := (uint16(d[i+1]&0x1f) << 8) | uint16(d[i+2])
 		if pid == pmtPid {
-			p = d[i+4 : i+mpegTsSize]
+			p = d[i+4 : i+PacketSize]
 			return
 		}
 	}
@@ -146,9 +146,9 @@ func FindPMT(d []byte) (p []byte, err error) {
 // FillPayload takes a channel and fills the packets Payload field until the
 // channel is empty or we've the packet reaches capacity
 func (p *Packet) FillPayload(data []byte) int {
-	currentPktLength := 6 + asInt(p.PCRF)*6 + asInt(p.OPCRF)*6 +
+	currentPktLen := 6 + asInt(p.PCRF)*6 + asInt(p.OPCRF)*6 +
 		asInt(p.SPF)*1 + asInt(p.TPDF)*1 + len(p.TPD)
-	p.Payload = make([]byte, mpegtsPayloadSize-currentPktLength)
+	p.Payload = make([]byte, PayloadSize-currentPktLen)
 	return copy(p.Payload, data)
 }
 
@@ -168,7 +168,11 @@ func asByte(b bool) byte {
 
 // ToByteSlice interprets the fields of the ts packet instance and outputs a
 // corresponding byte slice
-func (p *Packet) Bytes() []byte {
+func (p *Packet) Bytes(buf []byte) []byte {
+	if buf == nil || cap(buf) != PacketSize {
+		buf = make([]byte, 0, PacketSize)
+	}
+	buf = buf[:0]
 	stuffingLength := 182 - len(p.Payload) - len(p.TPD) - asInt(p.PCRF)*6 -
 		asInt(p.OPCRF)*6 - asInt(p.SPF)
 	var stuffing []byte
@@ -179,7 +183,6 @@ func (p *Packet) Bytes() []byte {
 		stuffing[i] = 0xFF
 	}
 	afl := 1 + asInt(p.PCRF)*6 + asInt(p.OPCRF)*6 + asInt(p.SPF) + asInt(p.TPDF) + len(p.TPD) + len(stuffing)
-	buf := make([]byte, 0, mpegTsSize)
 	buf = append(buf, []byte{
 		0x47,
 		(asByte(p.TEI)<<7 | asByte(p.PUSI)<<6 | asByte(p.Priority)<<5 | byte((p.PID&0xFF00)>>8)),

stream/mts/pes/pes.go

@@ -26,7 +26,7 @@ LICENSE
 
 package pes
 
-const maxPesSize = 10000
+const MaxPesSize = 10000
 
 /*
 The below data struct encapsulates the fields of an PES packet. Below is
@@ -92,8 +92,11 @@ type Packet struct {
 	Data         []byte // Pes packet data
 }
 
-func (p *Packet) Bytes() []byte {
-	buf := make([]byte, 0, maxPesSize)
+func (p *Packet) Bytes(buf []byte) []byte {
+	if buf == nil || cap(buf) != MaxPesSize {
+		buf = make([]byte, 0, MaxPesSize)
+	}
+	buf = buf[:0]
 	buf = append(buf, []byte{
 		0x00, 0x00, 0x01,
 		p.StreamID,

stream/mts/pes/pes_test.go

@@ -46,7 +46,7 @@ func TestPesToByteSlice(t *testing.T) {
 		Stuff:        []byte{0xFF, 0xFF},
 		Data:         []byte{0xEA, 0x4B, 0x12},
 	}
-	got := pkt.Bytes()
+	got := pkt.Bytes(nil)
 	want := []byte{
 		0x00, // packet start code prefix byte 1
 		0x00, // packet start code prefix byte 2

stream/mts/psi/helpers.go

@@ -68,6 +68,13 @@ func UpdateTime(dst []byte, t uint64) error {
 	return nil
 }
 
+// SyntaxSecLenFrom takes a byte slice representation of a psi and extracts
+// it's syntax section length
+func SyntaxSecLenFrom(p []byte) (l uint8) {
+	l = uint8(p[syntaxSecLenIndx]) - crcSize
+	return
+}
+
 // TimeFrom takes a byte slice representation of a psi-pmt and extracts it's
 // timestamp, returning as a uint64 if it exists, otherwise returning 0 and nil
 // if it does not exist
@@ -121,3 +128,15 @@ func trimTo(d []byte, t byte) []byte {
 	}
 	return d
 }
+
+// addPadding adds an appropriate amount of padding to a pat or pmt table for
+// addition to an mpegts packet
+func addPadding(d []byte) []byte {
+	t := make([]byte, PacketSize)
+	copy(t, d)
+	padding := t[len(d):]
+	for i := range padding {
+		padding[i] = 0xff
+	}
+	return d
+}

stream/mts/psi/psi.go

@@ -26,6 +26,10 @@ LICENSE
 
 package psi
 
+const (
+	PacketSize = 184 // packet size of a psi.
+)
+
 // Lengths of section definitions
 const (
 	ESSDDefLen = 5
@@ -58,6 +62,12 @@ const (
 	LocationDataSize = 32 // bytes
 )
 
+// Other misc consts
+const (
+	syntaxSecLenIndx = 3
+	crcSize          = 4
+)
+
 // Program specific information
 type PSI struct {
 	Pf  byte   // Point field
@@ -126,6 +136,7 @@ func (p *PSI) Bytes() []byte {
 	out[3] = byte(p.Sl)
 	out = append(out, p.Tss.Bytes()...)
 	out = addCrc(out)
+	out = addPadding(out)
 	return out
 }
 

stream/mts/psi/psi_test.go

@@ -130,7 +130,7 @@ const (
 
 // err message
 const (
-	errCmp = "Incorrect output, for: %v wanted: %v, got: %v"
+	errCmp = "Incorrect output, for: %v \nwant: %v, \ngot:  %v"
 )
 
 // Test time to bytes test Data
@@ -282,9 +282,7 @@ var bytesTests = []struct {
 func TestBytes(t *testing.T) {
 	for _, test := range bytesTests {
 		got := test.input.Bytes()
-		// Remove crc32s
-		got = got[:len(got)-4]
-		if !bytes.Equal(got, test.want) {
+		if !bytes.Equal(got, addPadding(addCrc(test.want))) {
 			t.Errorf("unexpected error for test %v: got:%v want:%v", test.name, got,
 				test.want)
 		}