package/gortsplib
1
0
Fork 0
mirror of https://github.com/aler9/gortsplib synced 2025-10-04 23:02:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

improve performance

Browse Source
This commit is contained in:
aler9
2022-02-17 22:25:01 +01:00
committed by Alessandro Ros
parent 3a2f9adbd5
commit bca6756cd6
4 changed files with 61 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
client.go
View File

@@ -214,6 +214,7 @@ type Client struct {
tracksByChannel map[int]int
lastRange *headers.Range
tcpReadBuffer *multibuffer.MultiBuffer
tcpRTPPacketBuffer *rtpPacketMultiBuffer
writeMutex sync.RWMutex // publish
writeFrameAllowed bool // publish
udpReportTimer *time.Timer
@@ -760,14 +761,14 @@ func (c *Client) runReader() {
atomic.StoreInt64(c.tcpLastFrameTime, now.Unix())
if isRTP {
var pkt rtp.Packet
pkt := c.tcpRTPPacketBuffer.next()
err := pkt.Unmarshal(payload)
if err != nil {
return
}
c.tracks[trackID].rtcpReceiver.ProcessPacketRTP(now, &pkt)
c.OnPacketRTP(trackID, &pkt)
c.tracks[trackID].rtcpReceiver.ProcessPacketRTP(now, pkt)
c.OnPacketRTP(trackID, pkt)
} else {
packets, err := rtcp.Unmarshal(payload)
if err != nil {
@@ -1536,6 +1537,7 @@ func (c *Client) doSetup(
case TransportTCP:
if c.tcpReadBuffer == nil {
c.tcpReadBuffer = multibuffer.New(uint64(c.ReadBufferCount), uint64(c.ReadBufferSize))
c.tcpRTPPacketBuffer = newRTPPacketMultiBuffer(uint64(c.ReadBufferCount))
}
if c.tracksByChannel == nil {

28
serverconn.go
View File

@@ -11,7 +11,6 @@ import (
"time"
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/aler9/gortsplib/pkg/base"
"github.com/aler9/gortsplib/pkg/liberrors"
@@ -35,17 +34,18 @@ type ServerConn struct {
s *Server
conn net.Conn
ctx context.Context
ctxCancel func()
remoteAddr *net.TCPAddr
br *bufio.Reader
sessions map[string]*ServerSession
tcpFrameEnabled bool
tcpSession *ServerSession
tcpFrameTimeout bool
tcpReadBuffer *multibuffer.MultiBuffer
tcpProcessFunc func(int, bool, []byte)
tcpWriterRunning bool
ctx context.Context
ctxCancel func()
remoteAddr *net.TCPAddr
br *bufio.Reader
sessions map[string]*ServerSession
tcpFrameEnabled bool
tcpSession *ServerSession
tcpFrameTimeout bool
tcpReadBuffer *multibuffer.MultiBuffer
tcpRTPPacketBuffer *rtpPacketMultiBuffer
tcpProcessFunc func(int, bool, []byte)
tcpWriterRunning bool
// in
sessionRemove chan *ServerSession
@@ -260,7 +260,7 @@ func (sc *ServerConn) tcpProcessPlay(trackID int, isRTP bool, payload []byte) {
func (sc *ServerConn) tcpProcessRecord(trackID int, isRTP bool, payload []byte) {
if isRTP {
var pkt rtp.Packet
pkt := sc.tcpRTPPacketBuffer.next()
err := pkt.Unmarshal(payload)
if err != nil {
return
@@ -270,7 +270,7 @@ func (sc *ServerConn) tcpProcessRecord(trackID int, isRTP bool, payload []byte)
h.OnPacketRTP(&ServerHandlerOnPacketRTPCtx{
Session: sc.tcpSession,
TrackID: trackID,
Packet: &pkt,
Packet: pkt,
})
}
} else {

1
serversession.go
View File

@@ -1019,6 +1019,7 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
ss.tcpConn.tcpFrameEnabled = true
ss.tcpConn.tcpFrameTimeout = true
ss.tcpConn.tcpReadBuffer = multibuffer.New(uint64(sc.s.ReadBufferCount), uint64(sc.s.ReadBufferSize))
ss.tcpConn.tcpRTPPacketBuffer = newRTPPacketMultiBuffer(uint64(sc.s.ReadBufferCount))
ss.tcpConn.tcpProcessFunc = sc.tcpProcessRecord
// when recording, writeBuffer is only used to send RTCP receiver reports,

60
serverudpl.go
View File

@@ -15,6 +15,26 @@ import (
"github.com/aler9/gortsplib/pkg/multibuffer"
)
type rtpPacketMultiBuffer struct {
count uint64
buffers []rtp.Packet
cur uint64
}
func newRTPPacketMultiBuffer(count uint64) *rtpPacketMultiBuffer {
buffers := make([]rtp.Packet, count)
return &rtpPacketMultiBuffer{
count: count,
buffers: buffers,
}
}
func (mb *rtpPacketMultiBuffer) next() *rtp.Packet {
ret := &mb.buffers[mb.cur%mb.count]
mb.cur++
return ret
}
type clientData struct {
ss *ServerSession
trackID int
@@ -40,14 +60,15 @@ func (p *clientAddr) fill(ip net.IP, port int) {
type serverUDPListener struct {
s *Server
pc *net.UDPConn
listenIP net.IP
isRTP bool
writeTimeout time.Duration
readBuffer *multibuffer.MultiBuffer
clientsMutex sync.RWMutex
clients map[clientAddr]*clientData
processFunc func(*clientData, []byte)
pc *net.UDPConn
listenIP net.IP
isRTP bool
writeTimeout time.Duration
readBuffer *multibuffer.MultiBuffer
rtpPacketBuffer *rtpPacketMultiBuffer
clientsMutex sync.RWMutex
clients map[clientAddr]*clientData
processFunc func(*clientData, []byte)
readerDone chan struct{}
}
@@ -135,14 +156,15 @@ func newServerUDPListener(
}
u := &serverUDPListener{
s: s,
pc: pc,
listenIP: listenIP,
clients: make(map[clientAddr]*clientData),
isRTP: isRTP,
writeTimeout: s.WriteTimeout,
readBuffer: multibuffer.New(uint64(s.ReadBufferCount), uint64(s.ReadBufferSize)),
readerDone: make(chan struct{}),
s: s,
pc: pc,
listenIP: listenIP,
clients: make(map[clientAddr]*clientData),
isRTP: isRTP,
writeTimeout: s.WriteTimeout,
readBuffer: multibuffer.New(uint64(s.ReadBufferCount), uint64(s.ReadBufferSize)),
rtpPacketBuffer: newRTPPacketMultiBuffer(uint64(s.ReadBufferCount)),
readerDone: make(chan struct{}),
}
if isRTP {
@@ -196,7 +218,7 @@ func (u *serverUDPListener) runReader() {
}
func (u *serverUDPListener) processRTP(clientData *clientData, payload []byte) {
var pkt rtp.Packet
pkt := u.rtpPacketBuffer.next()
err := pkt.Unmarshal(payload)
if err != nil {
return
@@ -204,13 +226,13 @@ func (u *serverUDPListener) processRTP(clientData *clientData, payload []byte) {
now := time.Now()
atomic.StoreInt64(clientData.ss.udpLastFrameTime, now.Unix())
clientData.ss.announcedTracks[clientData.trackID].rtcpReceiver.ProcessPacketRTP(now, &pkt)
clientData.ss.announcedTracks[clientData.trackID].rtcpReceiver.ProcessPacketRTP(now, pkt)
if h, ok := u.s.Handler.(ServerHandlerOnPacketRTP); ok {
h.OnPacketRTP(&ServerHandlerOnPacketRTPCtx{
Session: clientData.ss,
TrackID: clientData.trackID,
Packet: &pkt,
Packet: pkt,
})
}
}