Merge pull request #12 from alpacahq/TRD-858

TRD-898: update quickfix with commits of original quickfix repo

.gitignore

@@ -1,6 +1,7 @@
 *~
 *.swp
 *.swo
+.idea
 vendor
 _test/test
 _test/echo_server

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,17 @@ 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 +101,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 +138,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 +165,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,18 +213,43 @@ 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)
 	}
 }
 
+//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)
+		clone[0] = f[0]
+		to.tagLookup[tag] = clone
+	}
+	to.tags = make([]Tag, len(m.tags))
+	copy(to.tags, m.tags)
+	to.compare = m.compare
+}
+
 // DeleteTag removes a tag's value from field map, if present
 func (m *FieldMap) DeleteTag(tag Tag) {
+	m.rwLock.RLock()
+	defer m.rwLock.RUnlock()
+
 	delete(m.tagLookup, tag)
 }
 
 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)
@@ -214,6 +259,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
@@ -234,6 +282,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())
@@ -248,6 +299,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)
@@ -256,6 +310,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 {
@@ -271,6 +328,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 {

parser.go

@@ -5,15 +5,12 @@ import (
 	"errors"
 	"io"
 	"time"
-
-	"github.com/alpacahq/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:]

repeating_group.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 {

resend_state.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

session.go

@@ -43,7 +43,6 @@ type session struct {
 	transportDataDictionary *datadictionary.DataDictionary
 	appDataDictionary       *datadictionary.DataDictionary
 
-	messagePool
 	timestampPrecision TimestampPrecision
 }
 
@@ -664,14 +663,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 {
@@ -724,12 +715,34 @@ func (s *session) onAdmin(msg interface{}) {
 
 func (s *session) run() {
 	s.Start(s)
+	var stopChan = make(chan struct{})
+	s.stateTimer = internal.NewEventTimer(func() {
+		select {
+		//deadlock in write to chan s.sessionEvent after s.Stopped()==true and end of loop session.go:766 because no reader of chan s.sessionEvent
+		case s.sessionEvent <- internal.NeedHeartbeat:
+		case <-stopChan:
+		}
+	})
+	s.peerTimer = internal.NewEventTimer(func() {
+		select {
+		//deadlock in write to chan s.sessionEvent after s.Stopped()==true and end of loop session.go:766 because no reader of chan s.sessionEvent
+		case s.sessionEvent <- internal.PeerTimeout:
+		case <-stopChan:
+		}
+
+	})
+
+	// Without this sleep the ticker will be aligned at the millisecond which
+	// corresponds to the creation of the session. If the session creation
+	// happened at 07:00:00.678 and the session StartTime is 07:30:00, any new
+	// connection received between 07:30:00.000 and 07:30:00.677 will be
+	// rejected. Aligning the ticker with a round second fixes that.
+	time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second)))
 
-	s.stateTimer = internal.NewEventTimer(func() { s.sessionEvent <- internal.NeedHeartbeat })
-	s.peerTimer = internal.NewEventTimer(func() { s.sessionEvent <- internal.PeerTimeout })
 	ticker := time.NewTicker(time.Second)
 
 	defer func() {
+		close(stopChan)
 		s.stateTimer.Stop()
 		s.peerTimer.Stop()
 		ticker.Stop()

session_state.go

@@ -73,17 +73,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)))
 }