Optimize performance

.gitignore

@@ -7,3 +7,4 @@ _test/echo_server
 _test/tmp
 _vendor*
 gen
+.idea

acceptor.go

@@ -10,12 +10,14 @@ import (
 	"strconv"
 	"sync"
 
-	"github.com/armon/go-proxyproto"
+	proxyproto "github.com/armon/go-proxyproto"
+
 	"github.com/quickfixgo/quickfix/config"
 )
 
 //Acceptor accepts connections from FIX clients and manages the associated sessions.
 type Acceptor struct {
+	sync.RWMutex
 	app                   Application
 	settings              *Settings
 	logFactory            LogFactory
@@ -84,7 +86,6 @@ func (a *Acceptor) Start() error {
 			return err
 		}
 	}
-
 	for sessionID := range a.sessions {
 		session := a.sessions[sessionID]
 		a.sessionGroup.Add(1)
@@ -125,7 +126,9 @@ func (a *Acceptor) Stop() {
 
 //Get remote IP address for a given session.
 func (a *Acceptor) RemoteAddr(sessionID SessionID) (net.Addr, bool) {
+	a.RLock()
 	addr, ok := a.sessionAddr[sessionID]
+	a.RUnlock()
 	return addr, ok
 }
 
@@ -303,10 +306,11 @@ func (a *Acceptor) handleConnection(netConn net.Conn) {
 		session = dynamicSession
 		defer session.stop()
 	}
-
+	a.Lock()
 	a.sessionAddr[sessID] = netConn.RemoteAddr()
-	msgIn := make(chan fixIn)
-	msgOut := make(chan []byte)
+	a.Unlock()
+	msgIn := make(chan fixIn, session.ReceiveQueueLength)
+	msgOut := make(chan []byte, session.SendQueueLength)
 
 	if err := session.connect(msgIn, msgOut); err != nil {
 		a.globalLog.OnEventf("Unable to accept %v", err.Error())
@@ -351,7 +355,9 @@ LOOP:
 		case id := <-complete:
 			session, ok := sessions[id]
 			if ok {
+				a.Lock()
 				delete(a.sessionAddr, session.sessionID)
+				a.Unlock()
 				delete(sessions, id)
 			} else {
 				a.globalLog.OnEventf("Missing dynamic session %v!", id)

config/configuration.go

@@ -64,4 +64,7 @@ const (
 	RejectInvalidMessage         string = "RejectInvalidMessage"
 	DynamicSessions              string = "DynamicSessions"
 	DynamicQualifier             string = "DynamicQualifier"
+	SendBufferSize               string = "SendBufferSize"
+	SendQueueLength              string = "SendQueueLength"
+	ReceiveQueueLength           string = "ReceiveQueueLength"
 )

in_session.go

@@ -239,6 +239,7 @@ func (state inSession) resendMessages(session *session, beginSeqNo, endSeqNo int
 
 		session.log.OnEventf("Resending Message: %v", sentMessageSeqNum)
 		msgBytes = msg.build()
+		session.log.OnOutgoing(msgBytes)
 		session.sendBytes(msgBytes)
 
 		seqNum = sentMessageSeqNum + 1
@@ -382,6 +383,7 @@ func (state *inSession) generateSequenceReset(session *session, beginSeqNo int,
 
 	msgBytes := sequenceReset.build()
 
+	session.log.OnOutgoing(msgBytes)
 	session.sendBytes(msgBytes)
 	session.log.OnEventf("Sent SequenceReset TO: %v", endSeqNo)
 

initiator.go

@@ -169,8 +169,8 @@ func (i *Initiator) handleConnection(session *session, tlsConfig *tls.Config, di
 			netConn = tlsConn
 		}
 
-		msgIn = make(chan fixIn)
-		msgOut = make(chan []byte)
+		msgIn = make(chan fixIn, session.ReceiveQueueLength)
+		msgOut = make(chan []byte, session.SendQueueLength)
 		if err := session.connect(msgIn, msgOut); err != nil {
 			session.log.OnEventf("Failed to initiate: %v", err)
 			goto reconnect

internal/session_settings.go

@@ -16,6 +16,9 @@ type SessionSettings struct {
 	SkipCheckLatency             bool
 	MaxLatency                   time.Duration
 	DisableMessagePersist        bool
+	SendBufferSize               int
+	SendQueueLength              int
+	ReceiveQueueLength           int
 
 	//required on logon for FIX.T.1 messages
 	DefaultApplVerID string

session.go

@@ -331,8 +331,20 @@ func (s *session) persist(seqNum int, msgBytes []byte) error {
 }
 
 func (s *session) sendQueued() {
+	var queueBuffer = make([]byte, 0, s.SendBufferSize)
 	for _, msgBytes := range s.toSend {
-		s.sendBytes(msgBytes)
+		s.log.OnOutgoing(msgBytes)
+		if len(queueBuffer)+len(msgBytes) < s.SendBufferSize {
+			queueBuffer = append(queueBuffer, msgBytes...)
+		} else {
+			s.sendBytes(queueBuffer)
+			queueBuffer = make([]byte, 0, s.SendBufferSize)
+			queueBuffer = append(queueBuffer, msgBytes...)
+		}
+	}
+
+	if len(queueBuffer) > 0 {
+		s.sendBytes(queueBuffer)
 	}
 
 	s.dropQueued()
@@ -343,7 +355,6 @@ func (s *session) dropQueued() {
 }
 
 func (s *session) sendBytes(msg []byte) {
-	s.log.OnOutgoing(msg)
 	s.messageOut <- msg
 	s.stateTimer.Reset(s.HeartBtInt)
 }

session_factory.go

@@ -282,6 +282,33 @@ func (f sessionFactory) newSession(
 		s.DisableMessagePersist = !persistMessages
 	}
 
+	s.SendBufferSize = 1000
+	if settings.HasSetting(config.SendBufferSize) {
+		var sendBufferSize int
+		if sendBufferSize, err = settings.IntSetting(config.SendBufferSize); err != nil {
+			return
+		}
+		s.SendBufferSize = sendBufferSize
+	}
+
+	s.SendQueueLength = 1
+	if settings.HasSetting(config.SendQueueLength) {
+		var sendQueueLength int
+		if sendQueueLength, err = settings.IntSetting(config.SendQueueLength); err != nil {
+			return
+		}
+		s.SendQueueLength = sendQueueLength
+	}
+
+	s.SendQueueLength = 1
+	if settings.HasSetting(config.ReceiveQueueLength) {
+		var receiveQueueLength int
+		if receiveQueueLength, err = settings.IntSetting(config.ReceiveQueueLength); err != nil {
+			return
+		}
+		s.ReceiveQueueLength = receiveQueueLength
+	}
+
 	if f.BuildInitiators {
 		if err = f.buildInitiatorSettings(s, settings); err != nil {
 			return