challenge/recorder part2 test

Makefile.ctrlg

@@ -0,0 +1,27 @@
+.PHONY: tidy install-deps build test coverage
+
+# Run go mod tidy
+tidy:
+	@echo "Running go mod tidy..."
+	go mod tidy
+
+# Install dependencies
+install-deps: tidy
+	@echo No dependency to install
+
+# Build the project
+build: tidy
+	@echo "Building project..."
+	go test -run=XXX_SHOULD_NEVER_MATCH_XXX ./...
+
+# Run tests
+test: tidy
+	@echo "Running tests..."
+	go test ./... -skip TestMutations -count=1 -v -timeout 60s
+	@echo "Running mutation tests..."
+	go test ./... -run TestMutations -count=1 -v -timeout 300s
+
+# Generate coverage report
+coverage: tidy
+	@echo "Generating coverage report..."
+	go test ./... -cover -timeout 90s

deluge/deluge.go

@@ -1,11 +1,12 @@
 package deluge
 
 import (
-	"github.com/ofux/deluge-dsl/ast"
-	"github.com/ofux/deluge-dsl/evaluator"
-	"github.com/ofux/deluge-dsl/object"
-	log "github.com/sirupsen/logrus"
 	"time"
+
+	"github.com/ofux/deluge/dsl/ast"
+	"github.com/ofux/deluge/dsl/evaluator"
+	"github.com/ofux/deluge/dsl/object"
+	log "github.com/sirupsen/logrus"
 )
 
 type Deluge struct {
@@ -79,7 +80,7 @@ func (d *delugeBuilder) CreateDeluge(node ast.Node, args ...object.Object) objec
 	}
 
 	for scenarioId, v := range conf.Pairs {
-		scenarioConf, ok := v.Value.(*object.Hash)
+		scenarioConf, ok := v.(*object.Hash)
 		if !ok {
 			log.Fatalf("Expected scenario configuration to be an object at %s\n", ast.PrintLocation(node))
 		}
@@ -88,7 +89,7 @@ func (d *delugeBuilder) CreateDeluge(node ast.Node, args ...object.Object) objec
 		if !ok {
 			log.Fatalf("Expected 'concurrent' value in configuration at %s\n", ast.PrintLocation(node))
 		}
-		concurrentClients, ok := concurrentClientsHashPair.Value.(*object.Integer)
+		concurrentClients, ok := concurrentClientsHashPair.(*object.Integer)
 		if !ok {
 			log.Fatalf("Expected 'concurrent' value to be an integer in configuration at %s\n", ast.PrintLocation(node))
 		}
@@ -97,7 +98,7 @@ func (d *delugeBuilder) CreateDeluge(node ast.Node, args ...object.Object) objec
 		if !ok {
 			log.Fatalf("Expected 'delay' value in configuration at %s\n", ast.PrintLocation(node))
 		}
-		delayHashStr, ok := delayHashPair.Value.(*object.String)
+		delayHashStr, ok := delayHashPair.(*object.String)
 		if !ok {
 			log.Fatalf("Expected 'concurrent' value to be a duration in configuration at %s\n", ast.PrintLocation(node))
 		}

deluge/deluge_test.go

@@ -1,9 +1,10 @@
 package deluge
 
 import (
-	"github.com/ofux/deluge-dsl/lexer"
-	"github.com/ofux/deluge-dsl/parser"
 	"testing"
+
+	"github.com/ofux/deluge/dsl/lexer"
+	"github.com/ofux/deluge/dsl/parser"
 )
 
 func BenchmarkNewDeluge(b *testing.B) {

deluge/recorder.go

@@ -0,0 +1,147 @@
+package deluge
+
+import (
+	"errors"
+	"sync"
+
+	hdr "github.com/ofux/hdrhistogram"
+)
+
+type RecordingState int
+
+const (
+	READY RecordingState = iota
+	RECORDING
+	TERMINATING
+	TERMINATED
+)
+
+// Histogram is an interface for recording and retrieving histogram data.
+// It abstracts the underlying histogram implementation.
+type Histogram interface {
+	RecordValue(value int64) error
+	TotalCount() int64
+}
+
+// HistogramConstructor is a function type that creates new Histogram instances.
+type HistogramConstructor func() Histogram
+
+type Recorder interface {
+	Record(iteration int, id string, value int64)
+	Close()
+	GetRecords() (map[string][]Histogram, error)
+}
+
+type QueuedRecorder struct {
+	recording           RecordingState
+	recordsQueue        chan Record
+	recordingWaitGroup  *sync.WaitGroup
+	processingWaitGroup *sync.WaitGroup
+	histograms          map[string][]Histogram
+	newHistogram        HistogramConstructor
+}
+
+type Record struct {
+	iteration int
+	id        string
+	value     int64
+}
+
+func NewRecorder(newHistogram HistogramConstructor) Recorder {
+	recorder := &QueuedRecorder{
+		recording:           READY,
+		recordsQueue:        make(chan Record, 10),
+		recordingWaitGroup:  new(sync.WaitGroup),
+		processingWaitGroup: new(sync.WaitGroup),
+		histograms:          make(map[string][]Histogram),
+		newHistogram:        newHistogram,
+	}
+	recorder.processRecords()
+	return recorder
+}
+
+// Record records a new value in the underlying appropriate HDRHistogram.
+// This is safe to call this method from different goroutines.
+// Calling this method on a closed Recorder will cause a panic.
+func (r *QueuedRecorder) Record(iteration int, id string, value int64) {
+	r.recordingWaitGroup.Add(1)
+	go func() {
+		defer r.recordingWaitGroup.Done()
+		r.recordsQueue <- Record{
+			iteration: iteration,
+			id:        id,
+			value:     value,
+		}
+	}()
+}
+
+// Close closes the Recorder, making the results available for read.
+// Trying to record some values on a closed Recorder will cause a panic.
+func (r *QueuedRecorder) Close() {
+	if r.recording == RECORDING {
+		r.recording = TERMINATING
+		// wait for all records to be taken
+		r.recordingWaitGroup.Wait()
+		// ensure listener won't stay blocked
+		close(r.recordsQueue)
+		// wait for the end of recording
+		r.processingWaitGroup.Wait()
+		r.recording = TERMINATED
+	}
+}
+
+func (r *QueuedRecorder) GetRecords() (map[string][]Histogram, error) {
+	if r.recording != TERMINATED {
+		return nil, errors.New("Cannot get records while recording. Did you forget to call the 'Close()' method?")
+	}
+	return r.histograms, nil
+}
+
+func (r *QueuedRecorder) processRecords() {
+	r.recording = RECORDING
+	r.processingWaitGroup.Add(1)
+
+	go func() {
+		defer r.processingWaitGroup.Done()
+
+		for {
+			rec, ok := <-r.recordsQueue
+			if !ok {
+				return
+			}
+
+			histograms, ok := r.histograms[rec.id]
+			if !ok {
+				histograms = make([]Histogram, 0)
+				r.histograms[rec.id] = histograms
+			}
+
+			// TODO: optimize this
+			if len(histograms) <= rec.iteration {
+				diff := rec.iteration + 1 - len(histograms)
+				histograms = append(histograms, r.createHistograms(diff)...)
+				r.histograms[rec.id] = histograms
+			}
+
+			histogram := histograms[rec.iteration]
+			histogram.RecordValue(rec.value)
+		}
+	}()
+
+}
+
+func (r *QueuedRecorder) createHistograms(count int) []Histogram {
+	histograms := make([]Histogram, count)
+	for i := 0; i < count; i++ {
+		histograms[i] = r.newHistogram()
+	}
+	return histograms
+}
+
+// DefaultHistogramConstructor returns a HistogramConstructor that creates
+// HDR histograms with default settings (min=0, max=3600000000, sigfigs=3).
+func DefaultHistogramConstructor() HistogramConstructor {
+	return func() Histogram {
+		return hdr.New(0, 3600000000, 3)
+	}
+}

deluge/scenario.go

@@ -1,11 +1,12 @@
 package deluge
 
 import (
-	"github.com/ofux/deluge-dsl/ast"
-	log "github.com/sirupsen/logrus"
 	"strconv"
 	"sync"
 	"time"
+
+	"github.com/ofux/deluge/dsl/ast"
+	log "github.com/sirupsen/logrus"
 )
 
 type Scenario struct {
@@ -14,6 +15,7 @@ type Scenario struct {
 	script             ast.Node
 	duration           time.Duration
 	TotalSimUsersCalls int64
+	recorder           Recorder
 }
 
 func NewScenario(name string, concurrent int, duration time.Duration, script ast.Node) *Scenario {
@@ -22,10 +24,11 @@ func NewScenario(name string, concurrent int, duration time.Duration, script ast
 		simUsers: make([]*SimUser, concurrent),
 		duration: duration,
 		script:   script,
+		recorder: NewRecorder(DefaultHistogramConstructor()),
 	}
 
 	for i := 0; i < concurrent; i++ {
-		s.simUsers[i] = NewSimUser(strconv.Itoa(i), s.script)
+		s.simUsers[i] = NewSimUser(strconv.Itoa(i), s.script, s.recorder)
 	}
 
 	return s
@@ -46,6 +49,7 @@ func (sc *Scenario) Run(duration time.Duration) {
 			ticker := time.NewTicker(sc.duration)
 			timer := time.NewTimer(duration)
 
+			i := 0
 			for {
 				if time.Now().Sub(start).Nanoseconds() > duration.Nanoseconds() {
 					log.Debugf("Terminate user simulation %s", su.Name)
@@ -54,7 +58,7 @@ func (sc *Scenario) Run(duration time.Duration) {
 
 				log.Debugf("Running user simulation %s", su.Name)
 				simUserCallCounter <- 1
-				su.Run()
+				su.Run(i)
 
 				if su.Status == DoneError {
 					return
@@ -66,10 +70,12 @@ func (sc *Scenario) Run(duration time.Duration) {
 					return
 				case <-ticker.C:
 				}
+				i++
 			}
 		}(su)
 	}
 	waitg.Wait()
+	sc.recorder.Close()
 
 	log.Infof("Scenario executed %d requests in %s", sc.TotalSimUsersCalls, time.Now().Sub(start).String())
 }

deluge/simuser.go

@@ -1,12 +1,14 @@
 package deluge
 
 import (
-	"github.com/ofux/deluge-dsl/ast"
-	"github.com/ofux/deluge-dsl/evaluator"
-	"github.com/ofux/deluge-dsl/object"
-	log "github.com/sirupsen/logrus"
 	"net/http"
+	"strconv"
 	"time"
+
+	"github.com/ofux/deluge/dsl/ast"
+	"github.com/ofux/deluge/dsl/evaluator"
+	"github.com/ofux/deluge/dsl/object"
+	log "github.com/sirupsen/logrus"
 )
 
 type SimUserStatus int
@@ -25,23 +27,27 @@ type SimUser struct {
 	client        *http.Client
 	Status        SimUserStatus
 	SleepDuration time.Duration
+	recorder      Recorder
+	iteration     int
 }
 
-func NewSimUser(name string, script ast.Node) *SimUser {
+func NewSimUser(name string, script ast.Node, recorder Recorder) *SimUser {
 	su := &SimUser{
 		Name:      name,
 		script:    script,
 		evaluator: evaluator.NewEvaluator(),
 		client:    http.DefaultClient,
 		Status:    Virgin,
+		recorder:  recorder,
 	}
 
 	su.evaluator.AddBuiltin("http", su.ExecHTTPRequest)
 
 	return su
 }
 
-func (su *SimUser) Run() {
+func (su *SimUser) Run(iteration int) {
+	su.iteration = iteration
 	su.Status = InProgress
 	env := object.NewEnvironment()
 	evaluated := su.evaluator.Eval(su.script, env)
@@ -62,40 +68,42 @@ func (su *SimUser) ExecHTTPRequest(node ast.Node, args ...object.Object) object.
 		return oErr
 	}
 
-	//name := args[0].(*object.String).Value
+	reqName := args[0].(*object.String).Value
 	reqObj := args[1].(*object.Hash)
 
 	jsUrl, ok := reqObj.Get("url")
 	if !ok {
 		return evaluator.NewError(node, "invalid HTTP request: missing 'url' field")
 	}
-	url, ok := jsUrl.Value.(*object.String)
+	url, ok := jsUrl.(*object.String)
 	if !ok {
 		return evaluator.NewError(node, "invalid HTTP request: 'url' should be a STRING")
 	}
 
 	var method = "GET"
 	if methodField, ok := reqObj.Get("method"); ok {
-		if methodFieldVal, ok := methodField.Value.(*object.String); ok {
+		if methodFieldVal, ok := methodField.(*object.String); ok {
 			method = methodFieldVal.Value
 		}
 	}
 
 	req, err := http.NewRequest(method, url.Value, nil)
 	if err != nil {
-		return evaluator.NewError(node, err.Error())
+		return evaluator.NewError(node, "%s", err.Error())
 	}
 
 	log.Debugf("Performing HTTP request: %s %s", req.Method, req.URL.String())
 	start := time.Now()
-	//res, err := su.client.Do(req)
+	res, err := su.client.Do(req)
 	end := time.Now()
 	duration := end.Sub(start)
+
 	if err != nil {
 		log.Debugf("Request error: %s", err.Error())
-		return evaluator.NewError(node, err.Error())
+		return evaluator.NewError(node, "%s", err.Error())
 	} else {
 		log.Debugf("Response status: %s in %s", "res.Status", duration.String())
+		su.recorder.Record(su.iteration, reqName+"->"+strconv.Itoa(res.StatusCode), duration.Nanoseconds()/1000)
 	}
 
 	return evaluator.NULL