<commit_before>/*******************************************************************************
 * c7a/data/channel_multiplexer.hpp
 *
 * Part of Project c7a.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef C7A_DATA_CHANNEL_MULTIPLEXER_HEADER
#define C7A_DATA_CHANNEL_MULTIPLEXER_HEADER

#include <c7a/common/atomic_movable.hpp>
#include <c7a/data/block_writer.hpp>
#include <c7a/data/channel.hpp>
#include <c7a/net/dispatcher_thread.hpp>
#include <c7a/net/group.hpp>

#include <algorithm>
#include <map>
#include <memory>
#include <vector>

namespace c7a {
namespace data {

//! \addtogroup data Data Subsystem
//! \{

/*!
 * Multiplexes virtual Connections on Dispatcher.
 *
 * A worker as a TCP conneciton to each other worker to exchange large amounts
 * of data. Since multiple exchanges can occur at the same time on this single
 * connection we use multiplexing. The slices are called Blocks and are
 * indicated by a \ref StreamBlockHeader. Multiple Blocks form a Stream on a
 * single TCP connection. The multiplexer multiplexes all streams on all
 * sockets.
 *
 * All sockets are polled for headers. As soon as the a header arrives it is
 * either attached to an existing channel or a new channel instance is
 * created.
 */
class ChannelMultiplexer
{
public:
    explicit ChannelMultiplexer(size_t num_workers_per_node)
        : dispatcher_("dispatcher"), next_id_(num_workers_per_node, 0), num_workers_per_node_(num_workers_per_node) { }

    //! non-copyable: delete copy-constructor
    ChannelMultiplexer(const ChannelMultiplexer&) = delete;
    //! non-copyable: delete assignment operator
    ChannelMultiplexer& operator = (const ChannelMultiplexer&) = delete;
    //! default move constructor
    ChannelMultiplexer(ChannelMultiplexer&&) = default;

    //! Closes all client connections
    ~ChannelMultiplexer() {
        if (group_ != nullptr) {
            // close all still open Channels
            for (auto& ch : channels_)
                ch.second->Close();
        }

        // terminate dispatcher, this waits for unfinished AsyncWrites.
        dispatcher_.Terminate();

        if (group_ != nullptr)
            group_->Close();
    }

    void Connect(net::Group* group) {
        group_ = group;
        for (size_t id = 0; id < group_->num_connections(); id++) {
            if (id == group_->my_connection_id()) continue;
            AsyncReadStreamBlockHeader(group_->connection(id));
        }
    }

    //! Indicates if a channel exists with the given id
    //! Channels exist if they have been allocated before
    //! \param local_worker_id of the local worker who requested the channel
    bool HasChannel(size_t id, size_t local_worker_id) {
        return channels_.find(std::make_pair(local_worker_id, id)) != channels_.end();
    }

    //TODO Method to access channel via queue -> requires vec<Queue> or MultiQueue
    //TODO Method to access channel via callbacks

    //! Allocate the next channel
    //! \param local_worker_id of the local worker who requested the channel
    size_t AllocateNext(size_t local_worker_id) {
        assert(local_worker_id < next_id_.size());
        return next_id_[local_worker_id] = next_id_[local_worker_id] + 1;
    }

    //! Get channel with given id, if it does not exist, create it.
    //! \param local_worker_id of the local worker who requested the channel
    ChannelPtr GetOrCreateChannel(size_t id, size_t local_worker_id) {
        std::lock_guard<std::mutex> lock(mutex_);
        return _GetOrCreateChannel(id, local_worker_id);
    }

private:
    static const bool debug = false;

    net::DispatcherThread dispatcher_;

    //! Channels have an ID in block headers<commit_msg>resolve todo<commit_after>/*******************************************************************************
 * c7a/data/channel_multiplexer.hpp
 *
 * Part of Project c7a.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef C7A_DATA_CHANNEL_MULTIPLEXER_HEADER
#define C7A_DATA_CHANNEL_MULTIPLEXER_HEADER

#include <c7a/common/atomic_movable.hpp>
#include <c7a/data/block_writer.hpp>
#include <c7a/data/channel.hpp>
#include <c7a/net/dispatcher_thread.hpp>
#include <c7a/net/group.hpp>

#include <algorithm>
#include <map>
#include <memory>
#include <vector>

namespace c7a {
namespace data {

//! \addtogroup data Data Subsystem
//! \{

/*!
 * Multiplexes virtual Connections on Dispatcher.
 *
 * A worker as a TCP conneciton to each other worker to exchange large amounts
 * of data. Since multiple exchanges can occur at the same time on this single
 * connection we use multiplexing. The slices are called Blocks and are
 * indicated by a \ref StreamBlockHeader. Multiple Blocks form a Stream on a
 * single TCP connection. The multiplexer multiplexes all streams on all
 * sockets.
 *
 * All sockets are polled for headers. As soon as the a header arrives it is
 * either attached to an existing channel or a new channel instance is
 * created.
 */
class ChannelMultiplexer
{
public:
    explicit ChannelMultiplexer(size_t num_workers_per_node)
        : dispatcher_("dispatcher"), next_id_(num_workers_per_node, 0), num_workers_per_node_(num_workers_per_node) { }

    //! non-copyable: delete copy-constructor
    ChannelMultiplexer(const ChannelMultiplexer&) = delete;
    //! non-copyable: delete assignment operator
    ChannelMultiplexer& operator = (const ChannelMultiplexer&) = delete;
    //! default move constructor
    ChannelMultiplexer(ChannelMultiplexer&&) = default;

    //! Closes all client connections
    ~ChannelMultiplexer() {
        if (group_ != nullptr) {
            // close all still open Channels
            for (auto& ch : channels_)
                ch.second->Close();
        }

        // terminate dispatcher, this waits for unfinished AsyncWrites.
        dispatcher_.Terminate();

        if (group_ != nullptr)
            group_->Close();
    }

    void Connect(net::Group* group) {
        group_ = group;
        for (size_t id = 0; id < group_->num_connections(); id++) {
            if (id == group_->my_connection_id()) continue;
            AsyncReadStreamBlockHeader(group_->connection(id));
        }
    }

    //! Indicates if a channel exists with the given id
    //! Channels exist if they have been allocated before
    //! \param local_worker_id of the local worker who requested the channel
    bool HasChannel(size_t id, size_t local_worker_id) {
        return channels_.find(std::make_pair(local_worker_id, id)) != channels_.end();
    }

    //TODO(ts) Method to access channel via callbacks

    //! Allocate the next channel
    //! \param local_worker_id of the local worker who requested the channel
    size_t AllocateNext(size_t local_worker_id) {
        assert(local_worker_id < next_id_.size());
        return next_id_[local_worker_id] = next_id_[local_worker_id] + 1;
    }

    //! Get channel with given id, if it does not exist, create it.
    //! \param local_worker_id of the local worker who requested the channel
    ChannelPtr GetOrCreateChannel(size_t id, size_t local_worker_id) {
        std::lock_guard<std::mutex> lock(mutex_);
        return _GetOrCreateChannel(id, local_worker_id);
    }

private:
    static const bool debug = false;

    net::DispatcherThread dispatcher_;

    //! Channels have an ID in block headers
{'fraction_non_alphanumeric': 0.130683294362172, 'fraction_numerical': 0.005455962587685113, 'mean_word_length': 3.4683691236215903, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2208588', 'n_tokens_mistral': 2263, 'n_tokens_neox': 2142, 'n_words': 699}