session_base.hpp

Go to the documentation of this file.

/* Flow-IPC: Sessions
 * 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/session/detail/session_fwd.hpp"
#include "ipc/session/detail/session_shared_name.hpp"
#include "ipc/session/schema/detail/session_master_channel.capnp.h"
#include "ipc/session/info_collector.hpp"
#include "ipc/session/app.hpp"
#include "ipc/session/error.hpp"
#include "ipc/transport/struc/channel.hpp"
#include "ipc/transport/native_socket_stream_acceptor.hpp"
#include "ipc/transport/sync_io/native_socket_stream.hpp"
#include <boost/thread/future.hpp>
 
namespace ipc::session
{
 
// Types.
 
/**
 * Internal type containing data and types common to internal types Server_session_impl and Client_session_impl
 * which are the respective true cores of #Server_session and #Client_session respectively.
 *
 * A Server_session_impl and Client_session_impl share many basic properties, some public.  For example consider
 * that a Server_session_impl on this side has 1 counterpart Client_session_impl on the other side (or vice versa); for
 * both it is salient which Server_app is on the server side and which Client_app is on the client side.
 * In terms of types/constants, on each side the two Session objects must be identically configured via various
 * template params which results in non-trivial type aliases like the highly significant #Channel_obj,
 * exposed in the Session concept, and constants like #S_SOCKET_STREAM_ENABLED.
 *
 * Regarding data: an operational (PEER-state) Session (on either end) will need various members to be set
 * before it is indeed in PEER state.  This object stores such data, usually unset at first, and features
 * `protected` setters to invoke once each, until all are set permanently for PEER state.
 *
 * Session_base also has a `protected` inner class, Session_base::Graceful_finisher which certain variations
 * (for which this is necessary) of Server_session_impl and Client_session_impl use to carry out a
 * graceful-session-end handshake procedure.  This is here in Session_base -- despite being somewhat more
 * algorithmic/stateful than the other stuff here -- essentially because the Graceful_finisher (if needed at
 * all) on each side acts completely symmetrically.  Other items tend to be (internally) asymmetrical in
 * behavior between Server_session_impl and Client_session_impl (and variants).  More info on
 * Session_base::Graceful_finisher in its own doc header.
 *
 * @todo Session_base is in detail/ and is formally documented at the top of its doc header as "internal"; and
 * indeed it is out-of-the-box; but the ipc::session hierarchies are also formally user-extensible, so Session_base
 * should be publicly exposed after all.  Hence remove the "internal" verbiage; move the file; and go over its
 * source code/design, in case something about its API is too "Inside Baseball."  To be clear, though, this is
 * not of the highest priority: 1, detail/ or not it can still be sub-classed in a pinch; 2, arguably more saliently,
 * extending ipc::session hierarchies is a rare thing for a user to want to do in practice.  It's an
 * API-versus-SPI situation that isn't fully clean here.  (Generally in Flow-IPC we endeavour to keep that fully clean.)
 *
 * @todo The to-do preceding this one in the source code has corollaries: There are probably some (not all)
 * `_impl`y classes/class templates that are in detail/ but should not be, given that formally ipc::session
 * is user-extensible.  Again, though: 1, detail/ or not it (for each "it") can still be accessed in a
 * pinch; 2, arguably more saliently, extending ipc::session hierarchies is a rare thing for a user to want to do
 * in practice.  It's an API-versus-SPI situation that isn't fully clean here.  (Generally in Flow-IPC we endeavour
 * to keep that fully clean.)
 *
 * @tparam Mdt_payload
 *         See #Server_session, #Client_session (or Session concept).
 * @tparam MQ_TYPE_OR_NONE
 *         See #Server_session, #Client_session.
 * @tparam TRANSMIT_NATIVE_HANDLES
 *         See #Server_session, #Client_session.
 */
template<schema::MqType MQ_TYPE_OR_NONE, bool TRANSMIT_NATIVE_HANDLES, typename Mdt_payload>
class Session_base :
  private boost::noncopyable
{
public:
  // Constants.
 
  /// See Session_mv.
  static constexpr bool S_MQS_ENABLED = MQ_TYPE_OR_NONE != schema::MqType::NONE;
 
  /// See Session_mv.
  static constexpr bool S_SOCKET_STREAM_ENABLED = (!S_MQS_ENABLED) || TRANSMIT_NATIVE_HANDLES;
 
  // Types.
 
  static_assert(MQ_TYPE_OR_NONE != schema::MqType::END_SENTINEL,
                "Do not use the value END_SENTINEL for MQ_TYPE_OR_NONE; it is only a sentinel.  Did you mean NONE?");
 
  // Ensure the definitions immediately following are based on correct assumptions.
  static_assert(std::is_enum_v<schema::MqType>,
                "Sanity-checking capnp-generated MqType enum (must be an enum).");
  static_assert(std::is_unsigned_v<std::underlying_type_t<schema::MqType>>,
                "Sanity-checking capnp-generated MqType enum (backing type must be unsigned).");
  static_assert((int(schema::MqType::END_SENTINEL) - 1) == 2,
                "Code apparently must be updated -- expected exactly 2 MqType enum values plus NONE + sentinel.");
  static_assert((int(schema::MqType::POSIX) != int(schema::MqType::BIPC))
                 && (int(schema::MqType::POSIX) > int(schema::MqType::NONE))
                 && (int(schema::MqType::BIPC) > int(schema::MqType::NONE))
                 && (int(schema::MqType::POSIX) < int(schema::MqType::END_SENTINEL))
                 && (int(schema::MqType::BIPC) < int(schema::MqType::END_SENTINEL)),
                "Code apparently must be updated -- "
                  "expected exactly 2 particular MqType enum values plus NONE + sentinel.");
 
  /**
   * Relevant only if #S_MQS_ENABLED, this is the Persistent_mq_handle-concept impl type specified by
   * the user via `MQ_TYPE_OR_NONE`.
   */
  using Persistent_mq_handle_from_cfg
    = std::conditional_t<!S_MQS_ENABLED,
                         transport::Null_peer,
                         std::conditional_t<MQ_TYPE_OR_NONE == schema::MqType::POSIX,
                                            transport::Posix_mq_handle,
                                            transport::Bipc_mq_handle>>;
 
  /// See Session_mv (or Session concept).
  using Channel_obj = std::conditional_t<S_MQS_ENABLED,
                                         std::conditional_t
                                           <TRANSMIT_NATIVE_HANDLES,
                                            transport::Mqs_socket_stream_channel<true, Persistent_mq_handle_from_cfg>,
                                            transport::Mqs_channel<true, Persistent_mq_handle_from_cfg>>,
                                         std::conditional_t
                                           <TRANSMIT_NATIVE_HANDLES,
                                            transport::Socket_stream_channel<true>,
                                            transport::Socket_stream_channel_of_blobs<true>>>;
 
  /// See Session_mv.  Note: If changed from `vector` please update those doc headers too.
  using Channels = std::vector<Channel_obj>;
 
  /// See Session_mv (or Session concept).
  using Mdt_payload_obj = Mdt_payload;
 
  /// See Session_mv (or Session concept).
  using Mdt_reader_ptr
    = boost::shared_ptr<typename transport::struc::schema::Metadata<Mdt_payload_obj>::Reader>;
 
  /// See Session_mv (or Session concept).
  using Mdt_builder
    = typename transport::struc::schema::Metadata<Mdt_payload_obj>::Builder;
 
  /// See Session_mv (or Session concept).
  using Mdt_builder_ptr = boost::shared_ptr<Mdt_builder>;
 
  /// See Session_mv (or Session concept).
  using Structured_msg_builder_config = transport::struc::Heap_fixed_builder::Config;
 
  /// See Session_mv (or Session concept).
  using Structured_msg_reader_config = transport::struc::Heap_reader::Config;
 
  // Methods.
 
  /**
   * See Server_session_impl, Client_session_impl.  However in Session_base this value may be not-yet-set (empty), or
   * set (and immutable from then on).
   *
   * This value shall be known and immutable from construction for Server_session_impl, since server namespace -- PID
   * as of this writing -- can be determined from the start on the server side and applies to every server session
   * that Session_server produces.  Client_session_impl, however, determines
   * it at the latest possible moment which is at Client_session_impl::async_connect() time, at which point it needs
   * to determine the PID via PID file.
   *
   * @return See above.
   */
  const Shared_name& srv_namespace() const;
 
  /**
   * See Server_session_impl, Client_session_impl.  However in Session_base this value may be not-yet-set (empty), or
   * set (and immutable from then on).
   *
   * This value shall be generated uniquely -- process-wide, across all `Session_server`s in the process (see
   * next_cli_namespace()) -- for each new `Server_session` produced by Session_server; and Client_session_impl
   * shall find that out while logging in (last part of entering PEER state).
   *
   * @return See above.
   */
  const Shared_name& cli_namespace() const;
 
  /**
   * See Server_session_impl, Client_session_impl.  However in Session_base this value may be not-yet-set (null), or set
   * (and immutable from then on).  Note the value to which we refer is the actual pointer.  (The Client_app that
   * is pointed-to, itself, is certainly immutable too.)
   *
   * This value shall be set from the start in a Client_session_impl but determined during a given Server_session_impl's
   * log-in (the opposing Client_session_impl will transmit the Client_app::m_name).  The log-in shall complete the
   * `Server_session`'s entry to PEER state.
   *
   * @return See above.
   */
  const Client_app* cli_app_ptr() const;
 
  /**
   * Computes the name of the interprocess named-mutex used to control reading/writing to the file
   * storing (written by server, read by client) the value for srv_namespace().  The file
   * is located as cur_ns_store_absolute_path().
   *
   * This may be called anytime.
   *
   * @return See above.
   */
  Shared_name cur_ns_store_mutex_absolute_name() const;
 
  /**
   * Computes the absolute path to file storing (written by server, read by client) the value for srv_namespace().
   * The file is located as cur_ns_store_absolute_path().
   *
   * This may be called anytime.
   *
   * @return See above.
   */
  fs::path cur_ns_store_absolute_path() const;
 
  /**
   * Computes the absolute name at which the server shall set up a transport::Native_socket_stream_acceptor
   * to which client shall transport::Native_socket_stream::sync_connect() in order to establish a PEER-state
   * session.
   *
   * This must be called no earlier than set_srv_namespace(); otherwise behavior undefined (assertion
   * may trip).
   *
   * @return See above.
   */
  Shared_name session_master_socket_stream_acceptor_absolute_name() const;
 
  /**
   * See Session_mv::heap_reader_config() (1-arg).
   *
   * @param logger_ptr
   *        See above.
   * @return See above.
   */
  static Structured_msg_reader_config heap_reader_config(flow::log::Logger* logger_ptr);
 
  // Data.
 
  /**
   * Reference to Server_app (referring to local process in Server_session_impl, opposing process in
   * Client_session_impl).
   * This is known from construction and immutable (both the reference, of course, and the Server_app itself).
   */
  const Server_app& m_srv_app_ref;
 
protected:
  // Constants.
 
  /**
   * Internal timeout for `open_channel()`.
   *
   * ### The value ###
   * We initially tried something much less generous, 500ms.  It worked fine, but some people encountered
   * the timeout due to unrelated reasons, and it was natural to blame it on the timeout.  This much longer
   * timeout should make it obvious, in such a situation, that it's not about some slowness inside Flow-IPC
   * but a pathological application problem -- particularly around session-open time.  For example not
   * calling Server_session_impl::init_handlers(), or calling it late, as of this writing can cause issues with this.
   *
   * The downside is it makes Session::open_channel() potentially blocking formally speaking, whereas 500ms
   * could still claim to be non-blocking.  It's a matter of perspective really.  This value just seems to
   * cause less confusion.  We might reconsider the whole thing however.
   */
  static constexpr util::Fine_duration S_OPEN_CHANNEL_TIMEOUT = boost::chrono::seconds{60};
 
  /**
   * The default max sendable MQ message size as decided by Server_session_impl::make_channel_mqs() (and imposed on
   * both sides, both directions), if #S_MQS_ENABLED *and* Server_session_impl::S_SHM_ENABLED is `false`, when a
   * channel is opened (regardless of which side did the active-open or requested pre-opening at session start).
   * By *default* we mean this is the knob's value unless overridden via Session_server::mq_msg_size_limit().
   *
   * If `*this` belongs to Server_session_impl, that's what this is.
   * If it belongs to Client_session_impl, then this is what the opposing process -- if they're using the same code! --
   * will have decided (again, unless overriden in the opposing process via Session_server::mq_msg_size_limit()).
   *
   * @note If Server_session_impl::S_SHM_ENABLED is `true`, then a different (much smaller) MQ message size limit
   *       default is configured.  (Again, though, it can be overridden via Session_server::mq_msg_size_limit().)
   *
   * While it looks simple, there is a number of subtleties one must understand if *ever* considering changing it.
   *
   * ### bipc versus POSIX MQs ###
   * As of this writing the same constant value is used for both types of MQ configurable.  The actual value is
   * chosen due to a certain aspect of POSIX MQs (discussed just below); and we reuse it for bipc, though we absolutely
   * do not have to, for simplicity/for lack of better ideas at the moment.  It would be possible to bifurcate those
   * two cases if really desired.
   *
   * ### Why 8Ki? ###
   * By default in Linux POSIX MQs this happens to be the actual limit for message size --
   * visible in `/proc/sys/fs/mqueue/msgsize_max` -- so we cannot go higher typically.  However that file can be
   * modified.  For now we assume a typical environment; or at least that it will not go *below* this typical default.
   * If we did try a higher number here, opening of MQs by server will likely emit an error and refuse
   * (Server_session_impl::make_channel_mqs()).
   *
   * ### Things to consider if changing the value away from the above ###
   * - Contemplate why you're doing it.  bipc MQs are seen (in Boost source) to be a simple zero-copy data structure
   *   in an internally maintained kernel-persistent SHM pool; while I (ygoldfel) haven't verified via kernel source,
   *   likely Linux POSIX MQ impl is something very similar (reasoning omitted but trust me).  So copy perf should
   *   not be a factor; only RAM use and the functional ability to transmit messages of a certain size.
   *   - If the plan is to use `transport::struc::Heap_fixed_builder`-backed for structured messages
   *     (via transport::struc::Channel), then the max size is quite important: if a *leaf* in your message exceeds
   *     this size when serialized, it is a fatal error.
   *   - If, however, the plan is to use SHM-backing (e.g., via `shm::classic::*_session`), then this constant
   *     does not get used (a much smaller value does -- so small it would be ~meaningless to decrease it).
   * - Suppose you have changed this value, and suppose the `Heap_fixed_builder`-based use case *does* matter.
   *   If ::ipc has not yet been released in production, ever, then it's fine (assuming, that is, it'll work
   *   in the first place given the aforementioned `/proc/sys/...` limit for POSIX MQs).  If it *has* been
   *   released, but you can guarantee (via release process) that the client and server will always use the same ::ipc
   *   software, then you're still fine.  Just change this value; done.  (Note: A simple way to assure the opposing
   *   peer is appropriately upgraded is to carry out a `Protocol_negotiator` version(s) bump -- without allowing
   *   backward-compatibility, meaning min-supported-version=max-supported-version.  There are at least a couple
   *   potential layers at which to do this bump, so just pick the right one is all.)
   */
  static constexpr size_t S_MQS_MAX_MSG_SZ = 8 * 1024;
 
  // Types.
 
  /**
   * The (internally used) session master channel is a transport::struc::Channel of this concrete type.
   *
   * Rationale for the chosen knob values:
   *   - To be able to open_channel() (and hence passive-open) a #Channel_obj that can transmit native handles,
   *     certainly a handles pipe is required to transmit half of each socket-pair.  That said, an `Mqs_channel`
   *     would work fine if #S_SOCKET_STREAM_ENABLED is `false`.  The real reason at least a handles pipe is
   *     required is that to establish this channel -- unlike subsequent channels in the session -- a client-server
   *     flow is required.  This connection establishment is not possible with MQs, so of the available options
   *     only a socket stream would work.
   *     - There is no bulky data transfers over this channel, so adding
   *       a parallel MQ-based blobs pipe is overkill even if it could help performance; more so since the
   *       master channel is unlikely to be used frequently (open_channel() on either side shouldn't be that frequent).
   *       Since, again, the minute differences in perf versus a Unix-domain-socket-based transport are unlikely to
   *       be significant in this use case, it is easier to use a socket stream which lacks any kernel-persistent
   *       cleanup considerations -- or even just distinct naming considerations for that matter.
   *     - So: use a transport::Socket_stream_channel.
   *   - Use the specially tailored (internal) session master channel capnp schema, whose key messages cover (at least):
   *     - session log-in at the start;
   *     - open-channel request/response.
   *   - As of this writing, for serializing/deserializing, it's either the heap-allocating engine on either side,
   *     or it's something SHM-based (which could be faster).  However at least some SHM-based builders/readers
   *     themselves (internally) require a transport::Channel to function, which would require a Session typically,
   *     creating a cyclical feature dependency.  Plus, SHM has cleanup considerations.  So a heap-based engine
   *     on either side is the natural choice.  The cost is some perf: one copies from heap into the session master
   *     transport; and from there into heap on the other side.  In this context that's not a significant perf loss.
   */
  using Master_structured_channel
    = transport::struc::Channel_via_heap<transport::Socket_stream_channel<true>,
                                         schema::detail::SessionMasterChannelMessageBody
                                           <Mdt_payload_obj>>;
 
  /**
   * Handle to #Master_structured_channel.
   *
   * ### Rationale for type chosen ###
   * It's a handle at all, because at first and possibly later there may be no session master channel, so a null
   * value is useful.  (Though, `std::optional` could be used instead.)  It's a ref-counted pointer as opposed
   * to `unique_ptr` so that it can be observed via #Master_structured_channel_observer (`weak_ptr`) which is not
   * (and cannot) be available for `unique_ptr`.  The observing is needed tactically for certain async lambda needs.
   */
  using Master_structured_channel_ptr = boost::shared_ptr<Master_structured_channel>;
 
  /// Observer of #Master_structured_channel_ptr.  See its doc header.
  using Master_structured_channel_observer = boost::weak_ptr<Master_structured_channel>;
 
  /// See Session_mv (or Session concept).
  using Info_collector = session::Info_collector<Master_structured_channel>;
 
  /// Concrete function type for the on-passive-open handler (if any), used for storage.
  using On_passive_open_channel_func = Function<void (Channel_obj&& new_channel,
                                                      Mdt_reader_ptr&& new_channel_mdt)>;
 
  /**
   * Optional to use by subclasses, this operates a simple state machine that carries out a graceful-session-end
   * handshake procedure.  A particular Client_session_impl and symmetrically its opposing Server_session_impl
   * shall instantiate a `*this` internally and invoke its methods on certain events, as described in their
   * contracts.
   *
   * Now we explain what we're solving (Rationale); then how we solve it (Design).  The latter is much simpler,
   * we think, to understand than the former.  Hence apologies in advance for the length of "Rationale" section.
   *
   * ### Rationale ###
   * Consider a particular Client_session_impl + Server_session_impl class/object pair, A and B.  In this case --
   * which is somewhat unusual (relative to most internal `Session` algorithms) but in a good, simplifying way --
   * it does not matter which is A and which is B.  So let's say A is us and B is the opposing guy.
   *
   * Also, for now, assume these are just the vanilla ones: session::Client_session_impl and
   * session::Server_session_impl; not SHM-enabled -- no SHM arenas in play.
   *
   * Consider what happens when A dtor is invoked, in PEER state.  The session ends, and the session-end trigger is us.
   * Without Graceful_finisher what happens is as follows.  The #Master_structured_channel in `*this` is closed;
   * and then `*this` goes away altogether right away/synchronously.  Soon B's #Master_structured_channel
   * throws the graceful-close error; B itself correctly interprets this as a session-end trigger; and B throws
   * an error via the `Session` error handler.  Now, the user is told by docs they informally *should* destroy
   * the B (invoke its dtor) ASAP, but nothing is formally forcing them to; it can sit around.  It will never
   * accept channel passive-opens; `open_channel()` will no-op / return null; etc.  It is useless, but it can stay
   * around.  Eventually B dtor will be invoked by user; it'll basically clean up data structures, and that's that.
   *
   * There is no problem there: A dtor and B dtor can run at quite different times, or more precisely B dtor
   * can run much later than A dtor which triggered session-end.  Nothing bad happens.
   *
   * Now we'll discuss this in specific terms around actual `Client/Server_session_impl` variants that do exist
   * at this time.  We could try to be formal and general, but it's easier to just be specific for exposition
   * purposes.
   *
   * Take, now, the shm::classic::Client_session_impl shm::classic::Server_session_impl variants.  The above
   * generally holds, but there are now SHM arenas in play.  We can forget about the app-scope arenas right off
   * the bat: That guy stays around as long as the `Session_server` does, and that guy is formally required
   * to be around until all child `Session`s have been destroyed.  So only session-scope arenas -- the ones
   * inside A and B -- are relevant.  In fact, for SHM-classic there really is only *one* arena: A holds a
   * handle to it (`a.session_shm()` points to it), and B does so symmetrically.
   *
   * Now reconsider the scenario above: A dtor runs.  (Insert vanilla `*_session_impl` text from above here.)
   * Also, though, A dtor will remove the underlying SHM-pool by its name.  (This is possibly a lie.  This is
   * technically so, only if A is a `Server_session_impl`; we've assigned the task to the server side as of this
   * writing.  But this technicality is irrelevant: unlinking the SHM-pool doesn't blow up the other side's
   * access to the arena; it remains in RAM until that guy closes.  So just bear with us here.)  Then indeed
   * B dtor can run sometime later.  Is there a problem?  For SHM-classic, which is what we're discussing, no:
   * The arena -- 1 shared SHM-pool in this case -- stays alive, until B's handle gets closed by B dtor.
   * What about allocations/deallocations?  No problem there either: By contract, the A user *must* drop all
   * cross-process `shared_ptr` handles to objects that it owns, before invoking A dtor; so A is fine, and the
   * internal cross-process ref-counts for each A-held object get decremented by 1, due to the `shared_ptr` handles
   * being nullified by the user by contract.  (For context: The guided manual `session_app_org` page describes
   * to the reader how to structure their program, so that this is guaranteed.  In short, declare the object
   * handle data members in their code after the `Session` data member, so they get deinitialized in the reverse
   * order, in the user's hypothetical recommended `App_session` class.)  Isn't there a problem in B, though?  No:
   * the B user shall drop *its* `shared_ptr` handles, before invoking B dtor; this will drop their internal
   * cross-process ref-counts to 0; at which point shm::Pool_arena logic will deallocate those objects.  Deallocation
   * is entirely distributed between A and B; in this scenario B's handles just happen to live longer and therefore
   * happen to be the ones triggering the deallocations -- by the B process itself, synchronously.
   * A is not "in charge" of deallocations of anything in any special way; it doesn't matter which process originally
   * allocated a given object either.  That's the beauty/simplicity of SHM-classic.  (There are of course trade-offs
   * in play; this is worse for segregation/safety... see Manual and such.)
   *
   * Now we come to the point.  Consider session::shm::arena_lend::jemalloc::Client_session_impl and
   * `Server_session_impl`.  There *is* a problem.  Interestingly this time it really does *not* matter whether
   * A is server or client-side.  In SHM-jemalloc the session-scope arenas are not 1 but 2:
   * A has an arena, from which process A can allocate, and in which process A shall deallocate.
   * B has an arena, from which process B can allocate, and in which process B shall deallocate.
   * If B borrows object X from A, and is later done with it -- the `shared_ptr` local ref-count reached 0 --
   * SHM-jemalloc-arranged `shared_ptr` deleter will send an internal message over a special channel to A.
   * A will deallocate X once that has occurred, *and* any A-held `shared_ptr` group for X has also reached
   * ref-count 0 (which may have happened before, or it may happen later).  Exactly the same is true of A borrowing
   * from B, conversely/symmetrically.
   *
   * The problem is this: Suppose B holds a borrowed handle X (to an A-allocated object).  Now A dtor runs;
   * its `.session_shm()` arena -- its constituent SHM-pools! -- is destroyed.  That in itself shouldn't be
   * a problem; again B presumably holds a SHM-pool handle, so it won't just disappear from RAM under it.
   * I (ygoldfel) am somewhat fuzzy on what exactly happens (not being the direct original author of SHM-jemalloc), but
   * basically as I understand it, in destroying the arena, a bunch of jemalloc deinit steps execute; perhaps
   * heap-marker data structures are modified all over the place... it's chaos.  The bottom line is:
   * A arena gets mangled, so the B-held borrowed handle X has a high chance of pointing, now, to garbage:
   * Not unmapped garbage; not into a SHM-pool that's magically gone -- it's still there -- so there's not necessarily
   * a SEGV as a result; just garbage in the ~sense of accessing `free()`d (not un-`mmap()`ped) memory.
   *
   * (Update: I've spoken to the SHM-jemalloc author echan, and the above is basically right with a couple
   * clarifications.  It's not so much that the arena object being destroyed is the issue; in fact that in itself would
   * be handled gracefully by SHM-jemalloc innards; they will keep the arena around, even if the outer object goes away
   * (or more precisely as of this writing the arena-handle -- a `shared_ptr` -- does), while it knows things are
   * pointing into it, even borrowed things in another process.  Really it's the fact that we kill the *channel*
   * it uses to communicate with the borrower process: the arena is told there's nothing out there pointing into
   * it: locally the user has to have dropped all constructed-object handles; and the borrower process no
   * longer has a way of contacting it; so it assumes that guy's done/gone, as are all the borrowed handles in it.
   * That clarification doesn't change the situation though: Whether it's the SHM-jemalloc-session, or the
   * SHM-jemalloc-session-used-channel, or the SHM-jemalloc-arena going away that makes borrowed objects unusable --
   * they are unusable, is the bottom line.)
   *
   * (Update 2/some months after the previous: Technically the situation is now somewhat different from how it was
   * as written in the previous "update."  (Incidentally it is I who coded this change.)  The dead channel, while still
   * a thing, is less directly relevant, as there is no more return-object command IPC-transmitted from B to A, when B
   * no longer needs borrowed object X.  Instead A and B atomically-decrement a separately stored, in SHM,
   * use-count value; and A opportunistically (<-at worst) checks for any use-counts that have reached zero; once
   * detected it can free the associated X.  However this change (which is for perf) doesn't affect the core issue here:
   * The point is still that A assumes B is "dead to A" and has ~no hope that a borrowed-from-A object X
   * will reach use-count 0.  Or, well, it would have to wait for that to happen somehow... and that's not a thing.
   * Hence it still assumes a guy like X is dead, and from that point it's the same
   * reasoning as before.  Kill everything A-side; now B handle to object X is unusable.)
   *
   * All that is to say: A dtor runs; and user handle to object X in B instantly becomes unusable.
   * For example, I (ygoldfel) have observed a simple thing: A SHM-jemalloc-backed transport::struc::Msg_in received
   * from A, in B, was absolutely fine.  Then it was running a long tight-loop verification of its contents -- verifying
   * hashes of millions of X-pointed objects in SHM, preventing B dtor from running.  Meanwhile, A in a test program
   * had nothing more to do and ran A dtor (closed session).  Suddenly, the hash-verifier code in the tight loop
   * was hitting hash mismatches, throwing capnp exceptions in trying to traverse the message, and so on.
   *
   * What this means, ultimately, is straightforward: A dtor must not destroy its SHM-jemalloc arena
   * (as accessible normally through `.session_shm()`), until not just *A* user has dropped all its object
   * `shared_ptr` handles; but *B* user -- that's in the other process, B -- has as well!  How can we enforce
   * this, though?  One approach is to just put it on the user: Tell them that they must design their protocol
   * in such a way as to arrange some kind of handshake, as appropriate, beyond which each side knows for a fact
   * that the session objects, and therefore arenas, are safe to destroy.
   *
   * We don't want to do that for obvious reasons.  What we already tell them should be sufficient: If you get
   * a session-hosing error handler execution in your `Session`, then don't use that `Session`, and you should
   * destroy it ASAP, for it is useless, and the other side's `Session` is not long for this life.  If they
   * listen to this, then A dtor will soon be followed by B dtor anyway.  It might be delayed by some long
   * tight-loop operation such as the test-program hash-verifying scenario above, but generally users avoid such
   * things; and when they don't they can expect other things to get blocked, possibly in the opposing process.
   * Subjectively, weird blocking -- caused by the other side acting pathologically -- is a lot more acceptable
   * at session-start or session-end than something like that occurring mid-session.
   *
   * So what's the exact proposal?  It's this: If A dtor begins executing, it must first block until it knows
   * B dtor has begun executing; but once it does know that, it knows that all B-held object handles have been
   * dropped -- which is what we want.  (It would be possible for B user to indicate it some other way -- another
   * API call, short of destruction -- but that seems unnecessarily complex/precious.  If the session is finished,
   * then there's no reason to keep the `Session` object around = the overall policy we've pursued so far, so why
   * make exceptions for this?)
   *
   * To be clear, this means certain behavior by B user can cause A dtor to block.  We can explain this in the docs,
   * of course, but perhaps much more effective is to make it clear in the logs: say the wait begins here and what
   * it'll take for it to end; and say when it has ended and why.
   *
   * Also to be clear, Graceful_finisher should be used only in `*_session_impl` pairs that actually have this
   * problem; so of the 3 pairs we've discussed above -- only SHM-jemalloc's ones.  The others can completely
   * forego all this.
   *
   * @todo Consider how to avoid having SHM-jemalloc ipc::session mechanism require one side's `Session` dtor
   * await (and possibly block) the start of the opposing side's `Session` dtor, before it can proceed.  The
   * reason it does this today is above this to-do in the code.  The blocking is not too bad -- if the user's
   * applications are properly written, it will not occur, and the existing logging should make clear why it
   * happens, if it happens; but it's still not ideal.  There are a couple known mitigating approaches,
   * and at least 1 ticket covering them is likely filed.  (Either way the Graceful_finisher mechanism could
   * then be removed probably.)  To summarize them: 1, there can be a mechanism deep inside SHM-jemalloc code that gives
   * (arbitrarily long, configurable) grace period for known borrowed objects whose session channels
   * have closed before those objects were returned; this was echan's idea on this topic.  Once the grace period
   * is reached, it would act as-if they got returned then (so in the above scenario the arena could get
   * jemalloc-deinitialized and all).  If program needs to exit, it could either block until the end of the
   * grace period or... who knows?  2, something similar could be done about the SHM-internal-use channel:
   * do not close it but add it to some list of such channels; they would be finally closed upon detection of
   * the other side's `Session` dtor being reached, in some background thread somewhere.  (On 2nd thought
   * this sounds a lot like Graceful_finisher -- but for this particular internal-use channel only.  The code
   * here would only get more complex, though maybe not too much more.  However, it would resolve the observed
   * `Session`-dtor-blocking user-visible issue, at least until the delayed-channel-death process had to exit
   * entirely.)  Beyond these basic ideas (which could even perhaps be combined) this requires some thinking;
   * it is an interesting problem.  In the meantime the dtor-cross-process-barrier-of-sorts existing solution is
   * pretty decent.
   *
   * ### Design / how to use ###
   * How to use: Subclass shall create a `*this` if and only if the feature is required; it shall call its methods
   * and ctor on certain events, as prescribed in their doc headers.
   *
   * Design: All this assumes PEER state.  Now:
   *
   * Again let's say we're A, and they're B -- and just like we are.  Our goal is to ensure A dtor, at its
   * start, blocks until B dtor has started executing; and to give their side the information necessary to do the
   * same.  That's on_dtor_start(): the method that'll do the blocking; when it returns the dtor of subclass can
   * proceed.  So what's needed for on_dtor_start() to `return`?  One, A dtor has to have started -- but that's true
   * by the contract of when the method is called; two, B dtor has to have started.  How do we get informed of the
   * latter?  Answer: B sends a special message along the #Master_structured_channel, from its own on_dtor_start().
   * Naturally, we do the same -- in full-duplex fashion in a sense -- so that they can use the same logic.  So then
   * the steps in on_dtor_start() are:
   *   -# Send `GracefulSessionEnd` message to B, if possible (#Master_structured_channel is up, and the send
   *      op doesn't return failure).
   *   -# Perform `m_opposing_session_done.get_future().wait()`, where `m_opposing_session_done` is a `promise<void>`
   *      that shall be fulfilled on these events (and only if it hasn't already been fulfilled before):
   *      - The #Master_structured_channel has received `GracefulSessionEnd` *from* B.
   *      - The #Master_structured_channel has emitted a channel-hosing error.
   *
   * Notes:
   *   - The mainstream case is receiving `GracefulSessionEnd`: In normal operation, with no acts of god, we will no
   *     longer destroy the session-master-channel, until returning from on_dtor_start().
   *     - This could happen after the `.wait()` starts -- meaning A dtor indeed ran before B dtor (our side is
   *       the one triggering session-end).
   *     - It could just as easily happen *before* the `.wait()` starts -- meaning the reverse.  Then `.wait()`
   *       will return immediately.
   *   - Of course who knows what's going on -- the #Master_structured_channel could still go down without
   *     the `GracefulSessionEnd`.  Is it really right to treat this the same, and let on_dtor_start() return?
   *     Answer: Yes.  Firstly, there's hardly any other choice: the channel's dead; there's no other way of knowing
   *     what to wait for; and we can't just sit there cluelessly.  But more to the point, this means it's a
   *     *not*-graceful scenario: Graceful_finisher can't hope to deal with it and can only proceed heuristically.
   *     This is not networking; so we should have a pretty good level of predictability at any rate.
   *
   * Last but not least consider that by delaying the destruction of the #Master_structured_channel until after
   * the dtor has started (assuming no channel-hosing acts of god), we've changed something: The error handler shall
   * no longer inform the user of a session-end trigger from the other side!  Left alone this way, we've broken
   * the whole system by introducing a chicken-egg paradox: We don't let our user be informed of the session-end
   * trigger when would normally, so they won't call our dtor; but because they won't call our dtor, we'll never
   * reach the point where the channel would get hosed and we'd know to inform the user.  How to fix this?
   * The answer is pretty obvious: Receiving the `GracefulSessionEnd` shall now trigger a special graceful-session-end
   * error.  Naturally this would only be usefully emitted if we're not in the dtor yet.
   * So the latter situation should kick-off the user invoking our dtor sometime soon, hopefully; we'll send our
   * `GracefulSessionEnd` to them; and so on.
   */
  class Graceful_finisher :
    public flow::log::Log_context,
    private boost::noncopyable
  {
  public:
    // Constructors/destructor.
 
    /**
     * You must invoke this ctor (instantiate us) if and only if synchronized dtor execution is indeed required;
     * and `*this_session` has just reached PEER state.  Invoke from thread W only.
     *
     * ### How `master_channel` shall be touched by `*this` ###
     * We're being very explicit about this, since there's some inter-class private-state sharing going on...
     * generally not the most stable or stylistically-amazing situation....
     *   - This ctor will `.expect_msg(GRACEFUL_SESSION_END)`, so we can detect its arrival and mark it down as needed
     *     and possibly invoke `this_session->hose()` to report it to the user.
     *     Hence there's no method you'll need to call nor any setup like this needed on your part.
     *     - That handler captures `this`; yet `*this` is typically destroyed *before* `*master_channel`.  This is
     *       safe not by lifetime but by protocol: `GracefulSessionEnd` arrives at most once, and on_dtor_start()
     *       returns only once it has arrived (one-shot handler spent) or the channel is hosed (no further handlers).
     *   - on_dtor_start() will attempt to send `GracefulSessionEnd` to the opposing Graceful_finisher.
     *
     * @param logger_ptr
     *        Logger to use for logging subsequently.  (Maybe `Session_base` should just subclass `Log_context`
     *        for all of us?  Add to-do?)
     * @param this_session
     *        The containing Session_base.
     * @param async_worker
     *        The thread W of the containing `*_session_impl`.  `GracefulSessionEnd` handler is at least partially
     *        invoked there (`hose(...)`); and on_dtor_start() posts onto it.
     * @param master_channel
     *        The containing `Session` master channel.  The pointee must exist until `*this` Graceful_finisher
     *        is gone.
     */
    explicit Graceful_finisher(flow::log::Logger* logger_ptr, Session_base* this_session,
                               flow::async::Single_thread_task_loop* async_worker,
                               Master_structured_channel* master_channel);
 
    // Methods.
 
    /**
     * Must be invoked if the `*_session_impl` detects that the master channel has emitted channel-hosing error.
     *
     * Invoke from thread W.
     */
    void on_master_channel_hosed();
 
    /**
     * The reason Graceful_finisher exists, this method must be called at the start of `*_session_impl` dtor; and
     * will block until it is appropriate to let that dtor proceed to shut down the `*_session_impl`.
     *
     * Invoke from thread U, not thread W.
     */
    void on_dtor_start();
 
  private:
    // Data.
 
    /// The containing Session_base.  It shall exist until `*this` is gone.
    Session_base* const m_this_session;
 
    /// The containing `Session` thread W loop.  It shall exist until `*this` is gone.
    flow::async::Single_thread_task_loop* m_async_worker;
 
    /// The containing `Session` master channel.  It shall exist until `*this` is gone.
    Master_structured_channel* m_master_channel;
 
    /**
     * A promise whose fulfillment is a necessary and sufficient condition for on_dtor_start() returning
     * (letting `Session` dtor complete).
     *
     * It is fulfilled once the #Master_structured_channel receives `GracefulSessionEnd` from opposing side
     * (indicating the opposing on_dtor_start() has started) or got hosed (indicating we'll now never know this
     * and must act as-if opposing on_dtor_start() has started).  See "Design" in our doc header.
     */
    boost::promise<void> m_opposing_session_done;
  }; // class Graceful_finisher
 
  // Constructors.
 
  /**
   * Constructs: Client_session_impl form (the user is the one constructing the object, though in NULL state).
   * The values taken as args are set permanently (undefined behavior/assertion may trip if an attempt is made to
   * modify one via mutator).  The remaining values must be set via mutator before PEER state.
   *
   * @param cli_app_ref
   *        See cli_app_ptr().
   * @param srv_app_ref
   *        See #m_srv_app_ref.
   * @param on_err_func
   *        On-error handler from user.
   * @param on_passive_open_channel_func_or_empty_arg
   *        On-passive-open handler from user (empty if user wishes to disable passive-opens on this side).
   */
  explicit Session_base(const Client_app& cli_app_ref, const Server_app& srv_app_ref,
                        flow::async::Task_asio_err&& on_err_func,
                        On_passive_open_channel_func&& on_passive_open_channel_func_or_empty_arg);
 
  /**
   * Constructs: Server_session_impl form (Session_server is the one constructing the object, though in NULL state,
   * before log-in has completed, but after the socket-stream connection has been established).
   * The values taken as args are set permanently (undefined behavior/assertion may trip if an attempt is made to
   * modify one via mutator).  The remaining values must be set via mutator before PEER state.
   *
   * @param srv_app_ref
   *        See #m_srv_app_ref.
   */
  explicit Session_base(const Server_app& srv_app_ref);
 
  // Methods.
 
  /**
   * Sets srv_namespace() (do not call if already set).
   * @param srv_namespace_new
   *        Value.
   */
  void set_srv_namespace(Shared_name&& srv_namespace_new);
 
  /**
   * Sets cli_namespace() (do not call if already set).
   * @param cli_namespace_new
   *        Value.
   */
  void set_cli_namespace(Shared_name&& cli_namespace_new);
 
  /**
   * Restores srv_namespace() and cli_namespace() to the never-yet-set (empty) state.
   *
   * When to use: A `Client_session` connect attempt populates these along the way (set_srv_namespace() upon
   * reading the server's CNS a/k/a PID file; set_cli_namespace() upon receiving the log-in response); and by
   * public contract a failed attempt leaves the `Client_session` in NULL state, from which another attempt on
   * the same object is permitted.  Since these values are set-at-most-once-from-empty *per attempt* -- and
   * the next attempt may well obtain different values (e.g., the server restarted, hence a new
   * srv-namespace/PID) -- the failed attempt's go-back-to-NULL rollback must invoke this.
   * (Server-side: a `Server_session` never legitimately re-sets these; do not use this from server-side code.)
   */
  void reset_namespaces();
 
  /**
   * Sets cli_app_ptr() (do not call if already set).
   * @param cli_app_ptr_new
   *        Value.
   */
  void set_cli_app_ptr(const Client_app* cli_app_ptr_new);
 
  /**
   * Sets on_passive_open_channel_func_or_empty() (do not call if already set; do not call if user intends for
   * passive-opens to be disabled on this side).
   *
   * @param on_passive_open_channel_func
   *        Value.
   */
  void set_on_passive_open_channel_func(On_passive_open_channel_func&& on_passive_open_channel_func);
 
  /**
   * Sets on_err_func() (do not call if already set).
   *
   * @param on_err_func_arg
   *        Value.
   */
  void set_on_err_func(flow::async::Task_asio_err&& on_err_func_arg);
 
  /**
   * Returns `true` if and only if set_on_err_func() has been called, or equivalently if error-handler was set
   * via ctor.
   *
   * @return See above.
   */
  bool on_err_func_set() const;
 
  /**
   * The on-passive-open handler (may be empty even in final state, meaning user wants passive-opens disabled
   * on this side).
   * @return See above.
   */
  const On_passive_open_channel_func& on_passive_open_channel_func_or_empty() const;
 
  /**
   * Marks this session as hosed for (truthy) reason `err_code`; and *synchronously* invokes on-error handler;
   * only invoke if not already hosed().
   *
   * This utility is important and has certain pre-conditions (behavior undefined if not met; assertion may trip):
   *   - `*this` must be in PEER state.  In particular on_err_func_set() must return `true`.
   *   - hosed() must return `false`.
   *   - `err_code` must be truthy (non-success).
   *   - This must be invoked from a thread such that it is OK to *synchronously* invoke on-error handler.
   *     As of this writing this is Server_session_impl's or Client_session_impl's thread W in practice, which is how
   *     thread safety of hose() versus hosed() is also guaranteed by those users.
   *
   * @param err_code
   *        Truthy error.
   */
  void hose(const Error_code& err_code);
 
  /**
   * Returns `true` if and only if hose() has been called.  If so then #m_on_err_func has been executed already.
   * `*this` must be in PEER state.  In particular on_err_func_set() must return `true`.
   *
   * @return Ditto.
   */
  bool hosed() const;
 
  /**
   * Marks the `*_session_impl` dtor as having begun; from then on dtor_started() returns `true`, and callers
   * shall not hose() (emit session-hosing errors to the user).  See #m_dtor_started doc header for rationale
   * and the (essential) requirement as to when to call this: as the dtor's first act, in thread W terms.
   * Call from thread W only, at most once.
   */
  void set_dtor_started();
 
  /**
   * Returns `true` if and only if set_dtor_started() has been called.  Call from thread W only.
   * @return See above.
   */
  bool dtor_started() const;
 
private:
  // Data.
 
  /**
   * See cli_app_ptr().
   *
   * ### Rationale for it being `atomic<>` ###
   * It is for the following specific reason.  Consider `ostream<<` of `*this`.  `*this` is really either
   * in Client_session_impl or Server_session_impl.  In both cases these `ostream<<`s access only #m_srv_app_ref, which
   * is immutable throughout (so no problem there) and #m_cli_app_ptr.  For Client_session_impl #m_cli_app_ptr is
   * also immutable throughout (so no problem there).  For Server_session_impl however it does change in one spot:
   * Server_session_impl::async_accept_log_in() internal async handler for the log-in request shall, on success,
   * call set_cli_app_ptr() and change it from null to non-null.  This could cause concurrent access to the
   * data member, even though it's a mere pointer (but we don't count on the alleged "atomicity" of this; it is
   * generally considered not safe).
   *
   * Now, as of this writing, there's exactly one spot where `ostream<<` could be invoked from thread U while
   * it is being assigned in thread W: `async_accept_log_in()` is called by our internal code in
   * Session_server and only once; the only thing that can occur in thread U until the log-in response
   * handler is executed is that Server_session_impl dtor is called.  Before that dtor stops that thread W,
   * it does print the Server_session_impl once as of this writing.  Therefore, out of sheer caution, this guy
   * is `atomic<>`.  That said there could be other such invocations, as code might change during maintenance
   * in the future, in which case this `atomic<>`ness will quietly come in handy.
   */
  std::atomic<const Client_app*> m_cli_app_ptr;
 
  /// See srv_namespace().
  Shared_name m_srv_namespace;
 
  /// See cli_namespace().
  Shared_name m_cli_namespace;
 
  /// See set_on_err_func().
  flow::async::Task_asio_err m_on_err_func;
 
  /// See on_passive_open_channel_func_or_empty().
  On_passive_open_channel_func m_on_passive_open_channel_func_or_empty;
 
  /**
   * Starts falsy; becomes forever truthy (with a specific #Error_code that will not change thereafter)
   * once hose() is called (with that truthy value).  Note hose() may not be called before PEER state, which implies
   * #m_on_err_func is non-empty.
   *
   * ### Concurrency ###
   * See hose() and hosed() doc headers.  TL;DR: It is up to the caller to only call those, basically, from
   * thread W only.
   */
  Error_code m_peer_state_err_code_or_ok;
 
  /**
   * Read/written in thread W only; `false` until the `*_session_impl` dtor -- as its first act -- posts a task
   * calling set_dtor_started(); from then on session-hosing errors are not emitted to the user's on-error handler.
   *
   * ### Rationale ###
   * Nothing can -- or tries to -- prevent the on-error handler firing *during* dtor execution in general:
   * an async error can always lose a race against the user's decision to destroy the `Session`, and the user
   * must be able to handle that regardless (via standard async hygiene; or by using the `sync_io` variant,
   * where nothing fires uninvited).  However: (1) with Graceful_finisher in play, the graceful session-end
   * protocol *reliably* delivers events -- the opposing `GracefulSessionEnd`, then the opposing side's channel
   * closure -- while our dtor is blocked in `Graceful_finisher::on_dtor_start()`: absent this guard those would
   * fire "errors" at the user as a deterministic part of every orderly shutdown; and (2) sans-Graceful_finisher,
   * this shortens the coincidental-timing window as a by-product.  (1) is what drove this.  Either way an emission
   * suppressed hereby is pointless by definition: the user, having entered the dtor, no longer benefits from any
   * notification.
   *
   * ### Why one flag, here, and not one per interested party ###
   * The interested parties are the `*_session_impl` (its hose() sites, chiefly its on-master-channel-error
   * handler) and Graceful_finisher (its `GracefulSessionEnd` handler).  Both must observe the *same* moment as
   * the guarantee point -- the dtor's first W-post -- or a `GracefulSessionEnd` queued in thread W between
   * that post and any later, party-specific, flag-setting post would slip through and hose() after all.
   * (Graceful_finisher's on_dtor_start() posts at least one task after the dtor's first post(); so that gap is a
   * real thing.)  Hence the single flag lives here, in the class both parties can see.
   */
  bool m_dtor_started;
}; // class Session_base
 
/// Internally used macro; public API users should disregard (same deal as in struc/channel.hpp).
#define TEMPLATE_SESSION_BASE \
  template<schema::MqType MQ_TYPE_OR_NONE, bool TRANSMIT_NATIVE_HANDLES, typename Mdt_payload>
/// Internally used macro; public API users should disregard (same deal as in struc/channel.hpp).
#define CLASS_SESSION_BASE \
  Session_base<MQ_TYPE_OR_NONE, TRANSMIT_NATIVE_HANDLES, Mdt_payload>
 
// Template implementations.
 
TEMPLATE_SESSION_BASE
CLASS_SESSION_BASE::Session_base(const Client_app& cli_app_ref, const Server_app& srv_app_ref,
                                 flow::async::Task_asio_err&& on_err_func,
                                 On_passive_open_channel_func&& on_passive_open_channel_func_or_empty_arg) :
  /* This is, basically, the (protected) ctor for Client_session, followed by Client_session_impl::async_connect().
   * By the time we're constructed, by definition Client_app and, therefore-ish, m_on_*_func are known.
   * However: m_srv_namespace is looked-up (in file system, as of this writing) at the forthcoming async_connect();
   * empty for now; set_srv_namespace() invoked at that time.   m_cli_namespace is returned by server in
   * log-in response; set_cli_namespace() invoked at that time. */
 
  m_srv_app_ref(srv_app_ref), // Not copied!
  m_cli_app_ptr(&cli_app_ref), // Ditto!
  m_on_err_func(std::move(on_err_func)),
  m_on_passive_open_channel_func_or_empty(std::move(on_passive_open_channel_func_or_empty_arg)),
  // m_srv_namespace and m_cli_namespace remain .empty() for now.
  m_dtor_started(false)
{
  // Yep.
}
 
TEMPLATE_SESSION_BASE
CLASS_SESSION_BASE::Session_base(const Server_app& srv_app_ref) :
  /* This is, basically, the (protected) ctor for Server_session, followed by
   * Server_session_impl::async_accept_log_in(), itself `protected` as of this writing.  Various items are not
   * known until that async-succeeds; Client_app is not known (until it is identified in the log-in exchange), and,
   * therefore-ish, m_on_*_func are also unknown.
   * Server_app is still known from the very start.  Moreover, due to the way m_srv_namespace is determined (our own
   * PID), we can confidently initialize that from the start (ourselves being the server).
   * m_cli_namespace is generated on server side during log-in proceedings; set_cli_namespace() invoked at that time. */
 
  m_srv_app_ref(srv_app_ref), // Not copied!
  m_cli_app_ptr(nullptr),
  // As promised, we know our own srv-namespace:
  m_srv_namespace(Shared_name::ct(std::to_string(util::Process_credentials::own_process_id()))),
  // m_cli_namespace + m_on_err_func and m_on_passive_open_channel_func_or_empty remain .empty() for now.
  m_dtor_started(false)
{
  // Yep.
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_cli_app_ptr(const Client_app* cli_app_ptr_new)
{
  assert(cli_app_ptr_new && "As of this writing cli_app_ptr should be set at most once, from null to non-null.");
 
#ifndef NDEBUG
  const auto prev = m_cli_app_ptr.exchange(cli_app_ptr_new);
  assert((!prev) && "As of this writing cli_app_ptr should be set at most once, from null to non-null.");
#else
  m_cli_app_ptr = cli_app_ptr_new;
#endif
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_srv_namespace(Shared_name&& srv_namespace_new)
{
  assert(!srv_namespace_new.empty());
  assert((srv_namespace_new != Shared_name::S_SENTINEL) && "That is a reserved sentinel value.");
  assert(m_srv_namespace.empty() && "As of this writing srv_namespace should be set at most once, from empty.");
 
  m_srv_namespace = std::move(srv_namespace_new);
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_cli_namespace(Shared_name&& cli_namespace_new)
{
  assert(!cli_namespace_new.empty());
  assert((cli_namespace_new != Shared_name::S_SENTINEL) && "That is a reserved sentinel value.");
  assert(m_cli_namespace.empty() && "As of this writing cli_namespace should be set at most once, from empty.");
 
  m_cli_namespace = std::move(cli_namespace_new);
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::reset_namespaces()
{
  m_srv_namespace.clear();
  m_cli_namespace.clear();
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_on_passive_open_channel_func(On_passive_open_channel_func&& on_passive_open_channel_func)
{
  assert(m_on_passive_open_channel_func_or_empty.empty());
  m_on_passive_open_channel_func_or_empty = std::move(on_passive_open_channel_func);
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_on_err_func(flow::async::Task_asio_err&& on_err_func_arg)
{
  assert(!on_err_func_arg.empty());
  assert(m_on_err_func.empty() && "Call set_on_err_func() once at most and only if not already set through ctor.");
 
  m_on_err_func = std::move(on_err_func_arg);
}
 
TEMPLATE_SESSION_BASE
const Client_app* CLASS_SESSION_BASE::cli_app_ptr() const
{
  return m_cli_app_ptr.load(); // Fetch atomic<> payload and return it.  (Could write `return m_cli_app_ptr;` too.)
}
 
TEMPLATE_SESSION_BASE
const Shared_name& CLASS_SESSION_BASE::srv_namespace() const
{
  return m_srv_namespace;
}
 
TEMPLATE_SESSION_BASE
const Shared_name& CLASS_SESSION_BASE::cli_namespace() const
{
  return m_cli_namespace;
}
 
TEMPLATE_SESSION_BASE
const typename CLASS_SESSION_BASE::On_passive_open_channel_func&
  CLASS_SESSION_BASE::on_passive_open_channel_func_or_empty() const
{
  return m_on_passive_open_channel_func_or_empty;
}
 
TEMPLATE_SESSION_BASE
bool CLASS_SESSION_BASE::on_err_func_set() const
{
  return !m_on_err_func.empty();
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::hose(const Error_code& err_code)
{
  assert((!hosed()) && "By contract do not call unless hosed() is false.");
 
  m_peer_state_err_code_or_ok = err_code;
  m_on_err_func(err_code);
} // Session_base::hose()
 
TEMPLATE_SESSION_BASE
bool CLASS_SESSION_BASE::hosed() const
{
  assert(on_err_func_set() && "By contract do not call until PEER state -- when on-error handler must be known.");
  return bool(m_peer_state_err_code_or_ok);
}
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::set_dtor_started()
{
  assert((!m_dtor_started) && "By contract call at most once.");
  m_dtor_started = true;
}
 
TEMPLATE_SESSION_BASE
bool CLASS_SESSION_BASE::dtor_started() const
{
  return m_dtor_started;
}
 
TEMPLATE_SESSION_BASE
Shared_name CLASS_SESSION_BASE::cur_ns_store_mutex_absolute_name() const
{
  using std::to_string;
 
  // Global scope.  Pertains to Server_app::m_name, *all* processes thereof, *all* Client_apps, *all* sessions.
  auto mutex_name = build_conventional_shared_name(Shared_name::S_RESOURCE_TYPE_ID_MUTEX,
                                                   Shared_name::ct(m_srv_app_ref.m_name),
                                                   Shared_name::S_SENTINEL);
  /* That was the standard stuff.  Now the us-specific stuff:
   *
   * Above we specified it's a mutex; now for what's this mutex?  It's for the CNS (Current Namespace Store),
   * a/k/a the PID file. */
  mutex_name /= "cur_ns_store";
 
  /* Further differentiate the names by owner/permissions that will be used when creating the file.
   * This is in response to seeing, during testing, the situation where (in that particular case)
   * m_permissions_level_for_client_apps changed from an accidental S_NO_ACCESS to S_(something sane).
   * The file was created with 0000 permissions, then later no one could open it (or create it, since it existed)
   * due to it being untouchable, essentially forever without manual intervention (such as `rm`ing it).
   * The following, which should be otherwise harmless, should at least help avoid such conflicts.
   * (Is it harmless?  Almost, yes: more files can potentially be created and sit around until reboot;
   * but they are tiny.) */
  mutex_name /= 'u';
  mutex_name += to_string(m_srv_app_ref.m_user_id);
  mutex_name += 'g';
  mutex_name += to_string(m_srv_app_ref.m_group_id);
  mutex_name += 'p';
  mutex_name += to_string(int(m_srv_app_ref.m_permissions_level_for_client_apps));
 
  return mutex_name;
} // Session_base::cur_ns_store_mutex_absolute_name()
 
TEMPLATE_SESSION_BASE
fs::path CLASS_SESSION_BASE::cur_ns_store_absolute_path() const
{
  using Path = fs::path;
 
  Path path(m_srv_app_ref.m_kernel_persistent_run_dir_override.empty()
              ? util::IPC_KERNEL_PERSISTENT_RUN_DIR
              : m_srv_app_ref.m_kernel_persistent_run_dir_override);
  path /= m_srv_app_ref.m_name;
  path += ".libipc-cns.pid";
  return path;
}
 
TEMPLATE_SESSION_BASE
Shared_name CLASS_SESSION_BASE::session_master_socket_stream_acceptor_absolute_name() const
{
  assert((!m_srv_namespace.empty()) && "Serv-namespace must have been set by set_srv_namespace() by now.");
 
  // Global scope.  Pertains to Server_app::m_name, the particular process, *all* Client_apps, *all* sessions.
  auto acc_name = build_conventional_shared_name(transport::Native_socket_stream_acceptor::S_RESOURCE_TYPE_ID,
                                                 Shared_name::ct(m_srv_app_ref.m_name),
                                                 m_srv_namespace);
  /* That was the standard stuff.  Now the us-specific stuff:
   *
   * Above we specified it's a socket stream acceptor; but which one/for what?  It's the required one for
   * any further IPC; hence S_SENTINEL, `0`.  (As of this writing, in ipc::session anyway, we don't have any other use
   * for Native_socket_stream_acceptor (other socket streams are opened through the ones birthed through this one).
   * If we did, we could use `1`, `2`, etc., or something.  Anyway Shared_name::S_SENTINEL here is a solid choice.) */
  acc_name /= Shared_name::S_SENTINEL;
 
  return acc_name;
} // Session_base::session_master_socket_stream_acceptor_absolute_name()
 
TEMPLATE_SESSION_BASE
CLASS_SESSION_BASE::Graceful_finisher::Graceful_finisher(flow::log::Logger* logger_ptr, Session_base* this_session,
                                                         flow::async::Single_thread_task_loop* async_worker,
                                                         Master_structured_channel* master_channel) :
  flow::log::Log_context(logger_ptr, Log_component::S_SESSION),
  m_this_session(this_session),
  m_async_worker(async_worker),
  m_master_channel(master_channel)
{
  // We are in thread W.
 
  m_master_channel->expect_msg(Master_structured_channel::Msg_which_in::GRACEFUL_SESSION_END,
                               [this](auto&&)
                               // (The message doesn't matter and contains no fields; only that we received it matters.)
  {
    // We are in thread Wc (unspecified, really struc::Channel async callback thread).
    m_async_worker->post([this]()
    {
      // We are in thread W.  (We need to be to safely trigger hosed() and hose().)
 
      FLOW_LOG_INFO("Received GracefulSessionEnd from opposing Session object along session master channel "
                    "[" << *m_master_channel << "].  Will emit to local user as graceful-end "
                    "error; and mark it down.  If our Session dtor is running, it shall return soon.  If it has "
                    "not yet run, it shall return immediately upon starting to execute.");
 
      m_opposing_session_done.set_value();
      /* Can m_opposing_session_done already be set?  (That would throw promise exception, and we don't catch it
       * above.)  Answer: no, in the current protocol at least: Only once can it be sent (hence .expect_msg(), not
       * .expect_msgs()); and if *m_master_channel were to be hosed, then that would occur either after
       * expect_msg() firing handler, or it would occur instead of it.  (We *do* handle the "after" case in
       * on_master_channel_hosed(), where we do indeed catch the exception.) */
 
      if ((!m_this_session->dtor_started()) && (!m_this_session->hosed()))
      {
        m_this_session->hose(error::Code::S_SESSION_FINISHED); // It'll log.
      }
      /* else if (dtor_started()): Session is already being destroyed of the user's volition; don't bug them.
       *   (See Session_base::m_dtor_started doc header, particularly on why we consult *that* flag, set by the
       *   dtor's first W-post, rather than one we would set ourselves in on_dtor_start(): the latter would
       *   leave a window between the two post()s wherein we'd still hose() -- the very thing being prevented.)
       * else if (hosed()): No need to hose if already hosed (the usual pre-condition for hose()). */
    }); // m_async_worker->post()
  }); // m_master_channel->expect_msg(GRACEFUL_SESSION_END)
} // Session_base::Graceful_finisher::Graceful_finisher()
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::Graceful_finisher::on_master_channel_hosed()
{
  using boost::promise_already_satisfied;
 
  // We are in thread W.
  try
  {
    m_opposing_session_done.set_value();
  }
  catch (const promise_already_satisfied&)
  {
    // Interesting.  @todo Maybe log?
  }
} // Session_base::Graceful_finisher::on_master_channel_hosed()
 
TEMPLATE_SESSION_BASE
void CLASS_SESSION_BASE::Graceful_finisher::on_dtor_start()
{
  using flow::async::Synchronicity;
 
  // We are in thread U.  In fact we are in a Client/Server_session_impl dtor... but at its very start.
 
  /* Thread W must be fine and running; per algorithm (see Graceful_finisher class doc header).
   * It's only safe to access m_master_channel from there. */
  m_async_worker->post([&]()
  {
    FLOW_LOG_INFO("In Session object dtor sending GracefulSessionEnd to opposing Session object along "
                  "session master channel [" << *m_master_channel << "] -- if at all possible.");
    assert(m_this_session->dtor_started()
           && "By contract the *_session_impl dtor posts set_dtor_started() before any Graceful_finisher task.");
 
    auto msg = m_master_channel->create_msg();
    msg.body_root()->initGracefulSessionEnd();
 
    Error_code err_code_ignored;
    m_master_channel->send(&msg, nullptr, &err_code_ignored);
    /* Whatever happened, it logged.  We make a best effort; if channel is hosed or send fails, then the other
     * side shall detect it via its on_master_channel_hosed() presumably and set its m_opposing_session_done, hence
     * its on_dtor_start() will proceed. */
  }, Synchronicity::S_ASYNC_AND_AWAIT_CONCURRENT_COMPLETION); // m_async_worker->post()
 
  // That would've been non-blocking.  So now lastly:
 
  FLOW_LOG_INFO("In Session object dtor we now await sign that the opposing Session object dtor has also started; "
                "only then will we proceed with destroying our Session object: e.g., maybe they hold handles "
                "to objects in a SHM-arena we would deinitialize.  If their dtor has already been called, we will "
                "proceed immediately.  If not, we will now wait for that.  This would only block if the opposing "
                "side's user code neglects to destroy Session object on error; or has some kind of blocking "
                "operation in progress before destroying Session object.  "
                "Session master channel: [" << *m_master_channel << "].");
  m_opposing_session_done.get_future().wait();
  FLOW_LOG_INFO("In Session object dtor: Done awaiting sign that the opposing Session object dtor has also started.  "
                "Will now proceed with our Session object destruction.");
} // Session_base::Graceful_finisher::on_dtor_start()
 
TEMPLATE_SESSION_BASE
typename CLASS_SESSION_BASE::Structured_msg_reader_config
  CLASS_SESSION_BASE::heap_reader_config(flow::log::Logger* logger_ptr) // Static.
{
  return transport::struc::Heap_reader::Config
           { logger_ptr, 0,
             // Default rcv-stats target: pure-heap global singleton.
             &transport::struc::stat::Heap_serializer_global_stats::get().stats_mutable_default().m_rcv };
} // Session_base::heap_reader_config()
 
#undef CLASS_SESSION_BASE
#undef TEMPLATE_SESSION_BASE
 
} // namespace ipc::session