Merge pull request #5 from deelawn/feat/telemetry-new

Open telemetry and new op benchmarking

benchmarking/cmd/main.go

@@ -0,0 +1,101 @@
+package main
+
+import (
+	"encoding/binary"
+	"flag"
+	"fmt"
+	"os"
+	"sync"
+
+	"github.com/gnolang/gno/benchmarking"
+	"github.com/gnolang/gno/gnovm/pkg/gnolang"
+)
+
+const recordSize int = 10
+
+var pathFlag = flag.String("path", "", "the path to the benchmark file")
+
+func main() {
+	flag.Parse()
+
+	file, err := os.Open(*pathFlag)
+	if err != nil {
+		panic("could not create benchmark file: " + err.Error())
+	}
+
+	inputCh := make(chan []byte, 10000)
+	outputCh := make(chan string, 10000)
+	wg := sync.WaitGroup{}
+	numWorkers := 4
+	wg.Add(numWorkers)
+
+	doneCh := make(chan struct{})
+
+	for i := 0; i < numWorkers; i++ {
+		go func() {
+			for {
+				buf, ok := <-inputCh
+				if !ok {
+					break
+				}
+
+				opName := gnolang.Op(buf[0]).String()
+				if buf[1] != 0 {
+					opName = benchmarking.OpCodeString(buf[1])
+				}
+
+				elapsedTime := binary.LittleEndian.Uint32(buf[2:])
+				size := binary.LittleEndian.Uint32(buf[6:])
+				outputCh <- opName + "," + fmt.Sprint(elapsedTime) + "," + fmt.Sprint(size)
+			}
+			wg.Done()
+		}()
+	}
+
+	go func() {
+		out, err := os.Create("results.csv")
+		if err != nil {
+			panic("could not create readable output file: " + err.Error())
+		}
+
+		fmt.Fprintln(out, "op,elapsedTime,diskIOBytes")
+
+		for {
+			output, ok := <-outputCh
+			if !ok {
+				break
+			}
+
+			fmt.Fprintln(out, output)
+		}
+
+		out.Close()
+		doneCh <- struct{}{}
+	}()
+
+	var i int
+	bufSize := recordSize * 100000
+	for {
+		buf := make([]byte, bufSize)
+		if n, err := file.Read(buf); err != nil && n == 0 {
+			break
+		}
+
+		for j := 0; j < len(buf)/recordSize; j += recordSize {
+			inputCh <- buf[j : j+recordSize]
+		}
+
+		i += bufSize / recordSize
+		if i%1000 == 0 {
+			fmt.Println(i)
+		}
+	}
+
+	close(inputCh)
+	wg.Wait()
+	close(outputCh)
+	<-doneCh
+	close(doneCh)
+
+	fmt.Println("done")
+}

benchmarking/enabled.go

@@ -0,0 +1,13 @@
+package benchmarking
+
+var enabled bool
+
+func Enabled() bool {
+	return enabled
+}
+
+func Init(filepath string) {
+	enabled = true
+	initExporter(filepath)
+	initStack()
+}

benchmarking/exporter.go

@@ -0,0 +1,64 @@
+package benchmarking
+
+import (
+	"encoding/binary"
+	"os"
+	"time"
+)
+
+const flushTimerInterval = time.Duration(time.Second * 30)
+
+var fileWriter *exporter
+
+func initExporter(fileName string) {
+	file, err := os.Create(fileName)
+	if err != nil {
+		panic("could not create benchmark file: " + err.Error())
+	}
+
+	fileWriter = &exporter{
+		file:              file,
+		bytesToFlushAfter: 10 * 100000,
+		flushTimer:        *time.NewTimer(flushTimerInterval),
+	}
+
+	go func() {
+		for {
+			<-fileWriter.flushTimer.C
+			fileWriter.file.Sync()
+			fileWriter.flushTimer.Reset(flushTimerInterval)
+		}
+	}()
+}
+
+type exporter struct {
+	file              *os.File
+	bytesWritten      int
+	bytesToFlushAfter int
+	flushTimer        time.Timer
+}
+
+func (e *exporter) export(opCode OpCode, elapsedTime time.Duration, size uint32) {
+	buf := []byte{opCode[0], opCode[1], 0, 0, 0, 0, 0, 0, 0, 0}
+	binary.LittleEndian.PutUint32(buf[2:], uint32(elapsedTime))
+	binary.LittleEndian.PutUint32(buf[6:], size)
+	n, err := e.file.Write(buf)
+	if err != nil {
+		panic("could not write to benchmark file: " + err.Error())
+	}
+
+	e.bytesWritten += n
+	if e.bytesWritten > e.bytesToFlushAfter {
+		e.file.Sync()
+		e.bytesWritten = 0
+		e.flushTimer.Reset(flushTimerInterval)
+	}
+}
+
+func (e *exporter) close() {
+	e.file.Close()
+}
+
+func Finish() {
+	fileWriter.close()
+}

benchmarking/measurement.go

@@ -0,0 +1,37 @@
+package benchmarking
+
+import (
+	"time"
+)
+
+type measurement struct {
+	*timer
+	opCode     OpCode
+	allocation uint32
+}
+
+func startNewMeasurement(opCode OpCode) *measurement {
+	return &measurement{
+		timer:  &timer{startTime: time.Now()},
+		opCode: opCode,
+	}
+}
+
+func (m *measurement) pause() {
+	m.stop()
+}
+
+func (m *measurement) resume() {
+	m.start()
+}
+
+func (m *measurement) end(size uint32) {
+	m.stop()
+	if size != 0 && m.allocation != 0 {
+		panic("measurement cannot have both allocation and size")
+	} else if size == 0 {
+		size = m.allocation
+	}
+
+	fileWriter.export(m.opCode, m.elapsedTime, size)
+}

benchmarking/ops.go

@@ -0,0 +1,49 @@
+package benchmarking
+
+const (
+	OpStoreGetObject     byte = 0x01 // get value from store
+	OpStoreSetObject     byte = 0x02 // set value in store
+	OpStoreDeleteObject  byte = 0x03 // delete value from store
+	OpStoreGetPackage    byte = 0x04 // get package from store
+	OpStoreGetType       byte = 0x05 // get type from store
+	OpStoreSetType       byte = 0x06 // set type in store
+	OpStoreGetBlockNode  byte = 0x07 // get block node from store
+	OpStoreSetBlockNode  byte = 0x08 // set block node in store
+	OpStoreAddMemPackage byte = 0x09 // add mempackage to store
+	OpStoreGetMemPackage byte = 0x0A // get mempackage from store
+	OpFinalizeTx         byte = 0x0B // finalize realm transaction
+
+	invalidStorageOp string = "OpStoreInvalid"
+)
+
+var opCodeNames = []string{
+	invalidStorageOp,
+	"OpStoreGetObject",
+	"OpStoreSetObject",
+	"OpStoreDeleteObject",
+	"OpStoreGetPackage",
+	"OpStoreGetType",
+	"OpStoreSetType",
+	"OpStoreGetBlockNode",
+	"OpStoreSetBlockNode",
+	"OpStoreAddMemPackage",
+	"OpStoreGetMemPackage",
+	"OpFinalizeTx",
+}
+
+type OpCode [2]byte
+
+func VMOpCode(op byte) OpCode {
+	return [2]byte{op, 0x00}
+}
+
+func StorageOpCode(op byte) OpCode {
+	return [2]byte{0x00, op}
+}
+
+func OpCodeString(op byte) string {
+	if int(op) >= len(opCodeNames) {
+		return invalidStorageOp
+	}
+	return opCodeNames[op]
+}

benchmarking/stack.go

@@ -0,0 +1,51 @@
+package benchmarking
+
+const initStackSize int = 64
+
+var (
+	measurementStack []*measurement
+	stackSize        int
+)
+
+func initStack() {
+	measurementStack = make([]*measurement, initStackSize)
+}
+
+func StartMeasurement(opCode OpCode) {
+	if stackSize != 0 {
+		measurementStack[stackSize-1].pause()
+	}
+
+	if stackSize == len(measurementStack) {
+		newStack := make([]*measurement, stackSize*2)
+		copy(newStack, measurementStack)
+		measurementStack = newStack
+	}
+
+	measurementStack[stackSize] = startNewMeasurement(opCode)
+	stackSize++
+}
+
+// StopMeasurement ends the current measurement and resumes the previous one
+// if one exists. It accepts the number of bytes that were read/written to/from
+// the store. This value is zero if the operation is not a read or write.
+func StopMeasurement(size uint32) {
+	if stackSize == 0 {
+		return
+	}
+
+	stackSize--
+	measurementStack[stackSize].end(size)
+
+	if stackSize != 0 {
+		measurementStack[stackSize].resume()
+	}
+}
+
+func RecordAllocation(size uint32) {
+	if stackSize == 0 {
+		return
+	}
+
+	measurementStack[stackSize-1].allocation += size
+}

benchmarking/timer.go

@@ -0,0 +1,22 @@
+package benchmarking
+
+import "time"
+
+type timer struct {
+	startTime   time.Time
+	elapsedTime time.Duration
+	isStopped   bool
+}
+
+func (t *timer) start() {
+	t.startTime = time.Now()
+}
+
+func (t *timer) stop() {
+	if t.isStopped {
+		return
+	}
+
+	t.elapsedTime += time.Since(t.startTime)
+	t.isStopped = true
+}