Huge performance improvements

- Don't wait for messages to be consumed to continue parsing
- Queue up (almost) all the things!
- Configurable msgQueue buffer size
- But: Cancel now won't stop the already parsed info from being used

Author: markus-wa

demoinfocs_test.go

@@ -102,8 +102,6 @@ func TestCancelParseToEnd(t *testing.T) {
 		tix++
 		if tix == maxTicks {
 			p.Cancel()
-		} else if tix > maxTicks {
-			t.Fatal("Parsing continued after cancellation")
 		}
 	})
 

demopacket.go

@@ -11,18 +11,6 @@ import (
 	msg "github.com/markus-wa/demoinfocs-golang/msg"
 )
 
-var packetEntitiesPool sync.Pool = sync.Pool{
-	New: func() interface{} {
-		return new(msg.CSVCMsg_PacketEntities)
-	},
-}
-
-var gameEventPool sync.Pool = sync.Pool{
-	New: func() interface{} {
-		return new(msg.CSVCMsg_GameEvent)
-	},
-}
-
 var byteSlicePool sync.Pool = sync.Pool{
 	New: func() interface{} {
 		s := make([]byte, 0, 256)
@@ -46,15 +34,16 @@ func (p *Parser) parsePacket() {
 		var m proto.Message
 		switch cmd {
 		case int(msg.SVC_Messages_svc_PacketEntities):
-			m = packetEntitiesPool.Get().(*msg.CSVCMsg_PacketEntities)
-			defer packetEntitiesPool.Put(m)
+			// TODO: Find a way to pool SVC_Messages_svc_PacketEntities
+			// Need to make sure the message was consumed before pooling
+			// and the message's contents will be overridden (either by protobuf or manually)
+			m = new(msg.CSVCMsg_PacketEntities)
 
 		case int(msg.SVC_Messages_svc_GameEventList):
 			m = new(msg.CSVCMsg_GameEventList)
 
 		case int(msg.SVC_Messages_svc_GameEvent):
-			m = gameEventPool.Get().(*msg.CSVCMsg_GameEvent)
-			defer gameEventPool.Put(m)
+			m = new(msg.CSVCMsg_GameEvent)
 
 		case int(msg.SVC_Messages_svc_CreateStringTable):
 			m = new(msg.CSVCMsg_CreateStringTable)
@@ -101,9 +90,6 @@ func (p *Parser) parsePacket() {
 		p.bitReader.EndChunk()
 	}
 	p.bitReader.EndChunk()
-
-	// Make sure the created events are consumed so they can be pooled
-	p.msgDispatcher.SyncQueues(p.msgQueue)
 }
 
 // TODO: Find out what all this is good for and why we didn't use the removed functions on seVector, split & commandInfo

packet_handlers.go

@@ -673,3 +673,45 @@ func (p *Parser) handleUserMessage(um *msg.CSVCMsg_UserMessage) {
 		// Maybe msg.ECstrike15UserMessages_CS_UM_RadioText
 	}
 }
+
+type frameParsedTokenType struct{}
+
+var frameParsedToken = new(frameParsedTokenType)
+
+func (p *Parser) handleFrameParsed(*frameParsedTokenType) {
+	for k, rp := range p.rawPlayers {
+		if rp == nil {
+			continue
+		}
+
+		if pl := p.players[k]; pl != nil {
+			newPlayer := false
+			if p.connectedPlayers[rp.UserID] == nil {
+				p.connectedPlayers[rp.UserID] = pl
+				newPlayer = true
+			}
+
+			pl.Name = rp.Name
+			pl.SteamID = rp.XUID
+			pl.IsBot = rp.IsFakePlayer
+			pl.AdditionalPlayerInformation = &p.additionalPlayerInfo[pl.EntityID]
+
+			if pl.IsAlive() {
+				pl.LastAlivePosition = pl.Position
+			}
+
+			if newPlayer && pl.SteamID != 0 {
+				p.eventDispatcher.Dispatch(events.PlayerBindEvent{Player: pl})
+			}
+		}
+	}
+
+	p.currentFrame++
+	p.eventDispatcher.Dispatch(events.TickDoneEvent{})
+}
+
+type ingameTickNumber int
+
+func (p *Parser) handleIngameTickNumber(n ingameTickNumber) {
+	p.ingameTick = int(n)
+}

parser.go

@@ -162,6 +162,8 @@ func (ts TeamState) Flag() string {
 	return ts.flag
 }
 
+// TODO: Maybe we should use a channel instead of that WarnHandler stuff
+
 // WarnHandler is a function that handles warnings of a Parser.
 type WarnHandler func(string)
 
@@ -170,15 +172,26 @@ func WarnToStdErr(warning string) {
 	fmt.Fprintln(os.Stderr, warning)
 }
 
+// TODO: Change the New* methods (names + parameters)
+
 // NewParser creates a new Parser on the basis of an io.Reader
 // - like os.File or bytes.Reader - that reads demo data.
 // Any warnings that don't stop the Parser from doing it's job
 // will be passed to the warnHandler if it's not nil.
 func NewParser(demostream io.Reader, warnHandler WarnHandler) *Parser {
+	return NewParserWithBufferSize(demostream, -1, warnHandler)
+}
+
+// NewParserWithBufferSize returns a new Parser with a custom msgQueue buffer size.
+// For large demos, fast i/o and slow CPUs higher numbers are suggested and vice versa.
+// The buffer size can easily be in the hundred-thousands to low millions for the best performance.
+// A negative value will make the Parser automatically decide the buffer size during ParseHeader()
+// based on the number of ticks in the demo (nubmer of ticks = buffer size).
+// See also: NewParser()
+func NewParserWithBufferSize(demostream io.Reader, msgQueueBufferSize int, warnHandler WarnHandler) *Parser {
 	var p Parser
 	// Init parser
 	p.bitReader = bit.NewLargeBitReader(demostream)
-	p.msgQueue = make(chan interface{}, 8)
 	p.instanceBaselines = make(map[int][]byte)
 	p.preprocessedBaselines = make(map[int]map[int]st.PropValue)
 	p.equipmentMapping = make(map[*st.ServerClass]common.EquipmentElement)
@@ -195,7 +208,16 @@ func NewParser(demostream io.Reader, warnHandler WarnHandler) *Parser {
 	p.msgDispatcher.RegisterHandler(p.handleCreateStringTable)
 	p.msgDispatcher.RegisterHandler(p.handleUpdateStringTable)
 	p.msgDispatcher.RegisterHandler(p.handleUserMessage)
+	p.msgDispatcher.RegisterHandler(p.handleFrameParsed)
+	p.msgDispatcher.RegisterHandler(p.handleIngameTickNumber)
 
-	p.msgDispatcher.AddQueues(p.msgQueue)
+	if msgQueueBufferSize >= 0 {
+		p.initMsgQueue(msgQueueBufferSize)
+	}
 	return &p
 }
+
+func (p *Parser) initMsgQueue(buf int) {
+	p.msgQueue = make(chan interface{}, buf)
+	p.msgDispatcher.AddQueues(p.msgQueue)
+}

parsing.go

@@ -15,6 +15,10 @@ const (
 	playerWeaponPrePrefix = "bcc_nonlocaldata."
 )
 
+const (
+	msgHeaderNotParsed = "Tried to parse tick before parsing header"
+)
+
 // ParseHeader attempts to parse the header of the demo.
 // Returns error if the filestamp (first 8 bytes) doesn't match HL2DEMO.
 func (p *Parser) ParseHeader() error {
@@ -35,6 +39,12 @@ func (p *Parser) ParseHeader() error {
 		return errors.New("Invalid File-Type; expecting HL2DEMO in the first 8 bytes")
 	}
 
+	// Initialize queue if the buffer size wasn't specified, the amount of ticks
+	// seems to be a good indicator of how many events we'll get
+	if p.msgQueue == nil {
+		p.initMsgQueue(h.PlaybackTicks)
+	}
+
 	p.header = &h
 	p.eventDispatcher.Dispatch(events.HeaderParsedEvent{Header: h})
 	return nil
@@ -44,20 +54,30 @@ func (p *Parser) ParseHeader() error {
 // Aborts and returns an error if Cancel() is called before the end.
 // May panic if the demo is corrupt in some way.
 func (p *Parser) ParseToEnd() error {
+	if p.header == nil {
+		panic(msgHeaderNotParsed)
+	}
+
 	for {
 		select {
 		case <-p.cancelChan:
 			return errors.New("Parsing was cancelled before it finished")
 
 		default:
-			if !p.ParseNextFrame() {
+			if !p.parseFrame() {
+				// Make sure all the messages of the demo are handled
+				p.msgDispatcher.SyncQueues(p.msgQueue)
+
+				// Close msgQueue
+				close(p.msgQueue)
 				return nil
 			}
 		}
 	}
 }
 
-// Cancel aborts ParseToEnd() on the upcoming tick.
+// Cancel aborts ParseToEnd(). All information that was already read
+// up to this point will still be used (and new events may still be sent).
 func (p *Parser) Cancel() {
 	p.cancelChan <- struct{}{}
 }
@@ -67,56 +87,30 @@ func (p *Parser) Cancel() {
 // Panics if header hasn't been parsed yet - see Parser.ParseHeader().
 func (p *Parser) ParseNextFrame() bool {
 	if p.header == nil {
-		panic("Tried to parse tick before parsing header")
+		panic(msgHeaderNotParsed)
 	}
-	b := p.parseFrame()
-
-	for k, rp := range p.rawPlayers {
-		if rp == nil {
-			continue
-		}
-
-		if pl := p.players[k]; pl != nil {
-			newPlayer := false
-			if p.connectedPlayers[rp.UserID] == nil {
-				p.connectedPlayers[rp.UserID] = pl
-				newPlayer = true
-			}
 
-			pl.Name = rp.Name
-			pl.SteamID = rp.XUID
-			pl.IsBot = rp.IsFakePlayer
-			pl.AdditionalPlayerInformation = &p.additionalPlayerInfo[pl.EntityID]
-
-			if pl.IsAlive() {
-				pl.LastAlivePosition = pl.Position
-			}
-
-			if newPlayer && pl.SteamID != 0 {
-				p.eventDispatcher.Dispatch(events.PlayerBindEvent{Player: pl})
-			}
-		}
-	}
+	b := p.parseFrame()
 
-	p.eventDispatcher.Dispatch(events.TickDoneEvent{})
+	// Make sure all the messages of the frame are handled
+	p.msgDispatcher.SyncQueues(p.msgQueue)
 
+	// Close msgQueue if we are done
 	if !b {
 		close(p.msgQueue)
 	}
-
 	return b
 }
 
 func (p *Parser) parseFrame() bool {
 	cmd := demoCommand(p.bitReader.ReadSingleByte())
 
-	// Ingame tick number
-	p.ingameTick = p.bitReader.ReadSignedInt(32)
+	// Send ingame tick number update
+	p.msgQueue <- ingameTickNumber(p.bitReader.ReadSignedInt(32))
+
 	// Skip 'player slot'
 	p.bitReader.ReadSingleByte()
 
-	p.currentFrame++
-
 	switch cmd {
 	case dcSynctick:
 		// Ignore
@@ -130,6 +124,8 @@ func (p *Parser) parseFrame() bool {
 		p.bitReader.EndChunk()
 
 	case dcDataTables:
+		p.msgDispatcher.SyncQueues(p.msgQueue)
+
 		p.bitReader.BeginChunk(p.bitReader.ReadSignedInt(32) << 3)
 		p.stParser.ParsePacket(p.bitReader)
 		p.bitReader.EndChunk()
@@ -138,6 +134,8 @@ func (p *Parser) parseFrame() bool {
 		p.bindEntities()
 
 	case dcStringTables:
+		p.msgDispatcher.SyncQueues(p.msgQueue)
+
 		p.parseStringTables()
 
 	case dcUserCommand:
@@ -158,5 +156,9 @@ func (p *Parser) parseFrame() bool {
 	default:
 		panic(fmt.Sprintf("Canny handle it anymoe (command %v unknown)", cmd))
 	}
+
+	// Queue up some post processing
+	p.msgQueue <- frameParsedToken
+
 	return true
 }