Concurrent RPCs

[zeebo]

A common issue that everyone will hit when using DRPC is how to handle concurrent RPCs. We should have some answer for this.

[maxtruxa]

Some thoughts on the topic:

Connection pooling can be one such answer, that can even be used without modifying DRPC at all. But while connection pooling can be an effective technique, it is not always possible. I have a somewhat particular use case where an agent establishes a TCP connection to a server and offers RPC services that the server can then invoke. In other words, the TCP client is the RPC server and the TCP server is the RPC client. That means it's impossible for the RPC client to create new connections. (That is btw the reason I can't use gRPC at all and was very happy when I stumbled upon DRPC, which was extremely easy to hook up this way :))

A more integrated but also more complex solution is stream multiplexing, i.e. sending concurrent streams over a single transport. The wire format supports this already, because each frame contains a stream ID. Stream multiplexing is nice but brings its own set of problems (e.g. head-of-line blocking). For any kind of stream (server, client or bidirectional), some sort of back pressure mechanism is probably required to prevent the remote from being flooded with messages if it can't keep up - which would negatively impact or even kill other streams using the same connection. This is not an issue with the current non-multiplexing behavior, when a transport with its own pressure mechanism is used (e.g. TCP).

[zeebo]

Connection pooling can be one such answer, that can even be used without modifying DRPC at all. But while connection pooling can be an effective technique, it is not always possible. I have a somewhat particular use case where an agent establishes a TCP connection to a server and offers RPC services that the server can then invoke. In other words, the TCP client is the RPC server and the TCP server is the RPC client. That means it's impossible for the RPC client to create new connections.

Connection pooling is currently the answer that Storj uses (https://pkg.go.dev/storj.io/common/rpc/rpcpool) and works well. I think the "reverse connection" use case may be niche enough that an in-library answer need not apply to it. I totally didn't anticipate that, so it's great that it worked out :smile:.

Stream multiplexing is nice but brings its own set of problems (e.g. head-of-line blocking).

Yeah, this HoL blocking and buffering were the main problems I wanted to avoid by adding a multiplexing solution at the deepest layers. As you noticed, the wire format initially was meant to allow concurrent streams. At some point, I decided to remove concurrent stream support, and the readers now will error if received stream ids are not monotonically increasing. This greatly simplified the code to read and reassemble packets from the transport.

That said, multiplexing can be added on top of DRPC without changes as well by using something like https://pkg.go.dev/github.com/hashicorp/yamux. Or, if your transport is something like QUIC, you can open new streams from a single session.

In summary, I agree that connection pooling and multiplexing are the two main ways of solving this.

---

Some scattered thoughts:

• If we define an interface like interface { NewStream() (drpc.Transport, error) } we can make a drpcmulticonn (all names are strawmen) package that takes that and creates a new stream for every RPC. Then you can write a wrapper around yamux or your QUIC conn or whatever.

• There can be a drpcpoolconn package that exports a func New(dial func() (drpc.Transport, error)) drpc.Conn and calls and caches the results from dial (this is nearly the strategy Storj takes right now, except this would allow multiple pools that dial to the same destination).

• There can be a drpcpool package that exports a func (p *Pool) Get(key interface{}, dial func() (drpc.Transport, error)) (drpc.Conn, error) (this is the strategy Storj takes)

• I think drpcpoolconn is basically equivalent to drpcmulticonn in the sense that you can transform them into each other:

  • drpcmulticonn.New(streams) => drpcpoolconn.New(streams.NewStream)
  • drpcpoolconn.New(dial) => drpcmulticonn.New(streamsFunc(dial)) (streamsFunc implements the interface)
  • The only difference would be that drpcmulticonn is effectively a drpcpoolconn with cache size of 0.

• I think drpcpool is more general than drpcpoolconn

  • drpcpoolconn.New(dial) => drpcpool.New().Get(nil, dial)
  • The reverse transformation would require some effort.
  • Again, drpcpoolconn can be thought of as using a drpcpool with cache size 0.

• It seems to me the most common case that people will hit, especially when migrating from gRPC, will be adding concurrency to a system that uses basic TCP sockets, and I think that implies some sort of connection pooling to avoid a dial for every RPC.

[bcessa]

hey hi @zeebo, first of all DRPC is really cool :) thanks for sharing. I recently wrote a library of utilities to use it on production services. The package includes basic things like:

• Simplified TLS setup (for client and server)
• Server middleware, including basic components for logging, token-based auth, rate limit, panic recovery, etc
• Client middleware, including basic components for logging, custom metadata, panic recovery, etc
• Bi-directional streaming support over upgraded HTTP(S) connections using WebSockets
• Concurrent RPCs via connection pool

I was wondering if it's ok to add the link here (or somewhere else) for people looking for that kind of thing 😄 not trying to do any form of shameless plug so feel free to remove this message if it's not appropriate to do so 👍🏼

• https://pkg.go.dev/go.bryk.io/pkg/net/drpc
• https://github.com/bryk-io/pkg

[zeebo]

Awesome! I've added your description and a link to the package to the README. Let me know if it doesn't look right.

[Elara6331]

Hello, I recently wrote a module for concurrent RPC support with DRPC using stream multiplexing via yamux. It contains drop-in replacements for DRPC's server and conn.

Here are the links:

• https://gitea.arsenm.dev/Arsen6331/drpc
• https://pkg.go.dev/go.arsenm.dev/drpc/muxconn
• https://pkg.go.dev/go.arsenm.dev/drpc/muxserver

Can this be added to the README as well for people looking for something like this? Also, I can submit a PR adding it to DRPC as drpcmuxconn and drpcmuxserver if that seems like a good idea.

[zeebo]

Great! I'll add a link to the README. I don't think it's right to add it to the main repo at least in part due to the 3rd party dependency. An external package is great. I reviewed the code and I have a suggestion:

https://gitea.arsenm.dev/Arsen6331/drpc/src/commit/a85a9613bd6a/muxconn/muxconn.go#L39-L45

• This will panic if Close is called multiple times. Generally, DRPC tries to make Close be idempotent so you can use the "defer double close" pattern.
• If the session errors on closing, you should still close the conn, I think.

Everything else looked awesome. Thanks!

[Elara6331]

Thanks for your suggestions. I've just implemented them.

[zeebo]

There's now a bunch of different options for this.