text: /**
   *  Copyright (c) 2015 by Contributors
   */
  #ifndef PS_INTERNAL_POSTOFFICE_H_
  #define PS_INTERNAL_POSTOFFICE_H_
  #include <mutex>
  #include <algorithm>
  #include <vector>
  #include "ps/range.h"
  #include "ps/internal/env.h"
  #include "ps/internal/customer.h"
  #include "ps/internal/van.h"
  #include "ps/internal/elastic_training.h"
  
  namespace ps {
  /**
   * \brief the center of the system
   * 
   */
  class Postoffice {
   public:
    /**
     * \brief return the singleton object
     */
    static Postoffice* Get() {
      static Postoffice e; return &e;
    }
    /** \brief get the van */
    Van* van() { return van_; }
    /**
     * \brief start the system
     *
     * This function will block until every nodes are started.
     * \param argv0 the program name, used for logging.
     * \param do_barrier whether to block until every nodes are started.
     */
    void Start(int customer_id, const char* argv0, const bool do_barrier);
    /**
     * \brief terminate the system
     *
     * All nodes should call this function before existing. 
     * \param do_barrier whether to do block until every node is finalized, default true.
     */
    void Finalize(const int customer_id, const bool do_barrier = true);
    /**
     * \brief add an customer to the system. threadsafe
     */
    void AddCustomer(Customer* customer);
    /**
     * \brief remove a customer by given it's id. threasafe
     */
    void RemoveCustomer(Customer* customer);
    /**
     * \brief get the customer by id, threadsafe
     * \param app_id the application id
     * \param customer_id the customer id
     * \param timeout timeout in sec
     * \return return nullptr if doesn't exist and timeout
     */
    Customer* GetCustomer(int app_id, int customer_id, int timeout = 0) const;
    /**
     * \brief get the id of a node (group), threadsafe
     *
     * if it is a  node group, return the list of node ids in this
     * group. otherwise, return {node_id}
     */
    const std::unordered_set<int>& GetNodeIDs(int node_id) const {
      const auto it = node_ids_.find(node_id);
      CHECK(it != node_ids_.cend()) << "node " << node_id << " doesn't exist";
      return it->second;
    }
    /**
     * \brief return the key ranges of all server nodes
     */
    const std::vector<Range>& GetServerKeyRanges();
    /**
     * \brief the template of a callback
     */
    using Callback = std::function<void()>;
    /**
     * \brief Register a callback to the system which is called after Finalize()
     *
     * The following codes are equal
     * \code {cpp}
     * RegisterExitCallback(cb);
     * Finalize();
     * \endcode
     *
     * \code {cpp}
     * Finalize();
     * cb();
     * \endcode
     * \param cb the callback function
     */
    void RegisterExitCallback(const Callback& cb) {
      exit_callback_ = cb;
    }
    /**
     * \brief convert from a worker rank into a node id
     * \param rank the worker rank
     */
    static inline int WorkerRankToID(int rank) {
      return rank * 2 + 9;
    }
    /**
     * \brief convert from a server rank into a node id
     * \param rank the server rank
     */
    static inline int ServerRankToID(int rank) {
      return rank * 2 + 8;
    }
  
    static inline bool isWorkerId(int id){
      return id > 8 && ((id - 9) % 2 == 0);
    }
    /**
     * \brief convert from a node id into a server or worker rank
     * \param id the node id
     */
    static inline int IDtoRank(int id) {
  #ifdef _MSC_VER
  #undef max
  #endif
      return std::max((id - 8) / 2, 0);
    }
    /** \brief Returns the number of worker nodes */
    int num_workers() const { return num_workers_; }
    /** \brief Returns the number of server nodes */
    int num_servers() const { return num_servers_; }
  
    std::shared_ptr<ETNodeManager> et_node_manager() const {
      return et_node_manager_;
    }
  
    bool removed_hosts() const { return removed_hosts_; }
  
    /** \brief Returns the rank of this node in its group
     *
     * Each worker will have a unique rank within [0, NumWorkers()). So are
     * servers. This function is available only after \ref Start has been called.
     */
    int my_rank() const { return van_->GetMyRank();}
    /** \brief Returns true if this node is a worker node */
    int is_worker() const { return is_worker_; }
    /** \brief Returns true if this node is a server node. */
    int is_server() const { return is_server_; }
    /** \brief Returns true if this node is a scheduler node. */
    int is_scheduler() const { return is_scheduler_; }
    /** \brief Returns the verbose level. */
    int verbose() const { return verbose_; }
    /** \brief Return whether this node is a recovery node */
    bool is_recovery() const { return van_->my_node().is_recovery; }
  
    int is_new_worker() const {
      return is_new_worker_;
    }
    /**
     * \brief barrier
     * \param node_id the barrier group id
     */
    void Barrier(int customer_id, int node_group, bool is_membership_change_barrier=false, const std::vector<std::pair<std::string, std::string> >
                                & data={});
  
    /**
     * \brief process a control message, called by van
     * \param the received message
     */
    void Manage(const Message& recv);
    /**
     * \brief update the heartbeat record map
     * \param node_id the \ref Node id
     * \param t the last received heartbeat time
     */
    void UpdateHeartbeat(int node_id, time_t t) {
      std::lock_guard<std::mutex> lk(heartbeat_mu_);
      heartbeats_[node_id] = t;
    }
    /**
     * \brief get node ids that haven't reported heartbeats for over t seconds
     * \param t timeout in sec
     */
    std::vector<int> GetDeadNodes(int t = 60);
  
    /**
     * \brief update variables in memory which corresponds to env_var
     * currently supported num_worker
     */
    void updateEnvironmentVariable(const std::string& env_var, const std::string& val, const std::string& data, Meta* nodes);
    void notifyUpdateEnvReceived();
    /**
     * This will sync node_ids_ map with current workerIds.
     * if node_ids_ group has a worker which is not in workerIds, those worker will be removed from map
     * It will add the worker ids to their corresponding group if it is not in the map
    */
    void syncWorkerNodeIdsGroup(std::set<int> workerIds);
  
   private:
    Postoffice();
    ~Postoffice() { delete van_; }
  
    void InitEnvironment();
  
    void updateNumWorker(const char* val, const std::unordered_set<int>& removed_node_ids, Meta* nodes);
    void addWorkerNodeIdToGroups(int id);
    /**
     * This parses string data which is , separated to set of hostIps
     */
    std::unordered_set<int> parseRemovedNodeStringAndGetIds(const std::string& data);
  
    Van* van_;
    mutable std::mutex mu_;
    // app_id -> (customer_id -> customer pointer)
    std::unordered_map<int, std::unordered_map<int, Customer*>> customers_;
    std::unordered_map<int, std::unordered_set<int>> node_ids_;
    std::mutex server_key_ranges_mu_;
    std::vector<Range> server_key_ranges_;
    bool is_worker_, is_server_, is_scheduler_, is_new_worker_;
    int num_servers_, num_workers_;
    std::unordered_map<int, std::unordered_map<int, bool> > barrier_done_;
    int verbose_;
    std::mutex barrier_mu_;
    std::condition_variable barrier_cond_;
    std::atomic_int update_req_sent;
    std::mutex heartbeat_mu_;
    std::mutex start_mu_;
    int init_stage_ = 0;
  
    // This will keep track of if there are removed hosts. In case this is true,
    // then van will check if sender id is valid after  receiving messages.
    bool removed_hosts_ = false;
    std::unordered_map<int, time_t> heartbeats_;
    Callback exit_callback_;
    /** \brief Holding a shared_ptr to prevent it from being destructed too early */
    std::shared_ptr<Environment> env_ref_;
    std::shared_ptr<ETNodeManager> et_node_manager_;
    time_t start_time_;
    DISALLOW_COPY_AND_ASSIGN(Postoffice);
  };
  
  /** \brief verbose log */
  #define PS_VLOG(x) LOG_IF(INFO, x <= Postoffice::Get()->verbose())
  }  // namespace ps
  #endif  // PS_INTERNAL_POSTOFFICE_H_
  