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

#pragma once
#ifndef THRILL_NET_FLOW_CONTROL_CHANNEL_HEADER
#define THRILL_NET_FLOW_CONTROL_CHANNEL_HEADER

#include <thrill/common/functional.hpp>
#include <thrill/common/thread_barrier.hpp>
#include <thrill/net/group.hpp>

#include <functional>
#include <string>
#include <vector>

namespace thrill {
namespace net {

//! \addtogroup net Network Communication
//! \{

/**
 * \brief Provides a blocking collection for communication.
 * \details This wraps a raw net group and should be used for
 * flow control with integral types.
 *
 * Important notice on threading: It is not allowed to call two different methods of two different
 * instances of FlowControlChannel simultaniously by different threads, since the internal synchronization state
 * (the barrier) is shared globally.
 *
 * The implementations will be replaced by better/decentral versions.
 *
 * This class is probably the worst thing I've ever coded (ej).
 */
class FlowControlChannel
{
protected:

    static const bool self_verify = false;

    /**
     * The group associated with this channel.
     */
    net::Group& group;
    /**
     * The local id.
     */
    size_t id;

    /**
     * The count of all workers connected to this group.
     */
    size_t count;

    /**
     * The id of the worker thread associated with this flow channel.
     */
    size_t threadId;

    /**
     * The count of all workers connected to this group.
     */
    size_t threadCount;

    /**
     * The shared barrier used to synchronize between worker threads on this node.
     */
    common::ThreadBarrier& barrier;

    /**
     * A shared memory location to work upon.
     */
    void** shmem;

    /**
     * \brief Sends a value of an integral type T to a certain other worker.
     * \details This method can block if there is unsufficient space
     * in the send buffer. This method may only be called by thread with ID 0.
     *
     * \param destination The id of the worker to send the value to.
     * \param value The value to send.
     */
    template <typename T>
    void SendTo(size_t destination, T value) {

        assert(threadId == 0); //Only primary thread might send/receive.

        group.connection(destination).Send(value);
    }

    /**
     * \brief Receives a value of an integral type T from a certain other worker.
     * \details This method blocks until the data is received. This method may only be called by thread with ID 0.
     *
     * \param source The id of the worker to receive the value from.
     * \param value A pointer to a memory location where the
     * received value is stored.
     */
    template <typename T>
    void ReceiveFrom(size_t source, T* value) {

        assert(threadId == 0); // Only primary thread might send/receive.

        group.connection(source).Receive(value);
    }

    template <typename T>
    void SetLocalShared(T* value) {
        if(self_verify)
            assert(*shmem == nullptr);
        assert(threadId == 0);
        *shmem = value;
    }

    template <typename T>
    T * GetLocalShared() {
        assert(*shmem != nullptr);
        return *(reinterpret_cast<T**>(shmem));
    }

    void ClearLocalShared() {
        if(self_verify) {
            assert(threadId == 0);
            *shmem = nullptr;
            barrier.await();
        }
    }

public:
    /**
     * \brief Creates a new instance of this class, wrapping a group.
     */
    explicit FlowControlChannel(net::Group& group,
                                size_t threadId, size_t threadCount,
                                common::ThreadBarrier& barrier,
                                void** shmem)
        : group(group),
          id(group.my_host_rank()), count(group.num_hosts()),
          threadId(threadId), threadCount(threadCount),
          barrier(barrier), shmem(shmem) { }

    //Yes, I know its ugly, but it will work for now. <commit_msg>Fixed compiler error.<commit_after>/*******************************************************************************
 * thrill/net/flow_control_channel.hpp
 *
 * Part of Project Thrill.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef THRILL_NET_FLOW_CONTROL_CHANNEL_HEADER
#define THRILL_NET_FLOW_CONTROL_CHANNEL_HEADER

#include <thrill/common/functional.hpp>
#include <thrill/common/thread_barrier.hpp>
#include <thrill/net/group.hpp>

#include <functional>
#include <string>
#include <vector>

namespace thrill {
namespace net {

//! \addtogroup net Network Communication
//! \{

/**
 * \brief Provides a blocking collection for communication.
 * \details This wraps a raw net group and should be used for
 * flow control with integral types.
 *
 * Important notice on threading: It is not allowed to call two different methods of two different
 * instances of FlowControlChannel simultaniously by different threads, since the internal synchronization state
 * (the barrier) is shared globally.
 *
 * The implementations will be replaced by better/decentral versions.
 *
 * This class is probably the worst thing I've ever coded (ej).
 */
class FlowControlChannel
{
protected:

    static const bool self_verify = false;

    /**
     * The group associated with this channel.
     */
    net::Group& group;
    /**
     * The local id.
     */
    size_t id;

    /**
     * The count of all workers connected to this group.
     */
    size_t count;

    /**
     * The id of the worker thread associated with this flow channel.
     */
    size_t threadId;

    /**
     * The count of all workers connected to this group.
     */
    size_t threadCount;

    /**
     * The shared barrier used to synchronize between worker threads on this node.
     */
    common::ThreadBarrier& barrier;

    /**
     * A shared memory location to work upon.
     */
    void** shmem;

    /**
     * \brief Sends a value of an integral type T to a certain other worker.
     * \details This method can block if there is unsufficient space
     * in the send buffer. This method may only be called by thread with ID 0.
     *
     * \param destination The id of the worker to send the value to.
     * \param value The value to send.
     */
    template <typename T>
    void SendTo(size_t destination, T value) {

        assert(threadId == 0); //Only primary thread might send/receive.

        group.connection(destination).Send(value);
    }

    /**
     * \brief Receives a value of an integral type T from a certain other worker.
     * \details This method blocks until the data is received. This method may only be called by thread with ID 0.
     *
     * \param source The id of the worker to receive the value from.
     * \param value A pointer to a memory location where the
     * received value is stored.
     */
    template <typename T>
    void ReceiveFrom(size_t source, T* value) {

        assert(threadId == 0); // Only primary thread might send/receive.

        group.connection(source).Receive(value);
    }

    template <typename T>
    void SetLocalShared(T* value) {
        if(self_verify)
            assert(*shmem == nullptr);
        assert(threadId == 0);
        *shmem = value;
    }

    template <typename T>
    T * GetLocalShared() {
        assert(*shmem != nullptr);
        return *(reinterpret_cast<T**>(shmem));
    }

    void ClearLocalShared() {
        if(self_verify) {
            assert(threadId == 0);
            *shmem = nullptr;
            barrier.Await();
        }
    }

public:
    /**
     * \brief Creates a new instance of this class, wrapping a group.
     */
    explicit FlowControlChannel(net::Group& group,
                                size_t threadId, size_t threadCount,
                                common::ThreadBarrier& barrier,
                                void** shmem)
        : group(group),
          id(group.my_host_rank()), count(group.num_hosts()),
          threadId(threadId), threadCount(threadCount),
          barrier(barrier), shmem(shmem) { }

    //Yes, I know its ugly, but it will work for now. 
{'fraction_non_alphanumeric': 0.12573512114796517, 'fraction_numerical': 0.002352387673488591, 'mean_word_length': 2.4962993421052633, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, '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': '2214768', 'n_tokens_mistral': 2359, 'n_tokens_neox': 2241, 'n_words': 848}