supervision_manager.hpp#

Defined in header hpx/supervision/server/supervision_manager.hpp.

See Public API for a list of names and headers that are part of the public HPX API.

namespace hpx

namespace supervision

namespace server

Variables

constexpr char const *const supervision_manager_name = "supervision_manager/"#

struct observer_entry#

Public Members

hpx::id_type agent#

std::uint64_t epoch_filter#

struct supervision_manager : public hpx::components::fixed_component_base<supervision_manager>#

Public Types

using base_type = components::fixed_component_base<supervision_manager>#

Public Functions

supervision_manager()#

~supervision_manager()#

void register_server_instance(char const *service_name, std::uint32_t locality_id, error_code &ec = throws)#

void unregister_server_instance(error_code &ec = throws) const#

publish_result publish_event(hpx::id_type const &target, event ev, std::uint64_t epoch)#

Applies epoch/terminal-latch state-mutation rules, resolves any await_terminal() waiters affected by the transition, and invokes registered per-target lifecycle observers and activity observers for this call.

Parameters

target – Target whose supervision state is updated. Must live on this locality.
ev – Lifecycle event to apply.
epoch – Epoch associated with ev.

Returns

The result of applying ev to target’s state.

lifecycle_state query_state(hpx::id_type const &target)#

hpx::id_type register_observer(hpx::id_type const &target, hpx::id_type const &agent, std::uint64_t epoch_filter = static_cast<std::uint64_t>(-1))#

void unregister_observer(hpx::id_type const &observer_handle)#

void remove_target(hpx::id_type const &target)#

Clears all locally tracked state for target.

Unlike unregister_observer(), which removes a single previously registered observer handle (and leaves any recorded lifecycle state for its target(s) intact), remove_target() unconditionally forgets every piece of local bookkeeping this supervision manager holds for target - its recorded lifecycle state and current epoch (see publish_event()), and any per-target observers still registered for it (see register_observer()) - regardless of any specific observer handle. Intended for callers that know target will never be queried or observed again locally (e.g. after a failed registration that seeded some state for it, or once a peer has been evicted) and want to reclaim that local state instead of letting it accumulate indefinitely.

void tidy()#

Unconditionally clears all locally tracked state.

Snapshots every target currently present in states_ or observers_ under mtx_, releases the lock, then calls remove_target() for each one in turn, reusing its existing per-target teardown (waiter invalidation, activity-observer notifications, agent/state cleanup) instead of duplicating that logic here. Targets added concurrently with (or after) the snapshot are left untouched. If states_ and observers_ are empty at snapshot time, this is a no-op: tidy() never reports an error and never fires any notification of its own for the “nothing to do” case. Local-only: unlike remove_target(), this is not exposed as a remote action.

hpx::id_type register_activity_observer(hpx::id_type const &agent, std::uint64_t epoch_filter = static_cast<std::uint64_t>(-1))#

Registers a locality-scoped activity observer.

Replays targets that are already active when registration takes its snapshot.

The snapshot of the target’s tracked state and the insertion of the observer into the tracked set happen atomically under mtx_, which guarantees the observer receives exactly one notification for the target’s current state: either the replay (if already active at registration time) or a live transition that raced with registration, but never both and never neither.

Delivery of that notification (replay or live) happens after mtx_ is released and is therefore not ordered relative to any other concurrent live notification for the same target; callers must not assume replay is delivered before or after a racing live event.

void unregister_activity_observer(hpx::id_type const &observer_handle)#

Unregisters an activity observer.

The handle must have been returned by register_activity_observer(). As with unregister_observer(), no orphaned callbacks fire after this call completes. observer_handle must have been returned by register_activity_observer(); a handle returned by register_observer() instead (i.e. found in agents_ rather than activity_observers_) is rejected.

hpx::future<lifecycle_state> await_terminal(hpx::id_type const &target, std::uint64_t epoch = 0, std::chrono::steady_clock::duration timeout = (std::chrono::steady_clock::duration::max)())#

dispatch_outcome check_admission(hpx::id_type const &target, std::uint64_t epoch = 0) const noexcept#

Protected Types

using waiters_t = std::vector<waiter_entry>#

using stale_waiters_t = std::vector<std::pair<std::uint64_t, waiters_t>>#

using expired_waiters_t = std::vector<expired_waiter>#

Protected Functions

hpx::future<void> fire_events(hpx::id_type const &target, lifecycle_event_notification const &notification)#

hpx::future<void> fire_event(hpx::id_type const &target, hpx::id_type const &agent, lifecycle_event_notification notification)#

hpx::future<void> deliver_activity_notification(activity_notification const &notification, std::vector<observer_entry> const &observers)#

hpx::future<void> fire_activity_event(hpx::id_type const &agent, activity_notification notification)#

void record_error(hpx::id_type const &target, std::uint64_t expected_sequence_number, hpx::error_code const &ec)#

bool unregister_observer_target(hpx::id_type const &target, hpx::id_type const &observer_handle)#

std::uint64_t current_epoch_for(hpx::id_type const &target) const#

apply_event_outcome apply_event_and_resolve(hpx::id_type const &target, event ev, std::uint64_t epoch)#

waiters_t drain_terminal_waiters_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &target, std::uint64_t epoch, event last_event)#

stale_waiters_t drain_stale_waiters_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &target, std::uint64_t epoch)#

apply_result apply_new_epoch_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &target, supervision::event ev, std::uint64_t epoch)#

std::optional<apply_result> apply_current_epoch_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &target, supervision::event ev, std::uint64_t epoch)#

expired_waiters_t drain_expired_waiters_locked(std::unique_lock<hpx::spinlock> &l, std::chrono::steady_clock::time_point now)#

stale_waiters_t drain_all_waiters_for_target_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &target)#

void sweep_expired_waiters()#

void arm_sweep_timer_locked(std::unique_lock<hpx::spinlock> &&l)#

bool sweep_timer_callback()#

void recompute_earliest_deadline_locked(std::unique_lock<hpx::spinlock> &l)#

void remove_target_from_agents_locked(std::unique_lock<hpx::spinlock> &l, hpx::id_type const &agent, hpx::id_type const &target)#

Protected Static Functions

static void invalidate_stale_waiters(hpx::id_type const &target, std::uint64_t epoch, stale_waiters_t &stale)#

static void resolve_terminal_waiters(lifecycle_event_notification const &notification, waiters_t &to_resolve)#

static void invalidate_expired_waiters(expired_waiters_t &expired)#

Private Members

mutable hpx::spinlock mtx_#

hpx::spinlock sweep_timer_mtx_#

mutable std::string instance_name_#

std::map<hpx::id_type, lifecycle_state> states_#

std::map<hpx::id_type, std::vector<observer_entry>> observers_#

std::map<hpx::id_type, std::vector<hpx::id_type>> agents_#

std::vector<observer_entry> activity_observers_#

std::map<hpx::id_type, std::chrono::steady_clock::time_point> activated_at_#

std::map<waiter_key, waiters_t> waiters_#

std::chrono::steady_clock::time_point earliest_deadline_ = (std::chrono::steady_clock::time_point::max)()#

hpx::util::pool_timer sweep_timer_#

std::chrono::steady_clock::time_point armed_deadline_ = (std::chrono::steady_clock::time_point::max)()#

bool shutting_down_ = false#

std::size_t callbacks_in_flight_ = 0#

hpx::lcos::local::detail::condition_variable shutdown_cv_#

struct apply_event_outcome#

Public Members

publish_result result#

lifecycle_event_notification notification#

bool had_state_before = false#

std::vector<observer_entry> activity_observers_snapshot#

struct apply_result#

Public Members

lifecycle_event_notification notification#

waiters_t to_resolve#

struct expired_waiter#

Public Members

hpx::id_type target#

std::uint64_t epoch#

hpx::promise<lifecycle_state> promise#

struct waiter_entry#

Public Members

hpx::promise<lifecycle_state> promise#

std::chrono::steady_clock::time_point deadline#

struct waiter_key#

Public Members

hpx::id_type target#

std::uint64_t epoch#

Friends

inline friend bool operator<(waiter_key const &lhs, waiter_key const &rhs)#

inline friend bool operator==(waiter_key const &lhs, waiter_key const &rhs)#