client_context.hpp

Go to the documentation of this file.

/* Flow-IPC: Shared Memory
 * Copyright 2023 Akamai Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in
 * compliance with the License.  You may obtain a copy
 * of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in
 * writing, software distributed under the License is
 * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
 * CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing
 * permissions and limitations under the License. */
 
/// @file
#pragma once
 
#include "ipc/transport/struc/shm/rpc/session_vat_network.hpp"
#include "ipc/transport/asio_local_stream_socket.hpp"
#include "ipc/session/app.hpp"
#include "ipc/common.hpp"
#include <flow/log/log.hpp>
#include <boost/make_shared.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/future.hpp>
#include <optional>
#include <algorithm>
 
namespace ipc::transport::struc::shm::rpc
{
 
// Types.
 
/**
 * An object of this type facilitates the establishment, in client style, of a zero-copy-enabled capnp-RPC
 * conversation with an opposing process's Context_server.  On successful connect, it represents one such
 * conversation until its destruction; create another `*this` to begin another such conversation.
 *
 * @see ipc::transport::struc::shm::rpc namespace doc header for an overview of Flow-IPC capnp-RPC support, including
 *      how the present class relates to the others there.
 *
 * Before a successful connection (sync_connect()) a `*this` is said to be in NULL state.  After it and until
 * destruction it is said to be in PEER state.
 *
 * Once a `*this` is in a PEER state (on successful sync_connect()), its capabilities/semantics/API are exactly
 * identical to those of Server_context, objects of which type are generated by an opposing process's Context_server
 * object.  Thus the following relationships are analogous (if you're not familiar with Flow-IPC at large you may
 * disregard):
 *   - In zero-copy-enabled-RPC land:
 *     Client_context (in PEER state) - Context_server - Server_context (always in PEER state).
 *   - In general Flow-IPC land:
 *     session::Client_session (in PEER state) - session::Session_server - session::Server_session (in PEER state).
 *
 * ### Background ###
 * (This description will go down easier if you are familiar with capnp-RPC basics.)
 *
 * As introduced in ipc::transport::struc::shm::rpc namespace doc header, Flow-IPC's support of zero-copy RPC
 * centers on its implementation of the `capnp::VatNetwork` interface: Session_vat_network.  Consider its
 * non-zero-copy cousin shipped with capnp, `TwoPartyVatNetwork`.  To use it -- in the absence of simplifying
 * all-in-one helpers like `capnp::EzRpcClient` -- the following essential steps are usually involved,
 * on the *client end*:
 *   -# Connect a low-level transport stream (for local IPC, a Unix domain socket stream) to a listening server
 *      at some address (in Linux typically an abstract address; or more generally a file-addressed socket).
 *   -# Create a `TwoPartyVatNetwork`, giving it the stream endpoint (e.g., as a socket handle; from step 1)
 *      which it owns from that point on (up to and including closing it).
 *   -# Create an `RpcSystem`, passing it a reference to the `VatNetwork` from step 2; then perform RPC
 *      in various ways.  (Typically, if one connected as a client, they would also obtain an interface impl
 *      from the other side and go from there; but this can be reversed... various possibilities exist.)
 *   -# Once either this side decides to be done, or the other side closes the conversation (the latter might
 *      be detected via a `Promise` returned by `TwoPartyVatNetwork::onDisconnect()`):
 *      - Deinit in the opposite order: `RpcSystem`, `VatNetwork` (which now owns the stream).
 *
 * ### How to use Client_context ###
 * A Client_context object facilitates a very similar (almost identical) workflow.  The following points apply:
 *   - To specify the target opposing server, use the ipc::session::App system.  Don't worry: it is straightforward,
 *     yet it side-steps certain problems of portability/lack of abstraction and safety present when working with
 *     native IPC OS primitives directly.  In any case you'll setup a session::Client_app to ID this application; and
 *     session::Server_app to ID the opposing application.  (You'll give the same to Context_server on the
 *     opposing side.)
 *   - Client_context::sync_connect() will perform both steps 1 and 2 above...
 *     - ...except the `VatNetwork` -- which you should obtain via the vat_network() accessor of `*this` --
 *       is Session_vat_network.  That is: it will avoid all copying of underlying messages, meaning they can
 *       effectively be of any size at all with no difference in performance/latency.  Internally it will use
 *       SHared Memory (SHM) supplied by Flow-IPC.
 *   - You specify the SHM-provider to be used (behind the scenes!) via your choice of template parameter
 *     `Client_session_t`.  As of this writing the following values are available out of the box.
 *     - (SHM-classic) session::shm::classic::rpc::Client_session.  A good default choice.
 *     - (SHM-jemalloc) session::shm::arena_lend::jemalloc::rpc::Client_session.  For more advanced considerations
 *       particularly in terms of safety (e.g., each side holds its own segregated SHM arena, and the other side
 *       may -- at a kernel level -- only read from it).  There are also perf implications (the commercial-grade
 *       memory-manager, jemalloc, is used for individual allocations).
 *     - Please see the guided Manual and/or in-code doc headers such as for namespace ipc::shm.  99.9999% of this
 *       information will have no bearing on your use of capnp-RPC, semantically, but it may eventually be
 *       significant in terms of perf and safety considerations in production code.
 *
 * The `-> stream` feature (https://capnproto.org/news/#multi-stream-flow-control) has a potential impact on
 * shared RAM (SHM) use, and possibly on performance.  The way the related *flow control window* is determined
 * differs from vanilla (non-zero-copy) capnp-RPC's impl (as described in that link).
 *   - You may control and/or observe the relevant window knob via `this->vat_network()->streaming_flow_window_ki()`.
 *     See Session_vat_network::streaming_flow_window_ki() for further info.
 *
 * ### For Flow-IPC-savvy users / Direct SHM use ###
 * The above was written with a slant toward those who are familiar with capnp-RPC but may know next to nothing
 * about Flow-IPC at large.  However Client_context does provide access to the rest of Flow-IPC by exposing
 * most (not quite all) features of ipc::session::Session.  The latter is a (the!) gateway to the rest
 * of Flow-IPC which is full-featured.  To wit:
 *   - You may obtain the Flow-IPC `Session` object, after a successful sync_connect().  Call it `S`.
 *     - `S->session_shm()` and `S->app_shm()` give you direct access to SHM arenas, where you can directly
 *       construct C++ native objects (including arbitrary STL-compliant containers-of-containers-of-...),
 *       then transmit them to the opposing process and more.  (See ipc::shm::stl namespace doc header.)
 *       - As you can see in other docs, `S->lend_object()` and `S->borrow_object()` enable sharing native SHM-stored
 *         objects between processes.  To transmit the handle blobs returned by `->lend_object()`, you can use
 *         your shiny capnp-RPC session and/or any additional `Channel`s you opened via sync_connect()
 *         (or any other technique for that matter... but why would you need even more?!).
 *   - You may open direct-use channels (ipc::transport::Channel) via the optional args to sync_connect().
 *     These will be available immediately.  You can then use them directly (for low-level transport work),
 *     upgrade them to ipc::transport::struc::Channel for a simplified capnp-message exchange, and so on.
 *     See the relevant documentation.
 *     - By choosing the template parameter (which has a sensible default value also) for
 *       `"ipc::session::shm::*::Client_session"` (in setting our template-arg `Client_session_t`), you may make
 *       use of several transport mechanisms including (as of this writing) POSIX MQs, boost.interprocess (in-SHM)
 *       MQs, and socket-streams (of bytes and/or native-handles/FDs).
 *
 * Some features of ipc::session::Session are not usable through a Client_context.
 *   - The capnp-metadata exchange (ipc::session::Session::mdt_builder() et al) at session-open is not available.
 *     This would have complicated the API, and we figure the capnp-RPC layer is more than sufficient for any
 *     negotiation or information exchange you could possibly desire!
 *   - While you can call `S->open_channel()` (which would open a new `Channel` *after* the session is connected),
 *     it will not work, as there is no way to specify a passive-channel-open handler.  This would have, again,
 *     complicated the API -- and we'd want to perhaps make it available in a KJ-style way (promises, promises).
 *     - This is a possibility for future work.  However (1) capnp-RPC is available, even for native-handle exchange
 *       if desired ("capability passing" -- see capnp-RPC's `getFd()`); and (2) even for vanilla Flow-IPC work
 *       most prefer to use the at-session-open channels (which are available in Client_context), so open_channel()
 *       et al are in the first place an advanced feature.
 *
 * @internal
 *
 * @todo There is a tiny protocol involved between rpc::Client_context and rpc::Context_server, but (internally)
 * we lack a util::Protocol_negotiator; arguably we should add it, lest there is compatibility issue should we need
 * to modify this protocol.  The protocol is simple: the agreement that the first client-requested init-channel
 * is actually for capnp-RPC purposes and not to be emitted to the `sync_connect()`/`async_accept()` invoker.
 * Instead the channel is for this: Client sends one native handle (half of a socket-pair used for the capnp-RPC
 * conversation by the `Session_vat_network`) over it.  Having received it the server closes the channel.  On
 * the client side, here, `Client_context` retains its side of the channel indefinitely (reason: see
 * sync_connect_impl() body; but basically because it does not know when opposing server has actually consumed
 * the socket-pair). / Without `Protocol_negotiator`, which admittedly would complicate the impl annoyingly,
 * needing to change this protocol in the future (such as to avoid keeping the aforementioned channel open
 * indefinitely) is not a very civilized proposition.  To deal with this: Just in case, for now, we've thrown in
 * a reserved value along with the native handle.  Alternatively one can always increment an
 * ipc::session `Protocol_negotiator` version (both `local_(min|max)_proto_ver` <=> no compatibility with older
 * protocol), if a change to this subsequent RPC-establishment protocol is required.  After all both this module and
 * ipc::session are parts of Flow-IPC.
 *
 * @endinternal
 *
 * @tparam Client_session_t
 *         A concrete type a-la session::Client_session but SHM-enabled.  As of this writing
 *         out of the box the available types are listed above under: How to use Client_context.
 *         Naturally, the opposing Context_server must be parameterized in a compatible fashion.
 *         (E.g., `Client_context<ipc::session::shm::classic::Client_session<...knobs...>>` <=>
 *                `Context_server<ipc::session::shm::classic::Session_server<...same-knobs...>>`.)
 */
template<typename Client_session_t>
class Client_context :
  public flow::log::Log_context,
  private boost::noncopyable
{
public:
  // Types.
 
  /// Convenience alias for template arg.
  using Session_obj = Client_session_t;
  static_assert(Session_obj::S_SHM_ENABLED,
                "Client_context facilitates the use of Flow-IPC zero-copy (SHM) facilities for "
                  "zero-copy performance in using capnp-RPC; therefore the supplied Client_session_t template "
                  "parameter must be a SHM-enabled Client_session variant, session::shm::*::Client_session*.");
  static_assert(Session_obj::Channel_obj::S_HAS_NATIVE_HANDLE_PIPE,
                "Client_context, per its protocol with the opposing Context_server, sends a native handle -- "
                  "the stream-socket underlying the capnp-RPC conversation -- over the first init-channel; "
                  "therefore the supplied Client_session_t must be configured to emit Native_handle-bearing "
                  "channels (e.g., see aliases in session::shm::*::rpc namespaces).");
 
  /**
   * The Session_vat_network concrete type -- and `capnp::VatNetwork` interface impl -- that we produce.
   * See Session_vat_network docs; but in short, generally, once constructed this guy is used ~identically
   * to `capnp::TwoPartyVatNetwork`.
   */
  using Vat_network = typename Session_obj::Vat_network;
 
  // Constructors/destructor.
 
  /**
   * Constructs us without establishing the capnp-RPC session.  Use sync_connect() to establish it.
   *
   * @warning `cli_app_ref` and `srv_app_ref` must remain alive past this ctor: their *addresses*
   *          are stored and accessed at various points later.  As usual, unless null, same for `*logger_ptr`.
   *          **The required lifetime of these objects may exceed that of the constructed `*this`.**
   *          Therefore:
   * @see Please read Session_vat_network doc header section "Object lifetimes."  It explains how/why and more
   *      importantly how long these objects must stay alive.  Promise: it's pretty easy.  (Hard part is to not
   *      assume they just need to outlive `*this`.  The design of capnp-RPC/KJ is the reason for this.)
   * @note Spoiler alert: An easy approach is to keep `cli/srv_app_ref` and `*logger_ptr` objects alive past
   *       `*kj_io`.
   *
   * @param logger_ptr
   *        Logger to use for logging subsequently.  (You may use null to forego this completely.)
   *        See lifetime warning above.
   * @param kj_io
   *        A KJ event loop context.
   * @param cli_app_ref
   *        Properties of this client application.  See lifetime warning above.
   * @param srv_app_ref
   *        Properties of the opposing server application.  See lifetime warning above.
   * @param enable_hndl_transport
   *        `true` means native handles (a/k/a capabilities in capnp-RPC parlance) can
   *        be RPC-transmitted (i.e., your interface impls and clients can use `.getFd()` and yield something
   *        as opposed to nothing).  `false` means the opposite; in particular any native-handles arriving via
   *        capnp-RPC (as enabled via the #Vat_network returned by vat_network()) shall be disregarded.
   *        If you do not use `.getFd()` feature of the capnp-RPC system, it is best to set this to `false`,
   *        it is said, for safety and possibly even security.
   */
  explicit Client_context(flow::log::Logger* logger_ptr, kj::AsyncIoContext* kj_io,
                          const session::Client_app& cli_app_ref, const session::Server_app& srv_app_ref,
                          bool enable_hndl_transport = true);
 
  /**
   * Destructor; in particular (assuming sync_connect() succeeded) shuts down the #Vat_network and the
   * #Session_obj, in that order.  You must destroy any related #Rpc_system before invoking this dtor.
   * If you have any directly-obtained (via `->construct<T>()` or `->borrow_object<T>()`) SHM-handles from
   * `*this`, you must have nullified them before invoking this dtor.
   *
   * (Internally the #Session_obj may outlive this dtor briefly -- until the KJ event loop has disposed of any
   * message capnp-RPC has left in it -- but by then it is nobody's concern but its own.  See ctor doc header.)
   *
   * Informally we recommend, also, that once you disconnect your capnp-RPC conversation, or
   * the other side does so -- this can be detected via `vat_network()->on_disconnect()`-returned `kj::Promise` --
   * you destroy `*this` sans delay.  (Do not worry: there is no race condition.  But we digress.)
   */
  ~Client_context();
 
  // Methods.
 
  /**
   * To be invoked successfully at most once, it synchronously and non-blockingly attempts to
   * connect to an opposing `Context_server`; and synchronously reports failure or success, the latter
   * showing `*this` has entered PEER state.  Failure means `*this` remains in NULL state.
   *
   * Note that `*this` that has entered PEER state can never change state subsequently
   * (even on transmission error); once a PEER, always a PEER.
   *
   * Essentially this convenience operation has one main aspect:
   *   - It bundles together the initiation of a Flow-IPC session::Session and the creation of a `capnp::VatNetwork`
   *     (similar to a vanilla `capnp::TwoPartyVatNetwork` but zero-copy-enabled using Flow-IPC SHM resources from
   *     said `Session`).  So: one immediate op; and then you can make an `RpcSystem` and go capnp-RPC-ing happily; and
   *     (optionally) make use of certain Flow-IPC `Session` features as well (see below and class doc header).
   *
   * @note Upon entering PEER state (i.e., on success of sync_connect()): knobs of interest are available
   *       through vat_network() method which gives access to the core Session_vat_network.
   *
   * ### How does it know "where" the Session_server is listening? / Why non-blocking and synchronous? ###
   * Please see session::Client_session::sync_connect() doc header's similar section.  It is best to be aware
   * of this as background.
   *
   * ### Error conditions ###
   * If invoked already in PEER state, this returns `false` and otherwise does nothing.  The following assumes
   * otherwise (that it was invoked properly in NULL state).  Then:
   *
   * On success returns `true`.  On connect fail, throws an exception:
   *   - `flow::error::Runtime_error` (which is an `std::runtime_error` which is an `std::exception`)
   *     if it's from originating in Flow-IPC proper;
   *   - `kj::Exception` if from capnp internals;
   *   - generic `std::exception` and `...` might also be possible.
   *
   * In case of `Runtime_error`: The `Error_code` may be accessed via `.code()` (also `.code().message()`,
   * `.what()`) of the exception object.  #Error_code generated:
   *   - interprocess-mutex-related errors (probably from boost.interprocess) w/r/t reading the CNS (PID file),
   *     file-related system errors w/r/t reading the CNS (PID file) (see Session_server doc header for background),
   *   - error::Code::S_CLIENT_NAMESPACE_STORE_BAD_FORMAT (bad CNS contents),
   *     those emitted by transport::Native_socket_stream::sync_connect(),
   *     those emitted by transport::struc::Channel::send(),
   *     those emitted by transport::struc::Channel via on-error handler (most likely
   *     transport::error::Code::S_RECEIVES_FINISHED_CANNOT_RECEIVE indicating graceful shutdown of opposing process
   *     coincidentally during log-in procedure, prematurely ending session while it was starting),
   *     error::Code::S_CLIENT_MASTER_LOG_IN_RESPONSE_BAD,
   *     error::Code::S_INVALID_ARGUMENT (other side expected other sync_connect() overload with
   *     non-null `init_channels_by_srv_req` arg),
   *     session::error::Code::S_SESSION_OPEN_CHANNEL_SERVER_CANNOT_PROCEED_RESOURCE_UNAVAILABLE (SHM-jemalloc
   *     internal-use socket-connection could not be obtained from OS by opposing side),
   *   - possibly others.
   *
   * You may retry `this->sync_connect()`, if an exception is thrown.  (If an exception is thrown, post-conditions:
   * `*this` remains in NULL state; (if applicable) `*init_channels_by_srv_req` is empty;
   * (if applicable) `*init_channels_by_cli_req_pre_sized` is filled with NULL-state objects, container size
   * unchanged.)
   *
   * @note To comport with capnp-RPC/KJ style code flow, this API does not have a standard Flow-style
   *       optional `Error_code*` out-arg through which to communicate success/errors instead of exceptions if desired;
   *       it shall always fire an exception on error, otherwise return `true`.
   *
   * ### Seriously, why no `async_connect()`?  Isn't it common to perform an async socket-connect operation? ###
   * In a local-IPC context: In short, actually, no: there is no waiting involved.  (Even in a networked context
   * the potential asynchronicity -- waiting -- really comes from the fact that connection establishment with
   * a *listening* server involves networked round trips which are not instant; in our local case it is instant.)
   *
   * There is however a to-do elsewhere (which would propagate here as well) for a relatively exotic
   * wait-until-server-pops-up operation.
   *
   * @param init_channels_by_cli_req_pre_sized
   *        Null or pointer to container of `Session_obj::Channel_obj` with `.size()` specifying how many channels
   *        this side is requesting to be opened on its behalf; each element will be move-assigned
   *        a PEER-state `Channel_obj` on success.  Recommend simply ct-ing with `(n)` or `.resize(n)` which
   *        loads it with default-cted (NULL-state) objects to be replaced.  null is treated same as
   *        `.empty()`.
   * @param init_channels_by_srv_req
   *        Null or pointer to container of `Channel_obj` which shall be `.clear()`ed and replaced
   *        by a container of PEER-state `Channel_obj` on success the number being specified by the opposing
   *        (server) side.  The number may be zero.  null is allowed if and only if the number is zero;
   *        otherwise error::Code::S_INVALID_ARGUMENT is emitted.
   * @return `true` on successful connect; `false` if invoked in PEER state (already connected).
   *         Exception thrown on failure to connect (`.code()` indicating why).
   */
  bool sync_connect(typename Session_obj::Channels* init_channels_by_cli_req_pre_sized = nullptr,
                    typename Session_obj::Channels* init_channels_by_srv_req = nullptr);
 
  /**
   * Identical to sync_connect() with same signature but disables zero-copy transport.
   * That is underlying messages are sent in the mode compatible with vanilla capnp-RPC.
   * This requires some knobs to be set appropriately at the opposing process.  To wit:
   *   - If using Ez_rpc_server: use ctor that takes `transport_method_func` arg; it must return `false`,
   *     if the session::Client_app arg equals the `cli_app_ref` passed to `*this` ctor.
   *   - If using Context_server: use Context_server::accept() that takes `transport_method_func` arg;
   *     it must (...see previous bullet...).
   *   - If using Session_vat_network: use the "special mode" ctor arg (see Session_vat_network ctor doc headers).
   *   - Or use `capnp::VatNetwork` or `capnp::EzRpcServer`.
   *
   * @param init_channels_by_cli_req_pre_sized
   *        See sync_connect().
   * @param init_channels_by_srv_req
   *        See sync_connect().
   * @return See sync_connect().
   */
  bool sync_connect_sans_shm_transport(typename Session_obj::Channels* init_channels_by_cli_req_pre_sized = nullptr,
                                       typename Session_obj::Channels* init_channels_by_srv_req = nullptr);
 
  /**
   * Returns pointer to Session_vat_network established for use in your rpc::Rpc_system; or null if
   * no sync_connect() has succeeded or been invoked at all.  The #Vat_network is valid if and only if
   * `*this` exists.
   *
   * @see `vat_network()->streaming_flow_window_ki()` is a perf-knob of potential interest accessible through here.
   *       See Session_vat_network::streaming_flow_window_ki() doc header(s).
   *
   * See Session_vat_network docs; but in short, generally, once constructed this guy is used ~identically
   * to `capnp::TwoPartyVatNetwork`.
   *
   * @return See above.
   */
  Vat_network* vat_network();
 
  /**
   * Immutable counterpart to the other overload.
   * @return See above.
   */
  const Vat_network* vat_network() const;
 
  /**
   * Returns pointer to #Session_obj, if sync_connect() has succeeded; or null if no sync_connect() has succeeded
   * or been invoked at all.  The #Session_obj is valid if and only if `*this` exists.
   *
   * @see class doc header for overview of which features accessible through this #Session_obj are available
   *      (and which are not).
   *
   * @return See above.
   */
  Session_obj* session();
 
  /**
   * Immutable counterpart to the other overload.
   * @return See above.
   */
  const Session_obj* session() const;
 
private:
  // Types.
 
  /// Basically `shared_ptr<Session_obj>`.
  using Session_ptr = typename Vat_network::Shm_lender_borrower_ptr;
 
  // Methods.
 
  /**
   * Session-hosed handler for #m_session.  This only logs as opposed to reporting to the user of `*this` for
   * reasons explained inside (this is of some importance).  May run after the relevant `*this` dies; hence
   * the needed bits of info are to be captured and passed-in as args.
   *
   * @param logger_ptr
   *        See ctor.
   * @param cli_app
   *        See ctor.
   * @param srv_app
   *        See ctor.
   * @param err_code
   *        From error handler.
   */
  static void on_session_hosed(flow::log::Logger* logger_ptr,
                               const session::Client_app& cli_app, const session::Server_app& srv_app,
                               const Error_code& err_code);
 
  /**
   * Impl of sync_connect() and sync_connect_sans_shm_transport().
   *
   * @param init_channels_by_cli_req_pre_sized
   *        See `sync_connect*()`.
   * @param init_channels_by_srv_req
   *        See `sync_connect*()`.
   * @param sans_shm_transport
   *        Which method are we impl-ing?
   * @return See `sync_connect*()`.
   */
  bool sync_connect_impl(typename Session_obj::Channels* init_channels_by_cli_req_pre_sized,
                         typename Session_obj::Channels* init_channels_by_srv_req,
                         bool sans_shm_transport);
 
  // Data.
 
  /// KJ event loop context.
  kj::AsyncIoContext* const m_kj_io;
 
  /// See ctor.  Stored in case we must re-initialize #m_session, if sync_connect() only partially succeeds.
  const session::Client_app& m_cli_app_ref;
  /// See ctor.  Stored in case we must re-initialize #m_session, if sync_connect() only partially succeeds.
  const session::Server_app& m_srv_app_ref;
 
  /// See ctor.
  const bool m_enable_hndl_transport;
 
  /**
   * Session in NULL state (until sync_connect() succeeds) or PEER state (subsequently).
   *
   * Held via shared handle, as #m_network requires.  Corollary: `*m_session` may outlive `*this`.
   * See Session_vat_network doc header, "Object lifetimes."  That is also why we warn about `Logger` and `App`
   * lifetimes in our ctor doc header.
   */
  Session_ptr m_session;
 
  /**
   * Null until sync_connect() succeeds; then the idle channel being kept up, having been used
   * to convey the capnp-RPC transport handle to the opposing Context_server.  So: synchronously
   * in a successful sync_connect_impl(): grab the first init-channel (do not emit to user); establish
   * a socket-pair connection for capnp-RPC; sync-send half of that pair; and then just keep the
   * `Channel` -- still open, or at least not closed on our account -- around until destruction.
   *
   * ### Rationale: Why keep it open? ###
   * See sync_connect_impl() body for details.  In short: When we destroy the `Channel` (<=> close our-side
   * endpoint), the Context_server may well happen to not have quite gotten to consume the native-handle off
   * the us-to-them half-pipe yet.  It's a bidirectional pipe, and certain technicalities mean that
   * after our-side closing happens, their-side receiving won't succeed; so instead we simply keep it open:
   * they can read it whenever.  The cost is mainly the extra endpoint resource (in practice, a socket).
   * (It *is* possible to make it work, essentially by handling the channel with full minimalism as entirely
   * unidirectional.  Then no need for #m_rpc_setup_channel.  Again, though, for all that see sync_connect_impl().)
   */
  std::optional<typename Session_obj::Channel_obj::Async_io_obj> m_rpc_setup_channel;
 
  /// Null (until sync_connect() succeeds) or the established #Vat_network (subsequently).
  std::optional<Vat_network> m_network;
}; // class Client_context
 
// Template implementations.
 
template<typename Client_session_t>
Client_context<Client_session_t>::Client_context(flow::log::Logger* logger_ptr,
                                                 kj::AsyncIoContext* kj_io,
                                                 const session::Client_app& cli_app_ref,
                                                 const session::Server_app& srv_app_ref,
                                                 bool enable_hndl_transport) :
  flow::log::Log_context(logger_ptr, Log_component::S_RPC),
  m_kj_io(kj_io),
  m_cli_app_ref(cli_app_ref),
  m_srv_app_ref(srv_app_ref),
  m_enable_hndl_transport(enable_hndl_transport),
  m_session(boost::make_shared<Session_obj>(get_logger(), m_cli_app_ref, m_srv_app_ref,
                                            [logger_ptr, &cli_app_ref, &srv_app_ref](const Error_code& err_code)
                                              { on_session_hosed(logger_ptr, cli_app_ref, srv_app_ref, err_code); }))
{
  FLOW_LOG_INFO("rpc::Client_ctx [" << *this << "]: "
                "Created in NULL state (inactive).  Inactive ipc::session::Session [" << *m_session << "] exists.");
}
 
template<typename Client_session_t>
Client_context<Client_session_t>::~Client_context()
{
  FLOW_LOG_INFO("rpc::Client_ctx [" << *this << "]: "
                "Shutting down.  Session_vat_network shall shut down; then the ipc::session::Session "
                "(unless a straggling capnp-RPC message keeps the latter alive a bit longer).");
}
 
template<typename Client_session_t>
void Client_context<Client_session_t>::on_session_hosed(flow::log::Logger* logger_ptr,
                                                        const session::Client_app& cli_app,
                                                        const session::Server_app& srv_app,
                                                        const Error_code& err_code) // Static.
{
  FLOW_LOG_SET_CONTEXT(logger_ptr, Log_component::S_RPC);
  FLOW_LOG_INFO("rpc::Client_ctx-created ipc::session::Session [" << cli_app.m_name << "->" << srv_app.m_name << "]: "
                "session-hosed handler fired (code [" << err_code << "] [" << err_code.message() << "]).  "
                "This is likely normal, as the opposing process decided to end session; the RPC-system will "
                "have detected same, and user RPC session should be ended or ending imminently, at which point "
                "proper user code shall shut-down this Client_ctx which will shut down the Session_vat_network and "
                "lastly the ipc::session::Session.");
 
  /* Why do we merely log but in no way report this to `*this` user?
   * (After all with a vanilla Flow-IPC session we make a big deal of the session-hosed handler one must pass
   * to its ctor.)  Answer: The usefulness of the handler is, outside of emitting a reason error-code (which
   * actually is not *that* interesting typically; simply the other side decided to end the session), to *signal*
   * to the user that they cannot carry on.  Moreover, excepting crashes and the like, this is not a catastrophe:
   * it means the user *should* destroy the session and everything related to it, ideally soon; but behavior
   * remains not ill-defined until then.  (Details omitted; in short for some SHM-types this requires that each
   * of two opposing destructors waits until the other begins executing, before it can execute; for others this is not
   * necessary.)
   *
   * With a general session, users may generate `Channel`s which are independent of the session in terms of their
   * lifetimes -- even though informally it is recommended to follow the best practice of init/de-init order
   * (A up, B up, B down, A down), it is not required.  So formally speaking it makes sense to have a session-end
   * signal separately from channel-end signal(s).  In our case, however, we have a *mandatory* capnp-RPC session
   * we start; really it's the main point of a `*this` after all; and *that* system already provides a natural
   * signal as to the other side being done -- which is customary for capnp-RPC users to use.  That can be something
   * in their own protocol, or they can use vat_network()->on_disconnect() promise.  We indicate this explicitly
   * in the dtor doc header for instance.
   *
   * Therefore supplying a signal through the Flow-IPC session as well is an unnecessary complication of the API --
   * an asynchronous one at that which is particularly annoying.  It becomes ambiguous as to how to handle
   * disconnection.  So let's not make it hard. */
} // Client_context::on_session_hosed()
 
template<typename Client_session_t>
bool Client_context<Client_session_t>::sync_connect
       (typename Session_obj::Channels* init_channels_by_cli_req_pre_sized,
        typename Session_obj::Channels* init_channels_by_srv_req)
{
  return sync_connect_impl(init_channels_by_cli_req_pre_sized, init_channels_by_srv_req, false);
}
 
template<typename Client_session_t>
bool Client_context<Client_session_t>::sync_connect_sans_shm_transport
       (typename Session_obj::Channels* init_channels_by_cli_req_pre_sized,
        typename Session_obj::Channels* init_channels_by_srv_req)
{
  return sync_connect_impl(init_channels_by_cli_req_pre_sized, init_channels_by_srv_req, true);
}
 
template<typename Client_session_t>
bool Client_context<Client_session_t>::sync_connect_impl
       (typename Session_obj::Channels* init_channels_by_cli_req_pre_sized,
        typename Session_obj::Channels* init_channels_by_srv_req,
        bool sans_shm_transport)
{
  using Channels = typename Session_obj::Channels;
  using util::Blob_const;
  namespace asio_local = asio_local_stream_socket::local_ns;
  using asio_local::connect_pair;
  using Peer_socket = asio_local_stream_socket::Peer_socket<asio_local_stream_socket::Protocol_byte_stream>;
  using flow::util::Task_engine;
  using flow::error::Runtime_error;
  using boost::promise;
  using std::string;
  using std::exception;
 
  if (m_network)
  {
    FLOW_LOG_WARNING("rpc::Client_ctx [" << *this << "]: capnp-RPC connect requested, but we are already connected.  "
                     "Bug?  Ignoring.");
    return false;
  }
  // else
 
  /* We request 1 channel for capnp-RPC; plus any they might want for their own purposes.
   * Naturally the Server_context knows this agreement.
   * (See @todo in class doc header regarding protocol negotiation though.) */
  Channels actual_init_channels_by_cli{1 + (init_channels_by_cli_req_pre_sized
                                              ? init_channels_by_cli_req_pre_sized->size()
                                              : 0)};
  FLOW_LOG_INFO("rpc::Client_ctx [" << *this << "]: capnp-RPC connect requested.  Synchronous, non-blocking "
                "ipc::session::Session connect initiating; on success will hook up the "
                "[zero-copy-enabled? = [" << (!sans_shm_transport) << "]] Session_vat_network.");
 
  Native_handle local_hndl;
  Native_handle remote_hndl;
  try
  {
#ifndef NDEBUG
    const bool ok =
#endif
    m_session->sync_connect(m_session->mdt_builder(), &actual_init_channels_by_cli, nullptr,
                            init_channels_by_srv_req);
    assert(ok && "We had a NULL-state Client_session and tried .sync_connect(); why would it have reported otherwise?");
 
    /* .sync_connect() would have thrown on error (as advertised) (probably other guy isn't up; m_network remained
     * null).  Otherwise we carry on: */
 
    if (init_channels_by_cli_req_pre_sized && (!init_channels_by_cli_req_pre_sized->empty()))
    {
      assert(actual_init_channels_by_cli.size() > 1);
      std::move(++actual_init_channels_by_cli.begin(), actual_init_channels_by_cli.end(),
                init_channels_by_cli_req_pre_sized->begin());
      actual_init_channels_by_cli.resize(1);
    }
    assert(actual_init_channels_by_cli.size() == 1);
    auto& rpc_setup_channel = actual_init_channels_by_cli.front();
 
    /* Per our protocol with the opposing Context_server (see also the class doc header @todo regarding
     * Protocol_negotiator), rpc_setup_channel exists for one purpose: to convey -- from us to them -- a single
     * native handle, namely one end of a pre-connected stream-socket pair, over which the two Session_vat_network
     * peer objects (ours and theirs) shall subsequently transmit their (small) messages.  So: create said
     * socket pair; send one end over the channel and flush it (the channel's work is then done); and feed
     * the other end to our Session_vat_network.
     *
     * Why not just have Session_vat_network use rpc_setup_channel's own low-level transport (skipping this
     * dance)?  Answer: KJ/capnp async-I/O machinery (kj::AsyncIoStream et al) requires a stream-type
     * socket; whereas the channel's own transport is not, in general, that.  (As of this writing it is typically a
     * SOCK_SEQPACKET-type Unix-domain socket; depends on compile-time Native_socket_stream_cfg settings; but that
     * detail is not the point really: The point is we want it to work regardless of how Native_socket_stream might
     * itself be internally implemented/how it may or may not expose related resources to the world.  Cf.: this Channel
     * type can, generically, transmit `Native_handle`s, end of; so we use it for its purpose (one of them).)
     *
     * Compared to many other places in session-land of Flow-IPC, wherein typically the server-designated side
     * creates resources/client consumes them, here we do the reverse: we connect_pair(); they (briefly)
     * await one of the results of that.  Why?  Answer: Then we can keep everything synchronous -- no KJ async/promise
     * dance (we just sync_connect(), send_native_handle(), and *end_sending(); the latter is ~instant
     * in practice, so it's in effect synchronous too).  They already need to do the KJ async/promise dance
     * on account of the Session::async_accept() step; so the async_receive...() is tacked on to that. */
 
    {
      Task_engine task_engine; // Purely a formality required by the Peer_socket (boost.asio) API.
      Peer_socket local_sock{task_engine};
      Peer_socket remote_sock{task_engine};
      Error_code sys_err_code;
      connect_pair(local_sock, remote_sock, sys_err_code);
      /* (Could've used the throwing version of that API, but our message-and-such is arguably nicer.  Plus
       * technically we promised to throw Runtime_error, if the problem is within Flow-IPC proper which this
       * arguably is.) */
      if (sys_err_code)
      {
        throw Runtime_error{sys_err_code, "Client_context::sync_connect_impl(): connect_pair()"};
      }
      // else
      local_hndl = Native_handle{local_sock.release()};
      remote_hndl = Native_handle{remote_sock.release()};
    }
 
    {
      /* Send via the async-I/O API: less fuss than servicing the sync_io-pattern API for this one-off op.
       * Note async_io_obj() subsumes the channel's transport (and the m_rpc_setup_channel dtor shall close it). */
      m_rpc_setup_channel.emplace(rpc_setup_channel.async_io_obj());
 
      uint8_t protocol_dummy_payload = 0; // Reserved (see Protocol_negotiator to-do in class doc header).
      m_rpc_setup_channel->hndl_snd()->send_native_handle(remote_hndl,
                                                          Blob_const{&protocol_dummy_payload,
                                                                     sizeof(protocol_dummy_payload)});
      // (That throws Runtime_error on error, due to null Error_code* arg.)
 
      /* Send the graceful-close after that.  Neither the above nor this will internally hit would-block
       * (fresh channel), in any OS we know of at least, but that's irrelevant anyway: For reasons
       * we're about to explain, we're keeping *m_rpc_setup_channel alive indefinitely anyway.  So there's no
       * need for the typical async_end_sending(F) (where F() signals us to proceed <=> anything pending has been sent)
       * dance. */
      m_rpc_setup_channel->hndl_snd()->end_sending();
 
      /* Sent: the other side has (or on its schedule will have) its own copy of the handle;
       * our copy has done its job; close it (descriptor/handle goes away; description/kernel endpoint lives). */
      remote_hndl.close();
 
      /* The channel endpoint (reminder: of the channel we used to transmit remote_hndl, which is half of
       * capnp-RPC transmission-pipe socket-pair) itself, however, we deliberately keep around
       * in *m_rpc_setup_channel until *this dies.  Why though?
       *
       * To answer: First consider the minimal, most-elegant algorithm in purely algorithmic and resource-use
       * terms.  In that one, we have a Unix-domain socket S, and Context_server has socket D.  We send the
       * payload (remote_hndl basically), an EOF token; and now no longer need S.  Opposing server-side: we grab
       * D off the accept queue as soon as possible; async-read the payload; and now no longer need D.
       *
       * Can this be arranged in practice?  Yes.  The original init-channel, before .async_io_obj() conversion (which
       * we would forego in this hypothetical), is a sync_io-pattern Channel, so its ->hndl_snd() essentially has S.
       * To wrangle that, use it minimalistically: Client-side (here) do only start_send_...(); server-side only
       * start_receive_...().  Then, very similarly to the above, send_native_handle(), end_sending() -- and then
       * drop the init-channel object (close S); no m_rpc_setup_channel to store (no socket taken from OS
       * indefinitely).
       *
       * @todo We can and arguably should do it -- *if* an extra per-session socket (FD) is bothersome
       * in the field (seems very unlikely but conceivable).  The difficulty, in practice, is server-side.
       * Client-side in reality all the required ops are synchronous and will not require us to hook-up an
       * async-wait functor (it can be a no-op).  Server-side, though, there's the async-read which will mean
       * start_receive_...(F) will need an F() that'll actually do the wait; and in Context_server that needs to
       * be hooked to the KJ event-loop; hence an FdObserver or similar.  The best way to do *that* would be
       * to provide a utility module that links our sync_io-pattern API with KJ-style event loops -- itself
       * an excellent to-do potentially useful for KJ users orthogonally to anything else.  Until then it's too
       * hairy to custom-write it just for Context_server, just to save a socket taken from the OS.
       *
       * @todo (The "to-do" mentioned in preceding to-do) Elsewhere develop a public utility that links
       * KJ event loops to sync_io-pattern APIs, in spirit similarly to what we already provide for boost.asio
       * event loops.  (Make it available to users; and can also use in the preceding to-do ourselves, for the
       * Context_server side.)
       *
       * OK!  So then since doing the sync_io approach is, at best, a to-do: Why not just use an async-I/O
       * Channel?  Indeed that is what we do, as you can see; bringing us back to the original question:
       * why keep our side around (as m_rpc_setup_channel)?  Naively it would seem we're just doing the
       * async-I/O-ish version of the S/D algorithm above, the only difference being the signalling
       * from a Channel background thread ("thread W"), server-side only at that.  So what can go wrong?  Answer:
       * the wrongness comes from a technicality: ->hndl_snd/rcv() is really Native_socket_stream, and
       * although from our PoV server-side we are only doing receive-ops and no send-ops, internally an
       * async-I/O Native_socket_stream automatically sets itself up for bidirectional use, including a one-time
       * initial internal send of some Protocol_negotiator info.  Now, consider this fully-possible (and observed)
       * scenario:
       *
       * We send hndl, send EOF, close; then a split second later Context_server reaches the point where it can deal
       * with the capnp-RPC setup (perhaps it was setting up SHM stuff for a bit up to that point, so it's slightly
       * behind us).  So Context_server side of this channel will *internally* **send** that initial protocol-version
       * stuff -- but it can't; our side has closed for sending *and* (saliently) receiving.  In Linux at least
       * this is the broken-pipe error (EPIPE); but it makes sense generically/conceptually too.  Bottom line is:
       * transport::Native_socket_stream internally needs to write even when the user only needs to read; and
       * our client-side endpoint-closing behavior breaks that part.
       *
       * So that's why we solve it by simply not closing our side.  The to-dos above would tighten this up, but
       * the costs are (1) an extra per-session socket taken from OS, (2) slight bit of added code, and (3)
       * this endless comment. / Last thing: one can alternatively use a handshake protocol, wherein after
       * the send hndl/send EOF step we'd await an ack; server conversely would async-read and then send ack.
       * Upon receiving ack, or error, we can close our-side Channel.  It would require some concurrency
       * consideration though: perhaps a mostly-idle Single_thread_task_loop (our own thread W) for channel-closings
       * only.  Only a few more lines of code ultimately -- but it still adds *more* to the algorithm, whereas
       * the solution in the to-dos above makes it simpler (conceptually).  So: *if* there's anything to do at all
       * the to-dos are probably the better approach. */
    }
 
    if (m_enable_hndl_transport)
    {
      // Let it use the sensible default for last arg.
      m_network.emplace(get_logger(), m_kj_io,
                        sans_shm_transport ? Session_ptr{} : Session_ptr{m_session},
                        std::move(local_hndl));
    }
    else
    {
      // As requested disable handle-transmission (last arg).
      m_network.emplace(get_logger(), m_kj_io,
                        sans_shm_transport ? Session_ptr{} : Session_ptr{m_session},
                        std::move(local_hndl), 0);
    }
    assert(local_hndl.null() && "Session_vat_network ctor promises to consume the transport handle immediately.");
    /* (That *m_network construction might throw KJ or capnp error; we are not sure exactly all the specific
     * possibilities of what would throw.) */
  }
  catch (...)
  {
    /* Get the message for logging, if we can.  (This dispatcher fuss, as opposed to just catching
     * `const exception&`, is because: kj::Exception -- possible from Session_vat_network ctor -- does not
     * derive from std::exception.  KJ in fact *throws* an internal sub-class deriving from both, so the
     * middle arm below is only for a bare-thrown kj::Exception; can't hurt.) */
    string msg;
    try { throw; }
    catch (const exception& exc) { msg = exc.what(); }
    catch (const kj::Exception& exc) { msg = exc.getDescription().cStr(); }
    catch (...) { msg = "(unknown exception type)"; }
 
    FLOW_LOG_WARNING("rpc::Client_ctx [" << *this << "]: A step of capnp-RPC connect (the ipc::session::Session "
                     "[" << *m_session << "] connect itself; socket-pair creation; socket-handle transmission; "
                     "Session_vat_network ctor) threw exception [" << msg << "] which we shall re-throw; "
                     "getting `*this` and out-args back to pristine state first.");
    /* Close whatever socket-pair handles remain in our custody (each of these no-ops on null handle;
     * in particular Session_vat_network ctor -- if reached -- consumes (nullifies) local_hndl immediately;
     * in any case we cannot double-close no matter when it threw). */
    remote_hndl.close();
    local_hndl.close();
    m_rpc_setup_channel.reset(); // Get this back to pristine state too.
 
    /* As advertised: Undo anything we might have placed into the init-channels container(s) upon successful
     * sync_connect() if any.  (We don't bother trying to bring back whatever random stuff they had
     * in *init_channels_by_cli_req_pre_sized; placing NULL-state Channel{}s there is fine.) */
    if (init_channels_by_cli_req_pre_sized)
    {
      *init_channels_by_cli_req_pre_sized = Channels{init_channels_by_cli_req_pre_sized->size()};
    }
    if (init_channels_by_srv_req)
    {
      init_channels_by_srv_req->clear();
    }
 
    /* Whether Session connect failed (leaving it in NULL state) or succeeded (PEER state) before the throw:
     * replace it with a fresh NULL-state one; a subsequent retry starts from scratch. */
    m_session
      = boost::make_shared<Session_obj>(get_logger(), m_cli_app_ref, m_srv_app_ref,
                                        [logger_ptr = get_logger(), &cli_app = m_cli_app_ref, &srv_app = m_srv_app_ref]
                                          (const Error_code& err_code)
                                            { on_session_hosed(logger_ptr, cli_app, srv_app, err_code); });
    assert((!m_network) && "If Session_vat_network ctor (last step) threw -- m_network should still be null!  Bug?");
 
    throw;
  }
  // Got here: noice!  (m_network ctor logged just fine, so let's not.)
 
  return true;
} // Client_context::sync_connect_impl()
 
template<typename Client_session_t>
typename Client_context<Client_session_t>::Vat_network*
  Client_context<Client_session_t>::vat_network()
{
  return m_network ? &(*m_network) : nullptr;
}
 
template<typename Client_session_t>
const typename Client_context<Client_session_t>::Vat_network*
  Client_context<Client_session_t>::vat_network() const
{
  return const_cast<Client_context*>(this)->vat_network();
}
 
template<typename Client_session_t>
typename Client_context<Client_session_t>::Session_obj*
  Client_context<Client_session_t>::session()
{
  return m_network ? m_session.get() : nullptr;
}
 
template<typename Client_session_t>
const typename Client_context<Client_session_t>::Session_obj*
  Client_context<Client_session_t>::session() const
{
  return const_cast<Client_context*>(this)->session();
}
 
template<typename Client_session_t>
std::ostream& operator<<(std::ostream& os, const Client_context<Client_session_t>& val)
{
  const auto network = val.vat_network();
  os << '[';
  if (network)
  {
    os << "session[" << *(val.session()) << "] vat_netwk[" << *network << ']';
  }
  else
  {
    os << "null";
  }
  return os << "]@" << &val;
}
 
} // namespace ipc::transport::struc::shm::rpc