MPI server that manages the DET task queue and client dispatch. More...

#include <Server.h>

Classes
struct	RestartTask
	A metadata-only restart task: the data stays on the source client. More...

Public Member Functions
	Server (int rank, int viz_port=0)
	Set up the client table and, when `viz_port` is nonzero, start the visualization server; also runs Scheduler::initiationForSimulation(). More...

	~Server ()
	Defined in Server.cpp (VisualizationServer complete type required) More...

void	Shutdown ()
	Send client_shutdown to every client rank and stop the visualization server. Under –merge=now each client runs its self-merge in its shutdown handler and then waits in the MPI barrier that merge_outputs() enters, so merge_outputs() must follow. More...

void	Simulation ()
	Main server loop: poll for messages, dispatch tasks, until the tree is complete. More...

void	merge_outputs ()
	Merge per-rank Output/ directories into the shared project Output/. More...

int	getSequencesCompleted () const
	Number of sequences completed, including those restored from a resumed checkpoint. More...

bool	stoppedAtCheckpoint () const
	True when the run ended at a checkpoint (–checkpoint-stop, the interrupt menu's checkpoint option, or its exit option after an interval checkpoint was written). The outputs are then intermediate: –merge=defer must not skip the inline merge, and PE nodes keep their natural IDs. More...

int	find_next_free_client_round_robin ()
	Claim the next free client, round-robin from next_candidate. The returned client is marked busy and num_free_clients is decremented. More...

void	dispatch_available_tasks ()
	Make one pass over the task queue. More...

int	find_free_client_except (int exclude_rank)
	Claim the next free client via round-robin, skipping `exclude_rank`. More...

void	write_checkpoint ()
	Gather held branch data from every client and write `<checkpoint_dir>/checkpoint_seq_<sequences_completed>.dat`. More...

void	dispatch_resume_tasks ()
	Load the –resume checkpoint and dispatch its pending tasks. More...

Private Member Functions
bool	dispatch_one_resume_task ()
	Dispatch the last task of resume_pending_ to a free client. More...

Private Attributes
bool	checkpoint_pending_ = false
	Checkpoint requested, waiting for clients to drain. More...

bool	stopped_at_checkpoint_ = false
	Run ended at a checkpoint: outputs are intermediate. More...

int	num_free_clients
	Number of currently idle clients. More...

int	num_queue {0}
	Unused (the queue depth is tasks_queue.size()) More...

int	rank
	This process rank (always 0) More...

int	sequence_counter_ = 1
	Next sequence ID to assign (0 is the initial sequence) More...

int	sequences_completed_ = 0
	Total sequences finished so far. More...

int	last_checkpoint_seq_ = -1
	Sequence count at the last checkpoint written from the message path (-1 = none) More...

std::map< int, int >	resume_source_refs_
	Undispatched resume tasks per checkpoint source blob. More...

int	next_candidate = 1
	Round-robin pointer (starts at rank 1) More...

std::queue< RestartTask >	tasks_queue
	Pending metadata-only restart tasks. More...

std::unique_ptr< int[]>	clients_busy
	Per-client busy flag (0=free, 1=busy) More...

MPI_Status	status_server {}

std::ofstream	occupancy_log_

double	occupancy_t0_ = 0.0
	MPI_Wtime() when the log was opened [s]. More...

int	last_queue_ = -1
	last logged depth, to suppress no-op rows More...

int	last_busy_ = -1
	last logged busy-worker count More...

std::shared_ptr< TreeState >	viz_state_

std::unique_ptr< VisualizationServer >	viz_server_

int	viz_port_ = 0

std::unordered_map< int, int >	rank_current_sequence_
	rank -> sequence_id being run More...

std::unique_ptr< DETCheckpoint >	resume_ckpt_
	Loaded –resume checkpoint (null in a fresh run) More...

std::vector< PendingBranch >	resume_pending_
	Undispatched tasks; consumed from the back, index-aligned with resume_ckpt_->task_restart_states. More...

Detailed Description

MPI server that manages the DET task queue and client dispatch.

Runs on rank 0 only, single-threaded (the optional visualization HTTP server runs in its own thread and shares only TreeState). Call order used by parallelexec::parallel_init: construct, Simulation(), Shutdown(), then merge_outputs() unless the merge is deferred.

Definition at line 62 of file Server.h.

Constructor & Destructor Documentation

◆ Server()

Server::Server	(	int	rank,
		int	viz_port = `0`
	)

explicit

Set up the client table and, when viz_port is nonzero, start the visualization server; also runs Scheduler::initiationForSimulation().

Parameters

rank	MPI rank (always 0 for server).
viz_port	Port for visualization HTTP server (0=disabled).

Definition at line 50 of file Server.cpp.

References clients_busy, g_sim, Scheduler::initiationForSimulation(), PG::log, next_candidate, num_free_clients, rank, PG::size, SimulationContext::SScheduler, viz_port_, viz_server_, and viz_state_.

Here is the call graph for this function:

◆ ~Server()

Server::~Server ( )

default

Defined in Server.cpp (VisualizationServer complete type required)

Member Function Documentation

◆ dispatch_available_tasks()

void Server::dispatch_available_tasks ( )

Make one pass over the task queue.

A task whose source client is busy goes back to the end of the queue. Otherwise the source is told to send the held data to another free client (client_send_to_peer) or, if no other client is free, to run the branch itself (client_run_held). Stops when no client is free. Logs a "[MEM]" line with the server RSS after every pass.

Definition at line 406 of file Server.cpp.

References Server::RestartTask::branch_index, PG::client_run_held, PG::client_send_to_peer, clients_busy, find_free_client_except(), get_rss_mb(), Server::RestartTask::local_id, PG::log, num_free_clients, Server::RestartTask::parent_sequence_id, rank, rank_current_sequence_, Server::RestartTask::sequence_id, Server::RestartTask::source_rank, and tasks_queue.

Referenced by Simulation().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ dispatch_one_resume_task()

bool Server::dispatch_one_resume_task ( )

private

Dispatch the last task of resume_pending_ to a free client.

Sends the task's RestartNodeInfo (with its restart blob attached for the send only), the merged BranchRecord, the sequence id and the branch index with tag client_restart_run, and frees the source blob once its last task is dispatched. A task with no restart state is logged and dropped.

Returns: true if a task was sent; false if there is no task, no free client, or the task had no restart state.

Definition at line 1106 of file Server.cpp.

References PG::client_restart_run, clients_busy, find_next_free_client_round_robin(), PG::log, num_free_clients, rank, rank_current_sequence_, resume_ckpt_, resume_pending_, resume_source_refs_, and sequence_counter_.

Referenced by dispatch_resume_tasks(), and Simulation().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ dispatch_resume_tasks()

void Server::dispatch_resume_tasks ( )

Load the –resume checkpoint and dispatch its pending tasks.

Calls MPI_Abort if the configuration fingerprint differs or if the project Output/PE already holds a finalized (sequentially renamed) run. Overrides –compress-queue to the checkpoint's setting, restores the sequence counters, derives the branch-ID epoch for this run, and sends one task per free client; the rest are dispatched from Simulation() as clients free up, ahead of queued tasks.

Definition at line 1002 of file Server.cpp.

References SimConfig::compress_queue, SimulationContext::config, SimulationContext::config_fingerprint, dispatch_one_resume_task(), e, g_sim, Scheduler::getPEDirectory(), DETCheckpoint::load(), PG::log, num_free_clients, rank, resume_ckpt_, SimConfig::resume_file, resume_pending_, resume_source_refs_, sequence_counter_, sequences_completed_, SimulationContext::SScheduler, and viz_state_.

Referenced by Simulation().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ find_free_client_except()

int Server::find_free_client_except ( int exclude_rank )

Claim the next free client via round-robin, skipping exclude_rank.

Returns: Rank of the claimed client (marked busy), or -1 if none available.

Definition at line 391 of file Server.cpp.

References clients_busy, next_candidate, num_free_clients, and PG::size.

Referenced by dispatch_available_tasks().

Here is the caller graph for this function:

◆ find_next_free_client_round_robin()

int Server::find_next_free_client_round_robin ( )

Claim the next free client, round-robin from next_candidate. The returned client is marked busy and num_free_clients is decremented.

Returns: Rank of the claimed client, or -1 if none is free.

Definition at line 376 of file Server.cpp.

References clients_busy, next_candidate, num_free_clients, and PG::size.

Referenced by dispatch_one_resume_task(), and Simulation().

Here is the caller graph for this function:

◆ getSequencesCompleted()

int Server::getSequencesCompleted ( ) const

inline

Number of sequences completed, including those restored from a resumed checkpoint.

Definition at line 110 of file Server.h.

References sequences_completed_.

◆ merge_outputs()

void Server::merge_outputs ( )

Merge per-rank Output/ directories into the shared project Output/.

Reads "../rank_N/Output/" relative to the rank_0 working directory. –merge=now (default): enters an MPI barrier with the clients, which have renamed their ES/KPS files into place and staged logs as parts/<name>.rank_N; the server renames the remaining per-file PE nodes in ascending rank order (highest rank wins a name collision), concatenates the parts in rank order and deletes them. –merge=serial: the server copies every file itself. –merge=defer reaches this function only after a checkpoint stop; the server then renames and concatenates everything itself, with no barrier. In every mode it rebuilds ES_summary.txt and, unless the run stopped at a checkpoint, renumbers the PE nodes sequentially. Logs "Merged N sequences, M PE nodes into <dir> in X s (rename|serial copy)". A file that can be neither renamed nor copied is logged and skipped.

Definition at line 460 of file Server.cpp.

References adsidac::pe::aggregate_basename(), SimConfig::aggregate_pe_output, adsidac::merge::build_es_summary(), adsidac::pe::collect(), SimulationContext::config, e, adsidac::pe::finalize_sequential(), g_sim, Scheduler::getESDirectory(), Scheduler::getKPSDirectory(), Scheduler::getOutputDirectory(), Scheduler::getPEDirectory(), PG::log, SimConfig::merge_mode, rank, adsidac::pe::rewrite_sequential(), PG::size, SimulationContext::SScheduler, and stopped_at_checkpoint_.

Here is the call graph for this function:

◆ Shutdown()

void Server::Shutdown ( )

Send client_shutdown to every client rank and stop the visualization server. Under –merge=now each client runs its self-merge in its shutdown handler and then waits in the MPI barrier that merge_outputs() enters, so merge_outputs() must follow.

Definition at line 77 of file Server.cpp.

References PG::client_shutdown, PG::log, PG::size, and viz_server_.

◆ Simulation()

void Server::Simulation ( )

Main server loop: poll for messages, dispatch tasks, until the tree is complete.

Initializes the plant, control panel, procedures and crew modules (the server deserializes crew snapshots), installs the SIGUSR1 (checkpoint request) and SIGINT (interrupt menu) handlers, opens the occupancy log if –occupancy-log is set, then either resumes (–resume) or sends client_simulate_this for sequence 0. Polls with MPI_Iprobe and a 1 ms sleep. Returns when the queue and the resume list are empty, every client is free and no checkpoint is pending; after a checkpoint written under –checkpoint-stop; or when the user picks "exit" in the interrupt menu. A checkpoint is written only once every client is idle, so it waits for the longest running sequence.

Definition at line 86 of file Server.cpp.

References SimulationContext::Action_Taker, SimulationContext::All_procedure, SimConfig::checkpoint_dir, SimConfig::checkpoint_interval, checkpoint_pending_, SimConfig::checkpoint_stop, PG::client_peer_assign, PG::client_simulate_this, clients_busy, SimulationContext::config, SimulationContext::Decision_Maker, Operator::diagnosis_matrix, dispatch_available_tasks(), dispatch_one_resume_task(), dispatch_resume_tasks(), find_next_free_client_round_robin(), g_sim, get_rss_mb(), Scheduler::getProjectDirectory(), SimulationContext::Indicator, Operator::initiationForSimulation(), ConvertToString::initiationForSimulation(), Procedures::initiationForSimulation(), SystemModule::initiationForSimulation(), Operator::knowledge_base, last_busy_, last_checkpoint_seq_, last_queue_, PG::log, mpi_checkpoint_requested, mpi_sigint_handler(), mpi_sigint_received, mpi_sigusr1_handler(), num_free_clients, sim::Operator::OAT, SimConfig::occupancy_log, occupancy_log_, occupancy_t0_, sim::Operator::ODM, SimulationContext::Plant, rank, rank_current_sequence_, ControlPanel::readInfoFromFile(), SimConfig::resume_file, resume_pending_, sequence_counter_, SequenceOutputData::sequence_id, sequences_completed_, PG::server_branch_meta, PG::server_done_simulating, PG::server_done_with_report, PG::server_output_data, PG::server_send_complete, adsidac::set_shared_control_panel(), adsidac::set_shared_diagnosis_matrix(), adsidac::set_shared_kb(), adsidac::set_shared_procedures(), show_mpi_interrupt_menu(), PG::size, Server::RestartTask::source_rank, SimulationContext::SScheduler, status_server, stopped_at_checkpoint_, tasks_queue, SimulationContext::Translator, viz_state_, and write_checkpoint().

Here is the call graph for this function:

◆ stoppedAtCheckpoint()

bool Server::stoppedAtCheckpoint ( ) const

inline

True when the run ended at a checkpoint (–checkpoint-stop, the interrupt menu's checkpoint option, or its exit option after an interval checkpoint was written). The outputs are then intermediate: –merge=defer must not skip the inline merge, and PE nodes keep their natural IDs.

Definition at line 116 of file Server.h.

References stopped_at_checkpoint_.

◆ write_checkpoint()

void Server::write_checkpoint ( )

Gather held branch data from every client and write <checkpoint_dir>/checkpoint_seq_<sequences_completed>.dat.

The caller must ensure every client is idle (each one answers the client_checkpoint request from its message loop). Merges the gathered BranchRecords into one record keyed by cluster identity, carries over tasks of a resumed checkpoint that are still undispatched, and prints "Resume with: --resume=<path>" on stdout. Write failures are reported by DETCheckpoint::save() on stderr and do not stop the run.

Definition at line 612 of file Server.cpp.

References BranchCluster::all_branch, DETCheckpoint::branch_center_blob, BranchRecord::branch_cluster, PendingBranch::branch_index, DETCheckpoint::branch_record_blobs, SimConfig::checkpoint_dir, PendingBranch::cid, DETCheckpoint::cid_source, PG::client_checkpoint, DETCheckpoint::completed_reports, DETCheckpoint::compress_queue, SimConfig::compress_queue, SimulationContext::config, DETCheckpoint::config_fingerprint, SimulationContext::config_fingerprint, g_sim, BranchRecord::getIDOfRecentGeneratedES(), PG::log, DETCheckpoint::max_branch_id, PendingBranch::parent_sequence_id, DETCheckpoint::pending_tasks, rank, DETCheckpoint::restart_states, resume_ckpt_, resume_pending_, DETCheckpoint::save(), DETCheckpoint::sequence_counter, sequence_counter_, PendingBranch::sequence_id, DETCheckpoint::sequences_completed, sequences_completed_, BranchRecord::setIDOfRecentGeneratedBranch(), BranchRecord::setIDOfRecentGeneratedBranchCluster(), BranchRecord::setIDOfRecentGeneratedES(), PG::size, DETCheckpoint::source_blobs, DETCheckpoint::task_restart_states, tasks_queue, and viz_state_.

Referenced by Simulation().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

◆ checkpoint_pending_

bool Server::checkpoint_pending_ = false

private

Checkpoint requested, waiting for clients to drain.

Definition at line 158 of file Server.h.

Referenced by Simulation().

◆ clients_busy

std::unique_ptr<int[]> Server::clients_busy

private

Per-client busy flag (0=free, 1=busy)

Definition at line 179 of file Server.h.

Referenced by dispatch_available_tasks(), dispatch_one_resume_task(), find_free_client_except(), find_next_free_client_round_robin(), Server(), and Simulation().

◆ last_busy_

int Server::last_busy_ = -1

private

last logged busy-worker count

Definition at line 193 of file Server.h.

Referenced by Simulation().

◆ last_checkpoint_seq_

int Server::last_checkpoint_seq_ = -1

private

Sequence count at the last checkpoint written from the message path (-1 = none)

Definition at line 165 of file Server.h.

Referenced by Simulation().

◆ last_queue_

int Server::last_queue_ = -1

private

last logged depth, to suppress no-op rows

Definition at line 192 of file Server.h.

Referenced by Simulation().

◆ next_candidate

int Server::next_candidate = 1

private

Round-robin pointer (starts at rank 1)

Definition at line 167 of file Server.h.

Referenced by find_free_client_except(), find_next_free_client_round_robin(), and Server().

◆ num_free_clients

int Server::num_free_clients

private

Number of currently idle clients.

Definition at line 160 of file Server.h.

Referenced by dispatch_available_tasks(), dispatch_one_resume_task(), dispatch_resume_tasks(), find_free_client_except(), find_next_free_client_round_robin(), Server(), and Simulation().

◆ num_queue

int Server::num_queue {0}

private

Unused (the queue depth is tasks_queue.size())

Definition at line 161 of file Server.h.

◆ occupancy_log_

std::ofstream Server::occupancy_log_

private

Definition at line 190 of file Server.h.

Referenced by Simulation().

◆ occupancy_t0_

double Server::occupancy_t0_ = 0.0

private

MPI_Wtime() when the log was opened [s].

Definition at line 191 of file Server.h.

Referenced by Simulation().

◆ rank

int Server::rank

private

This process rank (always 0)

Definition at line 162 of file Server.h.

Referenced by dispatch_available_tasks(), dispatch_one_resume_task(), dispatch_resume_tasks(), merge_outputs(), Server(), Simulation(), and write_checkpoint().

◆ rank_current_sequence_

std::unordered_map<int, int> Server::rank_current_sequence_

private

rank -> sequence_id being run

Definition at line 199 of file Server.h.

Referenced by dispatch_available_tasks(), dispatch_one_resume_task(), and Simulation().

◆ resume_ckpt_

std::unique_ptr<DETCheckpoint> Server::resume_ckpt_

private

Loaded –resume checkpoint (null in a fresh run)

Definition at line 202 of file Server.h.

Referenced by dispatch_one_resume_task(), dispatch_resume_tasks(), and write_checkpoint().

◆ resume_pending_

std::vector<PendingBranch> Server::resume_pending_

private

Undispatched tasks; consumed from the back, index-aligned with resume_ckpt_->task_restart_states.

Definition at line 203 of file Server.h.

Referenced by dispatch_one_resume_task(), dispatch_resume_tasks(), Simulation(), and write_checkpoint().

◆ resume_source_refs_

std::map<int, int> Server::resume_source_refs_

private

Undispatched resume tasks per checkpoint source blob.

Definition at line 166 of file Server.h.

Referenced by dispatch_one_resume_task(), and dispatch_resume_tasks().

◆ sequence_counter_

int Server::sequence_counter_ = 1

private

Next sequence ID to assign (0 is the initial sequence)

Definition at line 163 of file Server.h.

Referenced by dispatch_one_resume_task(), dispatch_resume_tasks(), Simulation(), and write_checkpoint().

◆ sequences_completed_

int Server::sequences_completed_ = 0

private

Total sequences finished so far.

Definition at line 164 of file Server.h.

Referenced by dispatch_resume_tasks(), getSequencesCompleted(), Simulation(), and write_checkpoint().

◆ status_server

MPI_Status Server::status_server {}

private

Definition at line 180 of file Server.h.

Referenced by Simulation().

◆ stopped_at_checkpoint_

bool Server::stopped_at_checkpoint_ = false

private

Run ended at a checkpoint: outputs are intermediate.

Definition at line 159 of file Server.h.

Referenced by merge_outputs(), Simulation(), and stoppedAtCheckpoint().

◆ tasks_queue

std::queue<RestartTask> Server::tasks_queue

private

Pending metadata-only restart tasks.

Definition at line 178 of file Server.h.

Referenced by dispatch_available_tasks(), Simulation(), and write_checkpoint().

◆ viz_port_

int Server::viz_port_ = 0

private

Definition at line 198 of file Server.h.

Referenced by Server().

◆ viz_server_

std::unique_ptr<VisualizationServer> Server::viz_server_

private

Definition at line 197 of file Server.h.

Referenced by Server(), and Shutdown().

◆ viz_state_

std::shared_ptr<TreeState> Server::viz_state_

private

Definition at line 196 of file Server.h.

Referenced by dispatch_resume_tasks(), Server(), Simulation(), and write_checkpoint().

The documentation for this class was generated from the following files:

src/ADSIDAC/ParallelExec/Server.h
src/ADSIDAC/ParallelExec/Server.cpp

Classes

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ Server()

◆ ~Server()

Member Function Documentation

◆ dispatch_available_tasks()

◆ dispatch_one_resume_task()

◆ dispatch_resume_tasks()

◆ find_free_client_except()

◆ find_next_free_client_round_robin()

◆ getSequencesCompleted()

◆ merge_outputs()

◆ Shutdown()

◆ Simulation()

◆ stoppedAtCheckpoint()

◆ write_checkpoint()

Member Data Documentation

◆ checkpoint_pending_

◆ clients_busy

◆ last_busy_

◆ last_checkpoint_seq_

◆ last_queue_

◆ next_candidate

◆ num_free_clients

◆ num_queue

◆ occupancy_log_

◆ occupancy_t0_

◆ rank

◆ rank_current_sequence_

◆ resume_ckpt_

◆ resume_pending_

◆ resume_source_refs_

◆ sequence_counter_

◆ sequences_completed_

◆ status_server

◆ stopped_at_checkpoint_

◆ tasks_queue

◆ viz_port_

◆ viz_server_

◆ viz_state_