Parser performance & correctness improvements

Makefile

@@ -7,19 +7,19 @@ testnew:
 	go test -cover -run=TestMatchNew -v
 
 bench:
-	go test -run=XXX -bench=BenchmarkMatch -benchtime=1m -v
+	go test -run=XXX -bench=BenchmarkMatch8552595443 -benchmem -benchtime=10x -count=10 -v
 
 cover:
 	go test -cover -coverpkg github.com/dotabuff/manta,github.com/dotabuff/manta/vbkv -coverprofile /tmp/manta.cov -v
 	go tool cover -html=/tmp/manta.cov
 
 cpuprofile:
-	go test -v -run=TestMatch2159568145 -test.cpuprofile=/tmp/manta.cpuprof
+	go test -v -run=TestMatchNew8552595443 -test.cpuprofile=/tmp/manta.cpuprof
 	go tool pprof -svg -output=/tmp/manta.cpuprof.svg manta.test /tmp/manta.cpuprof
 	open /tmp/manta.cpuprof.svg
 
 memprofile:
-	go test -v -run=TestMatch2159568145 -test.memprofile=/tmp/manta.memprof -test.memprofilerate=1
+	go test -v -run=TestMatchNew8552595443 -test.memprofile=/tmp/manta.memprof -test.memprofilerate=1
 	go tool pprof --alloc_space manta.test /tmp/manta.memprof
 
 update: update-protobufs generate

class.go

@@ -16,6 +16,12 @@ type class struct {
 	classId    int32
 	name       string
 	serializer *serializer
+
+	// fpCache and fpNoop memoize name -> field-path resolution, which is
+	// class-invariant. They are shared by every entity of the class so a name
+	// is resolved once per class instead of once per entity.
+	fpCache map[string]*fieldPath
+	fpNoop  map[string]bool
 }
 
 func (c *class) getNameForFieldPath(fp *fieldPath) string {
@@ -67,6 +73,8 @@ func (p *Parser) onCDemoClassInfo(m *dota.CDemoClassInfo) error {
 			classId:    classId,
 			name:       networkName,
 			serializer: p.serializers[networkName],
+			fpCache:    make(map[string]*fieldPath),
+			fpNoop:     make(map[string]bool),
 		}
 		p.classesById[class.classId] = class
 		p.classesByName[class.name] = class
@@ -101,4 +109,8 @@ func (p *Parser) updateInstanceBaseline() {
 		}
 		p.classBaselines[classId] = item.Value
 	}
+
+	// Decoded baseline templates may now be stale; drop them so they are
+	// re-decoded lazily from the updated bytes on the next entity creation.
+	clear(p.classBaselineStates)
 }

demo_packet.go

@@ -14,7 +14,7 @@ type pendingMessage struct {
 }
 
 // Calculates the priority of the message. Lower is more important.
-func (m *pendingMessage) priority() int {
+func (m pendingMessage) priority() int {
 	switch m.t {
 	case
 		// These messages provide context needed for the rest of the tick
@@ -42,7 +42,7 @@ func (m *pendingMessage) priority() int {
 }
 
 // Provides a sortable structure for storing messages in the same packet.
-type pendingMessages []*pendingMessage
+type pendingMessages []pendingMessage
 
 func (ms pendingMessages) Len() int      { return len(ms) }
 func (ms pendingMessages) Swap(i, j int) { ms[i], ms[j] = ms[j], ms[i] }
@@ -60,9 +60,12 @@ func (ms pendingMessages) Less(i, j int) bool {
 // multiple inner packets from a single CDemoPacket. This is the main structure
 // that contains all other data types in the demo file.
 func (p *Parser) onCDemoPacket(m *dota.CDemoPacket) error {
-	// Create a slice to store pending mesages. Messages are read first as
-	// pending messages then sorted before dispatch.
-	ms := make(pendingMessages, 0, 2)
+	// Reuse a parser-level buffer to store pending messages. Messages are read
+	// first as pending messages then sorted before dispatch. onCDemoPacket is
+	// never re-entrant (it is dispatched only via callByDemoType, never nested
+	// within a callByPacketType call), so a single reused backing array is safe
+	// and avoids a heap allocation per embedded message.
+	ms := p.pendingMsgBuf[:0]
 
 	// Read all messages from the buffer. Messages are packed serially as
 	// {type, size, data}. We keep reading until until less than a byte remains.
@@ -71,21 +74,29 @@ func (p *Parser) onCDemoPacket(m *dota.CDemoPacket) error {
 		t := int32(r.readUBitVar())
 		size := r.readVarUint32()
 		buf := r.readBytes(size)
-		ms = append(ms, &pendingMessage{p.Tick, t, buf})
+		ms = append(ms, pendingMessage{p.Tick, t, buf})
 	}
 
 	// Sort messages to ensure dependencies are met. For example, we need to
-	// process string tables before game events that may reference them.
-	sort.Sort(ms)
-
-	// Dispatch messages in order, returning on handler error.
-	for _, m := range ms {
-		if err := p.Callbacks.callByPacketType(m.t, m.buf); err != nil {
-			return err
+	// process string tables before game events that may reference them. A
+	// stable sort keeps equal-priority messages in their original file order
+	// and avoids the reflection allocations of sort.Sort's interface path.
+	sort.Stable(ms)
+
+	// Dispatch messages in order, stopping on handler error.
+	var err error
+	for i := range ms {
+		if err = p.Callbacks.callByPacketType(ms[i].t, ms[i].buf); err != nil {
+			break
 		}
 	}
 
-	return nil
+	// Release the inner-packet buffer references and keep the slice at length
+	// zero so the reused backing array does not retain packet data. This runs on
+	// every path (success or error), before the single return.
+	clear(ms)
+	p.pendingMsgBuf = ms[:0]
+	return err
 }
 
 // Internal parser for callback OnCDemoFullPacket.

entity.go

@@ -49,25 +49,21 @@ type EntityHandler func(*Entity, EntityOp) error
 
 // Entity represents a single game entity in the replay
 type Entity struct {
-	index   int32
-	serial  int32
-	class   *class
-	active  bool
-	state   *fieldState
-	fpCache map[string]*fieldPath
-	fpNoop  map[string]bool
+	index  int32
+	serial int32
+	class  *class
+	active bool
+	state  *fieldState
 }
 
 // newEntity returns a new entity for the given index, serial and class
 func newEntity(index, serial int32, class *class) *Entity {
 	return &Entity{
-		index:   index,
-		serial:  serial,
-		class:   class,
-		active:  true,
-		state:   newFieldState(),
-		fpCache: make(map[string]*fieldPath),
-		fpNoop:  make(map[string]bool),
+		index:  index,
+		serial: serial,
+		class:  class,
+		active: true,
+		state:  newFieldState(),
 	}
 }
 
@@ -92,20 +88,20 @@ func (e *Entity) Dump() {
 
 // Get returns the current value of the Entity state for the given key
 func (e *Entity) Get(name string) interface{} {
-	if fp, ok := e.fpCache[name]; ok {
+	if fp, ok := e.class.fpCache[name]; ok {
 		return e.state.get(fp)
 	}
-	if e.fpNoop[name] {
+	if e.class.fpNoop[name] {
 		return nil
 	}
 
 	fp := newFieldPath()
 	if !e.class.getFieldPathForName(fp, name) {
-		e.fpNoop[name] = true
+		e.class.fpNoop[name] = true
 		fp.release()
 		return nil
 	}
-	e.fpCache[name] = fp
+	e.class.fpCache[name] = fp
 
 	return e.state.get(fp)
 }
@@ -218,9 +214,17 @@ func (p *Parser) FilterEntity(fb func(*Entity) bool) []*Entity {
 	return entities
 }
 
+// entityOpTuple pairs an entity with the operation performed on it. Updates are
+// buffered during a PacketEntities message, then dispatched to handlers.
+type entityOpTuple struct {
+	e  *Entity
+	op EntityOp
+}
+
 // Internal Callback for OnCSVCMsg_PacketEntities.
 func (p *Parser) onCSVCMsg_PacketEntities(m *dota.CSVCMsg_PacketEntities) error {
-	r := newReader(m.GetEntityData())
+	r := &p.entityReader
+	r.reset(m.GetEntityData())
 
 	var index = int32(-1)
 	var updates = int(m.GetUpdatedEntries())
@@ -237,11 +241,7 @@ func (p *Parser) onCSVCMsg_PacketEntities(m *dota.CSVCMsg_PacketEntities) error
 		p.entityFullPackets++
 	}
 
-	type tuple struct {
-		e  *Entity
-		op EntityOp
-	}
-	tuples := make([]tuple, 0, updates)
+	tuples := p.entityTuples[:0]
 
 	for ; updates > 0; updates-- {
 		index += int32(r.readUBitVar()) + 1
@@ -259,15 +259,24 @@ func (p *Parser) onCSVCMsg_PacketEntities(m *dota.CSVCMsg_PacketEntities) error
 					_panicf("unable to find new class %d", classId)
 				}
 
-				baseline := p.classBaselines[classId]
+				// Decode the class baseline once into a reusable template and
+				// clone it for each new entity, instead of re-decoding the raw
+				// baseline bytes on every creation.
+				baseline := p.classBaselineStates[classId]
 				if baseline == nil {
-					_panicf("unable to find new baseline %d", classId)
+					raw := p.classBaselines[classId]
+					if raw == nil {
+						_panicf("unable to find new baseline %d", classId)
+					}
+					baseline = newFieldState()
+					p.fpBuf = readFields(newReader(raw), class.serializer, baseline, p.fpBuf)
+					p.classBaselineStates[classId] = baseline
 				}
 
 				e = newEntity(index, serial, class)
+				e.state = baseline.clone()
 				p.entities[index] = e
-				readFields(newReader(baseline), class.serializer, e.state)
-				readFields(r, class.serializer, e.state)
+				p.fpBuf = readFields(r, class.serializer, e.state, p.fpBuf)
 				op = EntityOpCreated | EntityOpEntered
 
 			} else {
@@ -281,7 +290,7 @@ func (p *Parser) onCSVCMsg_PacketEntities(m *dota.CSVCMsg_PacketEntities) error
 					op |= EntityOpEntered
 				}
 
-				readFields(r, e.class.serializer, e.state)
+				p.fpBuf = readFields(r, e.class.serializer, e.state, p.fpBuf)
 			}
 
 		} else {
@@ -300,18 +309,26 @@ func (p *Parser) onCSVCMsg_PacketEntities(m *dota.CSVCMsg_PacketEntities) error
 			}
 		}
 
-		tuples = append(tuples, tuple{e, op})
+		tuples = append(tuples, entityOpTuple{e, op})
 	}
 
+	var err error
+dispatch:
 	for _, h := range p.entityHandlers {
-		for _, t := range tuples {
-			if err := h(t.e, t.op); err != nil {
-				return err
+		for i := range tuples {
+			if err = h(tuples[i].e, tuples[i].op); err != nil {
+				break dispatch
 			}
 		}
 	}
 
-	return nil
+	// Release the entity references and keep the buffer at length zero so a
+	// reused backing array does not retain (possibly deleted) entities and their
+	// state. This runs on every path (success or error), before the single
+	// return.
+	clear(tuples)
+	p.entityTuples = tuples[:0]
+	return err
 }
 
 // OnEntity registers an EntityHandler that will be called when an entity

field.go

@@ -252,7 +252,7 @@ func (f *field) getFieldPaths(fp *fieldPath, state *fieldState) []*fieldPath {
 		if sub, ok := state.get(fp).(*fieldState); ok {
 			fp.last++
 			for i, v := range sub.state {
-				if v != nil {
+				if !v.isEmpty() {
 					fp.path[fp.last] = i
 					x = append(x, fp.copy())
 				}
@@ -271,7 +271,7 @@ func (f *field) getFieldPaths(fp *fieldPath, state *fieldState) []*fieldPath {
 		if sub, ok := state.get(fp).(*fieldState); ok {
 			fp.last++
 			for i, v := range sub.state {
-				if v != nil {
+				if !v.isEmpty() {
 					fp.path[fp.last] = i
 					x = append(x, fp.copy())
 				}
@@ -283,7 +283,7 @@ func (f *field) getFieldPaths(fp *fieldPath, state *fieldState) []*fieldPath {
 		if sub, ok := state.get(fp).(*fieldState); ok {
 			fp.last += 2
 			for i, v := range sub.state {
-				if vv, ok := v.(*fieldState); ok {
+				if vv, ok := v.sub(); ok {
 					fp.path[fp.last-1] = i
 					x = append(x, f.serializer.getFieldPaths(fp, vv)...)
 				}