perf(cmpp): process inbound submits within client window

gateway/third_party/gocmpp/server.go

@@ -20,6 +20,7 @@ import (
 	"log"
 	"net"
 	"os"
+	"sync"
 	"sync/atomic"
 	"time"
 )
@@ -80,6 +81,9 @@ type Server struct {
 	// standard logger.
 	ErrorLog *log.Logger
 	OnClose  func(*Conn)
+	// SubmitWindow resolves the authenticated client's allowed in-flight Submit
+	// count. A nil resolver, or a value below two, preserves serial handling.
+	SubmitWindow func(*Conn) int
 }
 
 // A conn represents the server side of a Cmpp connection.
@@ -394,7 +398,14 @@ func (c *conn) serve() {
 		}
 	}()
 
+	var submitGroup sync.WaitGroup
+	var submitSlots chan struct{}
+	fatal := make(chan error, 1)
 	defer func() {
+		// Why wait: a handler may persist the message and register receipt routing
+		// after the peer disconnects. Session cleanup must run after every accepted
+		// in-flight request finishes, otherwise a late handler can recreate stale state.
+		submitGroup.Wait()
 		c.close()
 		if c.server.OnClose != nil {
 			c.server.OnClose(c.Conn)
@@ -408,6 +419,8 @@ func (c *conn) serve() {
 		select {
 		case <-c.exceed:
 			return // close the connection.
+		case <-fatal:
+			return
 		default:
 		}
 
@@ -426,29 +439,81 @@ func (c *conn) serve() {
 			break
 		}
 
-		_, err = c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog)
-		err1 := c.finishPacket(r)
-		if r.AfterSend != nil {
-			r.AfterSend(err1)
+		if isSubmitPacket(r.Packet.Packer) && c.submitWindow() > 1 {
+			if submitSlots == nil {
+				submitSlots = make(chan struct{}, c.submitWindow())
+			}
+			select {
+			case submitSlots <- struct{}{}:
+			case <-c.exceed:
+				return
+			case <-fatal:
+				return
+			}
+			submitGroup.Add(1)
+			go func(response *Response) {
+				defer submitGroup.Done()
+				defer func() { <-submitSlots }()
+				if handleErr := c.handlePacket(response); handleErr != nil {
+					select {
+					case fatal <- handleErr:
+					default:
+					}
+				}
+			}(r)
+			continue
 		}
-		if err1 != nil {
-			c.server.ErrorLog.Printf(
-				"send response packet failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
-				c.Conn.RemoteAddr(), c.Conn.Typ, r.Packer, r.SeqId, err1, err1,
-			)
-			break
-		}
-
-		if err != nil {
-			c.server.ErrorLog.Printf(
-				"handler failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
-				c.Conn.RemoteAddr(), c.Conn.Typ, r.Packet.Packer, r.SeqId, err, err,
-			)
+		if err = c.handlePacket(r); err != nil {
 			break
 		}
 	}
 }
 
+func (c *conn) submitWindow() int {
+	if c.server.SubmitWindow == nil {
+		return 1
+	}
+	window := c.server.SubmitWindow(c.Conn)
+	if window < 1 {
+		return 1
+	}
+	if window > 1024 {
+		return 1024
+	}
+	return window
+}
+
+func isSubmitPacket(packet Packer) bool {
+	switch packet.(type) {
+	case *Cmpp2SubmitReqPkt, *Cmpp3SubmitReqPkt:
+		return true
+	default:
+		return false
+	}
+}
+
+func (c *conn) handlePacket(r *Response) error {
+	_, handlerErr := c.server.Handler.ServeCmpp(r, r.Packet, c.server.ErrorLog)
+	sendErr := c.finishPacket(r)
+	if r.AfterSend != nil {
+		r.AfterSend(sendErr)
+	}
+	if sendErr != nil {
+		c.server.ErrorLog.Printf(
+			"send response packet failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
+			c.Conn.RemoteAddr(), c.Conn.Typ, r.Packer, r.SeqId, sendErr, sendErr,
+		)
+		return sendErr
+	}
+	if handlerErr != nil {
+		c.server.ErrorLog.Printf(
+			"handler failed remote=%v protocol=%s packet_type=%T seq=%d err_type=%T err=%v",
+			c.Conn.RemoteAddr(), c.Conn.Typ, r.Packet.Packer, r.SeqId, handlerErr, handlerErr,
+		)
+	}
+	return handlerErr
+}
+
 // Create new connection from rwc.
 func (srv *Server) newConn(rwc net.Conn) (c *conn, err error) {
 	c = new(conn)
@@ -480,6 +545,12 @@ func ListenAndServe(addr string, typ Type, t time.Duration, n int32, logWriter i
 // ListenAndServeWithClose behaves like ListenAndServe and invokes onClose once
 // after an accepted client connection ends, including abrupt TCP disconnects.
 func ListenAndServeWithClose(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), handlers ...Handler) error {
+	return ListenAndServeWithCloseAndSubmitWindow(addr, typ, t, n, logWriter, onClose, nil, handlers...)
+}
+
+// ListenAndServeWithCloseAndSubmitWindow adds bounded per-connection Submit
+// concurrency while keeping login, heartbeat and acknowledgement handling serial.
+func ListenAndServeWithCloseAndSubmitWindow(addr string, typ Type, t time.Duration, n int32, logWriter io.Writer, onClose func(*Conn), submitWindow func(*Conn) int, handlers ...Handler) error {
 	if addr == "" {
 		return ErrEmptyServerAddr
 	}
@@ -504,7 +575,7 @@ func ListenAndServeWithClose(addr string, typ Type, t time.Duration, n int32, lo
 	}
 	server := &Server{Addr: addr, Handler: handler, Typ: typ,
 		T: t, N: n,
-		ErrorLog: log.New(logWriter, "cmppserver: ", log.LstdFlags), OnClose: onClose}
+		ErrorLog: log.New(logWriter, "cmppserver: ", log.LstdFlags), OnClose: onClose, SubmitWindow: submitWindow}
 	return server.listenAndServe()
 }