executor: fix cpu usage exhaust (pingcap#72)

executor/runtime/benchmark/operator.go

@@ -1,12 +1,14 @@
 package benchmark
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 	"time"
 
 	"github.com/hanfei1991/microcosm/executor/runtime"
@@ -37,16 +39,31 @@ func (f *fileWriter) Prepare() error {
 }
 
 func sprintPayload(r *pb.Record) string {
-	str := fmt.Sprintf("tid %d, pk %d, time tracer ", r.Tid, r.Pk)
+	var buf bytes.Buffer
+	buf.WriteString("tid ")
+	buf.WriteString(strconv.FormatInt(int64(r.Tid), 10))
+	buf.WriteString(", pk")
+	buf.WriteString(strconv.FormatInt(int64(r.Pk), 10))
+	buf.WriteString(", time tracer ")
 	for _, ts := range r.TimeTracer {
-		str += fmt.Sprintf("%s ", time.Unix(0, ts))
+		buf.WriteString(strconv.FormatInt(ts, 10) + " ")
 	}
-	return str
+	return buf.String()
 }
 
 func sprintRecord(r *runtime.Record) string {
 	start := time.Unix(0, r.Payload.(*pb.Record).TimeTracer[0])
-	return fmt.Sprintf("flowID %s start %s end %s payload: %s\n", r.FlowID, start.String(), r.End.String(), sprintPayload(r.Payload.(*pb.Record)))
+	var buf bytes.Buffer
+	buf.WriteString("flowID ")
+	buf.WriteString(r.FlowID)
+	buf.WriteString(" start ")
+	buf.WriteString(start.String())
+	buf.WriteString(" end ")
+	buf.WriteString(r.End.String())
+	buf.WriteString(" payload: ")
+	buf.WriteString(sprintPayload(r.Payload.(*pb.Record)))
+	buf.WriteString("\n")
+	return buf.String()
 }
 
 func (f *fileWriter) writeStats(s *recordStats) error {

executor/runtime/runtime.go

@@ -3,6 +3,7 @@ package runtime
 import (
 	"context"
 	"sync"
+	"time"
 
 	"github.com/hanfei1991/microcosm/model"
 	"github.com/hanfei1991/microcosm/test"
@@ -72,7 +73,9 @@ func (s *Runtime) runImpl(ctx context.Context) {
 		}
 		t := s.q.pop()
 		if t == nil {
-			// idle
+			// idle, sleep for sometime to avoid busy loop
+			// TODO: find better wake up mechanism way if needed
+			time.Sleep(time.Millisecond * 50)
 			continue
 		}
 		status := t.Poll()

executor/runtime/task.go

@@ -33,12 +33,13 @@ type Channel struct {
 
 func (c *Channel) readBatch(batch int) []*Record {
 	records := make([]*Record, 0, batch)
+readLoop:
 	for i := 0; i < batch; i++ {
 		select {
 		case record := <-c.innerChan:
 			records = append(records, record)
 		default:
-			break
+			break readLoop
 		}
 	}
 	if len(records) > 0 {