Browse Source

improve performance by avoiding copy() (#21)

pull/80/head
aler9 5 years ago
parent
commit
74f7e5df66
  1. 2
      Makefile
  2. 7
      client.go
  3. 6
      main.go
  4. 14
      server-udp.go
  5. 12
      source.go
  6. 31
      utils.go

2
Makefile

@ -69,7 +69,7 @@ define CONFIG_RUN
#rtpPort: 8002 #rtpPort: 8002
#rtcpPort: 8003 #rtcpPort: 8003
#metrics: yes #metrics: yes
logDestinations: [stdout, file] pprof: yes
paths: paths:
all: all:

7
client.go

@ -118,8 +118,6 @@ type client struct {
streamProtocol gortsplib.StreamProtocol streamProtocol gortsplib.StreamProtocol
streamTracks map[int]*clientTrack streamTracks map[int]*clientTrack
rtcpReceivers []*gortsplib.RtcpReceiver rtcpReceivers []*gortsplib.RtcpReceiver
readBuf *doubleBuffer
writeBuf *doubleBuffer
describeRes chan describeRes describeRes chan describeRes
events chan clientEvent // only if state = Play and gortsplib.StreamProtocol = TCP events chan clientEvent // only if state = Play and gortsplib.StreamProtocol = TCP
@ -136,7 +134,6 @@ func newClient(p *program, nconn net.Conn) *client {
}), }),
state: clientStateInitial, state: clientStateInitial,
streamTracks: make(map[int]*clientTrack), streamTracks: make(map[int]*clientTrack),
readBuf: newDoubleBuffer(clientTcpReadBufferSize),
done: make(chan struct{}), done: make(chan struct{}),
} }
@ -838,7 +835,6 @@ func (c *client) runPlay(path string) {
confp := c.p.findConfForPath(path) confp := c.p.findConfForPath(path)
if c.streamProtocol == gortsplib.StreamProtocolTcp { if c.streamProtocol == gortsplib.StreamProtocolTcp {
c.writeBuf = newDoubleBuffer(clientTcpWriteBufferSize)
c.events = make(chan clientEvent) c.events = make(chan clientEvent)
} }
@ -1030,11 +1026,12 @@ func (c *client) runRecord(path string) {
} else { } else {
frame := &gortsplib.InterleavedFrame{} frame := &gortsplib.InterleavedFrame{}
readBuf := newMultiBuffer(3, clientTcpReadBufferSize)
readDone := make(chan error) readDone := make(chan error)
go func() { go func() {
for { for {
frame.Content = c.readBuf.swap() frame.Content = readBuf.next()
frame.Content = frame.Content[:cap(frame.Content)] frame.Content = frame.Content[:cap(frame.Content)]
recv, err := c.conn.ReadFrameOrRequest(frame) recv, err := c.conn.ReadFrameOrRequest(frame)

6
main.go

@ -621,15 +621,11 @@ func (p *program) forwardFrame(path string, trackId int, streamType gortsplib.St
} }
} else { } else {
buf := c.writeBuf.swap()
buf = buf[:len(frame)]
copy(buf, frame)
c.events <- clientEventFrameTcp{ c.events <- clientEventFrameTcp{
frame: &gortsplib.InterleavedFrame{ frame: &gortsplib.InterleavedFrame{
TrackId: trackId, TrackId: trackId,
StreamType: streamType, StreamType: streamType,
Content: buf, Content: frame,
}, },
} }
} }

14
server-udp.go

@ -16,8 +16,7 @@ type serverUdp struct {
p *program p *program
nconn *net.UDPConn nconn *net.UDPConn
streamType gortsplib.StreamType streamType gortsplib.StreamType
readBuf *doubleBuffer readBuf *multiBuffer
writeBuf *doubleBuffer
writeChan chan *udpAddrBufPair writeChan chan *udpAddrBufPair
done chan struct{} done chan struct{}
@ -35,8 +34,7 @@ func newServerUdp(p *program, port int, streamType gortsplib.StreamType) (*serve
p: p, p: p,
nconn: nconn, nconn: nconn,
streamType: streamType, streamType: streamType,
readBuf: newDoubleBuffer(clientUdpReadBufferSize), readBuf: newMultiBuffer(3, clientUdpReadBufferSize),
writeBuf: newDoubleBuffer(clientUdpWriteBufferSize),
writeChan: make(chan *udpAddrBufPair), writeChan: make(chan *udpAddrBufPair),
done: make(chan struct{}), done: make(chan struct{}),
} }
@ -66,7 +64,7 @@ func (l *serverUdp) run() {
}() }()
for { for {
buf := l.readBuf.swap() buf := l.readBuf.next()
n, addr, err := l.nconn.ReadFromUDP(buf) n, addr, err := l.nconn.ReadFromUDP(buf)
if err != nil { if err != nil {
break break
@ -91,11 +89,5 @@ func (l *serverUdp) close() {
} }
func (l *serverUdp) write(pair *udpAddrBufPair) { func (l *serverUdp) write(pair *udpAddrBufPair) {
// replace input buffer with write buffer
buf := l.writeBuf.swap()
buf = buf[:len(pair.buf)]
copy(buf, pair.buf)
pair.buf = buf
l.writeChan <- pair l.writeChan <- pair
} }

12
source.go

@ -257,9 +257,9 @@ func (s *source) runUdp(terminate chan struct{}, conn *gortsplib.ConnClient) boo
go func(trackId int, l *gortsplib.ConnClientUdpListener) { go func(trackId int, l *gortsplib.ConnClientUdpListener) {
defer wg.Done() defer wg.Done()
doubleBuf := newDoubleBuffer(sourceUdpReadBufferSize) multiBuf := newMultiBuffer(3, sourceUdpReadBufferSize)
for { for {
buf := doubleBuf.swap() buf := multiBuf.next()
n, err := l.Read(buf) n, err := l.Read(buf)
if err != nil { if err != nil {
@ -274,9 +274,9 @@ func (s *source) runUdp(terminate chan struct{}, conn *gortsplib.ConnClient) boo
go func(trackId int, l *gortsplib.ConnClientUdpListener) { go func(trackId int, l *gortsplib.ConnClientUdpListener) {
defer wg.Done() defer wg.Done()
doubleBuf := newDoubleBuffer(sourceUdpReadBufferSize) multiBuf := newMultiBuffer(3, sourceUdpReadBufferSize)
for { for {
buf := doubleBuf.swap() buf := multiBuf.next()
n, err := l.Read(buf) n, err := l.Read(buf)
if err != nil { if err != nil {
@ -340,12 +340,12 @@ func (s *source) runTcp(terminate chan struct{}, conn *gortsplib.ConnClient) boo
s.p.events <- programEventSourceReady{s} s.p.events <- programEventSourceReady{s}
frame := &gortsplib.InterleavedFrame{} frame := &gortsplib.InterleavedFrame{}
doubleBuf := newDoubleBuffer(sourceTcpReadBufferSize) multiBuf := newMultiBuffer(3, sourceTcpReadBufferSize)
tcpConnDone := make(chan error) tcpConnDone := make(chan error)
go func() { go func() {
for { for {
frame.Content = doubleBuf.swap() frame.Content = multiBuf.next()
frame.Content = frame.Content[:cap(frame.Content)] frame.Content = frame.Content[:cap(frame.Content)]
err := conn.ReadFrame(frame) err := conn.ReadFrame(frame)

31
utils.go

@ -52,27 +52,28 @@ func ipEqualOrInRange(ip net.IP, ips []interface{}) bool {
return false return false
} }
type doubleBuffer struct { type multiBuffer struct {
buf1 []byte buffers [][]byte
buf2 []byte curBuf int
curBuf bool
} }
func newDoubleBuffer(size int) *doubleBuffer { func newMultiBuffer(count int, size int) *multiBuffer {
return &doubleBuffer{ buffers := make([][]byte, count)
buf1: make([]byte, size), for i := 0; i < count; i++ {
buf2: make([]byte, size), buffers[i] = make([]byte, size)
}
return &multiBuffer{
buffers: buffers,
} }
} }
func (db *doubleBuffer) swap() []byte { func (mb *multiBuffer) next() []byte {
var ret []byte ret := mb.buffers[mb.curBuf]
if !db.curBuf { mb.curBuf += 1
ret = db.buf1 if mb.curBuf >= len(mb.buffers) {
} else { mb.curBuf = 0
ret = db.buf2
} }
db.curBuf = !db.curBuf
return ret return ret
} }

Loading…
Cancel
Save