derecho/multicast_group.cpp
/**
 *
 * @param _members A list of node IDs of members in this group
 * @param my_node_id The rank (ID) of this node in the group
 * @param sst The SST this group will use; created by the GMS (membership
 * service) for this group.
 * @param _callbacks A set of functions to call when messages have reached
 * various levels of stability
 */
@param _members A list of node IDs of members in this group
@param my_node_id The rank (ID) of this node in the group
@param sst The SST this group will use; created by the GMS (membership
service) for this group.
@param _callbacks A set of functions to call when messages have reached
various levels of stability
@param _members A list of node IDs of members in this group @param my_node_id The rank ( ID ) of this node in the group @param sst The SST this group will use ; created by the GMS ( membership service ) for this group . @param _callbacks A set of functions to call when messages have reached various levels of stability
MulticastGroup::MulticastGroup(
        std::vector<node_id_t> _members, node_id_t my_node_id,
        std::shared_ptr<DerechoSST> sst,
        CallbackSet callbacks,
        uint32_t total_num_subgroups,
        const std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings_by_id,
        const DerechoParams derecho_params,
        const persistence_manager_callbacks_t& _persistence_manager_callbacks,
        std::vector<char> already_failed)
        : logger(spdlog::get("debug_log")),
          members(_members),
          num_members(members.size()),
          member_index(index_of(members, my_node_id)),
          block_size(derecho_params.block_size),
          max_msg_size(compute_max_msg_size(derecho_params.max_payload_size, derecho_params.block_size)),
          type(derecho_params.type),
          window_size(derecho_params.window_size),
          callbacks(callbacks),
          total_num_subgroups(total_num_subgroups),
          subgroup_settings(subgroup_settings_by_id),
          received_intervals(sst->num_received.size(), {-1, -1}),
          rdmc_group_num_offset(0),
          future_message_indices(total_num_subgroups, 0),
          next_sends(total_num_subgroups),
          pending_sends(total_num_subgroups),
          current_sends(total_num_subgroups),
          next_message_to_deliver(total_num_subgroups),
          sender_timeout(derecho_params.timeout_ms),
          sst(sst),
          sst_multicast_group_ptrs(total_num_subgroups),
          last_transfer_medium(total_num_subgroups),
          persistence_manager_callbacks(_persistence_manager_callbacks) {
    assert(window_size >= 1);
    for(uint i = 0; i < num_members; ++i) {
        node_id_to_sst_index[members[i]] = i;
    }
    for(const auto p : subgroup_settings_by_id) {
        auto num_shard_members = p.second.members.size();
        while(free_message_buffers[p.first].size() < window_size * num_shard_members) {
            free_message_buffers[p.first].emplace_back(max_msg_size);
        }
    }
    initialize_sst_row();
    bool no_member_failed = true;
    if(already_failed.size()) {
        for(uint i = 0; i < num_members; ++i) {
            if(already_failed[i]) {
                no_member_failed = false;
                break;
            }
        }
    }
    if(!already_failed.size() || no_member_failed) {
        rdmc_sst_groups_created = create_rdmc_sst_groups();
    }
    register_predicates();
    sender_thread = std::thread(&MulticastGroup::send_loop, this);
    timeout_thread = std::thread(&MulticastGroup::check_failures_loop, this);
}
MulticastGroup :: MulticastGroup ( std :: vector < node_id_t > _members , node_id_t my_node_id , std :: shared_ptr < DerechoSST > sst , CallbackSet callbacks , uint32_t total_num_subgroups , const std :: map < subgroup_id_t , SubgroupSettings > & subgroup_settings_by_id , const DerechoParams derecho_params , const persistence_manager_callbacks_t & _persistence_manager_callbacks , std :: vector < char > already_failed ) : logger ( spdlog :: get ( " " ) ) , members ( _members ) , num_members ( members . size ( ) ) , member_index ( index_of ( members , my_node_id ) ) , block_size ( derecho_params . block_size ) , max_msg_size ( compute_max_msg_size ( derecho_params . max_payload_size , derecho_params . block_size ) ) , type ( derecho_params . type ) , window_size ( derecho_params . window_size ) , callbacks ( callbacks ) , total_num_subgroups ( total_num_subgroups ) , subgroup_settings ( subgroup_settings_by_id ) , received_intervals ( sst -> num_received . size ( ) , { -1 , -1 } ) , rdmc_group_num_offset ( 0 ) , future_message_indices ( total_num_subgroups , 0 ) , next_sends ( total_num_subgroups ) , pending_sends ( total_num_subgroups ) , current_sends ( total_num_subgroups ) , next_message_to_deliver ( total_num_subgroups ) , sender_timeout ( derecho_params . timeout_ms ) , sst ( sst ) , sst_multicast_group_ptrs ( total_num_subgroups ) , last_transfer_medium ( total_num_subgroups ) , persistence_manager_callbacks ( _persistence_manager_callbacks ) { assert ( window_size >= 1 ) ; for ( uint i = 0 ; i < num_members ; ++ i ) { node_id_to_sst_index [ members [ i ] ] = i ; } for ( const auto p : subgroup_settings_by_id ) { auto num_shard_members = p . second . members . size ( ) ; while ( free_message_buffers [ p . first ] . size ( ) < window_size * num_shard_members ) { free_message_buffers [ p . first ] . emplace_back ( max_msg_size ) ; } } initialize_sst_row ( ) ; bool no_member_failed = true ; if ( already_failed . size ( ) ) { for ( uint i = 0 ; i < num_members ; ++ i ) { if ( already_failed [ i ] ) { no_member_failed = false ; break ; } } } if ( ! already_failed . size ( ) || no_member_failed ) { rdmc_sst_groups_created = create_rdmc_sst_groups ( ) ; } register_predicates ( ) ; sender_thread = std :: thread ( & MulticastGroup :: send_loop , this ) ; timeout_thread = std :: thread ( & MulticastGroup :: check_failures_loop , this ) ; }
@param _members A list of node IDs of members in this group
 @param my_node_id The rank (ID) of this node in the group
 @param sst The SST this group will use; created by the GMS (membership
 service) for this group.
@param _members A list of node IDs of members in this group @param my_node_id The rank ( ID ) of this node in the group @param sst The SST this group will use ; created by the GMS ( membership service ) for this group .
// if groups are created successfully, rdmc_sst_groups_created will be set to true