Memory-based communication for gRPC calls

Apache Dubbo has the injvm method of communication, which can avoid the latency caused by the network, and also does not occupy the local port, which is a more convenient way of RPC communication for testing and local verification.

I recently saw containerd’s code and found that it has similar requirements. But using ip port communication, there may be port conflict; using unix socket, there may be path conflict. I looked into whether gRPC has a memory-based communication method similar to injvm. Then I found that pipe works very well, so I recorded it.

Golang/gRPC abstraction of the network

First, let’s look at the architecture diagram for a single call to gRPC. Of course, this architecture diagram is currently focused only on the network abstraction distribution.

Golang/gRPC abstraction of the network

We focus on the socket part.

1. OS system abstraction

First, on top of the network packet, the system abstracts out socket, representing a virtual connection, which is unreliable for UDP and as reliable as it can be for TCP.

For network programming, it is not enough to have a connection, but also to tell the developer how to create and close it. For server side , the system provides accept method to receive connections. For the client , the system provides the connect method, which is used to establish a connection with the server.

2. Golang abstraction

In Golang, the concept of socket peer is called net.Conn, which represents a virtual connection.

Next, for the server side, the act of accept is wrapped in the net.Listener interface; for the client side, Golang provides the net.Dial method based on connect.

type Listener interface {
  // 接收来自客户端的网络连接
  Accept() (Conn, error)
  Close() error
  Addr() Addr
}

3. gRPC Usage

So how does gRPC use Listener and Dial?

For the gRPC server, the Serve method receives a Listener, indicating that the service is available on that Listener.

For the gRPC client, the network is essentially just something that can connect to somewhere, so all that is needed is a dialer func(context.Context, string) (net.Conn, error) function.

What is a pipe

At the operating system level, pipe represents a data pipe that is on both ends of this program and can be a good fit for our requirements: memory-based network communication.

Golang also provides net.Pipe() function based on pipe to create a bi-directional, memory-based communication pipe that, in terms of capability, can meet gRPC’s requirements for the underlying communication very well.

However, net.Pipe only generates two net.Conns, i.e. only two network connections, no Listner as mentioned before, and no Dial method.

So in combination with Golang’s channel, net.Pipe is packaged as Listner and also provides Dial methods.

Listener.Accept(), just listen to a channel, when the client connects, the connection can be passed through the channel
Dial method, call Pipe, give one end to the server through the channel (as the server connection), and the other end as the client connection

The code is as follows.

package main

import (
  "context"
  "errors"
  "net"
  "sync"
  "sync/atomic"
)

var ErrPipeListenerClosed = errors.New(`pipe listener already closed`)

type PipeListener struct {
  ch    chan net.Conn
  close chan struct{}
  done  uint32
  m     sync.Mutex
}

func ListenPipe() *PipeListener {
  return &PipeListener{
    ch:    make(chan net.Conn),
    close: make(chan struct{}),
  }
}

// Accept 等待客户端连接
func (l *PipeListener) Accept() (c net.Conn, e error) {
  select {
  case c = <-l.ch:
  case <-l.close:
    e = ErrPipeListenerClosed
  }
  return
}

// Close 关闭 listener.
func (l *PipeListener) Close() (e error) {
  if atomic.LoadUint32(&l.done) == 0 {
    l.m.Lock()
    defer l.m.Unlock()
    if l.done == 0 {
      defer atomic.StoreUint32(&l.done, 1)
      close(l.close)
      return
    }
  }
  e = ErrPipeListenerClosed
  return
}

// Addr 返回 listener 的地址
func (l *PipeListener) Addr() net.Addr {
  return pipeAddr(0)
}
func (l *PipeListener) Dial(network, addr string) (net.Conn, error) {
  return l.DialContext(context.Background(), network, addr)
}
func (l *PipeListener) DialContext(ctx context.Context, network, addr string) (conn net.Conn, e error) {
  // PipeListener是否已经关闭
  if atomic.LoadUint32(&l.done) != 0 {
    e = ErrPipeListenerClosed
    return
  }

  // 创建pipe
  c0, c1 := net.Pipe()
  // 等待连接传递到服务端接收
  select {
  case <-ctx.Done():
    e = ctx.Err()
  case l.ch <- c0:
    conn = c1
  case <-l.close:
    c0.Close()
    c1.Close()
    e = ErrPipeListenerClosed
  }
  return
}

type pipeAddr int

func (pipeAddr) Network() string {
  return `pipe`
}
func (pipeAddr) String() string {
  return `pipe`
}

How to use pipe as a connection for gRPC

With the above wrapper, we can create a gRPC server-side and client-side based on this for memory-based RPC communication.

First, we simply create a service that contains four invocations.

syntax = "proto3";

option go_package = "google.golang.org/grpc/examples/helloworld/helloworld";
option java_multiple_files = true;
option java_package = "io.grpc.examples.helloworld";
option java_outer_classname = "HelloWorldProto";

package helloworld;

// The greeting service definition.
service Greeter {
  // unary调用
  rpc SayHello(HelloRequest) returns (HelloReply) {}

  // 服务端流式调用
  rpc SayHelloReplyStream(HelloRequest) returns (stream HelloReply);

  // 客户端流式调用
  rpc SayHelloRequestStream(stream HelloRequest) returns (HelloReply);

  // 双向流式调用
  rpc SayHelloBiStream(stream HelloRequest) returns (stream HelloReply);
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

Then generate the relevant stub code.

1
2
3

protoc --go_out=. --go_opt=paths=source_relative \
  --go-grpc_out=. --go-grpc_opt=paths=source_relative \
  helloworld/helloworld.proto

Then start writing the server-side code, the basic logic is to achieve a demo version of the server is good.

package main

import (
  "context"
  "log"

  "github.com/robberphex/grpc-in-memory/helloworld"
  pb "github.com/robberphex/grpc-in-memory/helloworld"
)

// helloworld.GreeterServer 的实现
type server struct {
  // 为了后面代码兼容，必须聚合UnimplementedGreeterServer
  // 这样以后在proto文件中新增加一个方法的时候，这段代码至少不会报错
  pb.UnimplementedGreeterServer
}

// unary调用的服务端代码
func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
  log.Printf("Received: %v", in.GetName())
  return &pb.HelloReply{Message: "Hello " + in.GetName()}, nil
}

// 客户端流式调用的服务端代码
// 接收两个req，然后返回一个resp
func (s *server) SayHelloRequestStream(streamServer pb.Greeter_SayHelloRequestStreamServer) error {
  req, err := streamServer.Recv()
  if err != nil {
    log.Printf("error receiving: %v", err)
    return err
  }
  log.Printf("Received: %v", req.GetName())
  req, err = streamServer.Recv()
  if err != nil {
    log.Printf("error receiving: %v", err)
    return err
  }
  log.Printf("Received: %v", req.GetName())
  streamServer.SendAndClose(&pb.HelloReply{Message: "Hello " + req.GetName()})
  return nil
}

// 服务端流式调用的服务端代码
// 接收一个req，然后发送两个resp
func (s *server) SayHelloReplyStream(req *pb.HelloRequest, streamServer pb.Greeter_SayHelloReplyStreamServer) error {
  log.Printf("Received: %v", req.GetName())
  err := streamServer.Send(&pb.HelloReply{Message: "Hello " + req.GetName()})
  if err != nil {
    log.Printf("error Send: %+v", err)
    return err
  }
  err = streamServer.Send(&pb.HelloReply{Message: "Hello " + req.GetName() + "_dup"})
  if err != nil {
    log.Printf("error Send: %+v", err)
    return err
  }
  return nil
}

// 双向流式调用的服务端代码
func (s *server) SayHelloBiStream(streamServer helloworld.Greeter_SayHelloBiStreamServer) error {
  req, err := streamServer.Recv()
  if err != nil {
    log.Printf("error receiving: %+v", err)
    // 及时将错误返回给客户端，下同
    return err
  }
  log.Printf("Received: %v", req.GetName())
  err = streamServer.Send(&pb.HelloReply{Message: "Hello " + req.GetName()})
  if err != nil {
    log.Printf("error Send: %+v", err)
    return err
  }
  // 离开这个函数后，streamServer会关闭，所以不推荐在单独的goroute发送消息
  return nil
}

// 新建一个服务端实现
func NewServerImpl() *server {
  return &server{}
}

Then we create a client based on a pipe connection to call the server.

The following steps are included.

create the server-side implementation
create a listener based on the pipe, and then create a gRPC server based on it
create a client connection based on the pipe, and then create a gRPC client to call the service

The code is as follows.

package main

import (
  "context"
  "fmt"
  "log"
  "net"

  pb "github.com/robberphex/grpc-in-memory/helloworld"
  "google.golang.org/grpc"
)

// 将一个服务实现转化为一个客户端
func serverToClient(svc *server) pb.GreeterClient {
  // 创建一个基于pipe的Listener
  pipe := ListenPipe()

  s := grpc.NewServer()
  // 注册Greeter服务到gRPC
  pb.RegisterGreeterServer(s, svc)
  if err := s.Serve(pipe); err != nil {
    log.Fatalf("failed to serve: %v", err)
  }
  // 客户端指定使用pipe作为网络连接
  clientConn, err := grpc.Dial(`pipe`,
    grpc.WithInsecure(),
    grpc.WithContextDialer(func(c context.Context, s string) (net.Conn, error) {
      return pipe.DialContext(c, `pipe`, s)
    }),
  )
  if err != nil {
    log.Fatalf("did not connect: %v", err)
  }
  // 基于pipe连接，创建gRPC客户端
  c := pb.NewGreeterClient(clientConn)
  return c
}

func main() {
  svc := NewServerImpl()
  c := serverToClient(svc)

  ctx := context.Background()

  // unary调用
  for i := 0; i < 5; i++ {
    r, err := c.SayHello(ctx, &pb.HelloRequest{Name: fmt.Sprintf("world_unary_%d", i)})
    if err != nil {
      log.Fatalf("could not greet: %v", err)
    }
    log.Printf("Greeting: %s", r.GetMessage())
  }

  // 客户端流式调用
  for i := 0; i < 5; i++ {
    streamClient, err := c.SayHelloRequestStream(ctx)
    if err != nil {
      log.Fatalf("could not SayHelloRequestStream: %v", err)
    }
    err = streamClient.Send(&pb.HelloRequest{Name: fmt.Sprintf("SayHelloRequestStream_%d", i)})
    if err != nil {
      log.Fatalf("could not Send: %v", err)
    }
    err = streamClient.Send(&pb.HelloRequest{Name: fmt.Sprintf("SayHelloRequestStream_%d_dup", i)})
    if err != nil {
      log.Fatalf("could not Send: %v", err)
    }
    reply, err := streamClient.CloseAndRecv()
    if err != nil {
      log.Fatalf("could not Recv: %v", err)
    }
    log.Println(reply.GetMessage())
  }

  // 服务端流式调用
  for i := 0; i < 5; i++ {
    streamClient, err := c.SayHelloReplyStream(ctx, &pb.HelloRequest{Name: fmt.Sprintf("SayHelloReplyStream_%d", i)})
    if err != nil {
      log.Fatalf("could not SayHelloReplyStream: %v", err)
    }
    reply, err := streamClient.Recv()
    if err != nil {
      log.Fatalf("could not Recv: %v", err)
    }
    log.Println(reply.GetMessage())
    reply, err = streamClient.Recv()
    if err != nil {
      log.Fatalf("could not Recv: %v", err)
    }
    log.Println(reply.GetMessage())
  }

  // 双向流式调用
  for i := 0; i < 5; i++ {
    streamClient, err := c.SayHelloBiStream(ctx)
    if err != nil {
      log.Fatalf("could not SayHelloStream: %v", err)
    }
    err = streamClient.Send(&pb.HelloRequest{Name: fmt.Sprintf("world_stream_%d", i)})
    if err != nil {
      log.Fatalf("could not Send: %v", err)
    }
    reply, err := streamClient.Recv()
    if err != nil {
      log.Fatalf("could not Recv: %v", err)
    }
    log.Println(reply.GetMessage())
  }
}

Summarize

Of course, there are better ways to communicate as memory-based RPC calls, such as passing the object directly to the server and communicating directly by local calls. But this way breaks many conventions, such as object address, such as gRPC connection parameters do not take effect, and so on.

In this article, the communication method based on Pipe is consistent with the normal RPC communication behavior except that the network layer goes through memory transfer, such as the same experience of serialization, through the HTTP/2 flow control, etc.. Of course, the performance will be a little worse than the native call, but the good thing is that for testing and verification scenarios, the consistency in behavior is more important.

Table of Contents

Golang/gRPC abstraction of the network

1. OS system abstraction

2. Golang abstraction

3. gRPC Usage

What is a pipe

How to use pipe as a connection for gRPC

Summarize