Rpc.dispatch', which is like
dispatchbut returns an
Rpc_result.t, which has more information than the
Add some benchmarks for Pipe_rpc
Optimize the server side of Pipe RPC and add an expert interface.
This feature will benefit all Pipe RPC implementors.
│ Name │ Time/Run │ mWd/Run │ Percentage │
bench.mldirect write │ 148.87ns │ 126.00w │ 100.00% │
│ Name │ Time/Run │ mWd/Run │ Percentage │
bench.mldirect write │ 69.71ns │ 3.00w │ 94.99% │
bench.mldirect write expert │ 73.39ns │ │ 100.00% │
Note: the reason the expert version is slightly slower is identified:
if we replace the
rpc_transport_low_latency.ml, then the expert version becomes faster.
The RPC tests for Pipe RPC showed a nice improvement on the server
side: lower CPU consumption and roughly 20% faster
Remove the ~update argument from State_rpc.dispatch.
Before this feature, the dispatch method for State_rpcs looked like this:
: ('query, 'state, 'update, 'error) t
-> update : ('state -> 'update -> 'state)
-> ( 'state * ('state * 'update) Pipe.Reader.t * Metadata.t
) Result.t Or_error.t Deferred.t
There are a couple of cases where having the
updatemethod there can be
sometimes the type of the state that it sent over the wire is different
from the type that is naturally maintained on the client side.
sometimes the update method needs to perform some Async computation to
return a new state.
It feels like the method is trying to be helpful, but has more of a tendency
of getting in the way. This feature removes this method.
The old API may be reinstated by using a
State_rpc.dispatch rpc connection query
Add support for
Implementation and interface very closely resemble that of a
Cleans up the implementation-side interface for aborting
Deferred.tthat got passed to
gone. The situations where it would have been determined now close the reading
end of the user-supplied pipe instead.
Previously, when an RPC dispatcher decided to abort a query, the RPC
implementation would get its
Deferred.tfilled in, but would remain
free to write some final elements to the pipe.
This is a little bit more flexible than the new interface, but it's also
problematic in that the implementer could easily just not pay attention to
aborted. (They're not obligated to pay attention to when the pipe is closed,
either, but at least they can't keep writing to it.) We don't think the extra
flexibility was used at all.
In the future, we may also simplify the client side to remove the
function on the dispatch side (at least when not using
dispatch_iter). For the
time being it remains, but closing the received pipe is the preferred way of
aborting the query.
There are a couple of known ways this might have changed behavior from before.
Both of these appear not to cause problems in the jane repo.
In the past, an implementation could write to a pipe as long as the client
didn't disconnect, even if it closed its pipe. Now writes will raise after
a client closes its pipe (or calls
abort), since the implementor's pipe will
also be closed in this case. Doing this was already unsafe, though, since the
pipe was closed if the RPC connection was closed.
abortedwas only determined if a client aborted the query or the connection
was closed. The new alternative,
Pipe.closedcalled on the returned pipe,
will also be determined if the implementation closes the pipe itself. This is
unlikely to cause serious issues but could potentially cause some confusing
Blocking RPC implementations (i.e., ones made with
Rpc.implement') now capture
the backtrace when they raise. This is consistent with what happens in the
deferred implementation case, since in that case the implementation is run
Monitor.try_with, which captures backtraces as well.
Here's an example of what a new error looks like:
((location "server-side blocking rpc computation")
"Float.iround_up_exn: argument (100000000000000000000.000000) is too large"))
"Raised at file "pervasives.ml", line 31, characters 25-45
\nCalled from file "result.ml", line 43, characters 17-22
\nCalled from file "monad.ml", line 17, characters 20-28
(connection_description ("Client connected via TCP" (localhost 3333)))
(rpc_tag foo) (rpc_version 1))
Transfer.Writer.send*raises, send an error to the client.
Switched to ppx.
Expose the lower-level registration hook in
Allow custom handling of missed async_rpc heartbeats.
dispatchfunction does not behave well when the reader side
of the pipe is closed.
It should gracefully abort the rpc instead of raising exceptions, or whatever it currently does.
Rpc.Expert.implement'with a similar
One_way.Expert.implementbut for 2-way rpcs.
Exceptions raised by an expert implementations are handled as follow:
if the query has already been answered, stop the server (as for one-way expert)
if not, send a
Rpc_error.Uncaught_exn(as for 2-way rpc)
Rpc.Pipe_rpc.dispatch_iter, plus a bunch of additional types to support
it. The main reason for this is to reduce space usage:
Pipe.iter_without_pushbackconsumes ~105 words in the steady state
(i.e., when no messages are coming in) while
dispatch_iterconsumes ~15. I'm
dispatchcan be improved a lot, but a pipe by itself is 46 words, so it
can't possibly become as small as
Both cases can be made smaller by making
instead of a closure. I plan to do this later.
One annoying property of the interface is that the only way to cancel
a subscription is to use
Pipe_rpc.abort, which has a terrible interface.
The logical way to improve the interface is to return a record of
Connection.t, and a
Query_id.t, which allocates an
additional few words. I'd kind of like to do this but it seems counter to the
goal of reducing space usage.
Rpc.Pipe_rpc.implement_direct, which uses a "direct stream writer" to
write results to the other side, rather than using a
consumes much less memory, ~15 words per open query as opposed to ~225 for
A large part of the diff in this feature is the addition of a module
Implementation_types, which just contains copies of type definitions from
Implementations. This is necessary to handle some cyclic
type definitions (both of these modules now depend on types defined in the other
This is the implementation-side dual of
Versioned_rpc.Both_convertso that commands can be constructed without client-side
conversions. Clients remain free to use conversions or not, as appropriate.
Callee_convertsinterfaces because nothing appears to use it,
Both_convertdoes not provide it. Now
Both_convert.Scan be supplied to satisfy
Annotate errors returned by the async-rpc library with the name of the RPC for
which the error was returned (if it's an rpc-level error) and a description of
the remote side of the connection (the ip:host if connected via a network
val rpcsin versioned_rpc modules.
Fixed race in
Rpcthat caused double connection cleanup.
Connection_closedand a Writer error,
occurring at the same time will cleanup the connection twice and call
response_handler of open_queries twice with two different errors.
(((pid 31291) (thread_id 0))
"unhandled exception in Async scheduler"
("Ivar.fill of full ivar" (Full _)
("Raised at file "error.ml", line 7, characters 21-29"
"Called from file "rpc.ml", line 101, characters 8-31"
"Called from file "connection.ml", line 251, characters 8-172"
"Called from file "core_hashtbl.ml", line 244, characters 36-48"
"Called from file "connection.ml", line 248, characters 2-278"
"Called from file "async_stream.ml", line 49, characters 53-56"
"Called from file "async_stream.ml", line 21, characters 34-39"
"Called from file "job_queue.ml", line 124, characters 4-7" ""))
(((name main) (here ()) (id 1) (has_seen_error true)
(is_detached false) (kill_index 0))))))
((pid 31291) (thread_id 0)))))
Fixed bugs in
Rpcin which a TCP connection's reader was closed before its
Versioned_rpc, eliminated an unnecessary Async cycle when placing RPC
give the reason why a pipe returned by an RPC was closed.
These functions take the IDs that are returned along with the pipes by
the dispatch functions, so the interface of
dispatchdid not need to
Rpc.Expert.dispatchexpose that the connection was closed, just like
Expose the name of the
Async_extra.Rpcto its own library,
abstracted its transport layer.
Async_kernel_rpcdepends only on
Async_kernel. This allows
different use cases.
Versioned_rpcwas moved to
Rpc.One_waymodule, for RPCs that do not expect any response
We have been able to send 6_000_000 (contentless) one-way messages
per second under some idealized circumstances.
Rpc, added an optional
heartbeat_configargument to configure
the heartbeat interval and timeout.
Rpc, added some information to
connection_stateargument to the
Rpc.Connection.serve, one can put client addresses in
connection_state, so this makes it possible to identify who
sent the unknown RPC instead of just saying what the RPC's name and
Rpc.One_way.Expert, which has an
gives direct access to the internal buffer instead of using a
Rpcnot raise an exception if a connection is closed due to
Transportimplementation to not
transferfunction similar to
Changed the RPC connection to flush the pipe when the writer is
closed, which was the only behavior of
that had been relied on.
With this change, if someone closes the underlying writer by hand
the pipes won't be flushed, which should be expected anyway.
Fixed an issue where "normal"
Pipe_rpcerrors caused the connection
Such errors include querying an unknown RPC or getting an exception
raised by the RPC implementation shutdown. Now such errors behave
Rpcerrors, i.e. they are completely ignored besides being
put in the return value of
Errors that occur later in
Pipe_rpcstill cause the connection to
close, since these should only occur if there is a bug somewhere.
Rpc, connection-closing errors no longer raise top-level
One can now call
close_reason : t -> Info.t Deferred.tto get the
reason the connection was closed.
Rpcto not write if the transport is closed.
The fix doesn't check that the writer is closed, but instead ensure
that we don't try to write when we shouldn't. This means that it
will still fail if the user close the transport by hand, which they
This change requires transport implementations to fail after they
have been closed. This is different from the semantics of
Async_unix.Writer, but the latter is non-intuitive and makes it
hard to check the correctness of the code. Moreover it is only