#include <algorithm>
#include <cassert>
#include <chrono>
#include <limits>
#include <thread>

#include <derecho/core/detail/derecho_internal.hpp>
#include <derecho/core/detail/multicast_group.hpp>
#include <derecho/persistent/Persistent.hpp>
#include <derecho/rdmc/detail/util.hpp>
#include <derecho/utils/logger.hpp>
#include <derecho/utils/time.h>

namespace derecho {

/**
 * Helper function to find the index of an element in a container.
 */
template <class T, class U>
size_t index_of(T container, U elem) {
    size_t n = 0;
    for(auto it = begin(container); it != end(container); ++it) {
        if(*it == elem) return n;
        n++;
    }
    return container.size();
}

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,
        unsigned int sender_timeout,
        const subgroup_post_next_version_func_t& post_next_version_callback,
        PersistenceManager& persistence_manager_ref,
        std::vector<char> already_failed)
        : members(_members),
          num_members(members.size()),
          member_index(index_of(members, my_node_id)),
          callbacks(callbacks),
          total_num_subgroups(total_num_subgroups),
          subgroup_settings_map(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),
          committed_sst_index(total_num_subgroups, -1),
          num_nulls_queued(total_num_subgroups, 0),
          first_null_index(total_num_subgroups, -1),
          pending_sends(total_num_subgroups),
          current_sends(total_num_subgroups),
          next_message_to_deliver(total_num_subgroups),
          minimum_persisted_version(total_num_subgroups, persistent::INVALID_VERSION),
          minimum_verified_version(total_num_subgroups, persistent::INVALID_VERSION),
          sender_timeout(sender_timeout),
          sst(sst),
          sst_multicast_group_ptrs(total_num_subgroups),
          last_transfer_medium(total_num_subgroups),
          post_next_version_callback(post_next_version_callback),
          persistence_manager(persistence_manager_ref) {
    for(uint i = 0; i < num_members; ++i) {
        node_id_to_sst_index[members[i]] = i;
    }

    for(const auto p : subgroup_settings_by_id) {
        subgroup_id_t id = p.first;
        const SubgroupSettings& settings = p.second;
        auto num_shard_members = settings.members.size();
        while(free_message_buffers[id].size() < settings.profile.window_size * num_shard_members) {
            free_message_buffers[id].emplace_back(settings.profile.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) {
        // if groups are created successfully, rdmc_sst_groups_created will be set to true
        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,
        MulticastGroup&& old_group,
        uint32_t total_num_subgroups,
        const std::map<subgroup_id_t, SubgroupSettings>& subgroup_settings_by_id,
        const subgroup_post_next_version_func_t& post_next_version_callback,
        std::vector<char> already_failed)
        : members(_members),
          num_members(members.size()),
          member_index(index_of(members, my_node_id)),
          callbacks(old_group.callbacks),
          total_num_subgroups(total_num_subgroups),
          subgroup_settings_map(subgroup_settings_by_id),
          received_intervals(sst->num_received.size(), {-1, -1}),
          rpc_callback(old_group.rpc_callback),
          rdmc_group_num_offset(old_group.rdmc_group_num_offset + old_group.num_members),
          future_message_indices(total_num_subgroups, 0),
          next_sends(total_num_subgroups),
          committed_sst_index(total_num_subgroups, -1),
          num_nulls_queued(total_num_subgroups, 0),
          first_null_index(total_num_subgroups, -1),
          pending_sends(total_num_subgroups),
          current_sends(total_num_subgroups),
          next_message_to_deliver(total_num_subgroups),
          minimum_persisted_version(total_num_subgroups, persistent::INVALID_VERSION),
          minimum_verified_version(total_num_subgroups, persistent::INVALID_VERSION),
          sender_timeout(old_group.sender_timeout),
          sst(sst),
          sst_multicast_group_ptrs(total_num_subgroups),
          last_transfer_medium(total_num_subgroups),
          post_next_version_callback(post_next_version_callback),
          persistence_manager(old_group.persistence_manager) {
    // Make sure rdmc_group_num_offset didn't overflow.
    assert(old_group.rdmc_group_num_offset <= std::numeric_limits<uint16_t>::max() - old_group.num_members - num_members);

    // Just in case
    old_group.wedge();

    for(uint i = 0; i < num_members; ++i) {
        node_id_to_sst_index[members[i]] = i;
    }

    // Convience function that takes a msg from the old group and
    // produces one suitable for this group.
    auto convert_msg = [this](RDMCMessage& msg, subgroup_id_t subgroup_num) {
        msg.sender_id = members[member_index];
        msg.index = future_message_indices[subgroup_num]++;
        return std::move(msg);
    };

    // Convience function that takes a msg from the old group and
    // produces one suitable for this group.
    auto convert_sst_msg = [this](SSTMessage& msg, subgroup_id_t subgroup_num) {
        msg.sender_id = members[member_index];
        msg.index = future_message_indices[subgroup_num]++;
        return std::move(msg);
    };

    for(const auto p : subgroup_settings_by_id) {
        subgroup_id_t id = p.first;
        const SubgroupSettings& settings = p.second;
        auto num_shard_members = settings.members.size();
        while(free_message_buffers[id].size() < settings.profile.window_size * num_shard_members) {
            free_message_buffers[id].emplace_back(settings.profile.max_msg_size);
        }
    }

    // Reclaim RDMCMessageBuffers from the old group, and supplement them with
    // additional if the group has grown.
    std::lock_guard<std::recursive_mutex> lock(old_group.msg_state_mtx);
    for(const auto p : subgroup_settings_by_id) {
        const subgroup_id_t subgroup_num = p.first;
        const SubgroupSettings& settings = p.second;
        auto num_shard_members = settings.members.size();
        // for later: don't move extra message buffers
        free_message_buffers[subgroup_num].swap(old_group.free_message_buffers[subgroup_num]);
        while(free_message_buffers[subgroup_num].size() < settings.profile.window_size * num_shard_members) {
            free_message_buffers[subgroup_num].emplace_back(settings.profile.max_msg_size);
        }
    }

    for(auto& msg : old_group.current_receives) {
        free_message_buffers[msg.first.first].push_back(std::move(msg.second.message_buffer));
    }
    old_group.current_receives.clear();

    // Assume that any locally stable messages failed. If we were the sender
    // than re-attempt, otherwise discard. TODO: Presumably the ragged edge
    // cleanup will want the chance to deliver some of these.
    for(auto& p : old_group.locally_stable_rdmc_messages) {
        if(p.second.size() == 0) {
            continue;
        }

        for(auto& q : p.second) {
            if(q.second.sender_id == members[member_index]) {
                pending_sends[p.first].push(convert_msg(q.second, p.first));
            } else {
                free_message_buffers[p.first].push_back(std::move(q.second.message_buffer));
            }
        }
    }
    old_group.locally_stable_rdmc_messages.clear();

    for(const auto p : subgroup_settings_by_id) {
        subgroup_id_t id = p.first;
        const SubgroupSettings& settings = p.second;
        auto num_shard_members = settings.members.size();
        while(free_message_buffers[id].size() < settings.profile.window_size * num_shard_members) {
            free_message_buffers[id].emplace_back(settings.profile.max_msg_size);
        }
    }

    old_group.locally_stable_sst_messages.clear();

    // Any messages that were being sent should be re-attempted.
    for(const auto& p : subgroup_settings_by_id) {
        auto subgroup_num = p.first;
        if(old_group.current_sends.size() > subgroup_num && old_group.current_sends[subgroup_num]) {
            pending_sends[subgroup_num].push(convert_msg(*old_group.current_sends[subgroup_num], subgroup_num));
        }

        if(old_group.pending_sends.size() > subgroup_num) {
            while(!old_group.pending_sends[subgroup_num].empty()) {
                pending_sends[subgroup_num].push(convert_msg(old_group.pending_sends[subgroup_num].front(), subgroup_num));
                old_group.pending_sends[subgroup_num].pop();
            }
        }

        if(old_group.next_sends.size() > subgroup_num && old_group.next_sends[subgroup_num]) {
            next_sends[subgroup_num] = convert_msg(*old_group.next_sends[subgroup_num], subgroup_num);
        }

        for(auto& entry : old_group.non_persistent_messages[subgroup_num]) {
            non_persistent_messages[subgroup_num].emplace(entry.first,
                                                          convert_msg(entry.second, subgroup_num));
        }
        old_group.non_persistent_messages.clear();
        for(auto& entry : old_group.non_persistent_sst_messages[subgroup_num]) {
            non_persistent_sst_messages[subgroup_num].emplace(entry.first,
                                                              convert_sst_msg(entry.second, subgroup_num));
        }
        old_group.non_persistent_sst_messages.clear();
    }

    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) {
        // if groups are created successfully, rdmc_sst_groups_created will be set to true
        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);
}

bool MulticastGroup::create_rdmc_sst_groups() {
    for(const auto& p : subgroup_settings_map) {
        uint32_t subgroup_num = p.first;
        const SubgroupSettings& subgroup_settings = p.second;
        const std::vector<node_id_t>& shard_members = subgroup_settings.members;
        std::size_t num_shard_members = shard_members.size();
        std::vector<int> shard_senders = subgroup_settings.senders;
        uint32_t num_shard_senders = get_num_senders(shard_senders);
        auto shard_sst_indices = get_shard_sst_indices(subgroup_num);

        sst_multicast_group_ptrs[subgroup_num] = std::make_unique<sst::multicast_group<DerechoSST>>(
                sst, shard_sst_indices, subgroup_settings.profile.window_size, subgroup_settings.profile.sst_max_msg_size, subgroup_settings.senders,
                subgroup_settings.num_received_offset, subgroup_settings.slot_offset, subgroup_settings.index_offset);

        if(subgroup_settings.profile.max_msg_size > subgroup_settings.profile.sst_max_msg_size) {
            for(uint shard_rank = 0, sender_rank = -1; shard_rank < num_shard_members; ++shard_rank) {
                // don't create RDMC group if the shard member is never going to send
                if(!shard_senders[shard_rank]) {
                    continue;
                }
                sender_rank++;
                node_id_t node_id = shard_members[shard_rank];
                // When RDMC receives a message, it should store it in
                // locally_stable_rdmc_messages and update the received count
                rdmc::completion_callback_t rdmc_receive_handler;
                rdmc_receive_handler = [this, subgroup_num, shard_rank, sender_rank,
                                        subgroup_settings, node_id,
                                        num_shard_senders,
                                        shard_sst_indices](char* data, size_t size) {
                    assert(this->sst);
                    std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
                    header* h = (header*)data;
                    const int32_t index = h->index;
                    message_id_t sequence_number = index * num_shard_senders + sender_rank;

                    dbg_default_trace("Locally received message in subgroup {}, sender rank {}, index {}",
                                      subgroup_num, shard_rank, index);
                    // Move message from current_receives to locally_stable_rdmc_messages.
                    if(node_id == members[member_index]) {
                        assert(current_sends[subgroup_num]);
                        locally_stable_rdmc_messages[subgroup_num][sequence_number] = std::move(*current_sends[subgroup_num]);
                        current_sends[subgroup_num] = std::nullopt;
                    } else {
                        auto it = current_receives.find({subgroup_num, node_id});
                        assert(it != current_receives.end());
                        auto& msg = it->second;
                        msg.index = index;
                        // We set the size in this receive handler instead of in the incoming_message_handler
                        msg.size = size;
                        locally_stable_rdmc_messages[subgroup_num].emplace(sequence_number, std::move(msg));
                        current_receives.erase(it);
                    }

                    auto new_num_received = resolve_num_received(index, subgroup_settings.num_received_offset + sender_rank);
                    // NULL Send Scheme
                    // only if I am a sender in the subgroup and the subgroup is not in UNORDERED mode
                    if(subgroup_settings.sender_rank >= 0 && subgroup_settings.mode != Mode::UNORDERED) {
                        if(subgroup_settings.sender_rank < (int)sender_rank) {
                            while(future_message_indices[subgroup_num] <= new_num_received) {
                                get_buffer_and_send_auto_null(subgroup_num);
                            }
                        } else if(subgroup_settings.sender_rank > (int)sender_rank) {
                            while(future_message_indices[subgroup_num] < new_num_received) {
                                get_buffer_and_send_auto_null(subgroup_num);
                            }
                        }
                    }

                    // deliver immediately if in UNORDERED mode
                    if(subgroup_settings.mode == Mode::UNORDERED) {
                        // issue stability upcalls for the recently sequenced messages
                        for(int i = sst->num_received[member_index][subgroup_settings.num_received_offset + sender_rank] + 1;
                            i <= new_num_received; ++i) {
                            message_id_t seq_num = i * num_shard_senders + sender_rank;
                            if(!locally_stable_sst_messages[subgroup_num].empty()
                               && locally_stable_sst_messages[subgroup_num].begin()->first == seq_num) {
                                auto& msg = locally_stable_sst_messages[subgroup_num].begin()->second;
                                char* buf = const_cast<char*>(msg.buf);
                                header* h = (header*)(buf);
                                // no delivery callback for a NULL message
                                if(msg.size > h->header_size && callbacks.global_stability_callback) {
                                    callbacks.global_stability_callback(subgroup_num, msg.sender_id,
                                                                        msg.index,
                                                                        {{buf + h->header_size, msg.size - h->header_size}},
                                                                        persistent::INVALID_VERSION);
                                }
                                if(node_id == members[member_index]) {
                                    pending_message_timestamps[subgroup_num].erase(h->timestamp);
                                }
                                locally_stable_sst_messages[subgroup_num].erase(locally_stable_sst_messages[subgroup_num].begin());
                            } else {
                                assert(!locally_stable_rdmc_messages[subgroup_num].empty());
                                auto it2 = locally_stable_rdmc_messages[subgroup_num].begin();
                                assert(it2->first == seq_num);
                                auto& msg = it2->second;
                                char* buf = msg.message_buffer.buffer.get();
                                header* h = (header*)(buf);
                                // no delivery for a NULL message
                                if(msg.size > h->header_size && callbacks.global_stability_callback) {
                                    callbacks.global_stability_callback(subgroup_num, msg.sender_id,
                                                                        msg.index,
                                                                        {{buf + h->header_size, msg.size - h->header_size}},
                                                                        persistent::INVALID_VERSION);
                                }
                                free_message_buffers[subgroup_num].push_back(std::move(msg.message_buffer));
                                if(node_id == members[member_index]) {
                                    pending_message_timestamps[subgroup_num].erase(h->timestamp);
                                }
                                locally_stable_rdmc_messages[subgroup_num].erase(it2);
                            }
                        }
                    }
                    if(new_num_received > sst->num_received[member_index][subgroup_settings.num_received_offset + sender_rank]) {
                        sst->num_received[member_index][subgroup_settings.num_received_offset + sender_rank] = new_num_received;
                        // std::atomic_signal_fence(std::memory_order_acq_rel);
                        auto* min_ptr = std::min_element(&sst->num_received[member_index][subgroup_settings.num_received_offset],
                                                         &sst->num_received[member_index][subgroup_settings.num_received_offset + num_shard_senders]);
                        uint min_index = std::distance(&sst->num_received[member_index][subgroup_settings.num_received_offset], min_ptr);
                        auto new_seq_num = (*min_ptr + 1) * num_shard_senders + min_index - 1;
                        if(static_cast<message_id_t>(new_seq_num) > sst->seq_num[member_index][subgroup_num]) {
                            dbg_default_trace("Updating seq_num for subgroup {} to {}", subgroup_num, new_seq_num);
                            sst->seq_num[member_index][subgroup_num] = new_seq_num;
                            sst->put(shard_sst_indices,
                                     sst->seq_num, subgroup_num);
                        }
                        sst->put(shard_sst_indices,
                                 sst->num_received,
                                 subgroup_settings.num_received_offset + sender_rank);
                    }
                };
                // Capture rdmc_receive_handler by copy! The reference to it won't be valid after this constructor ends!
                auto receive_handler_plus_notify =
                        [this, rdmc_receive_handler](char* data, size_t size) {
                            rdmc_receive_handler(data, size);
                            // signal background writer thread
                            sender_cv.notify_all();
                        };

                // Create a "rotated" vector of members in which the currently selected shard member (shard_rank) is first
                std::vector<uint32_t> rotated_shard_members(shard_members.size());
                for(uint k = 0; k < num_shard_members; ++k) {
                    rotated_shard_members[k] = shard_members[(shard_rank + k) % num_shard_members];
                }

                // don't create rdmc group if there's only one member in the shard
                if(num_shard_members <= 1) {
                    singleton_shard_receive_handlers[subgroup_num] = receive_handler_plus_notify;
                    continue;
                }

                if(node_id == members[member_index]) {
                    //Create a group in which this node is the sender, and only self-receives happen
                    if(!rdmc::create_group(
                               rdmc_group_num_offset, rotated_shard_members, subgroup_settings.profile.block_size, subgroup_settings.profile.rdmc_send_algorithm,
                               [](size_t length) -> rdmc::receive_destination {
                                   assert_always(false);
                                   return {nullptr, 0};
                               },
                               receive_handler_plus_notify,
                               [](std::optional<uint32_t>) {})) {
                        return false;
                    }
                    subgroup_to_rdmc_group[subgroup_num] = rdmc_group_num_offset;
                    rdmc_group_num_offset++;
                } else {
                    if(!rdmc::create_group(
                               rdmc_group_num_offset, rotated_shard_members, subgroup_settings.profile.block_size, subgroup_settings.profile.rdmc_send_algorithm,
                               [this, subgroup_num, node_id](size_t length) {
                                   std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
                                   assert(!free_message_buffers[subgroup_num].empty());
                                   //Create a Message struct to receive the data into.
                                   RDMCMessage msg;
                                   msg.sender_id = node_id;
                                   // The length variable is not the exact size of the msg,
                                   // but it is the nearest multiple of the block size greater then the size
                                   // so we will set the size in the receive handler
                                   msg.message_buffer = std::move(free_message_buffers[subgroup_num].back());
                                   free_message_buffers[subgroup_num].pop_back();

                                   rdmc::receive_destination ret{msg.message_buffer.mr, 0};
                                   current_receives[{subgroup_num, node_id}] = std::move(msg);

                                   assert(ret.mr->buffer != nullptr);
                                   return ret;
                               },
                               rdmc_receive_handler, [](std::optional<uint32_t>) {})) {
                        return false;
                    }
                    rdmc_group_num_offset++;
                }
            }
        }
    }
    return true;
}

void MulticastGroup::initialize_sst_row() {
    auto num_received_size = sst->num_received.size();
    auto seq_num_size = sst->seq_num.size();
    // Initializes only the local row
    for(uint j = 0; j < num_received_size; ++j) {
        sst->num_received[member_index][j] = -1;
    }
    for(uint j = 0; j < seq_num_size; ++j) {
        sst->seq_num[member_index][j] = -1;
        sst->delivered_num[member_index][j] = -1;
        sst->persisted_num[member_index][j] = -1;
        sst->verified_num[member_index][j] = -1;
    }
    memset(const_cast<unsigned char*>(sst->signatures[member_index]), 0, sst->signatures.size());
    for(uint j = 0; j < total_num_subgroups; j++) {
        sst->index[member_index][j] = -1;
    }
    // No put(), no sync(). The caller will issue them later.
}

void MulticastGroup::deliver_message(RDMCMessage& msg, const subgroup_id_t& subgroup_num,
                                     const persistent::version_t& version,
                                     const uint64_t& msg_ts_us) {
    char* buf = msg.message_buffer.buffer.get();
    header* h = (header*)(buf);
    // cooked send
    if(h->cooked_send) {
        buf += h->header_size;
        auto payload_size = msg.size - h->header_size;
        post_next_version_callback(subgroup_num, version, msg_ts_us);
        rpc_callback(subgroup_num, msg.sender_id, buf, payload_size);
        if(callbacks.global_stability_callback) {
            callbacks.global_stability_callback(subgroup_num, msg.sender_id, msg.index, {},
                                                version);
        }
    } else if(msg.size > h->header_size && callbacks.global_stability_callback) {
        callbacks.global_stability_callback(subgroup_num, msg.sender_id, msg.index,
                                            {{buf + h->header_size, msg.size - h->header_size}},
                                            version);
    }
}

void MulticastGroup::deliver_message(SSTMessage& msg, const subgroup_id_t& subgroup_num,
                                     const persistent::version_t& version,
                                     const uint64_t& msg_ts_us) {
    char* buf = const_cast<char*>(msg.buf);
    header* h = (header*)(buf);
    // cooked send
    if(h->cooked_send) {
        buf += h->header_size;
        auto payload_size = msg.size - h->header_size;
        post_next_version_callback(subgroup_num, version, msg_ts_us);
        rpc_callback(subgroup_num, msg.sender_id, buf, payload_size);
        if(callbacks.global_stability_callback) {
            callbacks.global_stability_callback(subgroup_num, msg.sender_id, msg.index, {},
                                                version);
        }
    } else if(msg.size > h->header_size && callbacks.global_stability_callback) {
        callbacks.global_stability_callback(subgroup_num, msg.sender_id, msg.index,
                                            {{buf + h->header_size, msg.size - h->header_size}},
                                            version);
    }
}

bool MulticastGroup::version_message(RDMCMessage& msg, const subgroup_id_t& subgroup_num,
                                     const persistent::version_t& version, const uint64_t& msg_timestamp) {
    char* buf = msg.message_buffer.buffer.get();
    header* h = (header*)(buf);
    // null message filter
    if(msg.size == h->header_size) {
        return false;
    }
    if(msg.sender_id == members[member_index]) {
        pending_persistence[subgroup_num][locally_stable_rdmc_messages[subgroup_num].begin()->first] = msg_timestamp;
    }
    // make a version for persistent<t>/volatile<t>
    uint64_t msg_ts_us = msg_timestamp / 1e3;
    if(msg_ts_us == 0) {
        struct timespec now;
        clock_gettime(CLOCK_REALTIME, &now);
        msg_ts_us = (uint64_t)now.tv_sec * 1e6 + now.tv_nsec / 1e3;
    }
    persistence_manager.make_version(subgroup_num, version, HLC{msg_ts_us, 0});
    return true;
}

bool MulticastGroup::version_message(SSTMessage& msg, const subgroup_id_t& subgroup_num,
                                     const persistent::version_t& version, const uint64_t& msg_timestamp) {
    char* buf = const_cast<char*>(msg.buf);
    header* h = (header*)(buf);
    // null message filter
    if(msg.size == h->header_size) {
        return false;
    }
    if(msg.sender_id == members[member_index]) {
        pending_persistence[subgroup_num][locally_stable_sst_messages[subgroup_num].begin()->first] = msg_timestamp;
    }
    // make a version for persistent<t>/volatile<t>
    uint64_t msg_ts_us = msg_timestamp / 1e3;
    if(msg_ts_us == 0) {
        struct timespec now;
        clock_gettime(CLOCK_REALTIME, &now);
        msg_ts_us = (uint64_t)now.tv_sec * 1e6 + now.tv_nsec / 1e3;
    }
    persistence_manager.make_version(subgroup_num, version, HLC{msg_ts_us, 0});
    return true;
}

void MulticastGroup::deliver_messages_upto(
        const std::vector<int32_t>& max_indices_for_senders,
        subgroup_id_t subgroup_num, uint32_t num_shard_senders) {
    bool non_null_msgs_delivered = false;
    assert(max_indices_for_senders.size() == (size_t)num_shard_senders);
    {
        std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
        int32_t curr_seq_num = sst->delivered_num[member_index][subgroup_num];
        int32_t max_seq_num = curr_seq_num;
        for(uint sender = 0; sender < num_shard_senders; sender++) {
            max_seq_num = std::max(max_seq_num,
                                   static_cast<int32_t>(max_indices_for_senders[sender] * num_shard_senders + sender));
        }
        persistent::version_t assigned_version = persistent::INVALID_VERSION;
        for(int32_t seq_num = curr_seq_num + 1; seq_num <= max_seq_num; seq_num++) {
            //determine if this sequence number should actually be skipped
            int32_t index = seq_num / num_shard_senders;
            uint32_t sender_rank = seq_num % num_shard_senders;
            if(index > max_indices_for_senders[sender_rank]) {
                continue;
            }
            auto rdmc_msg_ptr = locally_stable_rdmc_messages[subgroup_num].find(seq_num);
            assigned_version = persistent::combine_int32s(sst->vid[member_index], seq_num);
            if(rdmc_msg_ptr != locally_stable_rdmc_messages[subgroup_num].end()) {
                auto& msg = rdmc_msg_ptr->second;
                char* buf = msg.message_buffer.buffer.get();
                uint64_t msg_ts = ((header*)buf)->timestamp;
                //Note: deliver_message frees the RDMC buffer in msg, which is why the timestamp must be saved before calling this
                deliver_message(msg, subgroup_num, assigned_version, msg_ts / 1000);
                non_null_msgs_delivered |= version_message(msg, subgroup_num, assigned_version, msg_ts);
                // free the message buffer only after it version_message has been called
                free_message_buffers[subgroup_num].push_back(std::move(msg.message_buffer));
                locally_stable_rdmc_messages[subgroup_num].erase(rdmc_msg_ptr);
            } else {
                dbg_default_trace("Subgroup {}, deliver_messages_upto delivering an SST message with seq_num = {}",
                                  subgroup_num, seq_num);
                auto& msg = locally_stable_sst_messages[subgroup_num].at(seq_num);
                char* buf = (char*)msg.buf;
                uint64_t msg_ts = ((header*)buf)->timestamp;
                deliver_message(msg, subgroup_num, assigned_version, msg_ts / 1000);
                non_null_msgs_delivered |= version_message(msg, subgroup_num, assigned_version, msg_ts);
                locally_stable_sst_messages[subgroup_num].erase(seq_num);
            }
        }
        gmssst::set(sst->delivered_num[member_index][subgroup_num], max_seq_num);
        if(non_null_msgs_delivered) {
            //Call the persistence_manager_post_persist_func
            persistence_manager.post_persist_request(subgroup_num, assigned_version);
        }
    }
    sst->put(get_shard_sst_indices(subgroup_num),
             sst->delivered_num, subgroup_num);
}

int32_t MulticastGroup::resolve_num_received(int32_t index, uint32_t num_received_entry) {
    auto it = received_intervals[num_received_entry].end();
    it--;
    while(*it > index) {
        it--;
    }
    if(std::next(it) == received_intervals[num_received_entry].end()) {
        if(*it == index - 1) {
            *it = index;
        } else {
            received_intervals[num_received_entry].push_back(index);
            received_intervals[num_received_entry].push_back(index);
        }
    } else {
        auto next_it = std::next(it);
        if(*it != index - 1) {
            received_intervals[num_received_entry].insert(next_it, index);
            if(*next_it != index + 1) {
                received_intervals[num_received_entry].insert(next_it, index);
            } else {
                received_intervals[num_received_entry].erase(next_it);
            }
        } else {
            if(*next_it != index + 1) {
                received_intervals[num_received_entry].insert(next_it, index);
            } else {
                received_intervals[num_received_entry].erase(next_it);
            }
            received_intervals[num_received_entry].erase(it);
        }
    }
    return *std::next(received_intervals[num_received_entry].begin());
}

bool MulticastGroup::receiver_predicate(const SubgroupSettings& subgroup_settings,
                                        const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                                        uint32_t num_shard_senders, const DerechoSST& sst) {
    for(uint sender_count = 0; sender_count < num_shard_senders; ++sender_count) {
        // Equivalent to read_seq_num[sender_count] > last_seq_num[sender_count]
        if((message_id_t)sst.index[node_id_to_sst_index.at(subgroup_settings.members[shard_ranks_by_sender_rank.at(sender_count)])][subgroup_settings.index_offset]
           > sst.num_received_sst[member_index][subgroup_settings.num_received_offset + sender_count]) {
            return true;
        }
    }
    return false;
}

void MulticastGroup::sst_receive_handler(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                         const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                                         uint32_t num_shard_senders, uint32_t sender_rank,
                                         volatile char* data, uint64_t size) {
    header* h = (header*)data;
    int32_t index = h->index;
    int32_t num_nulls = h->num_nulls;

    do {
        if (num_nulls > 0) {
	    size = h->header_size;
	}
        message_id_t sequence_number = index * num_shard_senders + sender_rank;
        node_id_t node_id = subgroup_settings.members[shard_ranks_by_sender_rank.at(sender_rank)];

        locally_stable_sst_messages[subgroup_num][sequence_number] = {node_id, index, size, data};

        auto new_num_received = resolve_num_received(index, subgroup_settings.num_received_offset + sender_rank);

        /* NULL Send Scheme */
        // only if I am a sender in the subgroup and the subgroup is not in UNORDERED mode
        if(subgroup_settings.sender_rank >= 0 && subgroup_settings.mode != Mode::UNORDERED) {
            if(subgroup_settings.sender_rank < (int)sender_rank) {
                while(future_message_indices[subgroup_num] <= new_num_received) {
                    get_buffer_and_send_auto_null(subgroup_num);
                }
            } else if(subgroup_settings.sender_rank > (int)sender_rank) {
                while(future_message_indices[subgroup_num] < new_num_received) {
                    get_buffer_and_send_auto_null(subgroup_num);
                }
            }
        }

        if(subgroup_settings.mode == Mode::UNORDERED) {
            // issue stability upcalls for the recently sequenced messages
            for(int i = sst->num_received[member_index][subgroup_settings.num_received_offset + sender_rank] + 1; i <= new_num_received; ++i) {
                message_id_t seq_num = i * num_shard_senders + sender_rank;
                if(!locally_stable_sst_messages[subgroup_num].empty()
                   && locally_stable_sst_messages[subgroup_num].begin()->first == seq_num) {
                    auto& msg = locally_stable_sst_messages[subgroup_num].begin()->second;
                    char* buf = const_cast<char*>(msg.buf);
                    header* h = (header*)(buf);
                    if(msg.size > h->header_size && callbacks.global_stability_callback) {
                        callbacks.global_stability_callback(subgroup_num, msg.sender_id,
                                                            msg.index,
                                                            {{buf + h->header_size, msg.size - h->header_size}},
                                                            persistent::INVALID_VERSION);
                    }
                    if(node_id == members[member_index]) {
                        pending_message_timestamps[subgroup_num].erase(h->timestamp);
                    }
                    locally_stable_sst_messages[subgroup_num].erase(locally_stable_sst_messages[subgroup_num].begin());
                } else {
                    assert(!locally_stable_rdmc_messages[subgroup_num].empty());
                    auto it2 = locally_stable_rdmc_messages[subgroup_num].begin();
                    assert(it2->first == seq_num);
                    auto& msg = it2->second;
                    char* buf = msg.message_buffer.buffer.get();
                    header* h = (header*)(buf);
                    if(msg.size > h->header_size && callbacks.global_stability_callback) {
                        callbacks.global_stability_callback(subgroup_num, msg.sender_id,
                                                            msg.index,
                                                            {{buf + h->header_size, msg.size - h->header_size}},
                                                            persistent::INVALID_VERSION);
                    }
                    free_message_buffers[subgroup_num].push_back(std::move(msg.message_buffer));
                    if(node_id == members[member_index]) {
                        pending_message_timestamps[subgroup_num].erase(h->timestamp);
                    }
                    locally_stable_rdmc_messages[subgroup_num].erase(it2);
                }
            }
        }
        sst->num_received[member_index][subgroup_settings.num_received_offset + sender_rank] = new_num_received;
        index++;
    } while(--num_nulls > 0);
}

void MulticastGroup::receiver_function(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                       const std::map<uint32_t, uint32_t>& shard_ranks_by_sender_rank,
                                       uint32_t num_shard_senders, DerechoSST& sst,
                                       const std::function<void(uint32_t, volatile char*, uint32_t)>& sst_receive_handler_lambda) {
    DerechoParams profile = subgroup_settings.profile;
    const uint64_t slot_width = profile.sst_max_msg_size + sizeof(uint64_t);

    bool put_new_seq_num = false;
    {
        std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
        for(uint sender_count = 0; sender_count < num_shard_senders; ++sender_count) {
            const uint32_t sender_sst_index = node_id_to_sst_index.at(subgroup_settings.members[shard_ranks_by_sender_rank.at(sender_count)]);
            uint32_t slot;
            message_id_t old_index = sst.num_received_sst[member_index][subgroup_settings.num_received_offset + sender_count];
            const message_id_t received_index = sst.index[sender_sst_index][subgroup_settings.index_offset];
            while(received_index > old_index) {
                old_index++;
                slot = old_index % profile.window_size;
                dbg_default_trace("receiver_trig calling sst_receive_handler_lambda. next_seq = {}, num_received = {}, sender rank = {}. Reading from SST row {}, slot {}",
                                  received_index, old_index, sender_count, sender_sst_index, subgroup_settings.slot_offset + slot_width * slot);
                sst_receive_handler_lambda(sender_count,
                                           &sst.slots[sender_sst_index]
                                                     [subgroup_settings.slot_offset + slot_width * slot],
                                           (uint64_t&)sst.slots[sender_sst_index]
                                                               [subgroup_settings.slot_offset + slot_width * (slot + 1) - sizeof(uint64_t)]);

                // I pretend I received all the nulls, when actually I have received only the first one
                header* h = (header*)&sst.slots[sender_sst_index][subgroup_settings.slot_offset + slot_width * slot];
                if(h->num_nulls > 0) {
                    old_index += h->num_nulls - 1;
                }
                sst.num_received_sst[member_index][subgroup_settings.num_received_offset + sender_count] = old_index;
            }
            // std::atomic_signal_fence(std::memory_order_acq_rel);
            auto* min_ptr = std::min_element(&sst.num_received[member_index][subgroup_settings.num_received_offset],
                                             &sst.num_received[member_index][subgroup_settings.num_received_offset + num_shard_senders]);
            int min_index = std::distance(&sst.num_received[member_index][subgroup_settings.num_received_offset], min_ptr);
            message_id_t new_seq_num = (*min_ptr + 1) * num_shard_senders + min_index - 1;
            if(new_seq_num > sst.seq_num[member_index][subgroup_num]) {
                dbg_default_trace("Updating seq_num for subgroup {} to {}", subgroup_num, new_seq_num);
                sst.seq_num[member_index][subgroup_num] = new_seq_num;
                put_new_seq_num = true;
            }
        }
    }
    // lock released: puts can happen.
    sst.put((char*)std::addressof(sst.num_received_sst[0][subgroup_settings.num_received_offset]) - sst.getBaseAddress(),
            sizeof(decltype(sst.num_received_sst)::value_type) * num_shard_senders);
    if(put_new_seq_num) {
        sst.put(sst.seq_num, subgroup_num);
    }
    sst.put((char*)std::addressof(sst.num_received[0][subgroup_settings.num_received_offset]) - sst.getBaseAddress(),
            sizeof(decltype(sst.num_received)::value_type) * num_shard_senders);
}

void MulticastGroup::delivery_trigger(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                      const uint32_t num_shard_members, DerechoSST& sst) {
    bool update_sst = false;
    {
        std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
        // compute the min of the seq_num
        message_id_t min_stable_num
                = sst.seq_num[node_id_to_sst_index.at(subgroup_settings.members[0])][subgroup_num];
        for(uint i = 0; i < num_shard_members; ++i) {
            // to avoid a race condition, do not read the same SST entry twice
            message_id_t stable_num_copy = sst.seq_num[node_id_to_sst_index.at(subgroup_settings.members[i])][subgroup_num];
            min_stable_num = std::min(min_stable_num, stable_num_copy);
        }
        bool non_null_msgs_delivered = false;
        persistent::version_t assigned_version = persistent::INVALID_VERSION;
        while(true) {
            if(locally_stable_rdmc_messages[subgroup_num].empty() && locally_stable_sst_messages[subgroup_num].empty()) {
                break;
            }
            int32_t least_undelivered_rdmc_seq_num, least_undelivered_sst_seq_num;
            least_undelivered_rdmc_seq_num = least_undelivered_sst_seq_num = std::numeric_limits<int32_t>::max();
            if(!locally_stable_rdmc_messages[subgroup_num].empty()) {
                least_undelivered_rdmc_seq_num = locally_stable_rdmc_messages[subgroup_num].begin()->first;
            }
            if(!locally_stable_sst_messages[subgroup_num].empty()) {
                least_undelivered_sst_seq_num = locally_stable_sst_messages[subgroup_num].begin()->first;
            }
            if(least_undelivered_rdmc_seq_num < least_undelivered_sst_seq_num && least_undelivered_rdmc_seq_num <= min_stable_num) {
                update_sst = true;
                dbg_default_trace("Subgroup {}, can deliver a locally stable RDMC message: min_stable_num={} and least_undelivered_seq_num={}",
                                  subgroup_num, min_stable_num, least_undelivered_rdmc_seq_num);
                RDMCMessage& msg = locally_stable_rdmc_messages[subgroup_num].begin()->second;
                char* buf = msg.message_buffer.buffer.get();
                uint64_t msg_ts = ((header*)buf)->timestamp;
                //Note: deliver_message frees the RDMC buffer in msg, which is why the timestamp must be saved before calling this
                assigned_version = persistent::combine_int32s(sst.vid[member_index], least_undelivered_rdmc_seq_num);
                deliver_message(msg, subgroup_num, assigned_version, msg_ts / 1000);
                non_null_msgs_delivered |= version_message(msg, subgroup_num, assigned_version, msg_ts);
                // free the message buffer only after version_message has been called
                free_message_buffers[subgroup_num].push_back(std::move(msg.message_buffer));
                sst.delivered_num[member_index][subgroup_num] = least_undelivered_rdmc_seq_num;
                locally_stable_rdmc_messages[subgroup_num].erase(locally_stable_rdmc_messages[subgroup_num].begin());
            } else if(least_undelivered_sst_seq_num < least_undelivered_rdmc_seq_num && least_undelivered_sst_seq_num <= min_stable_num) {
                update_sst = true;
                dbg_default_trace("Subgroup {}, can deliver a locally stable SST message: min_stable_num={} and least_undelivered_seq_num={}",
                                  subgroup_num, min_stable_num, least_undelivered_sst_seq_num);
                SSTMessage& msg = locally_stable_sst_messages[subgroup_num].begin()->second;
                char* buf = (char*)msg.buf;
                uint64_t msg_ts = ((header*)buf)->timestamp;
                assigned_version = persistent::combine_int32s(sst.vid[member_index], least_undelivered_sst_seq_num);
                deliver_message(msg, subgroup_num, assigned_version, msg_ts / 1000);
                non_null_msgs_delivered |= version_message(msg, subgroup_num, assigned_version, msg_ts);
                sst.delivered_num[member_index][subgroup_num] = least_undelivered_sst_seq_num;
                locally_stable_sst_messages[subgroup_num].erase(locally_stable_sst_messages[subgroup_num].begin());
            } else {
                break;
            }
        }
        if(update_sst) {
            // post persistence request for ordered mode.
            if(non_null_msgs_delivered) {
                persistence_manager.post_persist_request(subgroup_num, assigned_version);
            }
        }
    }
    if(update_sst) {
        sst.put(get_shard_sst_indices(subgroup_num),
                sst.delivered_num, subgroup_num);
    }
}

void MulticastGroup::sst_send_trigger(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                      const uint32_t num_shard_members, DerechoSST& sst) {
    int32_t current_committed_index;
    int32_t to_be_sent;
    int32_t current_first_null_index;
    uint32_t current_num_nulls_queued;
    {
        std::unique_lock<std::recursive_mutex> lock(msg_state_mtx);
        to_be_sent = committed_sst_index[subgroup_num] - sst.index[member_index][subgroup_settings.index_offset];
        if(to_be_sent > 0) {
            current_committed_index = sst_multicast_group_ptrs[subgroup_num]->commit_send(to_be_sent);
            // Save current values and reset null-related counters.
            current_first_null_index = first_null_index[subgroup_num];
            current_num_nulls_queued = num_nulls_queued[subgroup_num];
            first_null_index[subgroup_num] = -1;
            num_nulls_queued[subgroup_num] = 0;
        }
    }
    // Here lock is released
    if(to_be_sent > 0) {
        if(current_num_nulls_queued > 0) {
            DerechoParams profile = subgroup_settings.profile;
            const uint64_t slot_width = profile.sst_max_msg_size + sizeof(uint64_t);
            auto null_slot = current_first_null_index % profile.window_size;
            header* h = (header*)&sst.slots[member_index][subgroup_settings.slot_offset + slot_width * null_slot];
            h->num_nulls = current_num_nulls_queued;
        }

        sst_multicast_group_ptrs[subgroup_num]->send(current_committed_index, to_be_sent, current_num_nulls_queued,
                                                     current_first_null_index, sizeof(header));
    }
}

void MulticastGroup::update_min_persisted_num(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                              uint32_t num_shard_members, DerechoSST& sst) {
    std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
    // compute the min of the persisted_num
    persistent::version_t min_persisted_num
            = sst.persisted_num[node_id_to_sst_index.at(subgroup_settings.members[0])][subgroup_num];
    for(uint32_t i = 1; i < num_shard_members; ++i) {
        persistent::version_t persisted_num_copy = sst.persisted_num[node_id_to_sst_index.at(subgroup_settings.members[i])]
                                                                    [subgroup_num];
        min_persisted_num = std::min(min_persisted_num, persisted_num_copy);
    }
    // callbacks
    if(min_persisted_num > minimum_persisted_version[subgroup_num]) {
        if(callbacks.global_persistence_callback) {
            callbacks.global_persistence_callback(subgroup_num, min_persisted_num);
        }
        persistence_manager.post_verify_request(subgroup_num, min_persisted_num);
        minimum_persisted_version[subgroup_num] = min_persisted_num;
    }
}

void MulticastGroup::update_min_verified_num(subgroup_id_t subgroup_num, const SubgroupSettings& subgroup_settings,
                                             uint32_t num_shard_members, DerechoSST& sst) {
    //Do I need msg_state_mtx here? What does it guard?
    persistent::version_t min_verified_num
            = sst.verified_num[node_id_to_sst_index.at(subgroup_settings.members[0])][subgroup_num];
    for(uint32_t i = 1; i < num_shard_members; ++i) {
        persistent::version_t member_verified_num = sst.verified_num[node_id_to_sst_index.at(subgroup_settings.members[i])]
                                                                    [subgroup_num];
        min_verified_num = std::min(min_verified_num, member_verified_num);
    }
    if(min_verified_num > minimum_verified_version[subgroup_num]) {
        callbacks.global_verified_callback(subgroup_num, min_verified_num);
        minimum_verified_version[subgroup_num] = min_verified_num;
    }
}

void MulticastGroup::register_predicates() {
    for(const auto& p : subgroup_settings_map) {
        subgroup_id_t subgroup_num = p.first;
        const SubgroupSettings& subgroup_settings = p.second;
        auto num_shard_members = subgroup_settings.members.size();
        std::vector<int> shard_senders = subgroup_settings.senders;
        auto num_shard_senders = get_num_senders(shard_senders);
        std::map<uint32_t, uint32_t> shard_ranks_by_sender_rank;
        for(uint j = 0, l = 0; j < num_shard_members; ++j) {
            if(shard_senders[j]) {
                shard_ranks_by_sender_rank[l] = j;
                l++;
            }
        }

        auto receiver_pred = [=](const DerechoSST& sst) {
            return receiver_predicate(subgroup_settings,
                                      shard_ranks_by_sender_rank, num_shard_senders, sst);
        };
        auto sst_receive_handler_lambda = [=](uint32_t sender_rank, volatile char* data, uint64_t size) {
            sst_receive_handler(subgroup_num, subgroup_settings,
                                shard_ranks_by_sender_rank, num_shard_senders,
                                sender_rank, data, size);
        };
        auto receiver_trig = [=](DerechoSST& sst) mutable {
            receiver_function(subgroup_num, subgroup_settings,
                              shard_ranks_by_sender_rank, num_shard_senders, sst,
                              sst_receive_handler_lambda);
        };
        receiver_pred_handles.emplace_back(sst->predicates.insert(receiver_pred, receiver_trig,
                                                                  sst::PredicateType::RECURRENT));

        auto sst_send_pred = [](const DerechoSST& sst) {
            return true;
        };
        auto sst_send_trig = [this, subgroup_num, subgroup_settings, num_shard_members](DerechoSST& sst) mutable {
            sst_send_trigger(subgroup_num, subgroup_settings, num_shard_members, sst);
        };
        receiver_pred_handles.emplace_back(sst->predicates.insert(sst_send_pred, sst_send_trig,
                                                                  sst::PredicateType::RECURRENT));

        if(subgroup_settings.mode != Mode::UNORDERED) {
            auto delivery_pred = [](const DerechoSST& sst) {
                return true;
            };
            auto delivery_trig = [=](DerechoSST& sst) mutable {
                delivery_trigger(subgroup_num, subgroup_settings, num_shard_members, sst);
            };

            delivery_pred_handles.emplace_back(sst->predicates.insert(delivery_pred, delivery_trig,
                                                                      sst::PredicateType::RECURRENT));

            //This predicate should be "current min over persisted_num is greater than the last
            //observed minimum persisted_num," but computing the current min in the predicate is
            //redundant with computing it in the trigger. To save time, we just make the predicate
            //do nothing, and make the trigger both compute and update the minimum persisted_num
            auto persistence_pred = [](const DerechoSST& sst) {
                return true;
            };
            auto persistence_trig = [=](DerechoSST& sst) mutable {
                update_min_persisted_num(subgroup_num, subgroup_settings, num_shard_members, sst);
            };

            persistence_pred_handles.emplace_back(sst->predicates.insert(persistence_pred, persistence_trig, sst::PredicateType::RECURRENT));

            //In case there are persistent objects with signatures, add a similar predicate to check/update the minimum verified_num
            auto verified_pred = [](const DerechoSST& sst) {
                return true;
            };
            auto verified_trig = [=](DerechoSST& sst) {
                update_min_verified_num(subgroup_num, subgroup_settings, num_shard_members, sst);
            };

            if(callbacks.global_verified_callback) {
                persistence_pred_handles.emplace_back(sst->predicates.insert(verified_pred, verified_trig, sst::PredicateType::RECURRENT));
            }

            if(subgroup_settings.sender_rank >= 0) {
                auto sender_pred = [=](const DerechoSST& sst) {
                    message_id_t seq_num = next_message_to_deliver[subgroup_num] * num_shard_senders + subgroup_settings.sender_rank;
                    for(uint i = 0; i < num_shard_members; ++i) {
                        if(sst.delivered_num[node_id_to_sst_index.at(subgroup_settings.members[i])][subgroup_num] < seq_num) {
                            return false;
                        }
                    }
                    return true;
                };
                auto sender_trig = [=](DerechoSST& sst) {
                    sender_cv.notify_all();
                    next_message_to_deliver[subgroup_num]++;
                };
                sender_pred_handles.emplace_back(sst->predicates.insert(sender_pred, sender_trig,
                                                                        sst::PredicateType::RECURRENT));
            }
        } else {
            //This subgroup is in UNORDERED mode
            if(subgroup_settings.sender_rank >= 0) {
                auto sender_pred = [=](const DerechoSST& sst) {
                    for(uint i = 0; i < num_shard_members; ++i) {
                        uint32_t num_received_offset = subgroup_settings.num_received_offset;
                        if(sst.num_received[node_id_to_sst_index.at(subgroup_settings.members[i])][num_received_offset + subgroup_settings.sender_rank]
                           < static_cast<int32_t>(future_message_indices[subgroup_num] - 1 - subgroup_settings.profile.window_size)) {
                            return false;
                        }
                    }
                    return true;
                };
                auto sender_trig = [this](DerechoSST& sst) {
                    sender_cv.notify_all();
                };
                sender_pred_handles.emplace_back(sst->predicates.insert(sender_pred, sender_trig,
                                                                        sst::PredicateType::RECURRENT));
            }
        }
    }
}

MulticastGroup::~MulticastGroup() {
    wedge();
    if(timeout_thread.joinable()) {
        timeout_thread.join();
    }
}

void MulticastGroup::wedge() {
    bool thread_shutdown_existing = thread_shutdown.exchange(true);
    if(thread_shutdown_existing) {  // Wedge has already been called
        return;
    }

    //Consume and remove all the predicate handles
    for(auto handle_iter = sender_pred_handles.begin(); handle_iter != sender_pred_handles.end();) {
        sst->predicates.remove(*handle_iter);
        handle_iter = sender_pred_handles.erase(handle_iter);
    }
    for(auto handle_iter = receiver_pred_handles.begin(); handle_iter != receiver_pred_handles.end();) {
        sst->predicates.remove(*handle_iter);
        handle_iter = receiver_pred_handles.erase(handle_iter);
    }
    for(auto handle_iter = delivery_pred_handles.begin(); handle_iter != delivery_pred_handles.end();) {
        sst->predicates.remove(*handle_iter);
        handle_iter = delivery_pred_handles.erase(handle_iter);
    }
    for(auto handle_iter = persistence_pred_handles.begin(); handle_iter != persistence_pred_handles.end();) {
        sst->predicates.remove(*handle_iter);
        handle_iter = persistence_pred_handles.erase(handle_iter);
    }

    for(uint i = 0; i < num_members; ++i) {
        rdmc::destroy_group(i + rdmc_group_num_offset);
    }

    sender_cv.notify_all();
    if(sender_thread.joinable()) {
        sender_thread.join();
    }
}

void MulticastGroup::send_loop() {
    pthread_setname_np(pthread_self(), "sender_thread");
    subgroup_id_t subgroup_to_send = 0;
    auto should_send_to_subgroup = [&](subgroup_id_t subgroup_num) {
        if(!rdmc_sst_groups_created) {
            return false;
        }
        if(pending_sends[subgroup_num].empty()) {
            return false;
        }
        RDMCMessage& msg = pending_sends[subgroup_num].front();
        const SubgroupSettings& subgroup_settings = subgroup_settings_map.at(subgroup_num);

        int shard_sender_index = subgroup_settings.sender_rank;
        std::vector<int> shard_senders = subgroup_settings.senders;
        uint32_t num_shard_senders = get_num_senders(shard_senders);
        assert(shard_sender_index >= 0);

        if(sst->num_received[member_index][subgroup_settings.num_received_offset + shard_sender_index] < msg.index - 1) {
            return false;
        }

        std::vector<node_id_t> shard_members = subgroup_settings.members;
        auto num_shard_members = shard_members.size();
        assert(num_shard_members >= 1);
        if(subgroup_settings.mode != Mode::UNORDERED) {
            for(uint i = 0; i < num_shard_members; ++i) {
                if(sst->delivered_num[node_id_to_sst_index.at(shard_members[i])][subgroup_num] < static_cast<message_id_t>((msg.index - subgroup_settings.profile.window_size) * num_shard_senders + shard_sender_index)) {
                    return false;
                }
            }
        } else {
            for(uint i = 0; i < num_shard_members; ++i) {
                auto num_received_offset = subgroup_settings.num_received_offset;
                if(sst->num_received[node_id_to_sst_index.at(shard_members[i])][num_received_offset + shard_sender_index]
                   < static_cast<int32_t>(future_message_indices[subgroup_num] - 1 - subgroup_settings.profile.window_size)) {
                    return false;
                }
            }
        }

        return true;
    };
    auto should_send = [&]() {
        for(uint i = 1; i <= total_num_subgroups; ++i) {
            auto subgroup_num = (subgroup_to_send + i) % total_num_subgroups;
            if(should_send_to_subgroup(subgroup_num)) {
                subgroup_to_send = subgroup_num;
                return true;
            }
        }
        return false;
    };
    auto should_wake = [&]() { return thread_shutdown || should_send(); };
    std::unique_lock<std::recursive_mutex> lock(msg_state_mtx);
    while(!thread_shutdown) {
        sender_cv.wait(lock, should_wake);
        if(!thread_shutdown) {
            current_sends[subgroup_to_send] = std::move(pending_sends[subgroup_to_send].front());
            dbg_default_trace("Calling send in subgroup {} on message {} from sender {}",
                              subgroup_to_send, current_sends[subgroup_to_send]->index, current_sends[subgroup_to_send]->sender_id);
            // make sure there are > 1 members before issuing RDMC send
            if(subgroup_settings_map.at(subgroup_to_send).members.size() > 1) {
                if(!rdmc::send(subgroup_to_rdmc_group.at(subgroup_to_send),
                               current_sends[subgroup_to_send]->message_buffer.mr, 0,
                               current_sends[subgroup_to_send]->size)) {
                    throw std::runtime_error("rdmc::send returned false");
                }
            } else {
                // receive the message right here
                singleton_shard_receive_handlers.at(subgroup_to_send)(
                        current_sends[subgroup_to_send]->message_buffer.buffer.get(), current_sends[subgroup_to_send]->size);
            }
            pending_sends[subgroup_to_send].pop();
        }
    }
}

const uint64_t MulticastGroup::compute_global_stability_frontier(uint32_t subgroup_num) {
    uint64_t global_stability_frontier = sst->local_stability_frontier[member_index][subgroup_num];
    auto shard_sst_indices = get_shard_sst_indices(subgroup_num);
    for(auto index : shard_sst_indices) {
        uint64_t local_stability_frontier_copy = sst->local_stability_frontier[index][subgroup_num];
        global_stability_frontier = std::min(global_stability_frontier, local_stability_frontier_copy);
    }
    return global_stability_frontier;
}

void MulticastGroup::check_failures_loop() {
    pthread_setname_np(pthread_self(), "timeout_thread");
    while(!thread_shutdown) {
        std::this_thread::sleep_for(std::chrono::milliseconds(sender_timeout));
        if(sst) {
            {
                std::unique_lock<std::recursive_mutex> lock(msg_state_mtx);
                auto current_time = get_walltime();
                for(auto p : subgroup_settings_map) {
                    auto subgroup_num = p.first;
                    auto members = p.second.members;
                    auto sst_indices = get_shard_sst_indices(subgroup_num);
                    // clean up timestamps of persisted messages
                    auto min_persisted_num = sst->persisted_num[member_index][subgroup_num];
                    for(auto i : sst_indices) {
                        persistent::version_t persisted_num_copy = sst->persisted_num[i][subgroup_num];
                        min_persisted_num = std::min(min_persisted_num, persisted_num_copy);
                    }
                    while(!pending_persistence[subgroup_num].empty() && pending_persistence[subgroup_num].begin()->first <= min_persisted_num) {
                        auto timestamp = pending_persistence[subgroup_num].begin()->second;
                        pending_persistence[subgroup_num].erase(pending_persistence[subgroup_num].begin());
                        pending_message_timestamps[subgroup_num].erase(timestamp);
                    }
                    if(pending_message_timestamps[subgroup_num].empty()) {
                        sst->local_stability_frontier[member_index][subgroup_num] = current_time;
                    } else {
                        sst->local_stability_frontier[member_index][subgroup_num] = std::min(current_time,
                                                                                             *pending_message_timestamps[subgroup_num].begin());
                    }
                }
            }
            sst->put_with_completion((char*)std::addressof(sst->local_stability_frontier[0][0]) - sst->getBaseAddress(),
                                     sizeof(sst->local_stability_frontier[0][0]) * sst->local_stability_frontier.size());
        }
    }
}

// we already hold the lock on msg_state_mtx when we call this
void MulticastGroup::get_buffer_and_send_auto_null(subgroup_id_t subgroup_num) {
    // short-circuits most of the normal checks because
    // we know that we received a message and are sending a null
    long long unsigned int msg_size = sizeof(header);
    const DerechoParams& profile = subgroup_settings_map.at(subgroup_num).profile;
    // very unlikely that msg_size does not fit in the max_msg_size since we are sending a NULL
    // but the user might not be interested in using SSTMC at all, then sst::max_msg_size can be zero
    if(msg_size > profile.sst_max_msg_size) {
        // Create new Message
        RDMCMessage msg;
        msg.sender_id = members[member_index];
        msg.index = future_message_indices[subgroup_num];
        msg.size = msg_size;
        msg.message_buffer = std::move(free_message_buffers[subgroup_num].back());
        free_message_buffers[subgroup_num].pop_back();

        auto current_time = get_walltime();
        pending_message_timestamps[subgroup_num].insert(current_time);

        // Fill header
        char* buf = msg.message_buffer.buffer.get();
        ((header*)buf)->header_size = sizeof(header);
        ((header*)buf)->index = msg.index;
        ((header*)buf)->timestamp = current_time;
        ((header*)buf)->cooked_send = false;

        future_message_indices[subgroup_num]++;
        pending_sends[subgroup_num].push(std::move(msg));
        sender_cv.notify_all();
    } else {
        char* buf = (char*)sst_multicast_group_ptrs[subgroup_num]->get_buffer(msg_size);

        assert(buf);

        auto current_time = get_walltime();
        pending_message_timestamps[subgroup_num].insert(current_time);

        ((header*)buf)->header_size = sizeof(header);
        ((header*)buf)->index = future_message_indices[subgroup_num];
        ((header*)buf)->timestamp = current_time;
        ((header*)buf)->num_nulls = 0;
        ((header*)buf)->cooked_send = false;

        future_message_indices[subgroup_num]++;
        committed_sst_index[subgroup_num]++;

        if(first_null_index[subgroup_num] < 0) {
            first_null_index[subgroup_num] = committed_sst_index[subgroup_num];
        }
        num_nulls_queued[subgroup_num]++;
    }
}

char* MulticastGroup::get_sendbuffer_ptr(subgroup_id_t subgroup_num,
                                         long long unsigned int payload_size,
                                         bool cooked_send) {
    long long unsigned int msg_size = payload_size + sizeof(header);
    const SubgroupSettings& subgroup_settings = subgroup_settings_map.at(subgroup_num);
    if(msg_size > subgroup_settings.profile.max_msg_size) {
        std::string exp_msg("Can't send messages of size larger than the maximum message size which is equal to ");
        exp_msg += std::to_string(subgroup_settings.profile.max_msg_size);
        throw derecho_exception(exp_msg);
    }

    std::vector<node_id_t> shard_members = subgroup_settings.members;
    auto num_shard_members = shard_members.size();
    // if the current node is not a sender, shard_sender_index will be -1
    uint32_t num_shard_senders;
    std::vector<int> shard_senders = subgroup_settings.senders;
    int shard_sender_index = subgroup_settings.sender_rank;
    num_shard_senders = get_num_senders(shard_senders);
    assert(shard_sender_index >= 0);

    if(subgroup_settings.mode != Mode::UNORDERED) {
        for(uint i = 0; i < num_shard_members; ++i) {
            if(sst->delivered_num[node_id_to_sst_index.at(shard_members[i])][subgroup_num]
               < static_cast<int32_t>((future_message_indices[subgroup_num] - subgroup_settings.profile.window_size) * num_shard_senders + shard_sender_index)) {
                return nullptr;
            }
        }
    } else {
        for(uint i = 0; i < num_shard_members; ++i) {
            auto num_received_offset = subgroup_settings.num_received_offset;
            if(sst->num_received[node_id_to_sst_index.at(shard_members[i])][num_received_offset + shard_sender_index]
               < static_cast<int32_t>(future_message_indices[subgroup_num] - subgroup_settings.profile.window_size)) {
                return nullptr;
            }
        }
    }

    if(msg_size > subgroup_settings.profile.sst_max_msg_size) {
        if(thread_shutdown) {
            return nullptr;
        }

        if(free_message_buffers[subgroup_num].empty()) {
            return nullptr;
        }

        if(pending_sst_sends[subgroup_num] || next_sends[subgroup_num]) {
            return nullptr;
        }

        // Create new Message
        RDMCMessage msg;
        msg.sender_id = members[member_index];
        msg.index = future_message_indices[subgroup_num];
        msg.size = msg_size;
        msg.message_buffer = std::move(free_message_buffers[subgroup_num].back());
        free_message_buffers[subgroup_num].pop_back();

        auto current_time = get_walltime();
        pending_message_timestamps[subgroup_num].insert(current_time);

        // Fill header
        char* buf = msg.message_buffer.buffer.get();
        ((header*)buf)->header_size = sizeof(header);
        ((header*)buf)->index = msg.index;
        ((header*)buf)->timestamp = current_time;
        ((header*)buf)->cooked_send = cooked_send;

        next_sends[subgroup_num] = std::move(msg);
        future_message_indices[subgroup_num]++;

        last_transfer_medium[subgroup_num] = true;
        return buf + sizeof(header);
    } else {
        if(pending_sst_sends[subgroup_num] || next_sends[subgroup_num]) {
            return nullptr;
        }

        pending_sst_sends[subgroup_num] = true;
        if(thread_shutdown) {
            pending_sst_sends[subgroup_num] = false;
            return nullptr;
        }
        char* buf = (char*)sst_multicast_group_ptrs[subgroup_num]->get_buffer(msg_size);
        if(!buf) {
            pending_sst_sends[subgroup_num] = false;
            return nullptr;
        }
        auto current_time = get_walltime();
        pending_message_timestamps[subgroup_num].insert(current_time);

        ((header*)buf)->header_size = sizeof(header);
        ((header*)buf)->index = future_message_indices[subgroup_num];
        ((header*)buf)->timestamp = current_time;
        ((header*)buf)->num_nulls = 0;
        ((header*)buf)->cooked_send = cooked_send;
        future_message_indices[subgroup_num]++;
        dbg_default_trace("Subgroup {}: get_sendbuffer_ptr increased future_message_indices to {}",
                          subgroup_num, future_message_indices[subgroup_num]);

        last_transfer_medium[subgroup_num] = false;
        return buf + sizeof(header);
    }
}

bool MulticastGroup::send(subgroup_id_t subgroup_num, long long unsigned int payload_size,
                          const std::function<void(char* buf)>& msg_generator, bool cooked_send) {
    if(!rdmc_sst_groups_created) {
        return false;
    }
    std::unique_lock<std::recursive_mutex> lock(msg_state_mtx);
    char* buf = get_sendbuffer_ptr(subgroup_num, payload_size, cooked_send);
    while(!buf) {
        // Don't want any deadlocks. For example, this thread cannot get a buffer because delivery is lagging
        // but the SST detect thread cannot proceed (and deliver) because it requires the same lock
        // do not use defer_lock in the unique_lock declaration above and move unlock to the end of the loop.
        // That will cause a bug. We want to unlock only when we are sure that buf is nullptr.
        lock.unlock();
        if(thread_shutdown) {
            return false;
        }
        lock.lock();
        buf = get_sendbuffer_ptr(subgroup_num, payload_size, cooked_send);
    }
    // call to the user supplied message generator
    msg_generator(buf);

    if(last_transfer_medium[subgroup_num]) {
        assert(next_sends[subgroup_num]);
        pending_sends[subgroup_num].push(std::move(*next_sends[subgroup_num]));
        next_sends[subgroup_num] = std::nullopt;
        sender_cv.notify_all();
        return true;
    } else {
        committed_sst_index[subgroup_num]++;
        pending_sst_sends[subgroup_num] = false;
        return true;
    }
}

bool MulticastGroup::check_pending_sst_sends(subgroup_id_t subgroup_num) {
    std::lock_guard<std::recursive_mutex> lock(msg_state_mtx);
    return pending_sst_sends[subgroup_num];
}

std::vector<uint32_t> MulticastGroup::get_shard_sst_indices(subgroup_id_t subgroup_num) {
    std::vector<node_id_t> shard_members = subgroup_settings_map.at(subgroup_num).members;

    std::vector<uint32_t> shard_sst_indices;
    for(auto m : shard_members) {
        shard_sst_indices.push_back(node_id_to_sst_index.at(m));
    }
    return shard_sst_indices;
}

void MulticastGroup::debug_print() {
    using std::cout;
    using std::endl;
    cout << "SST has " << sst->get_num_rows()
         << " rows; member_index is " << member_index << endl;
    cout << "Printing SST" << endl;
    for(const auto& p : subgroup_settings_map) {
        uint32_t subgroup_num = p.first;
        auto subgroup_settings = p.second;
        cout << "Subgroup " << subgroup_num << endl;
        auto shard_sst_indices = get_shard_sst_indices(subgroup_num);
        cout << "Printing seq_num, delivered_num" << endl;
        for(auto i : shard_sst_indices) {
            cout << sst->seq_num[i][subgroup_num] << " " << sst->delivered_num[i][subgroup_num] << endl;
        }
        cout << "Printing last_received_messages" << endl;
        for(auto i : shard_sst_indices) {
            uint32_t num_shard_senders = subgroup_settings.senders.size();
            for(uint j = 0; j < num_shard_senders; ++j) {
                cout << sst->num_received[i][subgroup_settings.num_received_offset + j] << " ";
            }
            cout << endl;
        }
        cout << "Printing multicastSST fields" << endl;
        sst_multicast_group_ptrs[subgroup_num]->debug_print();
        cout << endl;
    }

    std::cout << "Printing memory usage of free_message_buffers" << std::endl;
    for(const auto& p : free_message_buffers) {
        std::cout << "Subgroup " << p.first << ", Number of free buffers " << p.second.size() << std::endl;
    }
}

}  // namespace derecho
