Merge pull request #485 from waheedoo/make_field_map_thread_safe

Solve concurrent map read and map write by making Fieldmap thread-safe & avoiding recycling of addresses

Author: ackleymi

field_map.go

@@ -3,6 +3,7 @@ package quickfix
 import (
 	"bytes"
 	"sort"
+	"sync"
 	"time"
 )
 
@@ -39,6 +40,7 @@ func (t tagSort) Less(i, j int) bool { return t.compare(t.tags[i], t.tags[j]) }
 type FieldMap struct {
 	tagLookup map[Tag]field
 	tagSort
+	rwLock *sync.RWMutex
 }
 
 // ascending tags
@@ -49,12 +51,16 @@ func (m *FieldMap) init() {
 }
 
 func (m *FieldMap) initWithOrdering(ordering tagOrder) {
+	m.rwLock = &sync.RWMutex{}
 	m.tagLookup = make(map[Tag]field)
 	m.compare = ordering
 }
 
 //Tags returns all of the Field Tags in this FieldMap
 func (m FieldMap) Tags() []Tag {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	tags := make([]Tag, 0, len(m.tagLookup))
 	for t := range m.tagLookup {
 		tags = append(tags, t)
@@ -70,12 +76,18 @@ func (m FieldMap) Get(parser Field) MessageRejectError {
 
 //Has returns true if the Tag is present in this FieldMap
 func (m FieldMap) Has(tag Tag) bool {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	_, ok := m.tagLookup[tag]
 	return ok
 }
 
 //GetField parses of a field with Tag tag. Returned reject may indicate the field is not present, or the field value is invalid.
 func (m FieldMap) GetField(tag Tag, parser FieldValueReader) MessageRejectError {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	f, ok := m.tagLookup[tag]
 	if !ok {
 		return ConditionallyRequiredFieldMissing(tag)
@@ -90,6 +102,9 @@ func (m FieldMap) GetField(tag Tag, parser FieldValueReader) MessageRejectError
 
 //GetBytes is a zero-copy GetField wrapper for []bytes fields
 func (m FieldMap) GetBytes(tag Tag) ([]byte, MessageRejectError) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	f, ok := m.tagLookup[tag]
 	if !ok {
 		return nil, ConditionallyRequiredFieldMissing(tag)
@@ -124,6 +139,9 @@ func (m FieldMap) GetInt(tag Tag) (int, MessageRejectError) {
 
 //GetTime is a GetField wrapper for utc timestamp fields
 func (m FieldMap) GetTime(tag Tag) (t time.Time, err MessageRejectError) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	bytes, err := m.GetBytes(tag)
 	if err != nil {
 		return
@@ -148,6 +166,9 @@ func (m FieldMap) GetString(tag Tag) (string, MessageRejectError) {
 
 //GetGroup is a Get function specific to Group Fields.
 func (m FieldMap) GetGroup(parser FieldGroupReader) MessageRejectError {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	f, ok := m.tagLookup[parser.Tag()]
 	if !ok {
 		return ConditionallyRequiredFieldMissing(parser.Tag())
@@ -193,6 +214,9 @@ func (m *FieldMap) SetString(tag Tag, value string) *FieldMap {
 
 //Clear purges all fields from field map
 func (m *FieldMap) Clear() {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
 	m.tags = m.tags[0:0]
 	for k := range m.tagLookup {
 		delete(m.tagLookup, k)
@@ -201,6 +225,9 @@ func (m *FieldMap) Clear() {
 
 //CopyInto overwrites the given FieldMap with this one
 func (m *FieldMap) CopyInto(to *FieldMap) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	to.tagLookup = make(map[Tag]field)
 	for tag, f := range m.tagLookup {
 		clone := make(field, 1)
@@ -213,6 +240,9 @@ func (m *FieldMap) CopyInto(to *FieldMap) {
 }
 
 func (m *FieldMap) add(f field) {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
 	t := fieldTag(f)
 	if _, ok := m.tagLookup[t]; !ok {
 		m.tags = append(m.tags, t)
@@ -222,6 +252,9 @@ func (m *FieldMap) add(f field) {
 }
 
 func (m *FieldMap) getOrCreate(tag Tag) field {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
 	if f, ok := m.tagLookup[tag]; ok {
 		f = f[:1]
 		return f
@@ -242,6 +275,9 @@ func (m *FieldMap) Set(field FieldWriter) *FieldMap {
 
 //SetGroup is a setter specific to group fields
 func (m *FieldMap) SetGroup(field FieldGroupWriter) *FieldMap {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
 	_, ok := m.tagLookup[field.Tag()]
 	if !ok {
 		m.tags = append(m.tags, field.Tag())
@@ -256,6 +292,9 @@ func (m *FieldMap) sortedTags() []Tag {
 }
 
 func (m FieldMap) write(buffer *bytes.Buffer) {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
 	for _, tag := range m.sortedTags() {
 		if f, ok := m.tagLookup[tag]; ok {
 			writeField(f, buffer)
@@ -264,6 +303,9 @@ func (m FieldMap) write(buffer *bytes.Buffer) {
 }
 
 func (m FieldMap) total() int {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	total := 0
 	for _, fields := range m.tagLookup {
 		for _, tv := range fields {
@@ -279,6 +321,9 @@ func (m FieldMap) total() int {
 }
 
 func (m FieldMap) length() int {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	length := 0
 	for _, fields := range m.tagLookup {
 		for _, tv := range fields {

internal/buffer_pool.go

@@ -5,15 +5,12 @@ import (
 	"errors"
 	"io"
 	"time"
-
-	"github.com/quickfixgo/quickfix/internal"
 )
 
 const (
 	defaultBufSize = 4096
 )
 
-var bufferPool internal.BufferPool
 
 type parser struct {
 	//buffer is a slice of bigBuffer
@@ -145,7 +142,7 @@ func (p *parser) ReadMessage() (msgBytes *bytes.Buffer, err error) {
 		return
 	}
 
-	msgBytes = bufferPool.Get()
+	msgBytes = new(bytes.Buffer)
 	msgBytes.Reset()
 	msgBytes.Write(p.buffer[:index])
 	p.buffer = p.buffer[index:]

message_pool.go

@@ -114,12 +114,13 @@ func (f RepeatingGroup) Write() []TagValue {
 
 	for _, group := range f.groups {
 		tags := group.sortedTags()
-
+		group.rwLock.RLock()
 		for _, tag := range tags {
 			if fields, ok := group.tagLookup[tag]; ok {
 				tvs = append(tvs, fields...)
 			}
 		}
+		group.rwLock.RUnlock()
 	}
 
 	return tvs
@@ -199,7 +200,9 @@ func (f *RepeatingGroup) Read(tv []TagValue) ([]TagValue, error) {
 			f.groups = append(f.groups, group)
 		}
 
+		group.rwLock.Lock()
 		group.tagLookup[tvRange[0].tag] = tvRange
+		group.rwLock.Unlock()
 	}
 
 	if len(f.groups) != expectedGroupSize {

parser.go

@@ -54,9 +54,6 @@ func (s resendState) FixMsgIn(session *session, msg *Message) (nextState session
 
 		delete(s.messageStash, targetSeqNum)
 
-		//return stashed message to pool
-		session.returnToPool(msg)
-
 		nextState = inSession{}.FixMsgIn(session, msg)
 		if !nextState.IsLoggedOn() {
 			return

repeating_group.go

@@ -43,7 +43,6 @@ type session struct {
 	transportDataDictionary *datadictionary.DataDictionary
 	appDataDictionary       *datadictionary.DataDictionary
 
-	messagePool
 	timestampPrecision TimestampPrecision
 }
 
@@ -678,14 +677,6 @@ type fixIn struct {
 	receiveTime time.Time
 }
 
-func (s *session) returnToPool(msg *Message) {
-	s.messagePool.Put(msg)
-	if msg.rawMessage != nil {
-		bufferPool.Put(msg.rawMessage)
-		msg.rawMessage = nil
-	}
-}
-
 func (s *session) onDisconnect() {
 	s.log.OnEvent("Disconnected")
 	if s.ResetOnDisconnect {

resend_state.go

@@ -68,17 +68,14 @@ func (sm *stateMachine) Incoming(session *session, m fixIn) {
 
 	session.log.OnIncoming(m.bytes.Bytes())
 
-	msg := session.messagePool.Get()
+	msg := NewMessage()
 	if err := ParseMessageWithDataDictionary(msg, m.bytes, session.transportDataDictionary, session.appDataDictionary); err != nil {
 		session.log.OnEventf("Msg Parse Error: %v, %q", err.Error(), m.bytes)
 	} else {
 		msg.ReceiveTime = m.receiveTime
 		sm.fixMsgIn(session, msg)
 	}
 
-	if !msg.keepMessage {
-		session.returnToPool(msg)
-	}
 	session.peerTimer.Reset(time.Duration(float64(1.2) * float64(session.HeartBtInt)))
 }