<commit_before>// Copyright 2016 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// Test mutator availability and end-to-end latency of an RPC-like system.
package main

// TODO: This benchmark doesn't target a particular aspect of the GC
// like other gcbench benchmarks do. It would be nice if we just had
// composable sets of metrics that a benchmark could opt in to without
// needing a whole "framework", but I'm not sure how to do that with
// the gctrace on stderr thing. Maybe we could dup our own FD over
// stderr to read our own gctrace? That's dangerous if something goes
// wrong since you'll lose the panic. We could pipe it by way of an
// external program that could report if something goes wrong (e.g.,
// tee if you just want to see the whole log). This might also make it
// easier to ignore initialization and warm-up in the GC log.

import (
	"bufio"
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net"
	"os"
	"os/exec"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aclements/go-gcbench/gcbench"
	"golang.org/x/benchmarks/gcbench.newer/heapgen"
)

var (
	flagDuration   = flag.Duration("benchtime", 10*time.Second, "steady state duration")
	flagBallast    = gcbench.FlagBytes("ballast", 64*gcbench.MB, "retain `x` bytes of global data")
	flagReqsPerSec = flag.Float64("reqs-per-sec", 10000, "send `rate` requests per second")
	flagClient     = flag.String("client", "", "internal flag to act as RPC client")
)

func main() {
	flag.Parse()
	if flag.NArg() > 0 {
		flag.Usage()
		os.Exit(2)
	}

	// TODO: Report end-to-end latency measured by the client as a
	// metric.

	if *flagClient != "" {
		doClient(*flagClient)
		return
	}

	name := "RPC"
	gcbench.NewBenchmark(name, benchMain).Config("reqs-per-sec", *flagReqsPerSec).Config("ballast", *flagBallast).Run()
}

var sink1 interface{}
var requestCount int64

func benchMain() {
	// Create the ballast.
	m := heapgen.Measure(heapgen.MakeAST)
	sink1 = heapgen.Generate(m.Gen, m.BytesRetained, int(*flagBallast))

	// Start the server.
	l := startServer()

	// Start the client.
	client := exec.Command(os.Args[0], "-client", l.Addr().String(), "-reqs-per-sec", fmt.Sprint(*flagReqsPerSec))
	client.Env = []string{}
	cin, err := client.StdinPipe()
	if err != nil {
		log.Fatal("creating client stdin pipe: ", err)
	}
	defer cin.Close()
	cout, err := client.StdoutPipe()
	if err != nil {
		log.Fatal("creating client stdout pipe: ", err)
	}
	defer cout.Close()
	client.Stderr = os.Stderr
	err = client.Start()
	if err != nil {
		log.Fatal("starting client: ", err)
	}

	// Run the server.
	done := make(chan bool)
	startTime := time.Now()
	time.AfterFunc(*flagDuration, func() {
		gcbench.ReportExtra("reqs/sec", float64(atomic.LoadInt64(&requestCount))/float64(time.Since(startTime))*float64(time.Second))

		// Shut down client.
		cin.Close()
		err := client.Wait()
		if err != nil {
			log.Fatal("client failed: ", err)
		}
		close(done)
	})
	go func() {
		for {
			c, err := l.Accept()
			if err != nil {
				log.Fatal("accept: ", err)
			}
			go handler(c)
		}
	}()
	<-done
}

func startServer() net.Listener {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		if l, err = net.Listen("tcp6", "[::1]:0"); err != nil {
			log.Fatalf("failed to listen: %v", err)
		}
	}
	return l
}

func handler(c net.Conn) {
	var header [1]byte
	r, w := bufio.NewReader(c), bufio.NewWriter(c)
	d := json.NewDecoder(r)
	for {
		// Read the request.
		n, err := r.Read(header[:])
		if err == io.EOF {
			c.Close()
			return
		}
		if n != 1 || header[0] != 'x' {
			log.Fatal("bad header: ", n, err, header)
		}

		var data interface{}
		err = d.Decode(&data)
		if err != nil {
			log.Fatal("decoding request: ", err)
		}

		atomic.AddInt64(&requestCount, 1)

		// Send the response.
		json.NewEncoder(w).Encode(data)
		w.Flush()
	}
}

func doClient(addr string) {
	var wg sync.WaitGroup
	stop := make(chan bool)

	var lat gcbench.LatencyDist

	// Wait for stdin to close, then shut down cleanly.
	go func() {
		var buf [1024]byte
		for {
			n, err := os.Stdin.Read(buf[:])
			if n <= 0 || err != nil {
				break
			}
		}
		log.Print("cleaning up client")
		close(stop)
		wg.Wait()
		log.Print("closing ", len(conns.q), " connections")
		for _, c := range conns.q {
			c.Close()
		}
		lat.Fprint(os.Stderr)
		//log.Print("client exiting")
		os.Exit(0)
	}()

	// Generate a Poisson process by spacing events out by an
	// exponential distribution and using open-loop control. To
	// minimize the bottleneck in the coordinator, we shard this
	// process across several goroutines.
	const shards = 16
	reqsPerNSPerShard := *flagReqsPerSec / 1e9 / shards
	for i := 0; i < shards; i++ {
		wg.Add(1)
		go func() {
		loop:
			for {<commit_msg>gcbench/progs: print header for request latency<commit_after>// Copyright 2016 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

// Test mutator availability and end-to-end latency of an RPC-like system.
package main

// TODO: This benchmark doesn't target a particular aspect of the GC
// like other gcbench benchmarks do. It would be nice if we just had
// composable sets of metrics that a benchmark could opt in to without
// needing a whole "framework", but I'm not sure how to do that with
// the gctrace on stderr thing. Maybe we could dup our own FD over
// stderr to read our own gctrace? That's dangerous if something goes
// wrong since you'll lose the panic. We could pipe it by way of an
// external program that could report if something goes wrong (e.g.,
// tee if you just want to see the whole log). This might also make it
// easier to ignore initialization and warm-up in the GC log.

import (
	"bufio"
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net"
	"os"
	"os/exec"
	"sync"
	"sync/atomic"
	"time"

	"github.com/aclements/go-gcbench/gcbench"
	"golang.org/x/benchmarks/gcbench.newer/heapgen"
)

var (
	flagDuration   = flag.Duration("benchtime", 10*time.Second, "steady state duration")
	flagBallast    = gcbench.FlagBytes("ballast", 64*gcbench.MB, "retain `x` bytes of global data")
	flagReqsPerSec = flag.Float64("reqs-per-sec", 10000, "send `rate` requests per second")
	flagClient     = flag.String("client", "", "internal flag to act as RPC client")
)

func main() {
	flag.Parse()
	if flag.NArg() > 0 {
		flag.Usage()
		os.Exit(2)
	}

	// TODO: Report end-to-end latency measured by the client as a
	// metric.

	if *flagClient != "" {
		doClient(*flagClient)
		return
	}

	name := "RPC"
	gcbench.NewBenchmark(name, benchMain).Config("reqs-per-sec", *flagReqsPerSec).Config("ballast", *flagBallast).Run()
}

var sink1 interface{}
var requestCount int64

func benchMain() {
	// Create the ballast.
	m := heapgen.Measure(heapgen.MakeAST)
	sink1 = heapgen.Generate(m.Gen, m.BytesRetained, int(*flagBallast))

	// Start the server.
	l := startServer()

	// Start the client.
	client := exec.Command(os.Args[0], "-client", l.Addr().String(), "-reqs-per-sec", fmt.Sprint(*flagReqsPerSec))
	client.Env = []string{}
	cin, err := client.StdinPipe()
	if err != nil {
		log.Fatal("creating client stdin pipe: ", err)
	}
	defer cin.Close()
	cout, err := client.StdoutPipe()
	if err != nil {
		log.Fatal("creating client stdout pipe: ", err)
	}
	defer cout.Close()
	client.Stderr = os.Stderr
	err = client.Start()
	if err != nil {
		log.Fatal("starting client: ", err)
	}

	// Run the server.
	done := make(chan bool)
	startTime := time.Now()
	time.AfterFunc(*flagDuration, func() {
		gcbench.ReportExtra("reqs/sec", float64(atomic.LoadInt64(&requestCount))/float64(time.Since(startTime))*float64(time.Second))

		// Shut down client.
		cin.Close()
		err := client.Wait()
		if err != nil {
			log.Fatal("client failed: ", err)
		}
		close(done)
	})
	go func() {
		for {
			c, err := l.Accept()
			if err != nil {
				log.Fatal("accept: ", err)
			}
			go handler(c)
		}
	}()
	<-done
}

func startServer() net.Listener {
	l, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		if l, err = net.Listen("tcp6", "[::1]:0"); err != nil {
			log.Fatalf("failed to listen: %v", err)
		}
	}
	return l
}

func handler(c net.Conn) {
	var header [1]byte
	r, w := bufio.NewReader(c), bufio.NewWriter(c)
	d := json.NewDecoder(r)
	for {
		// Read the request.
		n, err := r.Read(header[:])
		if err == io.EOF {
			c.Close()
			return
		}
		if n != 1 || header[0] != 'x' {
			log.Fatal("bad header: ", n, err, header)
		}

		var data interface{}
		err = d.Decode(&data)
		if err != nil {
			log.Fatal("decoding request: ", err)
		}

		atomic.AddInt64(&requestCount, 1)

		// Send the response.
		json.NewEncoder(w).Encode(data)
		w.Flush()
	}
}

func doClient(addr string) {
	var wg sync.WaitGroup
	stop := make(chan bool)

	var lat gcbench.LatencyDist

	// Wait for stdin to close, then shut down cleanly.
	go func() {
		var buf [1024]byte
		for {
			n, err := os.Stdin.Read(buf[:])
			if n <= 0 || err != nil {
				break
			}
		}
		log.Print("cleaning up client")
		close(stop)
		wg.Wait()
		log.Print("closing ", len(conns.q), " connections")
		for _, c := range conns.q {
			c.Close()
		}
		fmt.Fprintln(os.Stderr, "client-measured request latency:")
		lat.Fprint(os.Stderr)
		//log.Print("client exiting")
		os.Exit(0)
	}()

	// Generate a Poisson process by spacing events out by an
	// exponential distribution and using open-loop control. To
	// minimize the bottleneck in the coordinator, we shard this
	// process across several goroutines.
	const shards = 16
	reqsPerNSPerShard := *flagReqsPerSec / 1e9 / shards
	for i := 0; i < shards; i++ {
		wg.Add(1)
		go func() {
		loop:
			for {
{'fraction_non_alphanumeric': 0.15006162695152014, 'fraction_numerical': 0.01191454396055875, 'mean_word_length': 3.2894273127753304, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '727311', 'n_tokens_mistral': 3818, 'n_tokens_neox': 3475, 'n_words': 1061}