rueian

type slot struct {
	mark uint32
	req  request
	ch   chan response
}

rueian

type ring struct {
	_     [8]uint64 // padding to reduce false sharing
	write uint64
	_     [7]uint64
	read1 uint64
	_     [7]uint64
	read2 uint64
	_     [7]uint64
	slots [1024]slot
}

rueian

type PipeliningQueue interface {
	EnqueueRequest(req request) response
	NextRequestToSend() (req request, ok bool)
	ReplyToNextRequest(resp response)
}

rueian

▶ go test -bench=BenchmarkClient -benchmem -benchtime 2s .
goos: darwin
goarch: arm64
pkg: rueidis-benchmark
BenchmarkClient
BenchmarkClient/Rueidis-parallelism(64)-key(16)-value(64)
BenchmarkClient/Rueidis-parallelism(64)-key(16)-value(64)-10     4513908	       532.5 ns/op	      84 B/op	       2 allocs/op
BenchmarkClient/Channel-parallelism(64)-key(16)-value(64)
BenchmarkClient/Channel-parallelism(64)-key(16)-value(64)-10   	 2199741	       943.6 ns/op	     260 B/op	       4 allocs/op
PASS

rueian

func pipelining(outgoing *bufio.Writer, incoming *bufio.Reader) chan request {
	requests := make(chan request, 512)
	waiting := make(chan request, 512)
	go func() { writing(outgoing, requests, waiting) }()
	go func() { reading(incoming, waiting) }()
	return requests
}

func makeRequest(requests chan request, data []byte) response {
	respCh := make(chan response, 1)

rueian

func reading(incoming *bufio.Reader, waiting <-chan request) error {
	for {
		resp, err := readResponse(incoming)
		if err != nil {
			return err
		}
		
		if req := <-waiting; req.caller != nil {
			req.caller <- resp
		}

rueian

func writing(outgoing *bufio.Writer, requests <-chan request, waiting chan request) (err error) {
	var req request
	var more = true
	for more && err == nil {
		select {
		case req, more = <-requests:
			waiting <- req // <- push req into another waiting channel
			_, err = outgoing.Write(req.body)
			continue
		default:

rueian

func reading(incoming *bufio.Reader) error {
	for {
		resp, err := readResponse(incoming)
		if err != nil {
			return err
		}
		// find the original caller and send the response to it
	}
}

rueian

package main

import "bufio"

func writing(buf *bufio.Writer, input <-chan []byte) (err error) {
	var data []byte
	var more = true
	for more && err == nil {
		select {
		case data, more = <-input:

rueian

// https://pkg.go.dev/os/signal
func main() {
	c := make(chan os.Signal, 1)
	signal.Notify(c, os.Interrupt)
	
	s := <-c
	fmt.Println("Got signal:", s)
}