server: save RAM by releasing read buffers earlier

serverconn.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"errors"
 	"net"
 	"net/url"
 	"strings"
@@ -34,18 +35,13 @@ type ServerConn struct {
 	s    *Server
 	conn net.Conn
 
-	ctx                context.Context
+	ctx        context.Context
-	ctxCancel          func()
+	ctxCancel  func()
-	remoteAddr         *net.TCPAddr
+	remoteAddr *net.TCPAddr
-	br                 *bufio.Reader
+	br         *bufio.Reader
-	sessions           map[string]*ServerSession
+	sessions   map[string]*ServerSession
-	tcpFrameEnabled    bool
+	readFunc   func(readRequest chan readReq) error
-	tcpSession         *ServerSession
+	tcpSession *ServerSession
-	tcpFrameTimeout    bool
-	tcpReadBuffer      *multibuffer.MultiBuffer
-	tcpRTPPacketBuffer *rtpPacketMultiBuffer
-	tcpProcessFunc     func(int, bool, []byte)
-	tcpWriterRunning   bool
 
 	// in
 	sessionRemove chan *ServerSession
@@ -76,6 +72,8 @@ func newServerConn(
 		done:          make(chan struct{}),
 	}
 
+	sc.readFunc = sc.readFuncStandard
+
 	s.wg.Add(1)
 	go sc.run()
 
@@ -117,77 +115,7 @@ func (sc *ServerConn) run() {
 	readRequest := make(chan readReq)
 	readErr := make(chan error)
 	readDone := make(chan struct{})
-	go func() {
+	go sc.runReader(readRequest, readErr, readDone)
-		defer close(readDone)
-		err := func() error {
-			var req base.Request
-			var frame base.InterleavedFrame
-
-			for {
-				if sc.tcpFrameEnabled {
-					if sc.tcpFrameTimeout {
-						sc.conn.SetReadDeadline(time.Now().Add(sc.s.ReadTimeout))
-					}
-
-					frame.Payload = sc.tcpReadBuffer.Next()
-					what, err := base.ReadInterleavedFrameOrRequest(&frame, &req, sc.br)
-					if err != nil {
-						return err
-					}
-
-					switch what.(type) {
-					case *base.InterleavedFrame:
-						channel := frame.Channel
-						isRTP := true
-						if (channel % 2) != 0 {
-							channel--
-							isRTP = false
-						}
-
-						// forward frame only if it has been set up
-						if trackID, ok := sc.tcpSession.tcpTracksByChannel[channel]; ok {
-							sc.tcpProcessFunc(trackID, isRTP, frame.Payload)
-						}
-
-					case *base.Request:
-						cres := make(chan error)
-						select {
-						case readRequest <- readReq{req: &req, res: cres}:
-							err := <-cres
-							if err != nil {
-								return err
-							}
-
-						case <-sc.ctx.Done():
-							return liberrors.ErrServerTerminated{}
-						}
-					}
-				} else {
-					err := req.Read(sc.br)
-					if err != nil {
-						return err
-					}
-
-					cres := make(chan error)
-					select {
-					case readRequest <- readReq{req: &req, res: cres}:
-						err = <-cres
-						if err != nil {
-							return err
-						}
-
-					case <-sc.ctx.Done():
-						return liberrors.ErrServerTerminated{}
-					}
-				}
-			}
-		}()
-
-		select {
-		case readErr <- err:
-		case <-sc.ctx.Done():
-		}
-	}()
 
 	err := func() error {
 		for {
@@ -239,53 +167,165 @@ func (sc *ServerConn) run() {
 	}
 }
 
-func (sc *ServerConn) tcpProcessPlay(trackID int, isRTP bool, payload []byte) {
+var errSwitchReadFunc = errors.New("switch read function")
-	if !isRTP {
+
-		packets, err := rtcp.Unmarshal(payload)
+func (sc *ServerConn) runReader(readRequest chan readReq, readErr chan error, readDone chan struct{}) {
-		if err != nil {
+	defer close(readDone)
-			return
+
+	for {
+		err := sc.readFunc(readRequest)
+
+		if err == errSwitchReadFunc {
+			continue
 		}
 
-		if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTCP); ok {
+		select {
-			for _, pkt := range packets {
+		case readErr <- err:
-				h.OnPacketRTCP(&ServerHandlerOnPacketRTCPCtx{
+		case <-sc.ctx.Done():
-					Session: sc.tcpSession,
+		}
-					TrackID: trackID,
+		break
-					Packet:  pkt,
+	}
-				})
+}
+
+func (sc *ServerConn) readFuncStandard(readRequest chan readReq) error {
+	// reset deadline
+	sc.conn.SetReadDeadline(time.Time{})
+
+	var req base.Request
+
+	for {
+		err := req.Read(sc.br)
+		if err != nil {
+			return err
+		}
+
+		cres := make(chan error)
+		select {
+		case readRequest <- readReq{req: &req, res: cres}:
+			err = <-cres
+			if err != nil {
+				return err
 			}
+
+		case <-sc.ctx.Done():
+			return liberrors.ErrServerTerminated{}
 		}
 	}
 }
 
-func (sc *ServerConn) tcpProcessRecord(trackID int, isRTP bool, payload []byte) {
+func (sc *ServerConn) readFuncTCP(readRequest chan readReq) error {
-	if isRTP {
+	// reset deadline
-		pkt := sc.tcpRTPPacketBuffer.next()
+	sc.conn.SetReadDeadline(time.Time{})
-		err := pkt.Unmarshal(payload)
-		if err != nil {
-			return
-		}
 
-		if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTP); ok {
+	select {
-			h.OnPacketRTP(&ServerHandlerOnPacketRTPCtx{
+	case sc.tcpSession.startWriter <- struct{}{}:
-				Session: sc.tcpSession,
+	case <-sc.tcpSession.ctx.Done():
-				TrackID: trackID,
+	}
-				Packet:  pkt,
+
-			})
+	var tcpReadBuffer *multibuffer.MultiBuffer
+	var processFunc func(int, bool, []byte)
+
+	if sc.tcpSession.state == ServerSessionStateRead {
+		// when playing, tcpReadBuffer is only used to receive RTCP receiver reports,
+		// that are much smaller than RTP packets and are sent at a fixed interval.
+		// decrease RAM consumption by allocating less buffers.
+		tcpReadBuffer = multibuffer.New(8, uint64(sc.s.ReadBufferSize))
+
+		processFunc = func(trackID int, isRTP bool, payload []byte) {
+			if !isRTP {
+				packets, err := rtcp.Unmarshal(payload)
+				if err != nil {
+					return
+				}
+
+				if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTCP); ok {
+					for _, pkt := range packets {
+						h.OnPacketRTCP(&ServerHandlerOnPacketRTCPCtx{
+							Session: sc.tcpSession,
+							TrackID: trackID,
+							Packet:  pkt,
+						})
+					}
+				}
+			}
 		}
 	} else {
-		packets, err := rtcp.Unmarshal(payload)
+		tcpReadBuffer = multibuffer.New(uint64(sc.s.ReadBufferCount), uint64(sc.s.ReadBufferSize))
-		if err != nil {
+		tcpRTPPacketBuffer := newRTPPacketMultiBuffer(uint64(sc.s.ReadBufferCount))
-			return
+
+		processFunc = func(trackID int, isRTP bool, payload []byte) {
+			if isRTP {
+				pkt := tcpRTPPacketBuffer.next()
+				err := pkt.Unmarshal(payload)
+				if err != nil {
+					return
+				}
+
+				if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTP); ok {
+					h.OnPacketRTP(&ServerHandlerOnPacketRTPCtx{
+						Session: sc.tcpSession,
+						TrackID: trackID,
+						Packet:  pkt,
+					})
+				}
+			} else {
+				packets, err := rtcp.Unmarshal(payload)
+				if err != nil {
+					return
+				}
+
+				if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTCP); ok {
+					for _, pkt := range packets {
+						h.OnPacketRTCP(&ServerHandlerOnPacketRTCPCtx{
+							Session: sc.tcpSession,
+							TrackID: trackID,
+							Packet:  pkt,
+						})
+					}
+				}
+			}
+		}
+	}
+
+	var req base.Request
+	var frame base.InterleavedFrame
+
+	for {
+		if sc.tcpSession.state == ServerSessionStatePublish {
+			sc.conn.SetReadDeadline(time.Now().Add(sc.s.ReadTimeout))
 		}
 
-		if h, ok := sc.s.Handler.(ServerHandlerOnPacketRTCP); ok {
+		frame.Payload = tcpReadBuffer.Next()
-			for _, pkt := range packets {
+		what, err := base.ReadInterleavedFrameOrRequest(&frame, &req, sc.br)
-				h.OnPacketRTCP(&ServerHandlerOnPacketRTCPCtx{
+		if err != nil {
-					Session: sc.tcpSession,
+			return err
-					TrackID: trackID,
+		}
-					Packet:  pkt,
+
-				})
+		switch what.(type) {
+		case *base.InterleavedFrame:
+			channel := frame.Channel
+			isRTP := true
+			if (channel % 2) != 0 {
+				channel--
+				isRTP = false
+			}
+
+			// forward frame only if it has been set up
+			if trackID, ok := sc.tcpSession.tcpTracksByChannel[channel]; ok {
+				processFunc(trackID, isRTP, frame.Payload)
+			}
+
+		case *base.Request:
+			cres := make(chan error)
+			select {
+			case readRequest <- readReq{req: &req, res: cres}:
+				err := <-cres
+				if err != nil {
+					return err
+				}
+
+			case <-sc.ctx.Done():
+				return liberrors.ErrServerTerminated{}
 			}
 		}
 	}
@@ -503,15 +543,6 @@ func (sc *ServerConn) handleRequestOuter(req *base.Request) error {
 	sc.conn.SetWriteDeadline(time.Now().Add(sc.s.WriteTimeout))
 	sc.conn.Write(buf.Bytes())
 
-	// start writer after sending the response
-	if sc.tcpFrameEnabled && !sc.tcpWriterRunning {
-		sc.tcpWriterRunning = true
-		select {
-		case sc.tcpSession.startWriter <- struct{}{}:
-		case <-sc.tcpSession.ctx.Done():
-		}
-	}
-
 	return err
 }
 

serversession.go

@@ -17,7 +17,6 @@ import (
 	"github.com/aler9/gortsplib/pkg/base"
 	"github.com/aler9/gortsplib/pkg/headers"
 	"github.com/aler9/gortsplib/pkg/liberrors"
-	"github.com/aler9/gortsplib/pkg/multibuffer"
 	"github.com/aler9/gortsplib/pkg/ringbuffer"
 	"github.com/aler9/gortsplib/pkg/rtcpreceiver"
 )
@@ -883,17 +882,12 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
 		default: // TCP
 			ss.tcpConn = sc
 			ss.tcpConn.tcpSession = ss
-			ss.tcpConn.tcpFrameEnabled = true
+
-			ss.tcpConn.tcpFrameTimeout = false
+			ss.tcpConn.readFunc = ss.tcpConn.readFuncTCP
-			// when playing, tcpReadBuffer is only used to receive RTCP receiver reports,
+			err = errSwitchReadFunc
-			// that are much smaller than RTP packets and are sent at a fixed interval.
-			// decrease RAM consumption by allocating less buffers.
-			ss.tcpConn.tcpReadBuffer = multibuffer.New(8, uint64(sc.s.ReadBufferSize))
-			ss.tcpConn.tcpProcessFunc = sc.tcpProcessPlay
 
 			ss.writeBuffer = ringbuffer.New(uint64(ss.s.ReadBufferCount))
-			// run writer after sending the response
+			// runWriter() is called by conn after sending the response
-			ss.tcpConn.tcpWriterRunning = false
 		}
 
 		// add RTP-Info
@@ -1016,18 +1010,15 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
 		default: // TCP
 			ss.tcpConn = sc
 			ss.tcpConn.tcpSession = ss
-			ss.tcpConn.tcpFrameEnabled = true
+
-			ss.tcpConn.tcpFrameTimeout = true
+			ss.tcpConn.readFunc = ss.tcpConn.readFuncTCP
-			ss.tcpConn.tcpReadBuffer = multibuffer.New(uint64(sc.s.ReadBufferCount), uint64(sc.s.ReadBufferSize))
+			err = errSwitchReadFunc
-			ss.tcpConn.tcpRTPPacketBuffer = newRTPPacketMultiBuffer(uint64(sc.s.ReadBufferCount))
-			ss.tcpConn.tcpProcessFunc = sc.tcpProcessRecord
 
 			// when recording, writeBuffer is only used to send RTCP receiver reports,
 			// that are much smaller than RTP packets and are sent at a fixed interval.
 			// decrease RAM consumption by allocating less buffers.
 			ss.writeBuffer = ringbuffer.New(uint64(8))
-			// run writer after sending the response
+			// runWriter() is called by conn after sending the response
-			ss.tcpConn.tcpWriterRunning = false
 		}
 
 		return res, err
@@ -1089,9 +1080,10 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
 			case TransportUDPMulticast:
 
 			default: // TCP
+				ss.tcpConn.readFunc = ss.tcpConn.readFuncStandard
+				err = errSwitchReadFunc
+
 				ss.tcpConn.tcpSession = nil
-				ss.tcpConn.tcpFrameEnabled = false
-				ss.tcpConn.tcpReadBuffer = nil
 				ss.tcpConn = nil
 			}
 
@@ -1108,10 +1100,10 @@ func (ss *ServerSession) handleRequest(sc *ServerConn, req *base.Request) (*base
 			case TransportUDPMulticast:
 
 			default: // TCP
+				ss.tcpConn.readFunc = ss.tcpConn.readFuncStandard
+				err = errSwitchReadFunc
+
 				ss.tcpConn.tcpSession = nil
-				ss.tcpConn.tcpFrameEnabled = false
-				ss.tcpConn.tcpReadBuffer = nil
-				ss.tcpConn.conn.SetReadDeadline(time.Time{})
 				ss.tcpConn = nil
 			}
 		}