monitor.hpp

Go to the documentation of this file.

// Copyright (c) 2020-2024 Simons Foundation
//
// 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
//
//     http://www.apache.org/licenses/LICENSE-2.0.txt
//
// 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.
//
// Authors: Philipp Dumitrescu, Thomas Hahn, Olivier Parcollet, Nils Wentzell

 
#pragma once
 
#include "./communicator.hpp"
#include "./macros.hpp"
#include "./utils.hpp"
 
#include <mpi.h>
 
#include <vector>
#include <unistd.h>
 
namespace mpi {

  class monitor {
    // Future struct for non-blocking MPI communication.
    struct future {
      // MPI request of the non-blocking MPI call.
      MPI_Request request{};
 
      // 0 means that no event has occurred, 1 means that an event has occurred.
      int event = 0;
    };
 
    public:
    monitor(mpi::communicator c) : comm_(c.duplicate()) {
      if (comm_.rank() == 0) {
        root_futures_.resize(c.size() - 1);
        for (int rank = 1; rank < c.size(); ++rank) {
          check_mpi_call(MPI_Irecv(&(root_futures_[rank - 1].event), 1, MPI_INT, rank, rank, comm_.get(), &(root_futures_[rank - 1].request)),
                         "MPI_Irecv");
        }
      } else {
        check_mpi_call(MPI_Ibcast(&any_event_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_any_), "MPI_Ibcast");
        check_mpi_call(MPI_Ibcast(&all_events_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_all_), "MPI_Ibcast");
      }
    }

    monitor(monitor const &) = delete;

    monitor &operator=(monitor const &) = delete;

    ~monitor() { finalize_communications(); }

    void report_local_event() {
      // prevent sending multiple signals
      if (local_event_ or finalized_) { return; }
 
      // a local event has occurred
      local_event_ = 1;
      if (comm_.rank() == 0) {
        // on root process, check all other nodes and perform necessary broadcasts
        root_check_nodes_and_bcast();
      } else {
        // on non-root processes, let the root process know about the local event
        check_mpi_call(MPI_Isend(&local_event_, 1, MPI_INT, 0, comm_.rank(), comm_.get(), &req_isent_), "MPI_Isend");
      }
    }

    [[nodiscard]] bool event_on_this_rank() const noexcept { return local_event_; }

    [[nodiscard]] bool event_on_any_rank() {
      // if final_communications() has already been called, any_event == 0 if no event has occurred, otherwise it is 1
      if (finalized_) return any_event_;
 
      // if a local event has occurred, we return true
      if (local_event_) return true;
 
      // on the root process, we first check the status of all non-root processes, perform the necessary broadcasts and
      // return true if an event has occurred
      if (comm_.rank() == 0) {
        root_check_nodes_and_bcast();
        return any_event_;
      }
 
      // on non-root processes, we check the status of the corresponding broadcast and return true if an event has
      // occurred
      MPI_Status status;
      int has_received = 0;
      check_mpi_call(MPI_Test(&req_ibcast_any_, &has_received, &status), "MPI_Test");
      return has_received and any_event_;
    }

    [[nodiscard]] bool event_on_all_ranks() {
      // if final_communications() has already been called, all_events == 0 if an event has not occurred on every
      // process, otherwise it is 1
      if (finalized_) return all_events_;
 
      // on the root process, we first check the status of all non-root processes, perform the necessary broadcasts and
      // return true if an event has occurred on all of them
      if (comm_.rank() == 0) {
        root_check_nodes_and_bcast();
        return all_events_;
      }
 
      // on non-root processes, we check the status of the broadcast and return true if an event has occurred on all
      // processes
      MPI_Status status;
      int has_received = 0;
      check_mpi_call(MPI_Test(&req_ibcast_all_, &has_received, &status), "MPI_Test");
      return has_received and all_events_;
    }

    void finalize_communications() {
      // prevent multiple calls
      if (finalized_) return;
 
      if (comm_.rank() == 0) {
        // on root process, wait for all non-root processes to finish their MPI_Isend calls
        while (root_check_nodes_and_bcast()) {
          usleep(100); // 100 us (micro seconds)
        }
        // and perform broadcasts in case they have not been done yet
        if (not any_event_) { check_mpi_call(MPI_Ibcast(&any_event_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_any_), "MPI_Ibcast"); }
        if (not all_events_) { check_mpi_call(MPI_Ibcast(&all_events_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_all_), "MPI_Ibcast"); }
      } else {
        // on non-root processes, perform MPI_Isend call in case it has not been done yet
        if (not local_event_) { check_mpi_call(MPI_Isend(&local_event_, 1, MPI_INT, 0, comm_.rank(), comm_.get(), &req_isent_), "MPI_Isend"); }
      }
 
      // all nodes wait for the broadcasts to be completed
      MPI_Status status_any, status_all;
      check_mpi_call(MPI_Wait(&req_ibcast_any_, &status_any), "MPI_Wait");
      check_mpi_call(MPI_Wait(&req_ibcast_all_, &status_all), "MPI_Wait");
 
      // free the communicator
      comm_.free();
      finalized_ = true;
    }
 
    private:
    // Root process checks the status of all non-root processes, performs necessary broadcasts and returns a boolean
    // that is true if at least one non-root process has not performed its MPI_Isend call yet.
    bool root_check_nodes_and_bcast() {
      EXPECTS(!finalized_);
      EXPECTS(comm_.rank() == 0);
      bool any      = false;
      bool all      = true;
      bool finished = true;
      for (auto &[request, rank_event] : root_futures_) {
        MPI_Status status;
        int rank_received = 0;
        check_mpi_call(MPI_Test(&request, &rank_received, &status), "MPI_Test");
        any |= (rank_received and rank_event);
        all &= (rank_received and rank_event);
        finished &= rank_received;
      }
      if (not any_event_ and (any or local_event_)) {
        any_event_ = 1;
        check_mpi_call(MPI_Ibcast(&any_event_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_any_), "MPI_Ibcast");
      }
      if (not all_events_ and all and local_event_) {
        all_events_ = 1;
        check_mpi_call(MPI_Ibcast(&all_events_, 1, MPI_INT, 0, comm_.get(), &req_ibcast_all_), "MPI_Ibcast");
      }
      return not finished;
    }
 
    private:
    // MPI communicator.
    mpi::communicator comm_;
 
    // Future objects stored on the root process for local events on non-root processes.
    std::vector<future> root_futures_;
 
    // MPI request for the broadcasting done on the root process in case an event has occurred on any rank.
    MPI_Request req_ibcast_any_{};
 
    // MPI request for the broadcasting done on the root process in case an event has occurred on all ranks.
    MPI_Request req_ibcast_all_{};
 
    // MPI request for the sending done on non-root processes.
    MPI_Request req_isent_{};
 
    // Set to 1, if a local event has occurred on this process.
    int local_event_ = 0;
 
    // Set to 1, if an event has occurred on any process.
    int any_event_ = 0;
 
    // Set to 1, if an event has occurred on all processes.
    int all_events_ = 0;
 
    // Set to true, if finalize_communications() has been called.
    bool finalized_ = false;
  };
 
} // namespace mpi