benchmark_grpc

README

/examples/benchmark_grpc

cd /

[!NOTE] Asyncmachine-go is an AOP Actor Model library for distributed workflows, built on top of a clock-based state machine. It has atomic transitions, subscriptions, RPC, logging, TUI debugger, metrics, tracing, and soon diagrams.

Benchmark: aRPC vs gRPC

This is a simple and opinionated benchmark of a subscribe-get-process scenario, implemented in both gRPC and aRPC. Source code can be found in /examples/benchmark_grpc. It essentially manipulates a worker state machine via various transports.

Steps:

1. subscription: wait for notifications
2. getter: get a value from the worker
3. processing: call an operation based on the value

Plain Go Implementation

i := 0
worker.Subscribe(func() {

    // loop
    i++
    if i > limit {
        close(end)
        return
    }

    // value (getter)
    value := worker.GetValue()

    // call op from value (processing)
    switch value {
    case Value1:
        go worker.CallOp(Op1)
    case Value2:
        go worker.CallOp(Op2)
    case Value3:
        go worker.CallOp(Op3)
    default:
        // err
        b.Fatalf("Unknown value: %v", value)
    }
})

worker.Start()

Results

$ task benchmark-grpc
...
BenchmarkClientArpc
    client_arpc_test.go:136: Transferred: 609 bytes
    client_arpc_test.go:137: Calls: 4
    client_arpc_test.go:138: Errors: 0
    client_arpc_test.go:136: Transferred: 1,149,424 bytes
    client_arpc_test.go:137: Calls: 10,003
    client_arpc_test.go:138: Errors: 0
BenchmarkClientArpc-8              10000            248913 ns/op           28405 B/op        766 allocs/op
BenchmarkClientGrpc
    client_grpc_test.go:117: Transferred: 1,113 bytes
    client_grpc_test.go:118: Calls: 9
    client_grpc_test.go:119: Errors: 0
    client_grpc_test.go:117: Transferred: 3,400,812 bytes
    client_grpc_test.go:118: Calls: 30,006
    client_grpc_test.go:119: Errors: 0
BenchmarkClientGrpc-8              10000            262693 ns/op           19593 B/op        391 allocs/op
BenchmarkClientLocal
BenchmarkClientLocal-8             10000               434.4 ns/op            16 B/op          1 allocs/op
PASS
ok      github.com/pancsta/asyncmachine-go/examples/benchmark_grpc      5.187s

aRPC

Worker's states can be found below. For handlers, please refer to the source.

// States structure defines relations and properties of states.
var States = am.Struct{
    // toggle
    Start: {},

    // ops
    CallOp: {
        Multi:   true,
        Require: S{Start},
    },

    // events
    Event: {
        Multi:   true,
        Require: S{Start},
    },

    // values
    Value1: {Remove: GroupValues},
    Value2: {Remove: GroupValues},
    Value3: {Remove: GroupValues},
}

// Groups of mutually exclusive states.

var (
    GroupValues = S{Value1, Value2, Value3}
)

monorepo

• /pkg/rpc/README.md
• /examples/arpc
• /pkg/rpc/HOWTO.md

Go back to the monorepo root to continue reading.

Documentation

Index

• type Op
• type Value
• type Worker
  • func (s *Worker) CallOp(op Op)
  • func (s *Worker) GetValue() Value
  • func (s *Worker) Start()
  • func (s *Worker) Subscribe(lis func())
• type WorkerArpcServer
  • func NewWorkerArpcServer(ctx context.Context, addr string, worker *Worker) (*WorkerArpcServer, error)
  • func (w *WorkerArpcServer) CallOpEnter(e *am.Event) bool
  • func (w *WorkerArpcServer) CallOpState(e *am.Event)
  • func (w *WorkerArpcServer) EventState(_ *am.Event)
  • func (w *WorkerArpcServer) StartState(_ *am.Event)
• type WorkerServiceServer
  • func NewWorkerServiceServer(worker *Worker) *WorkerServiceServer
  • func (s *WorkerServiceServer) CallOp(ctx context.Context, req *pb.CallOpRequest) (*pb.CallOpResponse, error)
  • func (s *WorkerServiceServer) GetValue(ctx context.Context, req *pb.Empty) (*pb.GetValueResponse, error)
  • func (s *WorkerServiceServer) Start(ctx context.Context, req *pb.Empty) (*pb.Empty, error)
  • func (s *WorkerServiceServer) Subscribe(req *pb.Empty, stream pb.WorkerService_SubscribeServer) error

Types

type Op

type Op int

const (
	Op1 Op = iota + 1
	Op2
	Op3
)

type Value

type Value int

const (
	Value1 Value = iota + 1
	Value2
	Value3
)

type Worker

type Worker struct {
	ErrCount     int
	SuccessCount int
	// contains filtered or unexported fields
}

func (*Worker) CallOp

func (s *Worker) CallOp(op Op)

func (*Worker) GetValue

func (s *Worker) GetValue() Value

func (*Worker) Start

func (s *Worker) Start()

func (*Worker) Subscribe

func (s *Worker) Subscribe(lis func())

type WorkerArpcServer

type WorkerArpcServer struct {
	Worker *Worker
	Mach   *am.Machine

	RPC *arpc.Server
}

func NewWorkerArpcServer

func NewWorkerArpcServer(
	ctx context.Context, addr string, worker *Worker,
) (*WorkerArpcServer, error)

func (*WorkerArpcServer) CallOpEnter

func (w *WorkerArpcServer) CallOpEnter(e *am.Event) bool

func (*WorkerArpcServer) CallOpState

func (w *WorkerArpcServer) CallOpState(e *am.Event)

func (*WorkerArpcServer) EventState

func (w *WorkerArpcServer) EventState(_ *am.Event)

func (*WorkerArpcServer) StartState

func (w *WorkerArpcServer) StartState(_ *am.Event)

type WorkerServiceServer

type WorkerServiceServer struct {
	pb.UnimplementedWorkerServiceServer
	// contains filtered or unexported fields
}

func NewWorkerServiceServer

func NewWorkerServiceServer(worker *Worker) *WorkerServiceServer

func (*WorkerServiceServer) CallOp

func (s *WorkerServiceServer) CallOp(ctx context.Context, req *pb.CallOpRequest) (*pb.CallOpResponse, error)

func (*WorkerServiceServer) GetValue

func (s *WorkerServiceServer) GetValue(ctx context.Context, req *pb.Empty) (*pb.GetValueResponse, error)

func (*WorkerServiceServer) Start

func (s *WorkerServiceServer) Start(ctx context.Context, req *pb.Empty) (*pb.Empty, error)

func (*WorkerServiceServer) Subscribe

func (s *WorkerServiceServer) Subscribe(req *pb.Empty, stream pb.WorkerService_SubscribeServer) error