added the ttrpc stress utility

This change adds a stress utility which can be used for stress testing
the ttrpc connection. This tool represents a simple client-server
interaction where the client sends continuous requests to the server,
and the server responds with the same data, allowing for testing of
concurrent request handling and response verification.

This tool is adapted from https://github.com/kevpar/ttrpcstress.git
which was written by Kevin Parsons.

Signed-off-by: Harsh Rawat <[email protected]>

Author: rawahars

Makefile

@@ -24,7 +24,7 @@ WHALE = "🇩"
 ONI = "👹"
 
 # Project binaries.
-COMMANDS=protoc-gen-go-ttrpc protoc-gen-gogottrpc
+COMMANDS=protoc-gen-go-ttrpc protoc-gen-gogottrpc ttrpc-stress
 
 ifdef BUILDTAGS
     GO_BUILDTAGS = ${BUILDTAGS}

cmd/ttrpc-stress/README.md

@@ -0,0 +1,62 @@
+# ttrpc-stress
+
+This is a simple client/server utility designed to stress-test a TTRPC connection. It aims to identify potential deadlock cases during repeated, rapid TTRPC requests.
+
+## Overview of the tool
+
+- The **server** listens for connections and exposes a single, straightforward method. It responds immediately to any requests.
+- The **client** creates multiple worker goroutines that send a large number of requests to the server as quickly as possible.
+
+This tool represents a simple client-server interaction where the client sends requests to the server, and the server responds with the same data, allowing for testing of concurrent request handling and response verification. By utilizing core TTRPC facilities like `ttrpc.(*Client).Call` and `ttrpc.(*Server).Register` instead of generated client/server code, the test remains straightforward and effective.
+
+## Usage
+
+The `ttrpc-stress` command provides two modes of operation: server and client.
+
+### Run the Server
+
+To start the server, specify a Unix socket or named pipe as the address:
+```bash
+ttrpc-stress server <ADDR>
+```
+- `<ADDR>`: The Unix socket or named pipe to listen on.
+
+### Run the Client
+
+To start the client, specify the address, number of iterations, and number of workers:
+```bash
+ttrpc-stress client <ADDR> <ITERATIONS> <WORKERS>
+```
+- `<ADDR>`: The Unix socket or named pipe to connect to.
+- `<ITERATIONS>`: The total number of iterations to execute.
+- `<WORKERS>`: The number of workers handling the iterations.
+
+## Version Compatibility Testing
+
+One of the primary motivations for developing this stress utility was to identify potential deadlock scenarios when using different versions of the server and client. The goal is to test the current version of TTRPC, which is used to build `ttrpc-stress`, against the following older versions in both server and client scenarios:
+
+- `v1.0.2`
+- `v1.1.0`
+- `v1.2.0`
+- `v1.2.4`
+- `latest`
+
+### Known Issues in TTRPC Versions
+
+| Version Range       | Description                              |  Comments                                                 |
+|---------------------|-------------------------------------------|------------------------------------------------------------------|
+| **v1.0.2 and before**   | Original deadlock bug                                   | [#94](https://github.com/containerd/ttrpc/pull/94) for fixing deadlock in `v1.1.0`                                         |
+| **v1.1.0 - v1.2.0** | No known deadlock bugs                                  |  |
+| **v1.2.0 - v1.2.4** | Streaming with a new deadlock bug                       | [#107](https://github.com/containerd/ttrpc/pull/107) introduced deadlock in `v1.2.0` |
+| **After v1.2.4**    | No known deadlock bugs                                  | [#168](https://github.com/containerd/ttrpc/pull/168) for fixing deadlock in `v1.2.4` |
+---
+
+Clients before `v1.1.0` and between `v1.2.0`-`v1.2.3` can encounted the deadlock.
+
+However, if the **server** version is `v1.2.0` or later, deadlock issues in the client may be avoided.
+
+Please refer to https://github.com/containerd/ttrpc/issues/72 for more information about the deadlock bug.
+
+---
+
+Happy testing!

cmd/ttrpc-stress/connection_other.go

@@ -0,0 +1,32 @@
+//go:build !windows
+// +build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package main
+
+import "net"
+
+// listenConnection listens for incoming Unix domain socket connections at the specified address.
+func listenConnection(addr string) (net.Listener, error) {
+	return net.Listen("unix", addr)
+}
+
+// dialConnection dials a Unix domain socket connection to the specified address.
+func dialConnection(addr string) (net.Conn, error) {
+	return net.Dial("unix", addr)
+}

cmd/ttrpc-stress/connection_windows.go

@@ -0,0 +1,41 @@
+//go:build windows
+// +build windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package main
+
+import (
+	"net"
+
+	"github.com/Microsoft/go-winio"
+)
+
+// listenConnection listens for incoming named pipe connections at the specified address.
+func listenConnection(addr string) (net.Listener, error) {
+	return winio.ListenPipe(addr, &winio.PipeConfig{
+		// 0 buffer sizes for pipe is important to help deadlock to occur.
+		// It can still occur if there is buffering, but it takes more IO volume to hit it.
+		InputBufferSize:  0,
+		OutputBufferSize: 0,
+	})
+}
+
+// dialConnection dials a named pipe connection to the specified address.
+func dialConnection(addr string) (net.Conn, error) {
+	return winio.DialPipe(addr, nil)
+}

cmd/ttrpc-stress/main.go

@@ -0,0 +1,203 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+
+	ttrpc "github.com/containerd/ttrpc"
+	payload "github.com/containerd/ttrpc/cmd/ttrpc-stress/payload"
+
+	"golang.org/x/sync/errgroup"
+)
+
+// main is the entry point of the stress utility.
+func main() {
+	// Define a flag for displaying usage information.
+	flagHelp := flag.Bool("help", false, "Display usage")
+	flag.Parse()
+
+	// Check if help flag is set or if there are insufficient arguments.
+	if *flagHelp || flag.NArg() < 2 {
+		usage()
+	}
+
+	// Switch based on the first argument to determine mode (server or client).
+	switch flag.Arg(0) {
+	case "server":
+		// Ensure correct number of arguments for server mode.
+		if flag.NArg() != 2 {
+			usage()
+		}
+
+		addr := flag.Arg(1)
+
+		// Run the server and handle any errors.
+		err := runServer(context.Background(), addr)
+		if err != nil {
+			log.Fatalf("error: %s", err)
+		}
+
+	case "client":
+		// Ensure correct number of arguments for client mode.
+		if flag.NArg() != 4 {
+			usage()
+		}
+
+		addr := flag.Arg(1)
+
+		// Parse iterations and workers arguments.
+		iters, err := strconv.Atoi(flag.Arg(2))
+		if err != nil {
+			log.Fatalf("failed parsing iters: %s", err)
+		}
+
+		workers, err := strconv.Atoi(flag.Arg(3))
+		if err != nil {
+			log.Fatalf("failed parsing workers: %s", err)
+		}
+
+		// Run the client and handle any errors.
+		err = runClient(context.Background(), addr, iters, workers)
+		if err != nil {
+			log.Fatalf("runtime error: %s", err)
+		}
+
+	default:
+		// Display usage information if the mode is unrecognized.
+		usage()
+	}
+}
+
+// usage prints the usage information and exits the program.
+// usage prints the usage information for the program and exits.
+func usage() {
+	fmt.Fprintf(os.Stderr, `Usage:
+	stress server <ADDR>
+		Run the server with the specified unix socket or named pipe.
+	stress client <ADDR> <ITERATIONS> <WORKERS>
+		Run the client with the specified unix socket or named pipe, number of ITERATIONS, and number of WORKERS.
+`)
+	os.Exit(1)
+}
+
+// runServer sets up and runs the server.
+func runServer(ctx context.Context, addr string) error {
+	log.Printf("Starting server on %s", addr)
+
+	// Listen for connections on the specified address.
+	l, err := listenConnection(addr)
+	if err != nil {
+		return fmt.Errorf("failed listening on %s: %w", addr, err)
+	}
+
+	// Create a new ttrpc server.
+	server, err := ttrpc.NewServer()
+	if err != nil {
+		return fmt.Errorf("failed creating ttrpc server: %w", err)
+	}
+
+	// Register a service and method with the server.
+	server.Register("ttrpc.stress.test.v1", map[string]ttrpc.Method{
+		"TEST": func(ctx context.Context, unmarshal func(interface{}) error) (interface{}, error) {
+			req := &payload.Payload{}
+			// Unmarshal the request payload.
+			if err := unmarshal(req); err != nil {
+				log.Fatalf("failed unmarshalling request: %s", err)
+			}
+			id := req.Value
+			log.Printf("got request: %d", id)
+			// Return the same payload as the response.
+			return &payload.Payload{Value: id}, nil
+		},
+	})
+
+	// Serve the server and handle any errors.
+	if err := server.Serve(ctx, l); err != nil {
+		return fmt.Errorf("failed serving server: %w", err)
+	}
+	return nil
+}
+
+// runClient sets up and runs the client.
+func runClient(ctx context.Context, addr string, iters int, workers int) error {
+	log.Printf("Starting client on %s", addr)
+
+	// Dial a connection to the specified pipe.
+	c, err := dialConnection(addr)
+	if err != nil {
+		return fmt.Errorf("failed dialing connection to %s: %w", addr, err)
+	}
+
+	// Create a new ttrpc client.
+	client := ttrpc.NewClient(c)
+	ch := make(chan int)
+	var eg errgroup.Group
+
+	// Start worker goroutines to send requests.
+	for i := 0; i < workers; i++ {
+		eg.Go(func() error {
+			for {
+				i, ok := <-ch
+				if !ok {
+					return nil
+				}
+				// Send the request and handle any errors.
+				if err := send(ctx, client, uint32(i)); err != nil {
+					return fmt.Errorf("failed sending request: %w", err)
+				}
+			}
+		})
+	}
+
+	// Send iterations to the channel.
+	for i := 0; i < iters; i++ {
+		ch <- i
+	}
+	close(ch)
+
+	// Wait for all goroutines to finish.
+	if err := eg.Wait(); err != nil {
+		return fmt.Errorf("failed waiting for goroutines: %w", err)
+	}
+	return nil
+}
+
+// send sends a request to the server and verifies the response.
+func send(ctx context.Context, client *ttrpc.Client, id uint32) error {
+	req := &payload.Payload{Value: id}
+	resp := &payload.Payload{}
+
+	log.Printf("sending request: %d", id)
+	// Call the server method and handle any errors.
+	if err := client.Call(ctx, "ttrpc.stress.test.v1", "TEST", req, resp); err != nil {
+		return err
+	}
+
+	ret := resp.Value
+	log.Printf("got response: %d", ret)
+	// Verify the response matches the request.
+	if ret != id {
+		return fmt.Errorf("expected return value %d but got %d", id, ret)
+	}
+	return nil
+}

cmd/ttrpc-stress/payload/doc.go

@@ -0,0 +1,17 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package payload