Server.h

Go to the documentation of this file.

 
 
#ifndef SERVER_H
#define SERVER_H
#include <mpi.h>
#include <memory>
#include <mutex>
#include <queue>
#include <vector>
#include <map>
#include <unordered_map>
#include <fstream>
 
struct DETCheckpoint;
struct PendingBranch;
#include "ParallelUtilities.h"
#include "ParallelGlobal.h"
#include <filesystem>
 
class VisualizationServer;
 
struct TreeState {
    std::mutex mu;                                  
    std::vector<SequenceReport> completed;          
    std::map<int, SequenceOutputData> output_data;  
    int total_clients = 0;                          
    int sequences_completed = 0;                    
    int queue_depth = 0;                            
    int num_free_clients = 0;                       
    long memory_mb = 0;                             
    double start_time = 0.0;                        
};
 
namespace fs = std::filesystem;
 
using PU = ParallelUtilities;
 
class Server {
public:
    explicit Server(int rank, int viz_port = 0);
    ~Server();  
 
    void Shutdown();
 
    void Simulation();
 
    void merge_outputs();
 
    int getSequencesCompleted() const { return sequences_completed_; }
 
    bool stoppedAtCheckpoint() const { return stopped_at_checkpoint_; }
 
    int find_next_free_client_round_robin();
 
    void dispatch_available_tasks();
 
    int find_free_client_except(int exclude_rank);
 
    void write_checkpoint();
 
    void dispatch_resume_tasks();
 
private:
    bool checkpoint_pending_ = false;  
    bool stopped_at_checkpoint_ = false; 
    int num_free_clients;               
    int num_queue{0};                   
    int rank;                           
    int sequence_counter_ = 1;          
    int sequences_completed_ = 0;       
    int last_checkpoint_seq_ = -1;      
    std::map<int, int> resume_source_refs_; 
    int next_candidate = 1;             
 
    struct RestartTask {
        int source_rank;                
        int local_id;                   
        int sequence_id;                
        int branch_index;               
        int parent_sequence_id;         
    };
 
    std::queue<RestartTask> tasks_queue;    
    std::unique_ptr<int[]> clients_busy;   
    MPI_Status status_server{};
 
    // Occupancy trace (--occupancy-log, default off). Wall-clock speedup cannot
    // distinguish a slow scheduler from an empty queue; this records which one
    // it is. CSV columns: elapsed_s, queue_depth, busy_workers, total_workers,
    // sequences_completed. The file is truncated at start and a row is written
    // on change rather than on a timer: queue depth and busy count are step
    // functions that only move at events, so change-logging is lossless and a
    // periodic sample would just alias it. If the file cannot be opened the
    // run continues without it (a warning is logged).
    std::ofstream occupancy_log_;
    double occupancy_t0_ = 0.0;         
    int last_queue_ = -1;               
    int last_busy_ = -1;                
 
    // Visualization state
    std::shared_ptr<TreeState> viz_state_;
    std::unique_ptr<VisualizationServer> viz_server_;
    int viz_port_ = 0;
    std::unordered_map<int, int> rank_current_sequence_;  
 
    // Resume state: the loaded checkpoint and its tasks not yet dispatched
    std::unique_ptr<DETCheckpoint> resume_ckpt_;   
    std::vector<PendingBranch> resume_pending_;    
 
    bool dispatch_one_resume_task();
};
 
#endif //SERVER_H