github ↗
docs menu
on this page

Module Rig_remote_proxy.Wire

The protocol between the processes of a job.

The protocol between the processes of a job.

A job's processes are its controller and an agent on each other machine (process, Rig_remote). Each pair of them has one TCP connection, which one end dials and the other accepts. The controller dials every agent, and each agent dials the agents after it. The protocol's values are encoded so:

  • an integer is little-endian;
  • a u64 above 262 - 1, OCaml's largest int, is malformed;
  • a process is a u32, 0 for the controller and i for agent i, from 1 to 232 - 1;
  • a string is its length (u32) and its bytes.

Handshake

The accepting end speaks first. Then each end proves that it holds the job's key over both ends' nonces, the dialing end first:

 accepting: "rig-job\n", version 12 (u32), 0, its nonce (32 bytes)
            or "rig-job\n", version 12 (u32), 1 and why
 dialing:   its nonce (32 bytes), its process, the accepting end's
            process, its proof (32 bytes)
 accepting: 0 and its proof (32 bytes)
            or 1 and why

The first why refuses a connection before any proof, such as one beyond those the accepting end can serve. The second refuses a dialing end that does not prove the key, or that the accepting end does not admit, such as a second controller. A reason has at most 4096 bytes. A longer one is cut there, and one announced longer is malformed.

A proof is HMAC-BLAKE2b-256, keyed by the BLAKE2b-256 hash of the job's key, over its end's label ("rig-job dialing" or "rig-job accepting"), the dialing end's process, the accepting end's process, the accepting end's nonce and the dialing end's nonce. The two ends' labels differ, so neither proof can stand for the other. The key itself never crosses the connection.

Messages

After the handshake, each direction is a sequence of messages, each its payload's length (u64), its kind (u8) and its payload. A message whose kind is unknown or not expected from its sender, or whose payload is malformed, fails the job, and so does one longer than the receiving process can hold.

A request that names an unknown id, or an id of the wrong kind, is refused in its answer, and the job goes on. A handover or a drop that does so fails the job, since neither has an answer. A rail message for a rail the receiving process does not hold on that connection is discarded, and the job goes on. Messages carry no integrity check after the handshake (Rig_remote's keys section states what a job trusts).

 kind  name      sent by     payload
 1     request   controller  a request ({!request})
 2     answer    agent       0 and the answer, or 1 and why
 3     handover  controller  a {!handover}, then the bytes of its copies
                             from the controller's memory
 4     drop      controller  id (u64)
 5     word      agent       device (u64), value (u64)
 6     bytes     agent       device (u64), value (u64), bytes of a copy
 7     rail      any         rail (u64), count (u64), the transfer's bytes
 8     beat      any         nothing
 9     abort     any         why
 10    close     any         nothing

The agent sends the answers to requests in the order the requests came, and applies the controller's messages in that order. The bytes of a handover's copies into the controller's memory come back as bytes messages, one per copy, in its parts' order, before the word message of its value.

A process sends a beat after a second in which it sent nothing on a connection, and fails the job when no byte has come on one for 10 seconds. An abort carries its sender's root cause, at most 4096 bytes, and fails the receiver's job. A connection ends in order once each end has sent a close and received the other's, and one that ends otherwise fails the job.

References

  • H. Krawczyk, M. Bellare and R. Canetti. RFC 2104, HMAC: Keyed-Hashing for Message Authentication: the proof, over a hash of 128-byte blocks.
  • M-J. Saarinen and J-P. Aumasson. RFC 7693, The BLAKE2 Cryptographic Hash and Message Authentication Code: BLAKE2b, its block and digest sizes.

Handshakes

#val version : int

version is the protocol's version, the handshake's: 12. Ends at other versions refuse each other. Adding or removing a level of an x86-64 host's Rig.arch, which an account carries and Rig_program.load orders, changes the version, so that no end reads a level its rig does not know. An arm64 feature does not: Rig_program.load compares features by name.

#val min_key : int

min_key is the fewest bytes of a job's key: 16.

#val max_key : int

max_key is the most bytes of a job's key: 4096.

#type process =
  1. #| Controller
    (*

    The controller, on its own machine.

    *)
  2. #| Agent of int
    (*

    Agent i, from 1.

    *)

The type for the processes of a job.

#val dial : Unix.file_descr -> key:string -> self:process -> peer:process -> (unit, string) Stdlib.result

dial fd ~key ~self ~peer runs the dialing end's handshake on the connected socket fd, as self, to peer. It waits at most 10 seconds for each reply.

Error why if the accepting end refuses, before the proofs or after them, why its reason; speaks another version, why naming both; is not a process of a job; proves another key; sends a malformed handshake; or does not reply in time, or if the stream fails.

Raises Invalid_argument if key has fewer than min_key or more than max_key bytes, self or peer is an agent outside 1 to 232 - 1, or peer is Controller or self.

#val accept : Unix.file_descr -> key:string -> admit:(process -> (unit, string) Stdlib.result) -> (process * process, string) Stdlib.result

accept fd ~key ~admit runs the accepting end's handshake on the connected socket fd, and is the dialing end's process and this end's, as the dialing end names it. Once the dialing end has proved key, admit p decides, where p is the dialing end's process, and its Error why refuses it with why, cut to 4096 bytes. It waits at most 10 seconds for each reply.

Error why if the dialing end proves another key, is not admitted, sends a malformed handshake, or fails to answer in time, or the stream fails.

Raises Invalid_argument if key has fewer than min_key or more than max_key bytes.

#val refuse : Unix.file_descr -> string -> unit

refuse fd why sends the refusal why, cut to 4096 bytes, in place of a greeting, then closes fd. It raises nothing: a failed send is the peer's loss.

Requests

#type account = {
  1. #id : int;
    (*

    Its id on its machine: 0 for the host, increasing as devices open.

    *)
  2. #name : string;
    (*

    Its name on its machine, such as "CUDA:3".

    *)
  3. #arch : string;
    (*

    Its Rig.arch there.

    *)
  4. #budget : int;
    (*

    Its Rig.budget there when it opened.

    *)
  5. #alignment : int;
    (*

    Its Rig.alignment there.

    *)
  6. #cache_bytes : int option;
    (*

    Its Rig.cache_bytes there, which crosses as a u64, 0 for None.

    *)
  7. #reaches : int list;
    (*

    The smaller ids of devices whose own memory it reaches there (Rig.reaches).

    *)
  8. #reached_by : int list;
    (*

    The smaller ids of devices that reach its own memory there.

    *)
}

The type for an agent's account of one of its devices. Between two devices, the account with the larger id describes both directions of reachability. Accounts remain unchanged when more devices open.

#type agent = {
  1. #name : string;
    (*

    Its machine's name, which the controller gives it: every process of the job names the machine by it.

    *)
  2. #host : string;
    (*

    The host the agent listens at.

    *)
  3. #port : int;
    (*

    The port the agent listens at.

    *)
}

The type for the agents of a job, as a join names them.

#type _ request =
  1. #| Join : {
    1. #agents : agent list;
    } -> account request
    (*

    Makes the agent one of agents, the job's agents in the order of their processes from 1. Its answer is its host's account, once it is connected to every other agent.

    *)
  2. #| Open : string -> account list request
    (*

    Opens the agent's devices of a kind, such as "CUDA". Its answer is their accounts, in the agent's index order.

    *)
  3. #| Alloc : {
    1. #id : int;
    2. #device : int;
    3. #memory : Rig_edge.memory;
    4. #bytes : int;
    } -> bool request
    (*

    Allocates bytes of device's memory as id: false if the device lacks the room.

    *)
  4. #| Map : {
    1. #id : int;
    2. #device : int;
    3. #region : int;
    } -> bool request
    (*

    Maps the memory region, of another device of the machine, on device as id: false if device cannot map it.

    *)
  5. #| Load : {
    1. #id : int;
    2. #binary : string;
    } -> int option array request
    (*

    Loads the program binary on the machine's host as id: the most threads a group of each of its launches may have, in the order of its steps with a loop's body in place.

    *)
  6. #| Entry : {
    1. #image : int;
    2. #name : string;
    } -> int option request
    (*

    The function name of the image image: None if it has none.

    *)
  7. #| Rail : {
    1. #id : int;
    2. #peer : process;
    3. #send : Rig_remote_abi.transfer array;
    4. #receive : Rig_remote_abi.transfer array;
    } -> unit request
    (*

    Makes the agent's end of the rail id to peer: its transfers send from the agent's machine and receive to it.

    *)

The type for requests, each answered by a value of its parameter. The ids of memory, images and rails are the controller's, from one count: each names one object of the job on the agent from its request until the controller drops it (Drop).

Work

#type side =
  1. #| Region of {
    1. #id : int;
    2. #offset : int;
    }
    (*

    Memory of the agent's machine: the object id, from its byte offset.

    *)
  2. #| Local
    (*

    Memory of the controller's process.

    *)

The type for the memory on one side of a copy.

#type part =
  1. #| Words of string
    (*

    A run's 32-bit words, little-endian.

    *)
  2. #| Copy of {
    1. #src : side;
    2. #dst : side;
    3. #width : int;
    4. #rows : int;
    5. #src_pitch : int;
    6. #dst_pitch : int;
    }
    (*

    A copy of rows rows of width bytes, row i from i * src_pitch bytes past src to i * dst_pitch bytes past dst, src and dst not both Local. Each pitch is at least width.

    *)
  3. #| Fill of {
    1. #dst : side;
    2. #bytes : int;
    3. #byte : char;
    }
    (*

    Sets bytes bytes of dst, which is not Local, to byte.

    *)

The type for the parts of a handover, which the agent runs in order.

#type handover = {
  1. #device : int;
    (*

    The device whose work it is.

    *)
  2. #value : int;
    (*

    The value on its device's timeline it is the work of.

    *)
  3. #waits : (int * int) array;
    (*

    The values of devices of the machine its work follows: device and value pairs.

    *)
  4. #parts : part array;
    (*

    Its work.

    *)
}

The type for handovers: the work of one value on a device's timeline.

 device (u64), value (u64), waits (u32) and each device (u64) and
 value (u64), parts (u32) and each part:
   0, words (u32) and the words
   1, width (u64), rows (u64), src pitch (u64), dst pitch (u64),
      src side, dst side
   2, bytes (u64), byte (u8), dst side
 side: 0, id (u64), offset (u64)   or   1

The bytes of a copy from Local follow the handover, and those of a copy into Local come back in a bytes message: width * rows bytes, its rows one after another.

Commands

#type command =
  1. #| Request : 'a request -> command
    (*

    A request, whose answer the agent sends in the order the requests came.

    *)
  2. #| Handover : handover * Rig_remote_abi.area array -> command
    (*

    A handover and the bytes of its copies from Local, one area per such copy, its rows one after another, in its parts' order.

    *)
  3. #| Drop : int -> command
    (*

    Releases the object id once the work submitted before no longer needs it.

    *)
  4. #| Close : command
    (*

    Ends the job in order, after the work submitted before.

    *)

The type for the messages an agent reads from its controller.