mirror of
https://github.com/zap-proto/ocaml-rpc.git
synced 2026-10-30 15:04:28 +00:00
Thomas Leonard
e9c59b54d8
Before, we also disconnected on receiving end-of-file, but that prevents sending the final abort message. Also, bootstrap now returns a broken cap if the connection got shut down, rather than raising an exception. This allows handling crossed calls more consistently.
1847 lines
74 KiB
OCaml
1847 lines
74 KiB
OCaml
module Log = Debug.Log
|
|
module IntMap = Map.Make(struct type t = int let compare (a:int) b = compare a b end)
|
|
|
|
let rec filter_map f = function
|
|
| [] -> []
|
|
| x :: xs ->
|
|
match f x with
|
|
| None -> filter_map f xs
|
|
| Some y -> y :: filter_map f xs
|
|
|
|
let pp_check check f (k, v) =
|
|
try check k v
|
|
with ex ->
|
|
Fmt.pf f "@,[%a] %a"
|
|
Fmt.(styled `Red string) "ERROR"
|
|
Debug.pp_exn ex
|
|
|
|
let pp_weak f = function
|
|
| None -> Fmt.pf f "(GC'd weak pointer)"
|
|
| Some x -> x#pp f
|
|
|
|
let hashtbl_dump ~key pp f items =
|
|
let add k v results = (k, v) :: results in
|
|
Hashtbl.fold add items []
|
|
|> List.sort (fun a b -> compare (key (snd a)) (key (snd b)))
|
|
|> Fmt.Dump.list pp f
|
|
|
|
module EmbargoId = Message_types.EmbargoId
|
|
|
|
module Make (EP : Message_types.ENDPOINT) = struct
|
|
module Core_types = EP.Core_types
|
|
module Wire = Core_types.Wire
|
|
module Cap_proxy = Cap_proxy.Make(Core_types)
|
|
module Struct_proxy = Struct_proxy.Make(Core_types)
|
|
module Local_struct_promise = Local_struct_promise.Make(Core_types)
|
|
module Out = EP.Out
|
|
module In = EP.In
|
|
|
|
let inc_ref = Core_types.inc_ref
|
|
let dec_ref = Core_types.dec_ref
|
|
|
|
let with_inc_ref x =
|
|
Core_types.inc_ref x;
|
|
x
|
|
|
|
open EP.Table
|
|
|
|
module PathSet = Set.Make(Wire.Path)
|
|
module Embargoes = Table.Allocating(EmbargoId)
|
|
|
|
(* Maps used cap indexes to their paths *)
|
|
let caps_used ~msg paths_used =
|
|
PathSet.elements paths_used
|
|
|> filter_map (fun path ->
|
|
match Wire.Response.cap_index msg path with
|
|
| None -> None
|
|
| Some i -> Some (i, path)
|
|
)
|
|
|> IntMap.of_list
|
|
|
|
module Questions = Table.Allocating(QuestionId)
|
|
module Answers = Table.Tracking(AnswerId)
|
|
module Exports = Table.Allocating(ExportId)
|
|
module Imports = Table.Tracking(ImportId)
|
|
|
|
module Import = struct
|
|
(* An entry in the imports table has a corresponding switchable, which the user of the library holds.
|
|
There are three possible events:
|
|
|
|
- The user reduces the switchable's ref-count to zero, indicating that
|
|
they don't need the import any longer.
|
|
|
|
- The peer resolves the import to something else (or the connection is
|
|
lost, resolving it to an exception).
|
|
|
|
- The peer quotes the same import ID again.
|
|
If we hadn't sent a release before this, it must be for the same object.
|
|
If we had sent a release then it might or might not be the same object.
|
|
|
|
We hold a weak-ref to the switchable so that if the user leaks it we will notice.
|
|
|
|
[ref_count] is zero iff [count] is zero:
|
|
- Initially, both are one.
|
|
- If we set [ref_count] to zero, we send a release and set [count] to zero.
|
|
- There is no other way for [count] to become zero.
|
|
*)
|
|
|
|
type t = {
|
|
id : ImportId.t;
|
|
mutable ref_count : RC.t; (* The switchable holds one until resolved, plus each [resolve_target] adds one. *)
|
|
mutable count : int; (* Number of times remote sent us this. *)
|
|
mutable used : bool; (* We have sent a message to this target (embargo needed on resolve). *)
|
|
settled : bool; (* This was a SenderHosted - it can't resolve except to an exception. *)
|
|
mutable resolution : disembargo_info;
|
|
proxy : Cap_proxy.resolver_cap Weak_ptr.t; (* Our switchable ([Weak_ptr.t] is mutable). *)
|
|
strong_proxy : < > option ref; (* Keeps the switchable alive if there are callbacks registered. *)
|
|
}
|
|
and disembargo_info = [
|
|
| `Unresolved
|
|
| `Local
|
|
| `Import of t (* Holds ref *)
|
|
| `Error
|
|
]
|
|
|
|
let id t = t.id
|
|
|
|
let pp f t =
|
|
Fmt.pf f "i%a" ImportId.pp t.id
|
|
|
|
let dump f t =
|
|
Fmt.pf f "%a -> %a"
|
|
RC.pp t.ref_count
|
|
pp_weak (Weak_ptr.get t.proxy)
|
|
|
|
let get_proxy t = Weak_ptr.get t.proxy
|
|
|
|
(* A new export from the peer. *)
|
|
let inc_count t =
|
|
assert (t.count > 0);
|
|
t.count <- t.count + 1
|
|
|
|
(* A new local reference (resolve_target). *)
|
|
let inc_ref t =
|
|
assert (t.count > 0);
|
|
let pp f = pp f t in
|
|
t.ref_count <- RC.succ ~pp t.ref_count
|
|
|
|
let clear_proxy t =
|
|
Weak_ptr.clear t.proxy
|
|
|
|
(* A [resolve_target] or our switchable no longer needs us. *)
|
|
let rec dec_ref t =
|
|
let pp f = pp f t in
|
|
t.ref_count <- RC.pred ~pp t.ref_count;
|
|
if RC.is_zero t.ref_count then (
|
|
assert (t.count > 0);
|
|
let count = t.count in
|
|
t.used <- false;
|
|
t.count <- 0;
|
|
let free_resolution =
|
|
match t.resolution with
|
|
| `Import i -> dec_ref i
|
|
| `Unresolved | `Local | `Error -> []
|
|
in
|
|
`Release (t, count) :: free_resolution
|
|
) else []
|
|
|
|
let mark_used t =
|
|
t.used <- true
|
|
|
|
let used t = t.used
|
|
|
|
let mark_resolved t ~get_import result =
|
|
if t.resolution <> `Unresolved then
|
|
Fmt.failwith "Got Resolve for already-resolved import %a" pp t
|
|
else match result with
|
|
| Error _ -> t.resolution <- `Error
|
|
| Ok desc ->
|
|
let info =
|
|
match desc with
|
|
| `None -> `Error (* Probably shouldn't happen *)
|
|
| `Error _ -> `Error (* Don't need to embargo errors *)
|
|
| `SenderPromise id | `SenderHosted id -> `Import (get_import id)
|
|
| `ThirdPartyHosted _ -> failwith "todo: disembargo_reply: ThirdPartyHosted"
|
|
| `ReceiverAnswer _ | `ReceiverHosted _ -> `Local
|
|
in
|
|
t.resolution <- info
|
|
|
|
let message_target t =
|
|
`ReceiverHosted t.id
|
|
|
|
let embargo_path t =
|
|
if t.used then Some (message_target t) else None
|
|
|
|
(* We have a reference pointing at this target and may need to perform an embargo.
|
|
This could be either because we pipelined messages to this import, or we pipelined
|
|
messages to another remote object which we know forwarded them to this one. *)
|
|
let disembargo_target t =
|
|
match t.resolution with
|
|
| `Unresolved -> None (* No embargoes needed yet *)
|
|
| `Error -> None (* Don't need to embargo errors *)
|
|
| `Import _ as i -> Some i
|
|
| `Local ->
|
|
(* The import resolved to a local service. Send a loopback disembargo to flush the path. *)
|
|
Some (message_target t)
|
|
|
|
let init_proxy t proxy =
|
|
assert (get_proxy t = None);
|
|
Weak_ptr.set t.proxy proxy
|
|
|
|
let check t =
|
|
let pp f = pp f t in
|
|
RC.check ~pp t.ref_count;
|
|
(* Count starts at one and is only incremented, or set to zero when ref_count is zero. *)
|
|
if t.count < 1 then
|
|
Debug.invariant_broken (fun f -> Fmt.pf f "Import remote count < 1, but still in table: %a" dump t);
|
|
match get_proxy t with
|
|
| Some x -> x#check_invariants
|
|
| None -> ()
|
|
|
|
let lost_connection t ~broken_cap =
|
|
match get_proxy t with
|
|
| Some switchable when switchable#problem = None -> switchable#resolve broken_cap
|
|
| _ -> ()
|
|
|
|
let v ~mark_dirty ~settled id = {
|
|
ref_count = RC.one;
|
|
count = 1;
|
|
id = id;
|
|
proxy = Weak_ptr.empty ();
|
|
strong_proxy = ref None;
|
|
settled;
|
|
resolution = `Unresolved;
|
|
used = mark_dirty;
|
|
}
|
|
end
|
|
|
|
module Question = struct
|
|
(* State model for questions:
|
|
|
|
1. waiting: returned=false, rc>0 (initial state)
|
|
We're waiting for some results. (Initially rc=1, from our remote promise)
|
|
On rc=0: send Finish, goto cancelled
|
|
On return: goto lingering
|
|
|
|
2. cancelled: returned=false, rc=0
|
|
We told the peer we don't care about the results.
|
|
It may return them anyway or return "cancelled".
|
|
We will never send another message referring to this question.
|
|
On return: goto complete (with results=cancelled)
|
|
|
|
3. lingering: returned=true, rc>0
|
|
We've got the results and want to finish, but we can't yet because rc>0.
|
|
On rc=0: send Finish, goto complete
|
|
|
|
4: complete: returned=true, rc=0 (final state)
|
|
Finish has been sent and we are no longer in the questions table.
|
|
We will never send or receive another message referring to this question.
|
|
|
|
The user releases the remote promise to indicate that they no longer care
|
|
about the result. The remote promise will then decrement rc.
|
|
|
|
The "return" event means the peer sent us a Return message.
|
|
|
|
rc>0 means that we may still want to send messages about this question.
|
|
This could be either messages sent via the remote promise, or disembargo
|
|
replies.
|
|
|
|
The disembargo case happens when e.g. we told the peer that a capability
|
|
in the answer to a question they asked us will come from this question
|
|
that we asked them.
|
|
|
|
Initially, our [answer] for the peer's question holds a reference to us
|
|
via our remote promise so that it can forward any calls. This will
|
|
prevent us from reaching rc=0. Once we get the results, the answer will
|
|
update to point directly at the new target and the remote promise may be
|
|
released then.
|
|
|
|
However, the peer may still need to send a disembargo request to the
|
|
answer, which must be sent back to its *original* target (this
|
|
questions's answer). For this case, we must keep our question alive a
|
|
little longer, until the peer finishes their question.
|
|
|
|
We expect this to happen soon because there's no point in the peer
|
|
holding on to their question now that they've got the answer.
|
|
|
|
Every path contains exactly one Finish, exactly one Return, and exactly
|
|
one release (rc=0) event.
|
|
*)
|
|
|
|
type disembargo_info = [
|
|
| `Error
|
|
| `Elsewhere
|
|
| `Local
|
|
| `Results of Wire.Response.t * [
|
|
| `Local
|
|
| `Import of Import.t (* Holds ref *)
|
|
| `None
|
|
] RO_array.t
|
|
]
|
|
|
|
type state =
|
|
| Waiting
|
|
| Cancelled
|
|
| Lingering of disembargo_info
|
|
| Complete
|
|
|
|
type t = {
|
|
id : QuestionId.t;
|
|
mutable remote_promise : [`Resolver of Core_types.struct_resolver | `Released];
|
|
mutable ref_count : RC.t;
|
|
mutable state : state;
|
|
params_for_release : ExportId.t list; (* For level 0 peers *)
|
|
mutable pipelined_fields : PathSet.t; (* Fields used while unresolved; will need embargoes *)
|
|
}
|
|
|
|
let id t = t.id
|
|
|
|
let pp f q =
|
|
Fmt.pf f "q%a" QuestionId.pp q.id
|
|
|
|
let pp_disembargo_desc f = function
|
|
| `Local -> Fmt.string f "local"
|
|
| `Import i -> Import.pp f i
|
|
| `None -> Fmt.string f "none"
|
|
|
|
let pp_disembargo_info f : disembargo_info -> unit = function
|
|
| `Error -> Fmt.string f "error"
|
|
| `Elsewhere -> Fmt.string f "elsewhere"
|
|
| `Local -> Fmt.string f "local"
|
|
| `Results (_, descs) ->
|
|
(RO_array.pp pp_disembargo_desc) f descs
|
|
|
|
let free_disembargo_info = function
|
|
| `Error -> []
|
|
| `Elsewhere -> []
|
|
| `Local -> []
|
|
| `Results (_, descs) ->
|
|
RO_array.fold_left (fun acc -> function
|
|
| `None | `Local -> acc
|
|
| `Import i -> `Release_import i :: acc
|
|
) [] descs
|
|
|
|
let pp_promise f q =
|
|
match q.remote_promise with
|
|
| `Released -> Fmt.string f "(released)"
|
|
| `Resolver p -> Fmt.pf f "%t" p#pp
|
|
|
|
let pp_state f = function
|
|
| Waiting -> Fmt.string f "waiting"
|
|
| Cancelled -> Fmt.string f "cancelled"
|
|
| Lingering i -> Fmt.pf f "lingering:%a" pp_disembargo_info i
|
|
| Complete -> Fmt.string f "complete"
|
|
|
|
let dump f t =
|
|
Fmt.pf f "(%a) %a" pp_state t.state pp_promise t
|
|
|
|
let sent_finish t =
|
|
match t.state with
|
|
| Waiting | Lingering _ -> false
|
|
| Cancelled | Complete -> true
|
|
|
|
let inc_ref t =
|
|
assert (not (sent_finish t));
|
|
let pp f = pp f t in
|
|
t.ref_count <- RC.succ t.ref_count ~pp
|
|
|
|
let dec_ref t =
|
|
let pp f = pp f t in
|
|
t.ref_count <- RC.pred t.ref_count ~pp;
|
|
if RC.is_zero t.ref_count then match t.state with
|
|
| Waiting -> t.state <- Cancelled; [`Send_finish]
|
|
| Lingering ret -> t.state <- Complete; `Send_finish :: `Release_table_entry :: free_disembargo_info ret
|
|
| Complete -> [] (* (only happens with lost_connection) *)
|
|
| Cancelled -> failwith "Can't hold refs while cancelled!"
|
|
else
|
|
[]
|
|
|
|
let release_proxy t =
|
|
t.remote_promise <- `Released;
|
|
dec_ref t
|
|
|
|
let set_cap_used t path =
|
|
t.pipelined_fields <- PathSet.add path t.pipelined_fields
|
|
|
|
let cap_used t path =
|
|
PathSet.mem path t.pipelined_fields
|
|
|
|
let message_target t path = `ReceiverAnswer (t.id, path)
|
|
|
|
(* Extract some useful parts of the response that we may need for embargoes. *)
|
|
let extract_resolution ~get_import = function
|
|
| #Error.t -> `Error
|
|
| `ResultsSentElsewhere -> `Elsewhere (* We don't care about the result, so should never embargo *)
|
|
| `TakeFromOtherQuestion _ -> `Local (* Remote will forward back to our answer *)
|
|
| `AcceptFromThirdParty -> failwith "todo: answer_cap_needs_disembargo: AcceptFromThirdParty"
|
|
| `Results (msg, descs) ->
|
|
let extract_cap = function
|
|
| `ReceiverAnswer _ | `ReceiverHosted _ -> `Local (* Remote will forward back to us *)
|
|
| `SenderPromise id | `SenderHosted id -> `Import (get_import id) (* Disembargo via import *)
|
|
| `ThirdPartyHosted _ -> failwith "todo: answer_cap_needs_disembargo: ThirdPartyHosted"
|
|
| `None -> `None
|
|
in
|
|
`Results (msg, RO_array.map extract_cap descs)
|
|
|
|
(* Something resolved to a promised answer, and we sent a disembargo (because the promise was local).
|
|
When the disembargo response arrived, the local promise had resolved to this question.
|
|
Do we need another embargo? Return the path down which to send the disembargo request, if so. *)
|
|
let answer_cap_disembargo t path =
|
|
if cap_used t path then (
|
|
match t.state with
|
|
| Waiting | Cancelled -> None (* Can't disembargo if not returned *)
|
|
| Complete -> failwith "Already finished!" (* Previous disembargo should have kept us alive *)
|
|
| Lingering ret ->
|
|
match ret with
|
|
| `Error -> None
|
|
| `Results (msg, descs) ->
|
|
(* If the answer was local, we need to embargo. *)
|
|
begin match Core_types.Wire.Response.cap_index msg path with
|
|
| None -> None
|
|
| Some i ->
|
|
match RO_array.get ~oob:`None descs i with
|
|
| `Local -> Some (message_target t path)
|
|
| `Import i -> Some (`Import i)
|
|
| `None -> None
|
|
end;
|
|
| `Elsewhere -> None (* We don't care about the result (except for forwarding) *)
|
|
| `Local -> Some (message_target t path)
|
|
) else None (* Not used, so no embargo needed *)
|
|
|
|
(* A [Return] message has arrived. *)
|
|
let return t ret =
|
|
match t.state with
|
|
| Waiting -> t.state <- Lingering ret; []
|
|
| Cancelled -> t.state <- Complete; `Release_table_entry :: free_disembargo_info ret
|
|
| Lingering _
|
|
| Complete -> failwith "Already returned!"
|
|
|
|
let paths_used t =
|
|
t.pipelined_fields
|
|
|
|
let check t =
|
|
let pp f = pp f t in
|
|
match t.state with
|
|
| Waiting | Lingering _ -> RC.check ~pp t.ref_count (* rc > 0 *)
|
|
| Cancelled -> assert (RC.is_zero t.ref_count)
|
|
| Complete -> Debug.invariant_broken (fun f -> Fmt.pf f "Complete question %a still in table!" dump t)
|
|
|
|
let lost_connection t ~ex =
|
|
(* The resolve might get a "release" event, but we'll ignore it in the Complete state.
|
|
We don't care about the final state, as the whole table is being dropped anyway. *)
|
|
t.state <- Complete;
|
|
match t.remote_promise with
|
|
| `Released -> ()
|
|
| `Resolver p -> Core_types.resolve_payload p (Error (`Exception ex))
|
|
|
|
let resolve t payload =
|
|
match t.remote_promise with
|
|
| `Resolver p -> Core_types.resolve_payload p payload
|
|
| `Released ->
|
|
match payload with
|
|
| Ok payload -> Core_types.Response_payload.release payload
|
|
| Error _ -> ()
|
|
|
|
let v ~params_for_release ~remote_promise id =
|
|
{
|
|
id;
|
|
remote_promise = `Resolver remote_promise;
|
|
ref_count = RC.one; (* Held by [remote_promise] *)
|
|
state = Waiting;
|
|
params_for_release;
|
|
pipelined_fields = PathSet.empty
|
|
}
|
|
end
|
|
|
|
type message_target_cap = [
|
|
| `Import of Import.t
|
|
| `QuestionCap of Question.t * Wire.Path.t
|
|
]
|
|
|
|
let pp_message_target_cap f = function
|
|
| `Import i -> Import.pp f i
|
|
| `QuestionCap (q, p) -> Fmt.pf f "%a[%a]" Question.pp q Wire.Path.pp p
|
|
|
|
(* When we answer a question or provide an export, we may tell the peer that the
|
|
resource is local to it. In that case, we need to keep a reference to the remote
|
|
object so that we can forward disembargo requests. *)
|
|
type resolve_target = [
|
|
message_target_cap
|
|
| `Local (* Resolved to a local object, not one at the peer. *)
|
|
| `None (* This capability is not resolved yet *)
|
|
]
|
|
|
|
module Answer = struct
|
|
(* State model for answers:
|
|
|
|
When a call (or bootstrap) arrives, we add a new answer to the answers table.
|
|
It is removed from the table when we receive a Finish message for it.
|
|
|
|
While in the table, any of the following may happen:
|
|
|
|
- We calculate the answer. We send it back to the peer, and also keep it locally
|
|
so that we can forward any pipelined messages or disembargo requests.
|
|
|
|
- We get a Finish from the peer. The question is cancelled and removed from the table.
|
|
We reply with a cancelled message.
|
|
|
|
- We tell the peer to take the answer from some question we asked.
|
|
We forward any pipelined messages to the new question until we get a Finish.
|
|
We respond to disembargo requests by sending the response back to the new question.
|
|
*)
|
|
|
|
type active = {
|
|
answer : Core_types.struct_ref;
|
|
|
|
resolution : [
|
|
| `Unresolved
|
|
| `Resolved of
|
|
resolve_target RO_array.t * (* Targets for disembargo replies and forwarding *)
|
|
ExportId.t list (* Caps to release if we get releaseResultCaps *)
|
|
| `Forwarded of Question.t (* We replied with TakeFromOtherQuestion *)
|
|
];
|
|
}
|
|
|
|
type t = {
|
|
id : AnswerId.t;
|
|
mutable state : [
|
|
| `Active of active (* No Finish message has arrived yet. *)
|
|
| `Finished (* We got a Finish. A return has been sent. *)
|
|
]
|
|
}
|
|
|
|
let id t = t.id
|
|
|
|
let answer_struct t =
|
|
match t.state with
|
|
| `Finished -> `Finished
|
|
| `Active x -> `Promise x.answer
|
|
|
|
let pp f t =
|
|
Fmt.pf f "a%a" AnswerId.pp t.id
|
|
|
|
let pp_resolution f = function
|
|
| `Unresolved -> Fmt.string f "(unresolved)"
|
|
| `Resolved _ -> Fmt.string f "(resolved)"
|
|
| `Forwarded q -> Fmt.pf f "(forwarded to %a)" Question.pp q
|
|
|
|
let dump f t =
|
|
match t.state with
|
|
| `Finished -> Fmt.pf f "(finished)"
|
|
| `Active {resolution; answer} -> Fmt.pf f "%a %t" pp_resolution resolution answer#pp
|
|
|
|
let check t =
|
|
match t.state with
|
|
| `Finished -> failwith "Finished answer still in table!"
|
|
| `Active {answer; resolution} ->
|
|
answer#check_invariants;
|
|
match resolution with
|
|
| `Forwarded q -> Question.check q
|
|
| `Unresolved | `Resolved _ -> ()
|
|
|
|
let needs_return t =
|
|
t.state <> `Finished
|
|
|
|
(* We're sending a return message. *)
|
|
let return_resolved t ~exports_for_release ~resolve_targets =
|
|
match t.state with
|
|
| `Finished -> Fmt.failwith "Can't return finished answer %a!" pp t
|
|
| `Active x ->
|
|
assert (x.resolution = `Unresolved);
|
|
t.state <- `Active {x with resolution = `Resolved (resolve_targets, exports_for_release)}
|
|
|
|
(* We're sending a TakeFromOtherQuestion return message. *)
|
|
let return_take_from_question t question =
|
|
match t.state with
|
|
| `Finished -> Fmt.failwith "Can't return finished answer %a!" pp t
|
|
| `Active x ->
|
|
assert (x.resolution = `Unresolved);
|
|
Question.inc_ref question;
|
|
t.state <- `Active {x with resolution = `Forwarded question}
|
|
|
|
(* Remove from Answers table after calling this. *)
|
|
let finish t ~release_result_caps =
|
|
match t.state with
|
|
| `Finished -> Fmt.failwith "Can't finish already-finished answer %a" pp t
|
|
| `Active {answer; resolution} ->
|
|
t.state <- `Finished;
|
|
dec_ref answer;
|
|
match resolution with
|
|
| `Unresolved -> [`Return_cancelled]
|
|
| `Resolved (resolve_targets, exports_for_release) ->
|
|
`Release_resolve_targets resolve_targets :: (
|
|
if release_result_caps then [`Release_exports exports_for_release]
|
|
else []
|
|
)
|
|
| `Forwarded q ->
|
|
begin match q.Question.remote_promise with
|
|
| `Resolver r ->
|
|
(* Notify anyone waiting on the results. *)
|
|
Core_types.resolve_exn r (Exception.v "Results sent elsewhere")
|
|
| `Released -> ()
|
|
end;
|
|
[`Question_actions (q, Question.dec_ref q)]
|
|
|
|
let lost_connection t =
|
|
finish ~release_result_caps:true t |> filter_map (function
|
|
| `Question_actions _
|
|
| `Release_resolve_targets _ as action -> Some action
|
|
| `Return_cancelled -> None
|
|
| `Release_exports (_ : ExportId.t list) -> None (* There are local to the connection anyway *)
|
|
)
|
|
|
|
(* The target of [path] that we returned as the answer.
|
|
[None] if we didn't return yet. *)
|
|
let resolve_target t path =
|
|
match t.state with
|
|
| `Finished -> Fmt.failwith "Answer %a is finished!" pp t
|
|
| `Active {answer; resolution} ->
|
|
match resolution with
|
|
| `Unresolved -> None
|
|
| `Forwarded q -> Some (Ok (`QuestionCap (q, path)))
|
|
| `Resolved (resolve_targets, _) ->
|
|
match answer#response with
|
|
| None -> Fmt.failwith "Answer %a is resolved, but no response recorded!" pp t
|
|
| Some (Error _) as e -> e
|
|
| Some (Ok msg) ->
|
|
match Core_types.Wire.Response.cap_index msg path with
|
|
| Some i when i >= 0 && i < RO_array.length resolve_targets ->
|
|
Some (Ok (RO_array.get_exn resolve_targets i))
|
|
| Some i -> Some (Error (Error.exn "Invalid answer cap index %d" i))
|
|
| None -> Some (Ok `None)
|
|
|
|
let disembargo_target t path =
|
|
match resolve_target t path with
|
|
| None -> Fmt.failwith "Got disembargo request for unresolved answer %a!" pp t
|
|
| Some (Error _) -> failwith "Got disembargo for an exception!"
|
|
| Some (Ok target) -> target
|
|
|
|
let create aid ~answer = {
|
|
id = aid;
|
|
state = `Active { answer; resolution = `Unresolved };
|
|
}
|
|
|
|
let create_uninitialised aid = {
|
|
id = aid;
|
|
state = `Finished;
|
|
}
|
|
|
|
let init t answer =
|
|
match t.state with
|
|
| `Finished -> t.state <- `Active { answer; resolution = `Unresolved }
|
|
| `Active _ -> Fmt.failwith "Answer %a already initialised!" pp t
|
|
end
|
|
|
|
module Export = struct
|
|
type t = {
|
|
id : ExportId.t;
|
|
mutable count : int; (* Number of times sent to remote and not yet released *)
|
|
mutable service : Core_types.cap;
|
|
mutable resolve_target : resolve_target; (* [`None] if not yet resolved *)
|
|
}
|
|
|
|
let dump f t =
|
|
Fmt.pf f "%t" t.service#pp
|
|
|
|
let pp f t =
|
|
Fmt.pf f "e%a" ExportId.pp t.id
|
|
|
|
let inc_ref t =
|
|
t.count <- t.count + 1
|
|
|
|
let id t = t.id
|
|
|
|
let count t = t.count
|
|
|
|
let resolve_target t = t.resolve_target
|
|
|
|
let service t = t.service
|
|
|
|
let resolve t target =
|
|
t.resolve_target <- target
|
|
|
|
let released = Core_types.broken_cap (Exception.v "(released)")
|
|
|
|
let release t ref_count =
|
|
assert (t.count >= ref_count);
|
|
let count = t.count - ref_count in
|
|
t.count <- count;
|
|
if count > 0 then `Do_nothing
|
|
else (
|
|
let service = t.service in
|
|
t.service <- released;
|
|
`Send_release (service, t.resolve_target)
|
|
)
|
|
|
|
let lost_connection t =
|
|
dec_ref t.service;
|
|
t.service <- released;
|
|
t.count <- 0
|
|
|
|
let check t = t.service#check_invariants
|
|
|
|
let v ~service id =
|
|
{ count = 1; service; id; resolve_target = `None }
|
|
end
|
|
|
|
type descr = [
|
|
message_target_cap
|
|
| `ThirdPartyHosted of Out.third_party_desc
|
|
| `Local of Core_types.cap
|
|
]
|
|
|
|
let pp_cap : [< descr] Fmt.t = fun f -> function
|
|
| `Import import -> Fmt.pf f "%a" Import.pp import
|
|
| `QuestionCap (question, p) -> Fmt.pf f "%a[%a]" Question.pp question Wire.Path.pp p
|
|
| `ThirdPartyHosted _third_party_desc -> Fmt.pf f "ThirdPartyHosted"
|
|
| `Local local -> Fmt.pf f "local:%t" local#pp
|
|
|
|
type restorer = ((Core_types.cap, Exception.t) result -> unit) -> string -> unit
|
|
|
|
type t = {
|
|
mutable queue_send : (EP.Out.t -> unit); (* (mutable for shutdown) *)
|
|
tags : Logs.Tag.set;
|
|
embargoes : (EmbargoId.t * Cap_proxy.resolver_cap) Embargoes.t;
|
|
restore : restorer;
|
|
fork : (unit -> unit) -> unit;
|
|
|
|
questions : Question.t Questions.t;
|
|
answers : Answer.t Answers.t;
|
|
exports : Export.t Exports.t;
|
|
imports : Import.t Imports.t;
|
|
exported_caps : (Core_types.cap, ExportId.t) Hashtbl.t;
|
|
|
|
mutable disconnected : Exception.t option; (* If set, this connection is finished. *)
|
|
}
|
|
|
|
type 'a S.brand += CapTP : (t * message_target_cap) S.brand
|
|
(* The [CapTP] message asks a capability to tell us its CapTP target, if any. *)
|
|
|
|
type 'a S.brand += CapTP_results : (t * Answer.t) S.brand
|
|
(* The [CapTP_results] message asks a struct_resolver to tell us if it's for a CapTP answer.
|
|
If so, we may be able to send the results directly. *)
|
|
|
|
let target_of (x : #Core_types.cap) = x#shortest#sealed_dispatch CapTP
|
|
|
|
let my_target_of t (x : #Core_types.cap) =
|
|
match target_of x with
|
|
| Some (t', target) when t == t' -> Some target
|
|
| _ -> None
|
|
|
|
let stats t =
|
|
{ Stats.
|
|
n_questions = Questions.active t.questions;
|
|
n_answers = Answers.active t.answers;
|
|
n_imports = Imports.active t.imports;
|
|
n_exports = Exports.active t.exports;
|
|
}
|
|
|
|
let default_restore k _object_id =
|
|
k @@ Error (Exception.v "This vat has no restorer")
|
|
|
|
let create ?(restore=default_restore) ~tags ~fork ~queue_send =
|
|
{
|
|
queue_send = (queue_send :> EP.Out.t -> unit);
|
|
tags;
|
|
restore = restore;
|
|
fork;
|
|
questions = Questions.make ();
|
|
answers = Answers.make ();
|
|
imports = Imports.make ();
|
|
exports = Exports.make ();
|
|
embargoes = Embargoes.make ();
|
|
exported_caps = Hashtbl.create 30;
|
|
disconnected = None;
|
|
}
|
|
[@@ocaml.warning "-16"] (* Too late to change the API now. *)
|
|
|
|
let with_qid qid t =
|
|
Logs.Tag.add Debug.qid_tag (QuestionId.uint32 qid) t.tags
|
|
|
|
let with_aid aid t =
|
|
Logs.Tag.add Debug.qid_tag (EP.Table.AnswerId.uint32 aid) t.tags
|
|
|
|
let tags t = t.tags
|
|
|
|
let pp_promise f = function
|
|
| Some q -> Question.pp f q
|
|
| None -> Fmt.string f "(not initialised)"
|
|
|
|
let check_connected t =
|
|
match t.disconnected with
|
|
| None -> ()
|
|
| Some ex -> Fmt.failwith "CapTP connection is disconnected (%a)" Exception.pp ex
|
|
|
|
module Send : sig
|
|
(** Converts struct pointers into integer table indexes, ready for sending.
|
|
The indexes are only valid until the next message is sent. *)
|
|
|
|
open EP.Core_types
|
|
|
|
val bootstrap : t -> struct_resolver -> Question.t
|
|
|
|
val call : t -> struct_resolver ->
|
|
message_target_cap -> Wire.Request.t ->
|
|
results_to:EP.Out.send_results_to ->
|
|
Question.t
|
|
|
|
val return : t -> Answer.t -> Core_types.Response_payload.t Core_types.or_error -> unit
|
|
|
|
val release : t -> Import.t -> int -> unit
|
|
(** [release t i count] tells the peer that [i] is no longer needed by us,
|
|
decreasing the ref-count by [count]. *)
|
|
|
|
val finish : t -> Question.t -> unit
|
|
(** [finish t q] sends a Finish message (with [releaseResultCaps=false]). *)
|
|
end = struct
|
|
|
|
(** We are sending [cap] to the client, either in a Return or Resolve message.
|
|
Return the [resolve_target] for it, and increment any required
|
|
ref-count to keep it alive. *)
|
|
let get_resolve_target t cap =
|
|
match target_of cap with
|
|
| Some (t', target) when t == t' ->
|
|
begin match target with
|
|
| `QuestionCap (question, _) as target ->
|
|
Question.inc_ref question;
|
|
target
|
|
| `Import import as target ->
|
|
Import.inc_ref import;
|
|
target
|
|
end
|
|
| Some _ (* todo: third-party *)
|
|
| None -> `Local
|
|
|
|
(** [export ~broken_caps t cap] is a descriptor for [cap].
|
|
If [cap] is a proxy object for a service at the peer, tell the peer the target directly.
|
|
Otherwise, export it to the peer (reusing an existing export, if any).
|
|
If the cap is broken and needs a fresh export, we queue up a suitable resolve message
|
|
on [broken_caps]. This is needed for return messages. *)
|
|
let rec export ?broken_caps : t -> Core_types.cap -> Out.desc = fun t cap ->
|
|
let cap = cap#shortest in
|
|
match my_target_of t cap with
|
|
| Some (`Import import) -> Import.message_target import
|
|
| Some (`QuestionCap (question, i)) -> Question.message_target question i
|
|
| None ->
|
|
let problem = cap#problem in
|
|
(* Exceptions are considered unsettled because we need to resolve them to
|
|
an error later (due to a protocol limitation). *)
|
|
let settled = problem = None && cap#blocker = None in
|
|
let ex =
|
|
match Hashtbl.find t.exported_caps cap with
|
|
| id ->
|
|
let ex = Exports.find_exn t.exports id in
|
|
Export.inc_ref ex;
|
|
ex
|
|
| exception Not_found ->
|
|
Core_types.inc_ref cap;
|
|
let ex = Exports.alloc t.exports (Export.v ~service:cap) in
|
|
let id = Export.id ex in
|
|
Hashtbl.add t.exported_caps cap id;
|
|
begin match problem, broken_caps with
|
|
| Some problem, Some broken_caps -> Queue.add (ex, problem) broken_caps
|
|
| Some _, _ -> failwith "Cap is broken, but [broken_caps] not provided!"
|
|
| None, _ when settled -> ()
|
|
| None, _ ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Monitoring promise export %a -> %a" Export.pp ex Export.dump ex);
|
|
cap#when_more_resolved (fun x ->
|
|
if Export.count ex > 0 then (
|
|
let x = x#shortest in
|
|
match x#problem with
|
|
| Some problem ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Export %a resolved to %t - sending exception"
|
|
Export.pp ex
|
|
x#pp
|
|
);
|
|
t.queue_send (`Resolve (Export.id ex, Error problem));
|
|
| None ->
|
|
let new_export = export t x in
|
|
Log.debug (fun f -> f ~tags:t.tags "Export %a resolved to %t - sending notification to use %a"
|
|
Export.pp ex
|
|
x#pp
|
|
Out.pp_desc new_export
|
|
);
|
|
Export.resolve ex (get_resolve_target t x);
|
|
t.queue_send (`Resolve (Export.id ex, Ok new_export));
|
|
); (* else: no longer an export *)
|
|
Core_types.dec_ref x
|
|
)
|
|
end;
|
|
ex
|
|
in
|
|
let id = Export.id ex in
|
|
if settled then `SenderHosted id
|
|
else `SenderPromise id
|
|
|
|
let bootstrap t remote_promise =
|
|
Questions.alloc t.questions (Question.v ~params_for_release:[] ~remote_promise)
|
|
|
|
(* This is for level 0 implementations, which don't understand about releasing caps. *)
|
|
let exports_of =
|
|
RO_array.fold_left (fun acc -> function
|
|
| `SenderPromise id | `SenderHosted id | `ThirdPartyHosted (_, id) -> id :: acc
|
|
| `None | `ReceiverAnswer _ | `ReceiverHosted _ -> acc
|
|
) []
|
|
|
|
(* For some reason, we can't send broken caps in a payload message. So, if
|
|
any of them are broken, we send a resolve for them immediately afterwards. *)
|
|
let resolve_broken t =
|
|
Queue.iter @@ fun (ex, problem) ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Sending resolve for already-broken export %a : %a"
|
|
Export.pp ex
|
|
Export.dump ex
|
|
);
|
|
t.queue_send (`Resolve (Export.id ex, Error problem))
|
|
|
|
let call t remote_promise (target : message_target_cap) msg ~results_to =
|
|
let broken_caps = Queue.create () in
|
|
let caps = Core_types.Request_payload.snapshot_caps msg in
|
|
let descs = RO_array.map (export ~broken_caps t) caps in
|
|
let question = Questions.alloc t.questions
|
|
(Question.v ~params_for_release:(exports_of descs) ~remote_promise)
|
|
in
|
|
let message_target =
|
|
match target with
|
|
| `Import import ->
|
|
Import.mark_used import;
|
|
Import.message_target import
|
|
| `QuestionCap (question, i) ->
|
|
Question.set_cap_used question i;
|
|
Question.message_target question i
|
|
in
|
|
let qid = Question.id question in
|
|
Log.debug (fun f -> f ~tags:(with_qid qid t) "Sending: (%a).call %a"
|
|
pp_cap target
|
|
Core_types.Request_payload.pp msg);
|
|
t.queue_send (`Call (qid, message_target, msg, descs, results_to));
|
|
resolve_broken t broken_caps;
|
|
question
|
|
|
|
let return_results t answer msg =
|
|
let caps = Core_types.Response_payload.snapshot_caps msg in
|
|
let aid = Answer.id answer in
|
|
let caps = RO_array.map (fun c -> c#shortest) caps in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Returning results: %a"
|
|
Core_types.Response_payload.pp msg);
|
|
RO_array.iter Core_types.inc_ref caps; (* Copy everything stored in [msg]. *)
|
|
let broken_caps = Queue.create () in
|
|
let descs = RO_array.map (export ~broken_caps t) caps in
|
|
let exports_for_release = exports_of descs in
|
|
let resolve_targets = RO_array.map (get_resolve_target t) caps in
|
|
Answer.return_resolved answer ~exports_for_release ~resolve_targets;
|
|
RO_array.iter Core_types.dec_ref caps;
|
|
let ret = `Results (msg, descs) in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Wire results: %a" Out.pp_return ret);
|
|
t.queue_send (`Return (aid, ret, false));
|
|
resolve_broken t broken_caps
|
|
|
|
let return t answer ret =
|
|
let aid = Answer.id answer in
|
|
match ret with
|
|
| Ok payload ->
|
|
return_results t answer payload
|
|
| Error err ->
|
|
let ret = (err : Error.t :> Out.return) in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Returning error: %a" Error.pp err);
|
|
Answer.return_resolved answer ~exports_for_release:[] ~resolve_targets:RO_array.empty;
|
|
t.queue_send (`Return (aid, ret, false))
|
|
|
|
let release t import count =
|
|
Imports.release t.imports (Import.id import);
|
|
Log.debug (fun f -> f ~tags:t.tags "Sending release of %a" Import.pp import);
|
|
t.queue_send (`Release (Import.id import, count))
|
|
|
|
let finish t question =
|
|
let qid = Question.id question in
|
|
Log.debug (fun f -> f ~tags:(with_qid qid t) "Send finish %a" Question.pp_promise question);
|
|
t.queue_send (`Finish (qid, false))
|
|
end
|
|
|
|
let apply_import_actions t =
|
|
List.iter @@ function
|
|
| `Release (i, count) -> Send.release t i count
|
|
|
|
let apply_question_actions t q =
|
|
List.iter @@ function
|
|
| `Send_finish -> Send.finish t q
|
|
| `Release_table_entry -> Questions.release t.questions (Question.id q)
|
|
|` Release_import i -> Import.dec_ref i |> apply_import_actions t
|
|
|
|
let release_resolve_target t = function
|
|
| `None | `Local -> ()
|
|
| `QuestionCap (q, _) -> Question.dec_ref q |> apply_question_actions t q;
|
|
| `Import i -> Import.dec_ref i |> apply_import_actions t
|
|
|
|
let release t export_id ~ref_count =
|
|
assert (ref_count > 0);
|
|
let export = Exports.find_exn t.exports export_id in
|
|
match Export.release export ref_count with
|
|
| `Do_nothing -> ()
|
|
| `Send_release (service, resolve_target) ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Releasing export %a" Export.pp export);
|
|
Hashtbl.remove t.exported_caps service;
|
|
Exports.release t.exports export_id;
|
|
dec_ref service;
|
|
release_resolve_target t resolve_target
|
|
|
|
let apply_answer_actions t answer =
|
|
List.iter @@ function
|
|
| `Return_cancelled ->
|
|
let aid = Answer.id answer in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Returning cancelled");
|
|
t.queue_send (`Return (aid, `Cancelled, false));
|
|
| `Release_exports exports -> List.iter (release t ~ref_count:1) exports
|
|
| `Release_resolve_targets targets -> RO_array.iter (release_resolve_target t) targets
|
|
| `Question_actions (q, actions) -> apply_question_actions t q actions
|
|
|
|
(* Note: takes ownership of [msg] *)
|
|
let rec send_call t target (results:Core_types.struct_resolver) msg =
|
|
match t.disconnected with
|
|
| Some ex -> results#resolve (Core_types.broken_struct (`Exception ex))
|
|
| None ->
|
|
match results#sealed_dispatch CapTP_results with
|
|
| Some (_, answer) when not (Answer.needs_return answer) ->
|
|
(* The only reason for making this call is to answer a cancelled question (we already
|
|
replied to it to confirm the cancellation). No-one cares about the answer, so do nothing. *)
|
|
Core_types.Request_payload.release msg;
|
|
Core_types.resolve_exn results Exception.cancelled (* (might be useful for debugging) *)
|
|
| Some (t', answer) when t == t' ->
|
|
let remote_promise = make_remote_promise t in
|
|
(* [results] is at [target]. We can tell the peer to send the results to itself directly. *)
|
|
let question = Send.call t (remote_promise :> Core_types.struct_resolver)
|
|
target msg ~results_to:`Yourself in
|
|
Core_types.Request_payload.release msg;
|
|
remote_promise#set_question question;
|
|
Answer.return_take_from_question answer question;
|
|
let aid = Answer.id answer in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Returning take-from-other-question %a" Question.pp question);
|
|
t.queue_send (`Return (aid, `TakeFromOtherQuestion (Question.id question), false));
|
|
(* Hook this up so we still forward any pipelined calls *)
|
|
results#resolve (remote_promise :> Core_types.struct_ref)
|
|
| _ ->
|
|
let remote_promise = make_remote_promise t in
|
|
let question = Send.call t (remote_promise :> Core_types.struct_resolver)
|
|
target msg ~results_to:`Caller in
|
|
Core_types.Request_payload.release msg;
|
|
remote_promise#set_question question;
|
|
results#resolve (remote_promise :> Core_types.struct_ref)
|
|
|
|
(* A cap that sends to a promised answer's cap at other *)
|
|
and make_remote_promise t =
|
|
object (self : #Core_types.struct_resolver)
|
|
inherit [Question.t option] Struct_proxy.t None
|
|
|
|
val mutable released_question = false
|
|
|
|
(* We send Finish in two cases:
|
|
- We are unresolved and the user wants to cancel.
|
|
- We got the result and want the server to free up the question.
|
|
Send Finish on whichever happens first, but not both.
|
|
*)
|
|
method private ensure_released q =
|
|
if not released_question then (
|
|
released_question <- true;
|
|
Question.release_proxy q |> apply_question_actions t q
|
|
)
|
|
|
|
method do_pipeline question i results msg =
|
|
match question with
|
|
| Some target_q ->
|
|
let target = `QuestionCap (target_q, i) in
|
|
send_call t target results msg
|
|
| None -> failwith "Not initialised!"
|
|
|
|
method on_resolve q _ =
|
|
match q with
|
|
| None -> failwith "Not initialised!"
|
|
| Some q -> self#ensure_released q
|
|
|
|
val name = "remote-promise"
|
|
|
|
method pp_unresolved = pp_promise
|
|
|
|
method set_question q =
|
|
self#update_target (Some q)
|
|
|
|
method send_cancel = function
|
|
| None -> failwith "Not initialised!"
|
|
| Some q -> self#ensure_released q
|
|
|
|
method field_sealed_dispatch : type a. Wire.Path.t -> a S.brand -> a option = fun path -> function
|
|
| CapTP ->
|
|
begin match state with
|
|
| Unresolved u ->
|
|
begin match u.target with
|
|
| None -> failwith "Not intialised!"
|
|
| Some target_q -> Some (t, `QuestionCap (target_q, path))
|
|
end
|
|
| _ -> failwith "Not a promise!"
|
|
end;
|
|
| _ -> None
|
|
end
|
|
|
|
let disembargo t request =
|
|
Log.debug (fun f -> f ~tags:t.tags "Sending disembargo %a" EP.Out.pp_disembargo_request request);
|
|
t.queue_send (`Disembargo_request request)
|
|
|
|
let bootstrap t object_id =
|
|
match t.disconnected with
|
|
| Some ex -> Core_types.broken_cap ex
|
|
| None ->
|
|
let result = make_remote_promise t in
|
|
let question = Send.bootstrap t (result :> Core_types.struct_resolver) in
|
|
result#set_question question;
|
|
let qid = Question.id question in
|
|
Log.debug (fun f -> f ~tags:(with_qid qid t) "Sending: bootstrap");
|
|
t.queue_send (`Bootstrap (qid, object_id));
|
|
let service = result#cap Wire.Path.root in
|
|
dec_ref result;
|
|
service
|
|
|
|
module Switchable = struct
|
|
class type handler = object
|
|
method pp : Format.formatter -> unit
|
|
method sealed_dispatch : 'a. 'a S.brand -> 'a option
|
|
method call : Core_types.struct_resolver -> Wire.Request.t -> unit
|
|
end
|
|
|
|
type unset = {
|
|
mutable rc : RC.t;
|
|
handler : handler; (* Will forward calls here until set *)
|
|
on_set : (Core_types.cap -> unit) Queue.t;
|
|
on_release : (unit -> unit) Queue.t;
|
|
}
|
|
|
|
type state =
|
|
| Unset of unset
|
|
| Set of Core_types.cap
|
|
|
|
let released = Core_types.broken_cap (Exception.v "(released)")
|
|
|
|
let pp_state f = function
|
|
| Unset x -> Fmt.pf f "(unset, %a) -> %t" RC.pp x.rc x.handler#pp
|
|
| Set x -> Fmt.pf f "(set) -> %t" x#pp
|
|
|
|
let target = function
|
|
| Unset x -> x.handler
|
|
| Set x -> (x :> handler)
|
|
|
|
type 'a S.brand += Gc : unit S.brand
|
|
|
|
(* Forwards messages to [init] until resolved.
|
|
Forces [release] when resolved or released.
|
|
Stores a reference to itself in [strong_proxy] when unresolved but with queued callbacks. *)
|
|
let make ~(release:unit Lazy.t) ~settled ~strong_proxy init =
|
|
object (self : #Core_types.cap)
|
|
val id = Debug.OID.next ()
|
|
val thread_id = Thread.(id (self ()))
|
|
|
|
val mutable state =
|
|
Unset { rc = RC.one; handler = init; on_set = Queue.create (); on_release = Queue.create () }
|
|
|
|
method call msg caps =
|
|
(target state)#call msg caps
|
|
|
|
method update_rc d =
|
|
match state with
|
|
| Unset u ->
|
|
u.rc <- RC.sum u.rc d ~pp:(fun f -> self#pp f);
|
|
if RC.is_zero u.rc then (
|
|
Lazy.force release;
|
|
let old_state = state in
|
|
strong_proxy := None;
|
|
state <- Set released;
|
|
match old_state with
|
|
| Unset u -> Queue.iter (fun f -> f ()) u.on_release
|
|
| Set x -> dec_ref x
|
|
)
|
|
| Set x -> x#update_rc d
|
|
|
|
method resolve cap =
|
|
match state with
|
|
| Set _ -> Fmt.failwith "Can't resolve already-set switchable %t to %t!" self#pp cap#pp
|
|
| Unset {handler = _; rc; on_set; on_release} ->
|
|
let pp f = self#pp f in
|
|
RC.check ~pp rc;
|
|
state <- Set cap;
|
|
strong_proxy := None;
|
|
begin match RC.to_int rc with
|
|
| Some rc -> cap#update_rc (rc - 1); (* Transfer our ref-count *)
|
|
| None -> ()
|
|
end;
|
|
Queue.iter (fun f -> f (with_inc_ref cap)) on_set;
|
|
Queue.iter (fun f -> cap#when_released f) on_release;
|
|
Lazy.force release
|
|
|
|
method break ex =
|
|
self#resolve (Core_types.broken_cap ex)
|
|
|
|
method shortest =
|
|
match state with
|
|
| Unset _ -> (self :> Core_types.cap) (* Can't shorten, as we may change later *)
|
|
| Set x -> x#shortest
|
|
|
|
method blocker =
|
|
match state with
|
|
| Unset _ when settled -> None
|
|
| Unset _ -> Some (self :> Core_types.base_ref)
|
|
| Set x -> x#blocker
|
|
|
|
method problem =
|
|
match state with
|
|
| Unset _ -> None
|
|
| Set x -> x#problem
|
|
|
|
method when_more_resolved fn =
|
|
match state with
|
|
| Unset x ->
|
|
Queue.add fn x.on_set;
|
|
strong_proxy := Some (self :> < >)
|
|
| Set x -> x#when_more_resolved fn
|
|
|
|
method when_released fn =
|
|
match state with
|
|
| Unset x ->
|
|
Queue.add fn x.on_release;
|
|
strong_proxy := Some (self :> < >)
|
|
| Set x -> x#when_released fn
|
|
|
|
(* When trying to find the target for export, it's OK to expose our current
|
|
target, even though [shortest] shouldn't.
|
|
In particular, we need this to deal with disembargo requests. *)
|
|
method sealed_dispatch : type a. a S.brand -> a option = function
|
|
| CapTP ->
|
|
begin match state with
|
|
| Unset x -> x.handler#sealed_dispatch CapTP
|
|
| Set x -> x#shortest#sealed_dispatch CapTP
|
|
end
|
|
| Gc ->
|
|
begin match state with
|
|
| Unset x ->
|
|
Core_types.Wire.ref_leak_detected thread_id (fun () ->
|
|
if RC.is_zero x.rc then (
|
|
Log.warn (fun f -> f "@[<v2>Reference GC'd with non-zero ref-count!@,%t@,\
|
|
But, ref-count is now zero, so a previous GC leak must have fixed it.@]"
|
|
self#pp);
|
|
) else (
|
|
Log.warn (fun f -> f "@[<v2>Reference GC'd with %a!@,%t@]"
|
|
RC.pp x.rc self#pp);
|
|
x.rc <- RC.leaked;
|
|
state <- Set released; (* Don't call resolve; rc is now invalid *)
|
|
assert (!strong_proxy = None); (* Otherwise we wouldn't have been GC'd *)
|
|
Queue.iter (fun f -> f released) x.on_set;
|
|
Lazy.force release
|
|
)
|
|
);
|
|
| Set _ -> ()
|
|
end;
|
|
Some ()
|
|
| _ -> None
|
|
|
|
method check_invariants =
|
|
match state with
|
|
| Unset u ->
|
|
let pp f = self#pp f in
|
|
assert ((!strong_proxy = None) = (Queue.is_empty u.on_set && Queue.is_empty u.on_release));
|
|
RC.check ~pp u.rc
|
|
| Set x ->
|
|
assert (!strong_proxy = None);
|
|
x#check_invariants
|
|
|
|
method pp f =
|
|
Fmt.pf f "switchable(%a) %a" Debug.OID.pp id pp_state state
|
|
|
|
initializer
|
|
Gc.finalise (fun (self:#Core_types.base_ref) -> ignore (self#sealed_dispatch Gc)) self
|
|
end
|
|
end
|
|
|
|
module Input : sig
|
|
open EP.Core_types
|
|
|
|
val call : t -> In.QuestionId.t -> In.message_target ->
|
|
Wire.Request.t -> In.desc RO_array.t -> results_to:In.send_results_to ->
|
|
unit
|
|
|
|
val bootstrap : t -> In.QuestionId.t -> string -> unit
|
|
val return : t -> In.AnswerId.t -> In.return -> release_param_caps:bool -> unit
|
|
val finish : t -> In.QuestionId.t -> release_result_caps:bool -> unit
|
|
val release : t -> In.ImportId.t -> ref_count:int -> unit
|
|
val disembargo_request : t -> In.disembargo_request -> unit
|
|
val disembargo_reply : t -> In.message_target -> Message_types.EmbargoId.t -> unit
|
|
val resolve : t -> In.ExportId.t -> (In.desc, Exception.t) result -> unit
|
|
end = struct
|
|
let release = release
|
|
|
|
let set_import_proxy t ~settled import =
|
|
let message_target = `Import import in
|
|
let cap =
|
|
object (_ : Switchable.handler)
|
|
val id = Debug.OID.next ()
|
|
|
|
method call results msg =
|
|
send_call t message_target results msg
|
|
|
|
method pp f =
|
|
if settled then
|
|
Fmt.pf f "far-ref(%a) -> %a" Debug.OID.pp id pp_cap message_target
|
|
else
|
|
Fmt.pf f "remote-promise(%a) -> %a" Debug.OID.pp id pp_cap message_target
|
|
|
|
method sealed_dispatch : type a. a S.brand -> a option = function
|
|
| CapTP -> Some (t, message_target)
|
|
| _ -> None
|
|
end
|
|
in
|
|
let release = lazy (
|
|
Import.clear_proxy import;
|
|
Import.dec_ref import |> apply_import_actions t;
|
|
) in
|
|
(* Imports can resolve to another cap (if unsettled) or break. *)
|
|
let switchable = Switchable.make ~release ~settled ~strong_proxy:import.strong_proxy cap in
|
|
Import.init_proxy import switchable;
|
|
switchable
|
|
|
|
let import_sender t ~mark_dirty ~settled id =
|
|
let new_import () =
|
|
let import = Import.v id ~mark_dirty ~settled in
|
|
Imports.set t.imports id import;
|
|
(set_import_proxy t ~settled import :> Core_types.cap)
|
|
in
|
|
match Imports.find t.imports id with
|
|
| None -> new_import ()
|
|
| Some import ->
|
|
Import.inc_count import;
|
|
if mark_dirty then Import.mark_used import;
|
|
match Import.get_proxy import with
|
|
| Some proxy ->
|
|
Core_types.inc_ref proxy;
|
|
(proxy :> Core_types.cap)
|
|
| None ->
|
|
(* The switchable got GC'd. It has already dec-ref'd the import.
|
|
Make a new proxy. *)
|
|
Import.inc_ref import;
|
|
(set_import_proxy t ~settled import :> Core_types.cap)
|
|
|
|
(* We previously pipelined messages to [old_path], which now turns out to be
|
|
local object [x]. Create an embargo proxy for [x] and send a disembargo
|
|
for it down [old_path]. *)
|
|
let local_embargo t ~old_path x =
|
|
if x#problem <> None then x (* Don't embargo errors *)
|
|
else (
|
|
let embargo = Cap_proxy.local_promise () in
|
|
(* Store in table *)
|
|
inc_ref embargo;
|
|
let (embargo_id, _) = Embargoes.alloc t.embargoes (fun id -> (id, embargo)) in
|
|
(* Send disembargo request *)
|
|
let disembargo_request = `Loopback (old_path, embargo_id) in
|
|
Log.debug (fun f -> f ~tags:t.tags "Embargo %t until %a is delivered"
|
|
x#pp
|
|
EP.Out.pp_disembargo_request disembargo_request
|
|
);
|
|
disembargo t disembargo_request;
|
|
(* We don't store [x], because when the reply comes back further disembargoes
|
|
may be needed. *)
|
|
dec_ref x;
|
|
(embargo :> Core_types.cap)
|
|
)
|
|
|
|
let maybe_embargo t ~old_path x =
|
|
match old_path with
|
|
| None -> x
|
|
| Some old_path -> local_embargo t ~old_path x
|
|
|
|
(* Turn a connection-scoped cap reference received from our peer into a general-purpose
|
|
cap for users. The caller owns the new reference and should [dec_ref] it when done.
|
|
|
|
This is used in several places:
|
|
|
|
- When we get a Resolve message (which must be for a SenderPromise or SenderHosted)
|
|
- When we get a Return message (resolving each PromisedAnswer from the question)
|
|
- When we receive a Call (no embargoes needed in that case; you can't pipeline over the
|
|
arguments of a call you haven't yet received)
|
|
- When we get a Disembargo_reply (might need another embargo)
|
|
|
|
If [embargo_path] is passed then we have already pipelined messages over this cap, and
|
|
may therefore need to embargo it (sending the disembargo via the old [embargo_path]).
|
|
This can only happen when a local promise at the peer has resolved, and we are given its
|
|
new target. *)
|
|
let import t ?embargo_path : In.desc -> Core_types.cap = function
|
|
| `SenderPromise id ->
|
|
import_sender t id ~settled:false ~mark_dirty:(embargo_path <> None)
|
|
| `SenderHosted id ->
|
|
(* We don't have to worry about embargoes here because anything we've already sent to the peer will
|
|
arrive before anything we send in the future, and it can't change. *)
|
|
import_sender t id ~settled:true ~mark_dirty:false
|
|
| `ReceiverHosted id ->
|
|
let export = Exports.find_exn t.exports id in
|
|
(* We host the target (which may be another promise).
|
|
We need to flush any pipelined calls still in flight before we can use it. *)
|
|
maybe_embargo t ~old_path:embargo_path (with_inc_ref (Export.service export))
|
|
| `ReceiverAnswer (id, path) ->
|
|
let answer = Answers.find_exn t.answers id in
|
|
begin match Answer.answer_struct answer with
|
|
| `Finished -> failwith "Got ReceiverAnswer for a finished promise!"
|
|
| `Promise answer_promise ->
|
|
match answer_promise#response with
|
|
| None ->
|
|
(* We don't know the answer, so we can't have replied yet.
|
|
We can send a disembargo request now and the peer will get it before
|
|
any answer and return it to us. *)
|
|
maybe_embargo t ~old_path:embargo_path (answer_promise#cap path)
|
|
| Some (Error _) -> answer_promise#cap path (* No need to embargo errors *)
|
|
| Some (Ok payload) ->
|
|
(* We've already replied to this question. If we returned a capability at the
|
|
requested path then send a disembargo request to it. If not, we can't embargo,
|
|
because the peer has no way to return the response, but we don't need to for errors. *)
|
|
match Core_types.Response_payload.field payload path with
|
|
| None -> answer_promise#cap path (* Don't embargo None *)
|
|
| Some c -> dec_ref c; maybe_embargo t ~old_path:embargo_path (answer_promise#cap path)
|
|
end
|
|
| `None -> Core_types.null
|
|
| `ThirdPartyHosted _ -> failwith "todo: import"
|
|
|
|
let make_answer_promise t answer ~results_to =
|
|
let aid = Answer.id answer in
|
|
let promise, resolver = Local_struct_promise.make () in
|
|
let () =
|
|
promise#when_resolved @@ fun x ->
|
|
if Answer.needs_return answer && t.disconnected = None then (
|
|
match results_to with
|
|
| `Caller -> Send.return t answer x
|
|
| `Yourself ->
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Returning results-sent-elsewhere");
|
|
Answer.return_resolved answer ~exports_for_release:[] ~resolve_targets:RO_array.empty;
|
|
t.queue_send (`Return (aid, `ResultsSentElsewhere, false))
|
|
| `ThirdParty _ -> failwith "todo: handle call by sending results to ThirdParty"
|
|
)
|
|
(* (else cancelled; reply already sent) *)
|
|
in
|
|
let resolver =
|
|
if results_to = `Caller then (
|
|
object (_ : Core_types.struct_resolver)
|
|
method pp f = Fmt.pf f "answer %a <- %t" AnswerId.pp aid resolver#pp
|
|
|
|
method resolve = resolver#resolve
|
|
|
|
method set_blocker = resolver#set_blocker
|
|
method clear_blocker = resolver#clear_blocker
|
|
|
|
method sealed_dispatch : type a. a S.brand -> a option = function
|
|
| CapTP_results -> Some (t, answer)
|
|
| _ -> None
|
|
end
|
|
) else resolver
|
|
in
|
|
promise, resolver
|
|
|
|
let call t aid (message_target : In.message_target) msg descs ~results_to =
|
|
(* todo: allowThirdPartyTailCall *)
|
|
let answer = Answer.create_uninitialised aid in
|
|
let answer_promise, answer_resolver = make_answer_promise t answer ~results_to in
|
|
Answer.init answer answer_promise;
|
|
Answers.set t.answers aid answer;
|
|
let target =
|
|
match message_target with
|
|
| `ReceiverHosted id ->
|
|
let export = Exports.find_exn t.exports id in
|
|
begin match Export.resolve_target export with
|
|
| #message_target_cap as target -> target
|
|
| `Local
|
|
| `None -> `Local (with_inc_ref (Export.service export))
|
|
end
|
|
| `ReceiverAnswer (id, path) ->
|
|
let answer = Answers.find_exn t.answers id in
|
|
match Answer.resolve_target answer path with
|
|
| Some (Ok (#message_target_cap as target)) -> target
|
|
| Some (Ok (`Local | `None))
|
|
| Some (Error _)
|
|
| None ->
|
|
(* Unresolved, error or local - queue on our promise *)
|
|
match Answer.answer_struct answer with
|
|
| `Finished -> failwith "Call to finished answer (shouldn't be in table!)"
|
|
| `Promise answer_promise -> `Local (answer_promise#cap path)
|
|
in
|
|
let caps = RO_array.map (import t) descs in
|
|
let msg = Core_types.Request_payload.with_caps caps msg in
|
|
match target with
|
|
| `Local target ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Handling call: (%t).call %a"
|
|
target#pp Core_types.Request_payload.pp msg);
|
|
t.fork (fun () ->
|
|
target#call answer_resolver msg; (* Takes ownership of [caps]. *)
|
|
dec_ref target
|
|
)
|
|
| #message_target_cap as target ->
|
|
Log.debug (fun f -> f ~tags:t.tags "Forwarding call: (%a).call %a"
|
|
pp_message_target_cap target Core_types.Request_payload.pp msg);
|
|
send_call t target answer_resolver msg
|
|
|
|
let bootstrap t id object_id =
|
|
let promise, answer_resolver = Local_struct_promise.make () in
|
|
let answer = Answer.create id ~answer:promise in
|
|
Answers.set t.answers id answer;
|
|
t.fork @@ fun () ->
|
|
object_id |> t.restore @@ fun service ->
|
|
if Answer.needs_return answer && t.disconnected = None then (
|
|
let results =
|
|
match service with
|
|
| Error ex -> Error (`Exception ex)
|
|
| Ok service ->
|
|
let msg =
|
|
Wire.Response.bootstrap ()
|
|
|> Core_types.Response_payload.with_caps (RO_array.of_list [service])
|
|
in
|
|
Ok msg
|
|
in
|
|
Core_types.resolve_payload answer_resolver results;
|
|
Send.return t answer results
|
|
) else (
|
|
Result.iter dec_ref service
|
|
)
|
|
|
|
let return_results t question msg descrs =
|
|
let caps_used = Question.paths_used question |> caps_used ~msg in
|
|
let import_with_embargoes cap_index d =
|
|
let embargo_path =
|
|
match IntMap.find_opt cap_index caps_used with
|
|
| None -> None
|
|
| Some path -> Some (Question.message_target question path)
|
|
in
|
|
import t d ?embargo_path
|
|
in
|
|
RO_array.mapi import_with_embargoes descrs
|
|
|
|
(* Turn wire descriptors into local objects, creating embargoes and sending
|
|
disembargo messages where necessary. *)
|
|
let import_return_caps t question ret =
|
|
match ret with
|
|
| `Results (msg, descs) ->
|
|
let caps = return_results t question msg descs in
|
|
`Results (Core_types.Response_payload.with_caps caps msg)
|
|
| #Error.t as err -> err
|
|
| `ResultsSentElsewhere -> `ResultsSentElsewhere
|
|
| `AcceptFromThirdParty -> failwith "todo: AcceptFromThirdParty"
|
|
| `TakeFromOtherQuestion aid ->
|
|
match Answer.answer_struct (Answers.find_exn t.answers aid) with
|
|
| `Finished -> Fmt.failwith "Can't take from answer %a - it's already finished!" AnswerId.pp aid
|
|
| `Promise other ->
|
|
match question.remote_promise with
|
|
| `Released -> `TakeFromCancelledQuestion
|
|
| `Resolver remote_promise ->
|
|
inc_ref other;
|
|
let paths_used = question.Question.pipelined_fields in
|
|
if PathSet.is_empty paths_used then `TakeFromOtherQuestion (other, remote_promise)
|
|
else (
|
|
(* Embargoes needed for paths that were used when we got the original return. *)
|
|
Question.inc_ref question;
|
|
`TakeFromOtherQuestionAndEmbargo (other, remote_promise, paths_used) (* Note: ref's question *)
|
|
)
|
|
|
|
let return t qid ret ~release_param_caps =
|
|
let question = Questions.find_exn t.questions qid in
|
|
if release_param_caps then List.iter (release t ~ref_count:1) question.params_for_release;
|
|
let ret2 = import_return_caps t question ret in
|
|
(* Any disembargo requests have now been sent, so we may finish the question. *)
|
|
let get_import id =
|
|
let i = Imports.find_exn t.imports id in
|
|
Import.inc_ref i;
|
|
i
|
|
in
|
|
let disembargo_info = Question.extract_resolution ~get_import ret in
|
|
Question.return question disembargo_info |> apply_question_actions t question;
|
|
begin match ret2 with
|
|
| `Results msg ->
|
|
Log.debug (fun f -> f ~tags:(with_qid qid t) "Got results: %a"
|
|
Core_types.Response_payload.pp msg
|
|
);
|
|
Question.resolve question (Ok msg)
|
|
| `TakeFromCancelledQuestion -> () (* Ignore response to cancelled Question *)
|
|
| `TakeFromOtherQuestion (other, remote_promise) ->
|
|
(* (we have added a ref on [other]) *)
|
|
remote_promise#resolve (other :> Core_types.struct_ref)
|
|
| `TakeFromOtherQuestionAndEmbargo (other, remote_promise, paths_used) ->
|
|
(* (we have added a ref on [other] and on [question]) *)
|
|
(* Embargoes needed for paths that were used when we got the original return. *)
|
|
(* We must resolve the remote promise to a local embargo now, so nothing else
|
|
gets sent to it. *)
|
|
let embargo, embargo_resolver = Local_struct_promise.make () in
|
|
begin match embargo_resolver#set_blocker (other :> Core_types.base_ref) with
|
|
| Error `Cycle -> assert false (* Can't happen; we just made it! *)
|
|
| Ok () ->
|
|
remote_promise#resolve embargo;
|
|
other#when_resolved (fun payload ->
|
|
(* XXX: Do we need a when_more_resolved here instead?
|
|
We might need multiple disembargoes as things resolve further... *)
|
|
embargo_resolver#clear_blocker;
|
|
begin match payload with
|
|
| Error _ as e ->
|
|
Core_types.resolve_payload embargo_resolver e;
|
|
| Ok msg ->
|
|
let embargoes_needed = caps_used ~msg paths_used in
|
|
let maybe_embargo cap_index cap =
|
|
inc_ref cap;
|
|
match IntMap.find_opt cap_index embargoes_needed with
|
|
| None -> cap
|
|
| Some path ->
|
|
let old_path = Question.message_target question path in
|
|
local_embargo t ~old_path cap
|
|
in
|
|
let caps = Core_types.Response_payload.snapshot_caps msg |> RO_array.mapi maybe_embargo in
|
|
Core_types.resolve_ok embargo_resolver
|
|
(Core_types.Response_payload.with_caps caps msg)
|
|
end;
|
|
dec_ref other;
|
|
Question.dec_ref question |> apply_question_actions t question
|
|
)
|
|
end;
|
|
| `ResultsSentElsewhere ->
|
|
(* Keep unresolved; we must continue forwarding *)
|
|
()
|
|
| #Error.t as err -> Question.resolve question (Error err)
|
|
end
|
|
|
|
let finish t aid ~release_result_caps =
|
|
let answer = Answers.find_exn t.answers aid in
|
|
Log.debug (fun f -> f ~tags:(with_aid aid t) "Received finish for %a" Answer.dump answer);
|
|
Answers.release t.answers aid;
|
|
Answer.finish answer ~release_result_caps |> apply_answer_actions t answer
|
|
|
|
let send_disembargo t embargo_id target =
|
|
let desc =
|
|
match target with
|
|
| `None -> Fmt.failwith "Protocol error: disembargo request for None cap"
|
|
| `Local -> Fmt.failwith "Protocol error: disembargo request for local target"
|
|
| `QuestionCap (question, path) -> Question.message_target question path
|
|
| `Import import -> Import.message_target import
|
|
in
|
|
Log.debug (fun f -> f ~tags:t.tags "Sending disembargo response to %a" EP.Out.pp_desc desc);
|
|
t.queue_send (`Disembargo_reply (desc, embargo_id))
|
|
|
|
(* If we're trying to disembargo something that resolved to an import, try to disembargo on
|
|
that instead. *)
|
|
let rec disembargo_imports = function
|
|
| None | Some #EP.Out.message_target as x -> x
|
|
| Some (`Import i) -> disembargo_imports (Import.disembargo_target i)
|
|
|
|
let disembargo_request t request =
|
|
Log.debug (fun f -> f ~tags:t.tags "Received disembargo request %a" EP.In.pp_disembargo_request request);
|
|
match request with
|
|
| `Loopback (old_path, embargo_id) ->
|
|
match old_path with
|
|
| `ReceiverHosted eid ->
|
|
send_disembargo t embargo_id (Exports.find_exn t.exports eid |> Export.resolve_target);
|
|
| `ReceiverAnswer (aid, path) ->
|
|
let answer = Answers.find_exn t.answers aid in
|
|
send_disembargo t embargo_id (Answer.disembargo_target answer path)
|
|
|
|
(* [double_disembargo_path resolve_target] is used to check whether we need a second embargo
|
|
after a previous disembargo arrived back at an answer or export which is now resolved to
|
|
[resolve_target]. If [resolve_target] is set and is remote and we have pipelined messages
|
|
to it then we return the path to use for the next disembargo step.
|
|
Returns [None] if no embargo is needed at the moment. *)
|
|
let double_disembargo_path = function
|
|
| `None ->
|
|
(* We haven't resolved yet. Any pipelined messages have been queued up locally and will
|
|
be forwarded when the answer or export does resolve. We can safely make calls directly
|
|
on the local promise. *)
|
|
None
|
|
| `Local ->
|
|
(* We resolved to a local service (possibly another promise).
|
|
Any pipelined messages have already been delivered to it and we can therefore send it
|
|
messages immediately. *)
|
|
None
|
|
| `QuestionCap (q, path) ->
|
|
(* We resolved to a question. If we have pipelined messages to the question and it has
|
|
resolved to a local service then we may need another embargo before using the new
|
|
local target. *)
|
|
Question.answer_cap_disembargo q path |> disembargo_imports
|
|
| `Import i ->
|
|
(* We resolved to an import. If we have pipelined messages to the remote export and it
|
|
has resolved to a local service then we will need another embargo before using the new
|
|
local target. *)
|
|
if Import.used i then disembargo_imports (Import.disembargo_target i)
|
|
else None
|
|
|
|
let disembargo_reply t target embargo_id =
|
|
let embargo = snd (Embargoes.find_exn t.embargoes embargo_id) in
|
|
Log.debug (fun f -> f ~tags:t.tags "Received disembargo response %a -> %t"
|
|
EP.In.pp_desc target
|
|
embargo#pp);
|
|
(* A remote export or answer resolved to [target] at some point in the
|
|
past and we sent a disembargo request to it to flush any pipelined
|
|
messages. This is now done - all pipelined messages that we care about
|
|
have been delivered to [target]. However, [target] may have forwarded
|
|
them on in turn, so another disembargo may be called for.
|
|
|
|
Release the old embargo, so that the cap now points at [target]'s
|
|
initial resolution. If [target] is already resolved, attempt to update
|
|
to its resolution, performing another embargo if necessary.
|
|
*)
|
|
let embargo_path =
|
|
match target with
|
|
| `ReceiverHosted id ->
|
|
let export = Exports.find_exn t.exports id in
|
|
double_disembargo_path (Export.resolve_target export)
|
|
| `ReceiverAnswer (id, path) ->
|
|
let answer = Answers.find_exn t.answers id in
|
|
match Answer.resolve_target answer path with
|
|
| None -> None (* No embargoes as still unresolved (target still remote) *)
|
|
| Some (Error _) -> None (* No embargoes for errors *)
|
|
| Some (Ok resolve_target) ->
|
|
double_disembargo_path resolve_target
|
|
in
|
|
let cap = import t ?embargo_path (target :> EP.In.desc) in
|
|
Log.debug (fun f -> f "Disembargo target is %t" cap#pp);
|
|
Embargoes.release t.embargoes embargo_id;
|
|
embargo#resolve cap;
|
|
dec_ref embargo
|
|
|
|
let resolve t import_id new_target =
|
|
Log.debug (fun f -> f ~tags:t.tags "Received resolve of import %a to %a"
|
|
ImportId.pp import_id
|
|
(Fmt.result ~ok:In.pp_desc ~error:Exception.pp) new_target
|
|
);
|
|
let import_new_target ~embargo_path =
|
|
match new_target with
|
|
| Error e -> Core_types.broken_cap e
|
|
| Ok desc -> import t desc ?embargo_path
|
|
in
|
|
match Imports.find t.imports import_id with
|
|
| None ->
|
|
let new_target = import_new_target ~embargo_path:None in
|
|
Log.debug (fun f -> f ~tags:t.tags "Import %a no longer used - releasing new resolve target %t"
|
|
ImportId.pp import_id new_target#pp);
|
|
dec_ref new_target
|
|
| Some im ->
|
|
(* Check we're not resolving a settled import. *)
|
|
if im.Import.settled then (
|
|
let new_target = import_new_target ~embargo_path:None in
|
|
let msg = Fmt.str "Got a Resolve (to %t) for settled import %a!" new_target#pp Import.dump im in
|
|
dec_ref new_target;
|
|
failwith msg
|
|
);
|
|
let get_import id =
|
|
let i = Imports.find_exn t.imports id in
|
|
Import.inc_ref i;
|
|
i
|
|
in
|
|
match Import.get_proxy im with
|
|
| Some x ->
|
|
(* This will also dec_ref the old remote-promise and the import. *)
|
|
let target = import_new_target ~embargo_path:(Import.embargo_path im) in
|
|
Import.mark_resolved im ~get_import new_target;
|
|
x#resolve target
|
|
| None ->
|
|
(* If we get here:
|
|
- The user released the switchable, but
|
|
- Some [resolve_target] kept the import in the table. *)
|
|
let target = import_new_target ~embargo_path:None in
|
|
Import.mark_resolved im ~get_import new_target;
|
|
Log.debug (fun f -> f ~tags:t.tags "Ignoring resolve of import %a, which we no longer need (to %t)"
|
|
ImportId.pp import_id target#pp);
|
|
dec_ref target
|
|
|
|
end
|
|
|
|
let handle_unimplemented t (msg : Out.t) =
|
|
match msg with
|
|
| `Resolve (_, Error _) -> ()
|
|
| `Resolve (_, Ok new_target) ->
|
|
(* If the peer doesn't understand resolve messages, we can just release target. *)
|
|
begin match new_target with
|
|
| `None
|
|
| `ReceiverHosted _
|
|
| `ReceiverAnswer _ -> ()
|
|
| `SenderHosted id
|
|
| `SenderPromise id
|
|
| `ThirdPartyHosted (_, id) -> Input.release t id ~ref_count:1
|
|
end
|
|
| `Bootstrap (qid, _) ->
|
|
(* If the peer didn't understand our question, pretend it returned an exception. *)
|
|
Input.return t qid ~release_param_caps:true
|
|
(Error.exn ~ty:`Unimplemented "Bootstrap message not implemented by peer")
|
|
| `Call (qid, _, _, _, _) ->
|
|
(* This could happen if we asked for the bootstrap object from a peer that doesn't
|
|
offer any services, and then tried to pipeline on the result. *)
|
|
Input.return t qid ~release_param_caps:true
|
|
(Error.exn ~ty:`Unimplemented "Call message not implemented by peer!")
|
|
| _ ->
|
|
failwith "Protocol error: peer unexpectedly responded with Unimplemented"
|
|
|
|
let disconnect t ex =
|
|
if t.disconnected = None then (
|
|
t.disconnected <- Some ex;
|
|
t.queue_send <- ignore;
|
|
Exports.drop_all t.exports (fun _ -> Export.lost_connection);
|
|
Hashtbl.clear t.exported_caps;
|
|
Questions.drop_all t.questions (fun _ -> Question.lost_connection ~ex);
|
|
Answers.drop_all t.answers (fun _ a -> Answer.lost_connection a |> apply_answer_actions t a);
|
|
let broken_cap = Core_types.broken_cap ex in
|
|
Imports.drop_all t.imports (fun _ -> Import.lost_connection ~broken_cap);
|
|
Embargoes.drop_all t.embargoes (fun _ (_, e) -> e#break ex; dec_ref e)
|
|
)
|
|
|
|
let handle_msg t (msg : [<In.t | `Unimplemented of Out.t]) =
|
|
check_connected t;
|
|
match msg with
|
|
| `Abort ex -> disconnect t ex
|
|
| `Call (aid, target,
|
|
msg, descs, results_to) -> Input.call t aid target msg descs ~results_to
|
|
| `Bootstrap (qid, oid) -> Input.bootstrap t qid oid
|
|
| `Return (aid, ret, release) -> Input.return t aid ret ~release_param_caps:release
|
|
| `Finish (aid, release) -> Input.finish t aid ~release_result_caps:release
|
|
| `Release (id, count) -> Input.release t id ~ref_count:count
|
|
| `Disembargo_request req -> Input.disembargo_request t req
|
|
| `Disembargo_reply (target, id) -> Input.disembargo_reply t target id
|
|
| `Resolve (id, target) -> Input.resolve t id target
|
|
| `Unimplemented x -> handle_unimplemented t x
|
|
|
|
let dump_embargo f (id, proxy) =
|
|
Fmt.pf f "%a: @[%t@]" EmbargoId.pp id proxy#pp
|
|
|
|
let check_embargo x = (snd x)#check_invariants
|
|
|
|
let check_exported_cap t cap export_id =
|
|
match Exports.find_exn t.exports export_id with
|
|
| export ->
|
|
if Export.service export <> cap then (
|
|
Debug.invariant_broken @@ fun f ->
|
|
Fmt.pf f "export_caps maps %t to export %a back to different cap %a!"
|
|
cap#pp Export.pp export Export.dump export
|
|
)
|
|
| exception ex ->
|
|
Debug.invariant_broken @@ fun f ->
|
|
Fmt.pf f "exported_caps for %t: %a" cap#pp Debug.pp_exn ex
|
|
|
|
let exported_sort_key export_id = export_id
|
|
|
|
let pp_exported_cap t f (cap, export_id) =
|
|
Fmt.pf f "%t => export %a%a" cap#pp ExportId.pp export_id (pp_check (check_exported_cap t)) (cap, export_id)
|
|
|
|
let dump f t =
|
|
match t.disconnected with
|
|
| Some reason -> Fmt.pf f "Disconnected: %a" Exception.pp reason
|
|
| None ->
|
|
Fmt.pf f "@[<v2>Questions:@,%a@]@,\
|
|
@[<v2>Answers:@,%a@]@,\
|
|
@[<v2>Exports:@,%a@]@,\
|
|
@[<v2>Imports:@,%a@]@,\
|
|
@[<v2>Embargoes:@,%a@]@,\
|
|
@[<v2>Exported caps:@,%a@]@,"
|
|
(Questions.dump ~check:Question.check Question.dump) t.questions
|
|
(Answers.dump ~check:Answer.check Answer.dump) t.answers
|
|
(Exports.dump ~check:Export.check Export.dump) t.exports
|
|
(Imports.dump ~check:Import.check Import.dump) t.imports
|
|
(Embargoes.dump ~check:check_embargo dump_embargo) t.embargoes
|
|
(hashtbl_dump ~key:exported_sort_key (pp_exported_cap t)) t.exported_caps
|
|
|
|
let check t =
|
|
Questions.iter (fun _ -> Question.check) t.questions;
|
|
Answers.iter (fun _ -> Answer.check) t.answers;
|
|
Imports.iter (fun _ -> Import.check) t.imports;
|
|
Exports.iter (fun _ -> Export.check) t.exports;
|
|
Embargoes.iter (fun _ -> check_embargo) t.embargoes;
|
|
Hashtbl.iter (check_exported_cap t) t.exported_caps
|
|
end
|