github ↗
docs menu
on this page

Module Rig_remote_proxy.Link

This process's ends of its job's connections.

Links: this process's ends of its job's connections, in C.

A link is this process's end of its connection to one other process of its job, once the handshake has admitted it (Wire). Whoever writes a message sends it from its own thread, a proxy's C sending a handover or a function of this module. The writer waits while an earlier message is sent, sends its own whole, and returns once it is sent, so the socket is the only back-pressure. A link owns the socket and two threads of C, which never hold the OCaml runtime:

  • the sending thread sends what the ready functions of the link's rails (rail) leave, and a beat after a second in which nothing was sent, between writers' messages. It waits on nothing but the writers and its socket.
  • the receiving thread reads each message. It places the transfers of the link's rails, and on the controller, writes the bytes of copies into this process's memory, advances the proxies' timeline words and delivers the answers to requests. It queues every other message for next, a handover with its bytes.

Messages carry no integrity check after the handshake, as Wire states.

Jobs and failure

The links of a process belong to its job (job), which fails or closes as a whole. The job fails once, with the first of these that this process learns of as its root cause:

  • a link's stream breaks or ends without a close: "NAME: closed its connection", or the system's error after "NAME: ";
  • no byte comes on a link for 10 seconds: "NAME: silent for 10 s";
  • a link's peer takes no byte of a send for 10 seconds: "NAME: read nothing for 10 s";
  • a message is malformed, as is one with a u64 above 262 - 1 (Wire): "NAME: a malformed frame";
  • a message is larger than this process can hold: "NAME: a frame larger than this process can hold";
  • a peer aborts: "NAME aborted: WHY", where WHY is the peer's root cause, unchanged;
  • fail.

NAME is the link's name. A reason is any bytes, at most 4096 of them. A longer one is cut there, here and in an agent's refusal (answer). Each process records its own root cause (Rig_remote's failure section).

Once the job fails, request and next return Error, the proxies' sleeps return, and every count of every rail of the job reads Int64.max_int, except the rails release_rail is releasing. Each link then ends its connection as Rig_remote's failure section states. A link that has sent no close sends its peer the abort and ends its stream, then reads and discards its peer's bytes until the peer ends its stream, which the peer does once it has read the abort. A link whose send failed sends no abort, its stream cut. A failed link sends no message it is given and handles no message it receives. It gives up the abort's send when it makes no progress for 10 seconds, and the reading 10 seconds after the later of the failure and the abort's send, or once the peer has sent nothing for 10 seconds, as a stopped process does. A peer misses the abort in the cases that section lists, and this process's end before close has returned is one of them. Nothing raises from a C thread, and nothing calls OCaml. The process learns of the failure from failed, wait or a function's Error.

A child of fork never uses its parent's links (Rig_remote's failure section), whose streams it would interleave with the parent's. Every function compares the process id with the one that started the job, and in a child the job is failed with "a child of fork does not use its parent's connections", in the child's copy alone. The child sends nothing on its parent's links and closes none of them, and a link it makes is failed, its socket closed.

Domains

Every function may be called from several domains in parallel. A function that waits, for the socket, an answer, a message or the job, releases the runtime while it waits.

next and answer each hold a lock of their link while they run, counted among the thread's locks (Drivers), except while next waits for a command. A callback that runs in that wait calls rig as any code does. Either, called from a callback inside a lock (Rig's domains section), raises Invalid_argument and changes nothing.

A job and its links are never freed, because proxies' C state holds their addresses. A link that has ended keeps its name and reason.

Jobs

#type job

The type for the jobs of this process. A job is open from job until it has failed or closed.

#val job : unit -> job

job () is a new open job, with no link.

Raises Invalid_argument if a job of the process is open.

#type state =
  1. #| Open
    (*

    Neither failed nor closed.

    *)
  2. #| Closed
    (*

    Closed in order (close).

    *)
  3. #| Failed of string
    (*

    Failed, with its root cause.

    *)

The type for the states of a job.

#val wait : job -> ms:int -> state

wait j ~ms is j's state once it is no longer Open, or after ms milliseconds.

#val failed : job -> string option

failed j is Some why iff wait j ~ms:0 is Failed why.

#val fail : job -> string -> unit

fail j why fails j with the root cause why, unless it has failed or closed. It then sends the abort of each link that is sending no message, unless the link's own thread saw the failure first and sends it. The send of an abort waits up to 10 seconds for a peer that reads nothing, and fail releases the runtime while it sends.

#val close : job -> unit

close j ends j. On an open job, each of its links sends a close after the messages sent before, and close returns once every peer's close has come and the links' threads have ended. A link's peer sends its close when it ends the job itself, so close waits for every peer to end it. Once a link has taken its close, it sends no message it is given. On a job that has failed, before or meanwhile, close returns once each link has sent its abort and read its peer's stream to the end, or given up (Jobs and failure), so a process that ends after close leaves each peer its abort. On a closed job it returns immediately.

Its wait returns to OCaml at least every 200 ms, where a pending signal's handler runs. The handler's exception ends close with the job as it was, the closes sent so far sent; a later close sends the others and waits again.

#type t

The type for links.

#val make : job -> Unix.file_descr -> name:string -> peer:Wire.process -> t

make j fd ~name ~peer is a link of j over the connected socket fd, whose handshake has admitted it, to the process peer named name: its machine's name, which starts its reasons. make takes fd, and its threads start immediately. On a failed job the link is failed and fd closed.

Raises Invalid_argument if j is closed, and Out_of_memory if the link cannot be allocated. fd is closed in either case.

#val name : t -> string

name l is l's name.

#val job_of : t -> job

job_of l is l's job.

#val fresh : unit -> int

fresh () is an id the process gave no object before: the controller names each object it makes on an agent by one (Wire.request).

The controller's end

#val request : t -> 'a Wire.request -> ('a, [ `Refused of string | `Failed of string ]) Stdlib.result

request l r sends r after every message sent before it and is the agent's answer. It is Error (`Refused why) if the agent refused r, and the job goes on. It is Error (`Failed why) if the job has failed, before or meanwhile, why its root cause, or if its close has begun (close), why then "the job is closed". An answer that does not decode as r's fails the job.

Raises Invalid_argument before sending anything if an id of r is negative.

#val drop : t -> int -> unit

drop l id sends the release of the agent's object id after every message sent before it, and returns once it is sent, or immediately if the job has failed or its close has begun. While another message holds the link's stream, it returns to OCaml at least every 200 ms, where a pending signal's handler runs, and the handler's exception ends drop with nothing sent; once its message has the stream, it sends it whole. So do word and bytes.

Raises Invalid_argument if id is negative.

The agent's end

#val next : t -> (Wire.command, string) Stdlib.result

next l is the next command l's peer sent, waiting for it. Error why once the job has failed, why its root cause. A message that does not decode as a command fails the job. Its wait returns to OCaml at least every 200 ms, where a pending signal's handler runs, and the handler's exception ends next having taken no command.

A Handover's areas hold its bytes until the next call of next on l, from any thread, so a thread must finish with a Handover's areas before any thread calls next. The link may then keep their memory, up to 128 MiB of it, for later commands to land in until next gives the close, so a stream of large Handover commands lands in memory already mapped.

An exception that ends next once it took a command other than a Drop, such as Out_of_memory or a Gc.Memprof callback's, fails the job: the command is lost, and the controller would otherwise wait for it, or pair later answers with the wrong requests.

Raises Invalid_argument from a callback inside a lock (Domains).

#val answer : t -> 'a Wire.request -> ('a, string) Stdlib.result -> unit

answer l r a sends a as the answer to r. Error why refuses r with why. From several threads, the check and the send are one step, so answers reach the peer in the order of their requests; a call for a request that is not the oldest unanswered raises, as below.

An exception that ends answer once it took r as answered, such as Out_of_memory for its frame, fails the job, which ends the controller's wait; a Sys.Break fails it with Sys.Break's text as its root cause. While another message holds the link's stream, answer returns to OCaml at least every 200 ms, where finalisers may run, and on Windows, which blocks no signal, a signal's handler. Elsewhere it blocks signals throughout.

Raises Invalid_argument if r is not, physically, the oldest request next gave and none answered, and from a callback inside a lock (Domains).

#val word : t -> device:int -> int -> unit

word l ~device v sends that the work of device up to v is done.

#val bytes : t -> device:int -> value:int -> Rig_remote_abi.area -> unit

bytes l ~device ~value b sends b's bytes as those of the next copy into the controller's memory of the work of device at value. It reads b in place and returns once its bytes are sent, or once the job has failed or its close has begun. The caller must not write b until it returns.

Rails

#val rail : t -> id:int -> send:Rig_remote_abi.transfer array -> receive:Rig_remote_abi.transfer array -> Rig_remote_abi.end_

rail l ~id ~send ~receive is this machine's end of the rail id to l's peer, which carries send to it and receive from it, with the rail's areas zeroed (Rails) and its ready function. Once ready reaches transfer j's count, its bytes are sent. If the link is sending nothing then, the ready function sends as many as the socket takes, and the sending thread sends the rest. sent is stored once they all are. The receiving thread places arriving transfers and stores arrived. The end's memory lives until release_rail, reachable or not.

Raises Invalid_argument if l has a rail id or Rig_remote_abi.check_transfers is Error, and Out_of_memory if the host cannot allocate the areas.

#val release_rail : t -> int -> unit

release_rail l id ends l's rail id here: once it returns, neither thread reads or writes its end, whose memory lives while it is reachable, and its ready function must not be called again. Its ready_fn may still store to its counts, as Rig_remote_abi.end_ states. Concurrent releases of the same rail all wait for its active transfers to finish. A transfer of it that arrives later is discarded, as is one of a rail l never had. It does nothing if l has no rail id. It takes no lock an OCaml thread holds, so a finaliser may call it. Its wait does not return to OCaml before it ends, so a signal's handler runs only then: rig's callers run it with signals blocked, from a finaliser or before any transfer of the rail began.