From 8c0ba4ad7ac808c86f70514e8445b22444660715 Mon Sep 17 00:00:00 2001 From: Nu0bG <145451708+Nu0bG@users.noreply.github.com> Date: Wed, 14 Oct 2026 13:55:18 +0700 Subject: [PATCH] Add OutChanCapacity: a buffered outgoing channel, and a session that waits for the writer when it is full With a buffer, a send that finds the channel full no longer makes the session loop come round at once. The session sets a flag and the writer wakes it when it takes a message. Unset (0), the channel stays unbuffered and the behaviour is unchanged. --- acceptor.go | 4 +- config/configuration.go | 12 +++ connection.go | 7 +- connection_internal_test.go | 2 +- initiator.go | 4 +- internal/session_settings.go | 1 + out_chan_capacity_test.go | 177 +++++++++++++++++++++++++++++++++++ session.go | 40 +++++++- session_factory.go | 9 ++ 9 files changed, 247 insertions(+), 9 deletions(-) create mode 100644 out_chan_capacity_test.go diff --git a/acceptor.go b/acceptor.go index 83384c5b8..6ffb3166c 100644 --- a/acceptor.go +++ b/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() { diff --git a/config/configuration.go b/config/configuration.go index 6282bab1e..f104bf2e1 100644 --- a/config/configuration.go +++ b/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 ( diff --git a/connection.go b/connection.go index 99a4c465e..968b09605 100644 --- a/connection.go +++ b/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()) diff --git a/connection_internal_test.go b/connection_internal_test.go index 081b3c110..b14ae6827 100644 --- a/connection_internal_test.go +++ b/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" diff --git a/initiator.go b/initiator.go index 9af90cb2b..c9c8fee8d 100644 --- a/initiator.go +++ b/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()) } diff --git a/internal/session_settings.go b/internal/session_settings.go index c45134822..c6d82aab3 100644 --- a/internal/session_settings.go +++ b/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 diff --git a/out_chan_capacity_test.go b/out_chan_capacity_test.go new file mode 100644 index 000000000..5fe46fbef --- /dev/null +++ b/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") +} diff --git a/session.go b/session.go index 7bcd3a112..b8f8dd313 100644 --- a/session.go +++ b/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] } diff --git a/session_factory.go b/session_factory.go index 60829ee42..e5acc74df 100644 --- a/session_factory.go +++ b/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