Support batching outgoing messages

[Yu-Xie]

Prototype for #555

• Support batching mode when sending application messages to target by adding a func SaveMessagesAndIncrNextSenderMsgSeqNum(seqNum int, msg [][]byte) in the storage interface, and then a func SendAppToTarget(m []Messagable, sessionID SessionID) error in registry.go to expose it.
• It's supported in SQLStore and MemoryStore in this PR. If we like this direction, I'm happy to implement FileStore and MongoStore too.

[Yu-Xie]

Posting some load testing results using local Postgres:

Output:

Spent 78.082917ms seconds to send 100 messages with 10 batch size and 10 batches
Spent 14.086666ms seconds to send 100 messages with 50 batch size and 2 batches
Spent 61.315208ms seconds to send 500 messages with 50 batch size and 10 batches
Spent 35.222625ms seconds to send 500 messages with 100 batch size and 5 batches
Spent 64.8035ms seconds to send 1000 messages with 100 batch size and 10 batches

Note that this is much faster compare to existing implementation (#555) which takes 1.5 SECONDS to prepare 1000 messages for sending.

Source code:

package quickfix

import (
	"fmt"
	"strconv"
	"sync/atomic"
	"testing"
	"time"

	"github.com/quickfixgo/quickfix/config"
	"github.com/stretchr/testify/require"

	_ "github.com/lib/pq"
)

func TestBatchSendMessage(t *testing.T) {
	const driver = "postgres"
	const dsn = `host=127.0.0.1 port=5432 user=xxx password=password dbname=xxx sslmode=disable`
	settings := NewSettings()
	sessionSettings := NewSessionSettings()
	sessionSettings.Set(config.BeginString, BeginStringFIX42)
	sessionSettings.Set(config.SenderCompID, "mock_initiator")
	sessionSettings.Set(config.TargetCompID, "mock_acceptor")
	sessionSettings.Set(config.HeartBtInt, "10")
	sessionSettings.Set(config.SocketConnectPort, "12345")
	sessionSettings.Set(config.SocketConnectHost, "127.0.0.1")
	sessionSettings.Set(config.SQLStoreDriver, driver)
	sessionSettings.Set(config.SQLStoreDataSourceName, dsn)
	sid, err := settings.AddSession(sessionSettings)
	require.NoError(t, err)

	factory := NewSQLStoreFactory(settings)

	initiator, err := NewInitiator(&app{}, factory, settings, NewNullLogFactory())
	require.NoError(t, err)

	err = initiator.Start()
	require.NoError(t, err)

	time.Sleep(10 * time.Second)

	require.NoError(t, batchSend(10, 100, sid))
	require.NoError(t, batchSend(50, 100, sid))
	require.NoError(t, batchSend(50, 500, sid))
	require.NoError(t, batchSend(100, 500, sid))
	require.NoError(t, batchSend(100, 1000, sid))

	time.Sleep(20 * time.Second)
	t.Fail() // to see logs
}

func batchSend(buf, n int, sessionID SessionID) error {
	if buf > n {
		buf = n
	}
	ch := make(chan Messagable, n)
	for _, o := range newOrders(n) {
		select {
		case ch <- o:
		}
	}
	// batch
	b := make([]Messagable, 0, buf)
	done := 0
	numBatches := 0
	start := time.Now()
	for {
		select {
		case o := <-ch:
			b = append(b, o)
			if len(b) == buf {
				if err := SendAppToTarget(b, sessionID); err != nil {
					return err
				}
				numBatches++
				done += buf
				b = make([]Messagable, 0, buf)
			}
		}
		if done == n {
			latency := time.Since(start)
			fmt.Printf("Spent %s seconds to send %d messages with %d batch size and %d batches\n", latency.String(), n, buf, numBatches)
			return nil
		}
	}
}

var cnt atomic.Int32

func newOrders(n int) []Messagable {
	res := make([]Messagable, 0, n)
	for i := 0; i < n; i++ {
		m := NewMessage()
		cnt.Add(1)
		m.Header.Set(w(35, "D"))
		m.Body.Set(w(38, "101.0"))
		m.Body.Set(w(11, strconv.Itoa(int(cnt.Load()))))
		m.Body.Set(w(55, "symbol"))
		m.Body.Set(w(54, "1"))
		m.Body.Set(w(59, "0"))
		m.Body.Set(w(40, "1"))
		res = append(res, m.ToMessage())
	}
	return res
}

func w(tag int, s string) FieldWriter {
	return &writer{tag: Tag(tag), s: s}
}

type writer struct {
	tag Tag
	s   string
}

func (w *writer) Tag() Tag {
	return w.tag
}

func (w *writer) Write() []byte {
	return []byte(w.s)
}

type app struct {
}

func (a app) OnCreate(sessionID SessionID) {
}

func (a app) OnLogon(sessionID SessionID) {
}

func (a app) OnLogout(sessionID SessionID) {
}

func (a app) ToAdmin(message *Message, sessionID SessionID) {
}

func (a app) ToApp(message *Message, sessionID SessionID) error {
	return nil
}

func (a app) FromAdmin(message *Message, sessionID SessionID) MessageRejectError {
	return nil
}

func (a app) FromApp(message *Message, sessionID SessionID) MessageRejectError {
	return nil
}

[ackleymi]

@Yu-Xie if you would change SendAppToTarget to SendBatchToTarget and SaveMessagesAndIncrNextSenderMsgSeqNum to SaveBatchAndIncrNextSenderMsgSeqNum I will merge. I'd like to keep func names immediately obvious to prevent accidental misuse. Also if the FileStore and MongoStore impls are easy enough to implement for you, please do so.

[Yu-Xie]

@Yu-Xie if you would change SendAppToTarget to SendBatchToTarget and SaveMessagesAndIncrNextSenderMsgSeqNum to SaveBatchAndIncrNextSenderMsgSeqNum I will merge. I'd like to keep func names immediately obvious to prevent accidental misuse. Also if the FileStore and MongoStore impls are easy enough to implement for you, please do so.

Makes sense -- I have addressed these feedbacks in #599

[Neal]

this can be moved out of the loop as we don't release the lock until we return so it is not very useful to notify each time