module Rpc_client:sig
..end
RPC clients
This module implements an RPC client, i.e. provides means to connect to an RPC service and call remote procedures. In general, this module works in an asynchronous way and is implemented event-driven. All events are handled by an event queue of type Unixqueue.t that must already exist and to which this module adds its own event handlers and event resources. This means that this module can co-exist with other services and share the same event queue with them.
You can push several procedure calls on the event queue at once. The queue serves then as a pipeline; the calls are sent to the server as long as the server accepts new calls. Replies are received in any order, and the return values of the remote procedures are delivered using a callback function.
You can set timeouts and force automatic retransmission if you want this; these features are enabled by default if the underlying transport mechanism is UDP. Timeouts and other exceptions are delivered to the callback functions, too.
The whole mechanism is designed to allow maximum parallelism without needing to use the multi-threading features of O'Caml. Especially, the following parallelisms can be done:
However, there are still some restrictions concerning asynchronous calls. Some of them will be removed in the future, but others are difficult to tackle:
Multi-threading: Only a single thread may use an RPC client at a time. There is a way so that several threads can share the same client without giving up concurrency, see Example: Threads sharing an RPC client for details.
exception Message_lost
got EOF when some pending procedure calls were not replied or even sent
exception Message_timeout
After all retransmissions, there was still no reply
exception Response_dropped
Drop reason: The response exceeded the configured maximum message size
exception Communication_error of exn
an I/O error happened
exception Client_is_down
The RPC call cannot be performed because the client has been shut down in the meantime. You can get this exception if you begin a new call, but the connection is closed now.
exception Keep_call
This exception can be raised by the callback function that is invoked when the server response arrives. It causes that the RPC call record is kept in the housekeeping structure of the client. If the server sends another response, the callback function will be invoked again. I.e. one call can be replied several times (server-driven batching).
exception Unbound_exception of exn
This exception can be raised by the callback function that is invoked
when the server response arrives. It simply causes that the inner
exception bypasses the exception handler, and falls through to the
caller of Unixqueue.run
. This is useful to jump out of the running RPC
routines.
type
t
The type of RPC clients
type
connector =
| |
Inet of |
(* | Hostname or IP address, port | *) |
| |
Internet of |
(* | The address plus port | *) |
| |
Unix of |
(* | Path to unix dom sock. Not supported on Win32. | *) |
| |
W32_pipe of |
(* | Path to named pipe (only Win32) | *) |
| |
Descriptor of |
(* | Pass an already open socket descriptor. The descriptor will not
be closed when the client is done! On Win32, the proxy descriptors
as returned by | *) |
| |
Dynamic_descriptor of |
(* | The function is called to get the socket descriptor.
Unlike | *) |
| |
Portmapped of |
(* | The portmapper on this host is queried to get address information | *) |
val connector_of_sockaddr : Unix.sockaddr -> connector
Converts the socket address into a connector
val connector_of_socksymbol : Netsockaddr.socksymbol -> connector
Converts the Netsockaddr.socksymbol
into a connector
val shutdown_connector : t ->
Rpc_transport.rpc_multiplex_controller -> (unit -> unit) -> unit
The default implementation to shut down the connector. Actions are triggered that will take the connector down at some time in the future. At this time, the callback function is invoked.
For Descriptor
connector the socket is shut down but not closed.
For the other connector types the socket is also closed.
Win32 named pipes are shut down.
class type socket_config =object
..end
Configuration for `Socket
(see below).
val default_socket_config : socket_config
Default configuration with non_blocking_connect
= true
class default_socket_config :socket_config
Default configuration as class
val blocking_socket_config : socket_config
Configuration with non_blocking_connect
= false
class blocking_socket_config :socket_config
blocking connect
configuration as class
val tls_socket_config : (module Netsys_crypto_types.TLS_CONFIG) -> socket_config
This configuration establishes TLS when connecting with the server.
It is (so far) only compatible with Rpc.Tcp
.
class tls_socket_config :(module Netsys_crypto_types.TLS_CONFIG) ->
socket_config
TLS configuration as class
typeinternal_pipe =
Netxdr.xdr_value Netsys_polypipe.polypipe
typeinternal_socket =
Netxdr.xdr_value Netsys_polysocket.polyclient
typemode2 =
[ `Internal_endpoint of internal_pipe * internal_pipe
| `Internal_socket of internal_socket
| `Multiplexer_endpoint of Rpc_transport.rpc_multiplex_controller
| `Socket of Rpc.protocol * connector * socket_config
| `Socket_endpoint of Rpc.protocol * Unix.file_descr ]
Determines the type of the client for create2
:
`Socket_endpoint(proto,fd)
: Socket fd
is a connected socket
descriptor used for communication. proto
determines the
encapsulation; should be Tcp
for stream sockets and Udp
for
datagram sockets. The descriptor will be closed when the client
terminates.`Multiplexer_endpoint m
: m
is an RPC multiplex controller.`Socket(proto, conn, config)
: Creates and connect a client
socket according to conn
. proto
determines the
encapsulation; should be Tcp
for stream sockets and Udp
for
datagram sockets. config
specifies configuration details.
In particular, use this option to enable TLS for the socket:
Get a tls_socket_config
and pass this as config
here.`Internal_endpoint(rd,wr)
: Creates a client that exchanges
data over the pair of polypipes (rd,wr)
(see Netsys_polypipe
).
The polypipes will be closed when the client terminates.`Internal_socket psock
: Creates a client that exchanges
data over the polysocket client psock
(see Netsys_polysocket
).
The client must already be connected to the server (i.e.
Netsys_polysocket.connect
has already been called).
The polysocket will be closed when the client terminates.val create2 : ?program_number:Netnumber.uint4 ->
?version_number:Netnumber.uint4 ->
?initial_xid:int ->
?shutdown:(t ->
Rpc_transport.rpc_multiplex_controller -> (unit -> unit) -> unit) ->
mode2 -> Rpc_program.t -> Unixqueue.event_system -> t
New style clients:
Opens a connection to the server specified by mode2
.
The server is assumed to implement an RPC program as specified by
the Rpc_program.t
argument. (You can override the program and version
numbers stored in this argument by the optional parameters
program_number
and version_number
. If you need to call several
programs/versions with the same client, use unbound_create
instead.)
All communication to the server is handled using the given queue
Unixqueue.event_system
. There is a limit of 2GB per message
or Sys.max_string_length
, whatever is lower.
If the protocol (passed along with mode2
) is Tcp, the communication
will be handled stream-oriented. In this case, no timeout is detected
and no retransmissions are done.
If the protocol is Udp, a datagram-oriented communication style is used. This works only for Internet UDP sockets because these are bidirectional (Unix domain sockets are unidirectional and do not work). For Udp, there is a timeout of 15 seconds and a maximum of 3 retransmissions (i.e. a total of 4 transmission trials). For connected UDP sockets there is a limit of 64K per message (max. size of an Internet packet). For unconnected UDP sockets there is a limit of 16K per message due to restrictions in the OCaml runtime.
program_number
: Overrides the program number in Rpc_program.t
version_number
: Overrides the version number in Rpc_program.t
initial_xid
: The initial value for the session identifier.shutdown
: This function is called when the client is shut down
to close the client socket. By default, shutdown_connector
is
called.val unbound_create : ?initial_xid:int ->
?shutdown:(t ->
Rpc_transport.rpc_multiplex_controller -> (unit -> unit) -> unit) ->
mode2 -> Unixqueue.event_system -> t
Creates an unbound client. This is like create2
, but the client is
not restricted to a particular RPC program.
One can convert an unbound client into a bound client by calling
bind
, see below. It is possible to bind several times, so several
programs can be called with the same client (provided the server is
also capable of dealing with several programs).
This function does not support Portmapped
connectors.
val bind : t -> Rpc_program.t -> unit
Binds this program additionally
val use : t -> Rpc_program.t -> unit
If there are no bound programs, this is a no-op. Otherwise it is checked whether the passed program is bound. If not, an exception is raised.
Programs are compared by comparing Rpc_program.id
. The program
must be the same value, but it is also allowed to
Rpc_program.update
it in the meantime, i.e. to change program
and version numbers.
val configure : t -> int -> float -> unit
configure client retransmissions timeout
:
sets the number of retransmissions and the timeout for the next calls.
(These values are defaults; the actual values are stored with each
call.)
Values of retransmissions > 0
are semantically only valid if the
called procedures are idempotent, i.e. invoking them several times
with the same values has the same effect as only one invocation.
Positive values for retransmissions
should only be used for Udp-style
communication.
The timeout value determines how long the client waits until the
next retransmission is done, or, if no more retransmissions are
permitted, a Message_timeout
exception is delivered to the receiving
callback function. A timeout
value of 0.0 means immediate timeout
(see next paragraph). A negative timeout
value means 'no timeout'.
Positive timeout
values are possible for both Udp and Tcp connections.
Timeout values are measured in seconds.
There is a special application for the timeout value 0.0: If you
don't expect an answer from the server at all ("batch mode"), this
timeout value will cause that the message handler will get
a Message_timeout
exception immediately. You should ignore this
exception for batch mode. The positive effect from the timeout is that
the internal management routines will remove the remote call from
the list of pending calls such that this list will not become too long.
(You can get a similar effect by calling set_batch_call
, however.)
Note that the meaning of timeouts for TCP connections is unclear. The TCP stream may be in an undefined state. Because of this, the client does not make any attempt to clean the state up for TCP. The user is advised to shut down the client, and reconnect.
There is another subtle difference between UDP and TCP. For UDP, the timer is started when the packet is sent. For TCP, however, the timer is already started when the RPC call is added to the queue, i.e. much earlier. This means that the time for connecting to the remote service is also bound by the timeout. The rationale is that TCP timeouts are usually set to catch total service failures rather than packet losses, and this behaviour is best for this purpose.
val configure_next_call : t -> int -> float -> unit
Same as configure
, but it only affects the next call
val set_dgram_destination : t -> Unix.sockaddr option -> unit
set_dgram_destination client addr_opt
: This function is required
for using the client in conjunction with unconnected UDP sockets.
For connected sockets, the destination of datagrams is implicitly
given. For unconnected sockets, one has to set the destination
explicitly. Do so by calling set_dgram_destination
with
Some addr
as addr_opt
argument before calling.
Passing None
as addr_opt
removes the explicit destination again.
Note that unconnected sockets differ from connected sockets also in
the relaxation that they can receive messages from any IP address,
and not only the one they are connected to.
The current destination is used for all following calls. It is
not automatically reset to None
after the next call.
val set_batch_call : t -> unit
The next call will be a batch call. The client does not wait for the response of a batch call. Instead, the client immediately fakes the response of a "void" return value.
It is required that the batch call has a "void" return type. Otherwise, the client raises an exception, and ignores the call.
This setting only affects the next call.
val set_user_name : t -> string option -> unit
Sets the user name, or None
(the default user name). This is only
meaningful for authentication.
val set_max_response_length : t -> int -> unit
Sets the maximum length of responses. By default, there is only the
implicit maximum of Sys.max_string_length
.
If the maximum is exceeded, the exception Response_dropped
is raised.
val set_exception_handler : t -> (exn -> unit) -> unit
sets an exception handler (the default prints the exception
with `Crit
level to the logger set in Netlog
).
Only exceptions resulting from invocations of a
callback function are forwarded to this handler (unless wrapped
by Unbound_exception
).
Exceptions occuring in the handler itself are not caught, and will fall through.
val set_mstring_factories : t -> Netxdr_mstring.named_mstring_factories -> unit
Sets the mstring factory configuration that is used for decoding responses containing managed strings.
val event_system : t -> Unixqueue.event_system
Returns the unixqueue to which the client is attached
val programs : t -> Rpc_program.t list
Returns the list of all bound programs
val get_socket_name : t -> Unix.sockaddr
val get_peer_name : t -> Unix.sockaddr
Return the addresses of the client socket and the server socket, resp. Note that these are only available when the client is already connected. The function calls fail otherwise. It is also possible that the underlying transport mechanism does not know these data.
val get_sender_of_last_response : t -> Unix.sockaddr
Return the address of the sender of the last received response.
val get_xid_of_last_call : t -> Netnumber.uint4
Returns the session identifier used in the just made call
val get_protocol : t -> Rpc.protocol
Get the protocol flavour
val get_tls_session_props : t -> Nettls_support.tls_session_props option
Get the TLS properties so far TLS is enabled
val get_gssapi_props : t -> Netsys_gssapi.client_props option
Get the GSSAPI properties of the last call (so far available)
val abandon_call : t -> Netnumber.uint4 -> unit
To be used in conjunction with Rpc_client.Keep_call
: The call
with this session identifier is no longer expected, and removed
from the internal data structures.
Restriction: for now, this does not work when there is authentication.
val is_up : t -> bool
Return whether the client is up
val get_stats : t -> int * int * int
Get stats (n_delayed, n_waiting, n_pending)
:
n_delayed
: Calls that wait for authenticationn_waiting
: Calls that wait for being sentn_pending
: Calls that wait for the responseval unbound_sync_call : t ->
Rpc_program.t -> string -> Netxdr.xdr_value -> Netxdr.xdr_value
unbound_sync_call client pgm proc arg
: Invoke the remote procedure
proc
of the program pgm
via client
. The input arguments are
arg
. The result arguments are returned (or an error is raised)
val unbound_async_call : t ->
Rpc_program.t ->
string -> Netxdr.xdr_value -> ((unit -> Netxdr.xdr_value) -> unit) -> unit
unbound_ssync_call client pgm proc arg emit
: Invoke the remote
procedure
proc
of the program pgm
via client
. The input arguments are
arg
. When the result r
is available, the client will call
emit (fun () -> r)
back. When an exception e
is available, the
client will call emit (fun () -> raise e)
back.
class unbound_async_call :t -> Rpc_program.t -> string -> Netxdr.xdr_value ->
[Netxdr.xdr_value]
Uq_engines.engine
Same as unbound_async_call
, but with an engine API.
val synchronize : Unixqueue.event_system -> ('a -> ((unit -> 'b) -> unit) -> unit) -> 'a -> 'b
Turns an async call into a synchronous call
val shut_down : t -> unit
Shuts down the connection. Any unprocessed calls get the exception
Message_lost
. It is no error to shut down a client that is already
down - nothing happens in this case.
Shutdowns can be complex operations. For this reason, this function implements some magic that is usually the right thing, but may also be wrong:
Descriptor
to the client.
You don't know when the client is finally down, and the descriptor
can be closed.The following functions allow more fine grained control of the shutdown.
val sync_shutdown : t -> unit
Enforces a synchronous shutdown of the connection. This is only possible if called from outside the event loop. This function fails if called from within the event loop.
You can be sure that the shutdown is completely done when this function returns normally.
val trigger_shutdown : t -> (unit -> unit) -> unit
Triggers the shutdown, and calls the passed function back when it is done.
The function is not only called when the client has to be taken down, but also if the client is already down.
val set_debug_name : t -> string -> unit
Set a name printed with debug messages
val get_debug_name : t -> string
Get the debug name
typereject_code =
[ `Fail | `Next | `Renew | `Retry ]
Reaction on authentication problems:
`Fail
: Stop here, and report to user`Retry
: Just try again with current session`Renew
: Drop the current session, and get a new session from
the current auth_method
`Next
: Try the next authentication methodclass type auth_session =object
..end
An auth_session
object is normally created for every client instance.
class type auth_protocol =object
..end
An authentication protocol is used for creating an authentication session.
class type auth_method =object
..end
An auth_method
object represents a method of authentication.
val auth_none : auth_method
The authentication method that does not perform authentication.
val set_auth_methods : t -> auth_method list -> unit
Set the authentication methods for this client. The passed methods
are tried in turn until a method is accepted by the server.
The default is auth_none
When the methods are set for an active client, the ongoing calls are continued with the old method. First new calls are ensured to use the new list.
module type USE_CLIENT =sig
..end
This module type is what the generated "clnt" module assumes about the client interface
val xdr_ctx : t -> Netxdr.ctx
Returns the recommended XDR context
val create : ?program_number:Netnumber.uint4 ->
?version_number:Netnumber.uint4 ->
?initial_xid:int ->
?shutdown:(t ->
Rpc_transport.rpc_multiplex_controller -> (unit -> unit) -> unit) ->
Unixqueue.event_system ->
connector -> Rpc.protocol -> Rpc_program.t -> t
create2
or unbound_create
.Opens a connection to the server specified by the connector
.
The server is assumed to implement an RPC program as specified by
the Rpc_program.t
argument. (You can override the program and version
numbers stored in this argument by the optional parameters
program_number
and version_number
.)
All communication to the server is handled using the given queue
Unixqueue.event_system
.
If the protocol is Tcp, the communication will be handled stream- oriented. In this case, no timeout is detected and no retransmissions are done.
If the protocol is Udp, a datagram-oriented communication style is used. This works only for Internet UDP sockets because these are bidirectional (Unix domain sockets are unidirectional and do not work). For Udp, there is a timeout of 15 seconds and a maximum of 3 retransmissions (i.e. a total of 4 transmission trials).
Unlike create2
, servers made with create
always use blocking
connect
for backwards compatibility.
program_number
: Overrides the program number in Rpc_program.t
version_number
: Overrides the version number in Rpc_program.t
initial_xid
: The initial value for the session identifier.shutdown
: This function is called when the client is shut down
to close the client socket. By default, shutdown_connector
is
called.val program : t -> Rpc_program.t
List.hd (Rpc_client.programs client)
Returns the program the client represents.
val add_call : t ->
string -> Netxdr.xdr_value -> ((unit -> Netxdr.xdr_value) -> unit) -> unit
add_call
is restricted to the case that there is only
one bound program. It will fail in other cases. Use
unbound_async_call
instead. Note also that there is no longer
the optional when_sent
argument. Use set_batch_call
insteadadd_call client proc_name arg f
: add the call to the procedure name
with argument arg
to the queue of unprocessed calls.
When the reply has arrived or an error situation is detected, the
function f
is called back. The argument of f
is another function
that will return the result or raise an exception:
let my_f get_result =
try
let result = get_result() in
...
with
exn -> ...
in
add_call client name arg my_f
If f
does not catch the exception, the pluggable exception handler
of the client is called (see set_exception_handler
). Exceptions are
either Message_lost
, Message_timeout
, or Communication_error
.
The function f
can raise the exception Keep_call
to indicate
the special handling that a further reply of the call is expected
(batching).
val sync_call : t -> string -> Netxdr.xdr_value -> Netxdr.xdr_value
Calls the procedure synchronously. Note that this implies that the underlying unixqueue is started and that all events are processed regardless of whether they have something to do with this call or not.
val verbose : bool -> unit
set whether you want debug messages or not (same as setting
Rpc_client.Debug.enable
)
module Debug:sig
..end