Add OutChanCapacity, and let the session wait for the writer when the outgoing buffer is full

acceptor.go

@@ -360,7 +360,7 @@ func (a *Acceptor) handleConnection(netConn net.Conn) {
 
 	a.sessionAddr.Store(sessID, netConn.RemoteAddr())
 	msgIn := make(chan fixIn, session.InChanCapacity)
-	msgOut := make(chan []byte)
+	msgOut := make(chan []byte, session.OutChanCapacity)
 
 	if err := session.connect(msgIn, msgOut); err != nil {
 		a.globalLog.OnEventf("Unable to accept session %v connection: %v", sessID, err.Error())
@@ -372,7 +372,7 @@ func (a *Acceptor) handleConnection(netConn net.Conn) {
 		readLoop(parser, msgIn, a.globalLog)
 	}()
 
-	writeLoop(netConn, msgOut, a.globalLog)
+	writeLoop(netConn, msgOut, a.globalLog, session.messageTaken)
 }
 
 func (a *Acceptor) dynamicSessionsLoop() {

config/configuration.go

@@ -499,6 +499,18 @@ const (
 	// Valid Values:
 	//  - A positive integer, or zero for an unbuffered channel
 	InChanCapacity string = "InChanCapacity"
+
+	// OutChanCapacity sets the maximum number of messages that can be buffered in the channel for outgoing FIX messages,
+	// between the session and the goroutine that writes them to the connection. With a buffer, a session that finds it
+	// full waits for the writer to take a message instead of retrying at once.
+	//
+	// Required: No
+	//
+	// Default: 0
+	//
+	// Valid Values:
+	//  - A positive integer, or zero for an unbuffered channel
+	OutChanCapacity string = "OutChanCapacity"
 )
 
 const (

connection.go

@@ -17,12 +17,17 @@ package quickfix
 
 import "io"
 
-func writeLoop(connection io.Writer, messageOut chan []byte, log Log) {
+// writeLoop writes each message taken from messageOut to the connection. taken, if set, is called once a
+// message has been taken, before it is written: the channel has room again from that moment.
+func writeLoop(connection io.Writer, messageOut chan []byte, log Log, taken func()) {
 	for {
 		msg, ok := <-messageOut
 		if !ok {
 			return
 		}
+		if taken != nil {
+			taken()
+		}
 
 		if _, err := connection.Write(msg); err != nil {
 			log.OnEvent(err.Error())

connection_internal_test.go

@@ -31,7 +31,7 @@ func TestWriteLoop(t *testing.T) {
 		msgOut <- []byte("test msg 3")
 		close(msgOut)
 	}()
-	writeLoop(writer, msgOut, nullLog{})
+	writeLoop(writer, msgOut, nullLog{}, nil)
 
 	expected := "test msg 1 test msg 2 test msg 3"
 

initiator.go

@@ -206,7 +206,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		}
 
 		msgIn = make(chan fixIn, session.InChanCapacity)
-		msgOut = make(chan []byte)
+		msgOut = make(chan []byte, session.OutChanCapacity)
 		if err := session.connect(msgIn, msgOut); err != nil {
 			session.log.OnEventf("Failed to initiate: %v", err)
 			goto reconnect
@@ -215,7 +215,7 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 		go readLoop(newParser(bufio.NewReader(netConn)), msgIn, session.log)
 		disconnected = make(chan interface{})
 		go func() {
-			writeLoop(netConn, msgOut, session.log)
+			writeLoop(netConn, msgOut, session.log, session.messageTaken)
 			if err := netConn.Close(); err != nil {
 				session.log.OnEvent(err.Error())
 			}

internal/session_settings.go

@@ -22,6 +22,7 @@ type SessionSettings struct {
 	ResetSeqTime                 time.Time
 	EnableResetSeqTime           bool
 	InChanCapacity               int
+	OutChanCapacity              int
 
 	// Required on logon for FIX.T.1 messages.
 	DefaultApplVerID string

out_chan_capacity_test.go

@@ -0,0 +1,177 @@
+// Copyright (c) quickfixengine.org  All rights reserved.
+//
+// This file may be distributed under the terms of the quickfixengine.org
+// license as defined by quickfixengine.org and appearing in the file
+// LICENSE included in the packaging of this file.
+//
+// This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING
+// THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A
+// PARTICULAR PURPOSE.
+//
+// See http://www.quickfixengine.org/LICENSE for licensing information.
+//
+// Contact ask@quickfixengine.org if any conditions of this licensing
+// are not clear to you.
+
+package quickfix
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/quickfixgo/quickfix/config"
+	"github.com/quickfixgo/quickfix/internal"
+)
+
+func newOutChanTestSession(capacity int) (*session, chan []byte) {
+	out := make(chan []byte, capacity)
+	s := &session{
+		log:          nullLog{},
+		messageOut:   out,
+		messageEvent: make(chan bool, 1),
+		stateTimer:   internal.NewEventTimer(func() {}),
+	}
+	s.OutChanCapacity = capacity
+	s.HeartBtInt = time.Minute
+	return s, out
+}
+
+// woken reports whether the session has been told to come round, and consumes the notice.
+func woken(s *session) bool {
+	select {
+	case <-s.messageEvent:
+		return true
+	default:
+		return false
+	}
+}
+
+func TestSendQueuedWithAnUnbufferedChannelComesRoundAtOnce(t *testing.T) {
+	s, _ := newOutChanTestSession(0)
+	defer s.stateTimer.Stop()
+
+	s.toSend = [][]byte{[]byte("1")}
+	s.sendQueued(false)
+
+	assert.Len(t, s.toSend, 1, "no writer stands at the channel, so the message stays queued")
+	assert.True(t, woken(s), "an unbuffered channel keeps today's behaviour: the session comes round at once")
+}
+
+func TestSendQueuedWaitsForTheWriterWhenTheBufferIsFull(t *testing.T) {
+	s, out := newOutChanTestSession(2)
+	defer s.stateTimer.Stop()
+
+	s.toSend = [][]byte{[]byte("1"), []byte("2"), []byte("3")}
+	s.sendQueued(false)
+	require.Len(t, s.toSend, 1, "two messages fit in the buffer and the third waits")
+	assert.False(t, woken(s), "nothing the session does makes room, so it must not come round")
+
+	s.sendQueued(false)
+	assert.False(t, woken(s), "coming round again before the writer took anything changes nothing")
+
+	<-out
+	s.messageTaken()
+	assert.True(t, woken(s), "the writer's take wakes the session")
+	s.sendQueued(false)
+	assert.Empty(t, s.toSend)
+
+	<-out
+	s.messageTaken()
+	assert.False(t, woken(s), "a take that no one waits for wakes nothing")
+}
+
+func TestWriteLoopTellsEachTake(t *testing.T) {
+	out := make(chan []byte, 3)
+	out <- []byte("a")
+	out <- []byte("b")
+	out <- []byte("c")
+	close(out)
+
+	takes := 0
+	var written bytes.Buffer
+	writeLoop(&written, out, nullLog{}, func() { takes++ })
+
+	assert.Equal(t, 3, takes)
+	assert.Equal(t, "abc", written.String())
+}
+
+type slowWriter struct {
+	delay   time.Duration
+	written [][]byte
+}
+
+func (w *slowWriter) Write(p []byte) (int, error) {
+	time.Sleep(w.delay)
+	w.written = append(w.written, append([]byte(nil), p...))
+	return len(p), nil
+}
+
+func TestABufferedChannelDeliversEveryMessageInOrderToASlowWriter(t *testing.T) {
+	const messages = 2000
+	s, out := newOutChanTestSession(4)
+	defer s.stateTimer.Stop()
+
+	w := &slowWriter{delay: 50 * time.Microsecond}
+	done := make(chan struct{})
+	go func() {
+		writeLoop(w, out, nullLog{}, s.messageTaken)
+		close(done)
+	}()
+
+	for i := 0; i < messages; i++ {
+		s.toSend = append(s.toSend, []byte(fmt.Sprintf("%05d", i)))
+	}
+
+	// The session goroutine's part of the run loop: come round when told to.
+	rounds := 0
+	s.sendQueued(false)
+	deadline := time.After(30 * time.Second)
+	for len(s.toSend) > 0 {
+		select {
+		case <-s.messageEvent:
+			rounds++
+			s.sendQueued(false)
+		case <-deadline:
+			t.Fatalf("%d messages still queued after %d rounds", len(s.toSend), rounds)
+		}
+	}
+	close(out)
+	<-done
+
+	require.Len(t, w.written, messages)
+	for i, msg := range w.written {
+		require.Equal(t, fmt.Sprintf("%05d", i), string(msg))
+	}
+	assert.LessOrEqual(t, rounds, messages, "each round follows a take, so there are no more rounds than messages")
+}
+
+func TestOutChanCapacitySetting(t *testing.T) {
+	build := func(value string) (*session, error) {
+		settings := NewSessionSettings()
+		settings.Set(config.BeginString, BeginStringFIX42)
+		settings.Set(config.SenderCompID, "S")
+		settings.Set(config.TargetCompID, "T")
+		settings.Set(config.HeartBtInt, "30")
+		if value != "" {
+			settings.Set(config.OutChanCapacity, value)
+		}
+		return sessionFactory{}.newSession(SessionID{BeginString: BeginStringFIX42, SenderCompID: "S", TargetCompID: "T"},
+			NewMemoryStoreFactory(), settings, NewNullLogFactory(), nil)
+	}
+
+	s, err := build("")
+	require.NoError(t, err)
+	assert.Equal(t, 0, s.OutChanCapacity, "unset, the channel stays unbuffered")
+
+	s, err = build("256")
+	require.NoError(t, err)
+	assert.Equal(t, 256, s.OutChanCapacity)
+
+	_, err = build("-1")
+	assert.Error(t, err, "a negative capacity is refused")
+}

session.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/quickfixgo/quickfix/datadictionary"
@@ -39,6 +40,10 @@ type session struct {
 	// Application messages are queued up for send here.
 	toSend [][]byte
 
+	// Set when a send found a buffered messageOut full; the writer clears it and wakes the session
+	// when it takes a message. See messageTaken.
+	waitingForRoom atomic.Bool
+
 	// Mutex for access to toSend.
 	sendMutex sync.Mutex
 	// Mutex to prevent messages being sent when resendRequest is active
@@ -411,16 +416,45 @@ func (s *session) persist(seqNum int, msgBytes []byte) error {
 
 func (s *session) sendQueued(blockUntilSent bool) {
 	for i, msgBytes := range s.toSend {
-		if !s.sendBytes(msgBytes, blockUntilSent) {
-			s.toSend = s.toSend[i:]
+		if s.sendBytes(msgBytes, blockUntilSent) {
+			continue
+		}
+		if s.OutChanCapacity > 0 && s.waitForRoom(msgBytes) {
+			continue
+		}
+		s.toSend = s.toSend[i:]
+		if s.OutChanCapacity == 0 {
 			s.notifyMessageOut()
-			return
 		}
+		return
 	}
 
 	s.dropQueued()
 }
 
+// waitForRoom is called when a send found a buffered messageOut full. Nothing the session goroutine does
+// makes room, so instead of coming round again at once it asks the writer to wake it when it takes a
+// message (messageTaken). The flag is set before the send is tried once more, so a message taken in
+// between is either seen by that try or wakes the session. It reports whether that try sent the message.
+func (s *session) waitForRoom(msg []byte) bool {
+	if s.messageOut == nil {
+		return false
+	}
+	s.waitingForRoom.Store(true)
+	if s.sendBytes(msg, false) {
+		s.waitingForRoom.Store(false)
+		return true
+	}
+	return false
+}
+
+// messageTaken is called by the writer each time it takes a message from messageOut.
+func (s *session) messageTaken() {
+	if s.waitingForRoom.CompareAndSwap(true, false) {
+		s.notifyMessageOut()
+	}
+}
+
 func (s *session) dropQueued() {
 	s.toSend = s.toSend[:0]
 }

session_factory.go

@@ -443,6 +443,15 @@ func (f sessionFactory) newSession(
 		s.InChanCapacity = 1
 	}
 
+	if settings.HasSetting(config.OutChanCapacity) {
+		if s.OutChanCapacity, err = settings.IntSetting(config.OutChanCapacity); err != nil {
+			return
+		} else if s.OutChanCapacity < 0 {
+			err = IncorrectFormatForSetting{Setting: config.OutChanCapacity, Value: []byte(strconv.Itoa(s.OutChanCapacity))}
+			return
+		}
+	}
+
 	if f.BuildInitiators {
 		if err = f.buildInitiatorSettings(s, settings); err != nil {
 			return