Ledger (sarek.Spoc_core.Transfer.Ledger)

Keep-alive ledger for stream-ordered work

A copy or launch enqueued on a stream may still be reading or writing its operands after the call returns. The backends see raw addresses, which are not GC roots, so SPOC roots the operands here — Bigarray sources and destinations, the vectors involved — until it KNOWS the work has completed:

the stream was synchronised (synchronize_stream, destroy_stream), or its device was (flush, Device.synchronize);
an event recorded after the work was synchronised, or queried complete;
transitively: when a stream b that waited on an event of stream a completes past that wait, a's work up to the event has completed too. This is what releases a copy stream that is never synchronised itself — the spoc_llm M3 pattern (copy on a, record e, b waits on e, sync b).

Entries are ordered by a per-stream sequence number. The ledger is protected by a mutex and is never touched from a GC finaliser (which could run inside Mutex.protect on the same domain and self-deadlock).

type entry = {

seq : int;
keep : keep;
(*
Rooted for liveness only; never inspected.
*)
vec_ids : int list;
host_ranges : (nativeint * int) list;
(*
(address, bytes) in flight.
*)

}

type stream_state = {

sid : int;
sdev : Device.t;
mutable next_seq : int;
mutable pending : entry list;
mutable waits : (int * stream_state * int) list;
(*
(seq of the wait on this stream, waited stream, its seq)
*)
nonblocking : bool;
(*
Created with ~non_blocking:true: the default stream does not wait for it, so implicit paths must settle it themselves.
*)
mutable drain : unit -> unit;
(*
Synchronise the stream; set by create_stream.
*)
mutable destroyed : bool;
(*
Set by destroy_stream. A destroyed handle is refused: the driver may have recycled it, and work enqueued through it would land in a ledger nothing consults any more.
*)

}

val lock : Stdlib.Mutex.t

val next_sid : int Stdlib.ref

val live : (int, stream_state) Stdlib.Hashtbl.t

val nonblocking_live : int Stdlib.Atomic.t

val has_nonblocking : unit -> bool

val new_stream : ?nonblocking:bool -> Device.t -> stream_state

val forget : stream_state -> unit

val check_live : op:string -> stream_state -> unit

val enqueue : 
  stream_state ->
  keep:keep ->
  vec_ids:int list ->
  host_ranges:(nativeint * int) list ->
  unit

val mark : stream_state -> int

Sequence number of the last operation enqueued so far: what an event recorded now covers.

val add_wait : stream_state -> on:(stream_state * int) -> unit

A wait is itself an operation on b: once b has completed past it, the waited work on the other stream has completed.

val release_locked : stream_state -> int -> unit

val release : stream_state -> int -> unit

val release_all : stream_state -> unit

val device_marks : Device.t -> (stream_state * int) list

The last operation enqueued so far on each live stream of dev: what a device sync started now covers.

val release_marks : (stream_state * int) list -> unit

val pending_count : stream_state -> int

val holds_vector : int -> bool

val nonblocking_holding : int -> stream_state list

Live non-blocking streams with pending work that references id.

val holds_host_range : nativeint -> int -> bool