group.hpp

/*
//@HEADER
// ************************************************************************
//
//
//            _|_|_|_|  _|_|_|_|  _|      _|  _|_|_|  _|      _|
//            _|        _|        _|_|    _|    _|      _|  _|
//            _|_|_|    _|_|_|    _|  _|  _|    _|        _|
//            _|        _|        _|    _|_|    _|      _|  _|
//            _|        _|_|_|_|  _|      _|  _|_|_|  _|      _|
//
//
//
//
// Copyright (C) 2016 Rutgers University and Sandia Corporation
//
// Under the terms of Contract DE-AC04-94AL85000 with Sandia Corporation,
// the U.S. Government retains certain rights in this software.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
// 1. Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
//
// 3. Neither the name of the Corporation nor the names of the
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY SANDIA CORPORATION "AS IS" AND ANY
// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SANDIA CORPORATION OR THE
// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
// Author Marc Gamell, Eric Valenzuela, Keita Teranishi, Manish Parashar,
//        Michael Heroux, and Matthew Whitlock
//
// Questions? Contact Keita Teranishi (knteran@sandia.gov) and
//                    Marc Gamell (mgamell@cac.rutgers.edu)
//
// ************************************************************************
//@HEADER
*/
#ifndef __FENIX_DATA_GROUP_H__
#define __FENIX_DATA_GROUP_H__
 
#include <unordered_map>
#include <map>
#include <memory>
#include <vector>
#include <set>
#include <source_location>
 
#include <mpi.h>
#include "fenix.h"
#include "fenix/data/member.hpp"
#include "fenix/data/subset.hpp"
#include "fenix/mpixx/comm.hpp"
 
#define __FENIX_DEFAULT_GROUP_SIZE 32
 
namespace fenix::data {
 
//We keep basic bookkeeping info here, policy specific
//information is kept by the policy's data type.
struct DataGroup {
  DataGroup(int groupid, MPI_Comm c, int timestart, int depth, int policy);
 
  virtual ~DataGroup();
 
  // Create the cohort group for this policy
  virtual MPI_Group create_cohort() = 0;
 
  // Synchronize timestamps across cohort members
  virtual void sync_timestamps();
 
  // Initialize group after construction (creates cohort_comm, syncs state)
  virtual void init() {
    // Free old cohort if re-initializing (e.g., during recovery)
    if (cohort != MPI_GROUP_NULL) {
      MPI_Group_free(&cohort);
    }
    cohort      = create_cohort();
    cohort_comm = mpixx::Comm::create_group(comm, cohort, 0);
    cohort_size = cohort_comm.size();
    cohort_rank = cohort_comm.rank();
    sync_timestamps();
  }
 
  int groupid;
  mpixx::CommRef comm;
  int comm_size;
  int current_rank;
  int timestart;
  int timestamp;
  int depth;
  int policy_name;
 
  std::set<std::shared_ptr<DataMember>, DataMemberIdComparator> members;
  std::vector<int> member_order;
 
  std::set<int, std::greater<int>> timestamps; // Reverse sorted: newest first
 
  MPI_Group cohort = MPI_GROUP_NULL;
  mpixx::Comm cohort_comm;
  int cohort_size = -1;
  int cohort_rank = -1;
 
  std::vector<int> get_member_ids();
  //Search for id, returning null if not found.
  DataMember* search_member(int id);
  //As search_member, but throw if not found
  DataMember* find_member(
    int id, std::source_location loc = std::source_location::current()
  );
  void member_create(int id, void* data, int count, MPI_Datatype datatype);
  void member_create(
    int id, void* data, int count, MPI_Datatype datatype, SerializeFunc& s
  );
  void member_create(const util::DataBuffer& serialized);
 
  virtual void member_delete(int memberid);
 
  // Create and emplace a policy-specific member into members map
  virtual void emplace_member(DataMember&& member);
 
  virtual void get_redundant_policy(int* name, void* value) = 0;
 
  virtual void member_repair(int member_id);
  virtual void member_restore_from_rank(
    int member_id, void* target_bugger, int max, int timestamp, int source_rank
  ) = 0;
 
  // Default implementations using timestamps
  virtual void commit();
  virtual void snapshot_delete(int timestamp);
  virtual int get_number_of_snapshots();
  virtual int get_snapshot_at_position(int position);
  virtual std::vector<int> get_snapshots();
 
  // Revoke cohort_comm (called before recovery to invalidate communicator)
  virtual void revoke();
 
  MPI_Group get_cohort() const { return cohort; }
 
  // String representation, for debugging
  std::string str();
};
 
struct DataGroupIdComparator {
  using is_transparent = void; // Enables heterogeneous lookup
 
  bool operator()(
    const std::shared_ptr<DataGroup>& a, const std::shared_ptr<DataGroup>& b
  ) const;
 
  bool operator()(const std::shared_ptr<DataGroup>& a, int id) const;
 
  bool operator()(int id, const std::shared_ptr<DataGroup>& a) const;
};
 
} //end namespace fenix::data
 
#endif // FENIX_DATA_GROUP_H