Module Uq_engines

module Uq_engines: sig .. end

An engine performs a certain task in an autonomous way. Engines are attached to a Unixqueue.event_system, and do their task by generating events for resources of the operating system, and by handling such events. Engines are in one of four states: They may be still working, they may be done, they may be aborted, or they may be in an error state. The three latter states a called final states, because they indicate that the engine has stopped operation.

It is possible to ask an engine to notify another object when it changes its state. For simplicity, notification is done by invoking a callback function, and not by issuing notification events.

THREAD SAFETY

Unclear.

Exceptions

exception Closed_channel

Raised when a method of a closed channel object is called (only channel methods count).

This exception should be regarded as equivalent to Netchannels.Closed_channel, but it is not the same exception.

exception Broken_communication

Engines indicate this error when they cannot continue because the other endpoint of communication signals an error.

This exception is not raised, but used as argument of the `Error state.

exception Watchdog_timeout

Used by the watchdog engine to indicate a timeout.

This exception is not raised, but used as argument of the `Error state.

exception Addressing_method_not_supported

Raised by client_socket_connector and server_socket_acceptor to indicate that the passed address is not supported by the class.

exception Cancelled

The callback function of a multiplex_controller is invoked with this exception if the operation is cancelled.

Engine definition

type 'a engine_state = [ `Aborted | `Done of 'a | `Error of exn | `Working of int ]

The type of states with result values of type 't:

`Working n: The engine is working. The number n counts the number of events that have been processed.
`Done arg: The engine has completed its task without errors. The argument arg is the result value of the engine
`Error exn: The engine has aborted because of an error. The argument exn describes the error as an exception.
`Aborted: The engine has aborted because the abort method was called

class type ['a] engine = object .. end

This class type defines the interface an engine must support.

Generic functions and classes

val when_state : ?is_done:('a -> unit) ->
       ?is_error:(exn -> unit) ->
       ?is_aborted:(unit -> unit) -> 'a #engine -> unit

Watches the state of the argument engine, and arranges that one of the functions is called when a final state is reached. After the function has been called, the engine is no longer watched.

is_done : The state transitions to `Done. The argument of is_done is the argument of the `Done state.

is_error : The state transitions to `Error. The argument of is_error is the argument of the `Error state.

is_aborted : The state transitions to `Aborted.

class [['a, 'b]] map_engine : map_done:('a -> 'b engine_state) -> ?map_error:exn -> 'b engine_state -> ?map_aborted:unit -> 'b engine_state -> 'a #engine -> ['b] engine

The map_engine observes the argument engine, and when the state changes to `Done, `Error, or `Aborted, the corresponding mapping function is called, and the resulting state becomes the state of the mapped engine.

class ['a] epsilon_engine : 'a engine_state -> Unixqueue.event_system -> ['a] engine

This engine transitions from its initial state `Working 0 in one step to the passed constant state.

class [['a, 'b]] seq_engine : 'a #engine -> ('a -> 'b #engine) -> ['b] engine

This engine runs two engines in sequential order.

class [['a, 'b]] sync_engine : 'a #engine -> 'b #engine -> [('a * 'b)] engine

This engine runs two engines in parallel, and waits until both are `Done (synchronization).

Fundamental engines

class poll_engine : ?extra_match:exn -> bool -> (Unixqueue.operation * float) list -> Unixqueue.event_system -> object .. end

This engine waits until one of the passed operations can be carried out, or until one of the operations times out.

class poll_process_engine : ?period:float -> pid:int -> Unixqueue.event_system -> [Unix.process_status] engine

This engine waits until the process with the ID pid terminates.

class watchdog : float -> 'a #engine -> [unit] engine

A watchdog engine checks whether the argument engine makes progress, and if there is no progress for the passed number of seconds, the engine is aborted, and the watchdog state changes to `Error Watchdog_timeout.

class ['a] engine_mixin : 'a engine_state -> object .. end

A useful class fragment that implements state and request_notification.

Transfer engines

Transfer engines copy data between file descriptors.

class type async_out_channel = object .. end

An asynchrounous output channel provides methods to output data to a stream descriptor.

class type async_in_channel = object .. end

An asynchrounous input channel provides methods to input data from a stream descriptor.

class receiver : src:Unix.file_descr -> dst:#async_out_channel -> ?close_src:bool -> ?close_dst:bool -> Unixqueue.event_system -> [unit] engine

This engine copies all data from the src file descriptor to the dst output channel.

class sender : src:#async_in_channel -> dst:Unix.file_descr -> ?close_src:bool -> ?close_dst:bool -> Unixqueue.event_system -> [unit] engine

This engine copies all data from the src input channel to the dst file descriptor.

class type async_out_channel_engine = object .. end

Combination of engine + async_out_channel

class type async_in_channel_engine = object .. end

Combination of engine + async_in_channel

class output_async_descr : dst:Unix.file_descr -> ?buffer_size:int -> ?close_dst:bool -> Unixqueue.event_system -> async_out_channel_engine

This engine implements an async_out_channel for the output descriptor dst.

class input_async_descr : src:Unix.file_descr -> ?buffer_size:int -> ?close_src:bool -> Unixqueue.event_system -> async_in_channel_engine

The corresponding class for asynchronous input channels.

type copy_task = [ `Bidirectional of Unix.file_descr * Unix.file_descr
       | `Tridirectional of Unix.file_descr * Unix.file_descr * Unix.file_descr
       | `Uni_socket of Unix.file_descr * Unix.file_descr
       | `Unidirectional of Unix.file_descr * Unix.file_descr ]

Specifies the task the copier class has to do:

`Unidirectional(src,dst): Data from src are copied to dst. EOF of src causes that both descriptors are closed.
`Uni_socket(src,dst): Data from src are copied to dst. EOF of src causes that dst is shut down for sending; all descriptors remain open. It is required that dst is a socket.
`Bidirectional(bi1,bi2): Data from bi1 are copied to bi2, and data from bi2 are copied to bi1. EOF of one descriptor causes that the other descriptor is shut down for sending. When both descriptors are at EOF, both are closed. It is required that bi1 and bi2 are sockets.
`Tridirectional(bi,dst,src): Data from bi are copied to dst, and data from src are copied to bi (i.e. a bidirectional descriptor is split up into two unidirectional descriptors). EOF of bi causes that dst is closed. EOF of src causes that bi is shut down for sending. EOF in both directions causes that all descriptors are closed. It is required that bi is a socket.

class copier : copy_task -> Unixqueue.event_system -> [unit] engine

This engine copies data between file descriptors as specified by the copy_task argument.

Socket engines

type sockspec = [ `Sock_inet of Unix.socket_type * Unix.inet_addr * int
       | `Sock_inet_byname of Unix.socket_type * string * int
       | `Sock_unix of Unix.socket_type * string ]

Extended names for socket addresses. Currently, these naming schemes are supported:

`Sock_unix(stype,path): Names the Unix domain socket at path. The socket type stype is an auxiliary piece of information, but not a distinguishing part of the name. path = "" refers to anonymous sockets. Otherwise, the path must be an absolute path name.
`Sock_inet(stype,addr,port): Names the Internet socket of type stype bound to the IP address addr and the port. If stype = Unix.SOCK_STREAM, a TCP socket is meant, and if stype = Unix.SOCK_DGRAM, a UDP socket is meant. It is allowed that addr = Unix.inet_addr_any. If port = 0, the name is to be considered as incomplete.
`Sock_inet_byname(stype,name,port): Names the Internet socket of type stype bound to the IP address corresponding to the name, and bound to the port. It is unspecified which naming service is used to resolve name to an IP address, and how it is used. If the name cannot be resolved, no socket is meant; this is usually an error. stype is interpreted as for `Sock_inet. If port = 0, the name is to be considered as incomplete.

It is currently not possible to name IP sockets that are bound to several IP addresses but not all IP addresses of the host.

type connect_address = [ `Command of string * (int -> Unixqueue.event_system -> unit)
       | `Socket of sockspec * connect_options ]

Specifies the service to connect to:

`Socket(addr,opts): Connect to the passed socket address
`Command(cmd,handler): The cmd is started with the shell, and stdin and stdout are used to transfer data to the process and from the process, respectively. Only SOCK_STREAM type is supported. Note that the passed file descriptors are normal pipes, not sockets (so the descriptors can be individually closed).
There is not any kind of error detection, so the command should be failsafe. stderr of the command is connected with stderr of the caller process.
No provisions are taken to wait for the process; this is the task of the caller. After the process has been started, the handler is invoked with the process ID and the event system to give the caller a chance to arrange that the process will be waited for.

type connect_options = {

conn_bind : sockspec option; (* Bind the connecting socket to this address (same family as the connected socket required). None: Use an anonymous port. *)

}

val default_connect_options : connect_options

Returns the default options

type connect_status = [ `Command of Unix.file_descr * int
       | `Socket of Unix.file_descr * sockspec ]

This type corresponds with Uq_engines.connect_address: An engine connecting with an address `X will return a status of `X.

`Socket(fd,addr): fd is the client socket connected with the service. addr is the socket address of the client that must be used by the server to reach the client.
`Command(fd, pid): fd is the Unix domain socket connected with the running command. pid is the process ID.

val client_socket : connect_status -> Unix.file_descr

Returns the client socket contained in the connect_status

type listen_address = [ `Socket of sockspec * listen_options ]

Specifies the resource to listen on:

`Socket(addr,opts): It is listened on a socket with address addr

type listen_options = {

	`lstn_backlog : int;`	`(*`	The length of the queue of not yet accepted connections.	`*)`
	`lstn_reuseaddr : bool;`	`(*`	Whether to allow that the address can be immediately reused after the previous listener has its socket shut down	`*)`

}

val default_listen_options : listen_options

Returns the default options

class type client_socket_connector = object .. end

This class type provides engines to connect to a service.

class type server_socket_acceptor = object .. end

This class type is for service providers that listen for connections.

class direct_socket_acceptor : Unix.file_descr -> Unixqueue.event_system -> server_socket_acceptor

An implementation of server_socket_acceptor for sockets

class type server_socket_listener = object .. end

This class type represents factories for service providers

val connector : ?proxy:#client_socket_connector ->
       connect_address ->
       Unixqueue.event_system -> connect_status engine

This engine connects to a socket as specified by the connect_address, optionally using the proxy, and changes to the state `Done(status) when the connection is established.

If the proxy does not support the connect_address, the class will raise Addressing_method_not_supported.

The descriptor fd (part of the connect_status) is in non-blocking mode, and the close-on-exec flag is set. It is the task of the caller to close this descriptor.

The engine attaches automatically to the event system, and detaches when it is possible to do so. This depends on the type of the connection method. For direct socket connections, the engine can often detach immediately when the conection is established. For proxy connections it is required that the engine copies data to and from the file descriptor. In this case, the engine detaches when the file descriptor is closed.

It is possible that name service queries block execution.

val listener : ?proxy:#server_socket_listener ->
       listen_address ->
       Unixqueue.event_system -> server_socket_acceptor engine

This engine creates a server socket listening on the listen_address. If passed, the proxy is used to create the server socket.

On success, the engine goes to state `Done acc, where acc is the acceptor object (see above). The acceptor object can be used to accept incoming connections.

type datagram_type = [ `Inet6_udp | `Inet_udp | `Unix_dgram ]

- `Unix_dgram: Datagrams over Unix domain sockets

`Inet_udp: Internet v4 UDP protocol
`Inet6_udp: Internet v6 UDP protocol

class type wrapped_datagram_socket = object .. end

A wrapped_datagram_socket allows datagrams to be sent via proxies.

class type datagram_socket_provider = object .. end

This is a factory for wrapped_datagram_socket objects.

val datagram_provider : ?proxy:#datagram_socket_provider ->
       datagram_type ->
       Unixqueue.event_system ->
       wrapped_datagram_socket engine

This engine creates a datagram socket as demanded by the datagram_type, optionally using proxy for sending and receiving datagrams.

The socket is unconnected.

The socket is in non-blocking mode, and the close-on-exec flag is set.

Multiplex Controllers

class type multiplex_controller = object .. end

A multiplex_controller is a quite low-level device to abstract bidirectional socket connections.

val create_multiplex_controller_for_connected_socket : ?close_inactive_descr:bool ->
       ?supports_half_open_connection:bool ->
       Unix.file_descr ->
       Unixqueue.unix_event_system -> multiplex_controller

Creates a multiplex controller for a bidirectional socket (e.g. a TCP socket). It is essential that the socket is in connected state.

Note that the file descriptor is not closed when the attached engines are terminated. One can call inactivate manually to do that.

close_inactive_descr: Whether inactivate closes the descriptor. True by default.

supports_half_open_connection: This implementation does not know how to find out whether the socket supports half-open connections. You can simply set this boolean because of this. Defaults to false. You can set it to true for TCP connections and for Unix-domain connections with stream semantics.

class type datagram_multiplex_controller = object .. end

Additional methods for unconnected datagram handling

val create_multiplex_controller_for_datagram_socket : ?close_inactive_descr:bool ->
       Unix.file_descr ->
       Unixqueue.unix_event_system -> datagram_multiplex_controller

Creates a multiplex controller for datagram sockets (e.g. UDP socket).

Note that the file descriptor is not closed when the attached engines are terminated. One can call inactivate manually to do that.

close_inactive_descr: Whether inactivate closes the descriptor. True by default.

type onshutdown_out_spec = [ `Action of
           async_out_channel_engine ->
           multiplex_controller -> unit engine_state -> unit
       | `Ignore
       | `Initiate_shutdown ]

See class output_async_mplex for explanations

type onshutdown_in_spec = [ `Action of
           async_in_channel_engine ->
           multiplex_controller -> unit engine_state -> unit
       | `Ignore
       | `Initiate_shutdown ]

See class input_async_mplex for explanations

class output_async_mplex : ?onclose:[ `Ignore | `Write_eof ] -> ?onshutdown:onshutdown_out_spec -> ?buffer_size:int -> multiplex_controller -> async_out_channel_engine

Creates an asynchronous output channel writing to the multiplex controller (see also output_async_descr for the corresponding class writing to a single descriptor).

class input_async_mplex : ?onshutdown:onshutdown_in_spec -> ?buffer_size:int -> multiplex_controller -> async_in_channel_engine

Creates an asynchronous input channel reading from the multiplex controller.

This web site is published by Informatikbüro Gerd Stolpmann

Plasma	GitLab	Archive
Projects	Blog	Knowledge