<commit_before>/******************************************************************************\
 *           ___        __                                                    *
 *          /\_ \    __/\ \                                                   *
 *          \//\ \  /\_\ \ \____    ___   _____   _____      __               *
 *            \ \ \ \/\ \ \ '__`\  /'___\/\ '__`\/\ '__`\  /'__`\             *
 *             \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_           *
 *             /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\          *
 *             \/____/ \/_/\/___/  \/____/ \ \ \/  \ \ \/  \/__/\/_/          *
 *                                          \ \_\   \ \_\                     *
 *                                           \/_/    \/_/                     *
 *                                                                            *
 * Copyright (C) 2011, 2012                                                   *
 * <NAME> <<EMAIL>>                     *
 *                                                                            *
 * This file is part of libcppa.                                              *
 * libcppa is free software: you can redistribute it and/or modify it under   *
 * the terms of the GNU Lesser General Public License as published by the     *
 * Free Software Foundation, either version 3 of the License                  *
 * or (at your option) any later version.                                     *
 *                                                                            *
 * libcppa is distributed in the hope that it will be useful,                 *
 * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                       *
 * See the GNU Lesser General Public License for more details.                *
 *                                                                            *
 * You should have received a copy of the GNU Lesser General Public License   *
 * along with libcppa. If not, see <http://www.gnu.org/licenses/>.            *
\******************************************************************************/


#include <set>
#include <stdexcept>

#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"

namespace {

using namespace cppa;

typedef std::lock_guard<util::shared_spinlock> exclusive_guard;
typedef util::shared_lock_guard<util::shared_spinlock> shared_guard;
typedef util::upgrade_lock_guard<util::shared_spinlock> upgrade_guard;

class local_group_module;

class local_group : public group {

 public:

    void enqueue(actor* sender, any_tuple msg) /*override*/ {
        shared_guard guard(m_shared_mtx);
        for (auto& s : m_subscribers) {
            s->enqueue(sender, msg);
        }
    }

    group::subscription subscribe(const channel_ptr& who) /*override*/ {
        exclusive_guard guard(m_shared_mtx);
        if (m_subscribers.insert(who).second) {
            return {who, this};
        }
        return {};
    }

    void unsubscribe(const channel_ptr& who) /*override*/ {
        exclusive_guard guard(m_shared_mtx);
        m_subscribers.erase(who);
    }

    void serialize(serializer* sink);

    local_group(local_group_module* mod, std::string id);

 private:

    util::shared_spinlock m_shared_mtx;
    std::set<channel_ptr> m_subscribers;

};

class local_group_module : public group::module {

    typedef group::module super;

 public:

    local_group_module() : super("local") { }

    group_ptr get(const std::string& identifier) {
        shared_guard guard(m_mtx);
        auto i = m_instances.find(identifier);
        if (i != m_instances.end()) {
            return i->second;
        }
        else {
            group_ptr tmp(new local_group(this, identifier));
            { // lifetime scope of uguard
                upgrade_guard uguard(guard);
                auto p = m_instances.insert(std::make_pair(identifier, tmp));
                // someone might preempt us
                return p.first->second;
            }
        }
    }

    intrusive_ptr<group> deserialize(deserializer* source) {
        auto gname = source->read_value(pt_u8string);
        return this->get(cppa::get<std::string>(gname));
    }

    void serialize(local_group* ptr, serializer* sink) {
        sink->write_value(ptr->identifier());
    }

 private:

    util::shared_spinlock m_mtx;
    std::map<std::string, group_ptr> m_instances;

};

local_group::local_group(local_group_module* mod, std::string id)
: group(mod, std::move(id)) { }

void local_group::serialize(serializer* sink) {
    // this cast is safe, because the only available constructor accepts
    // local_group_module* as module pointer
    static_cast<local_group_module*>(m_module)->serialize(this, sink);
}

std::atomic<size_t> m_ad_hoc_id;

} // namespace <anonymous>

namespace cppa { namespace detail {

group_manager::group_manager() {
    group::unique_module_ptr ptr(new local_group_module);<commit_msg>implemented proxies for local groups of other hosts<commit_after>/******************************************************************************\
 *           ___        __                                                    *
 *          /\_ \    __/\ \                                                   *
 *          \//\ \  /\_\ \ \____    ___   _____   _____      __               *
 *            \ \ \ \/\ \ \ '__`\  /'___\/\ '__`\/\ '__`\  /'__`\             *
 *             \_\ \_\ \ \ \ \L\ \/\ \__/\ \ \L\ \ \ \L\ \/\ \L\.\_           *
 *             /\____\\ \_\ \_,__/\ \____\\ \ ,__/\ \ ,__/\ \__/.\_\          *
 *             \/____/ \/_/\/___/  \/____/ \ \ \/  \ \ \/  \/__/\/_/          *
 *                                          \ \_\   \ \_\                     *
 *                                           \/_/    \/_/                     *
 *                                                                            *
 * Copyright (C) 2011, 2012                                                   *
 * <NAME> <<EMAIL>>                     *
 *                                                                            *
 * This file is part of libcppa.                                              *
 * libcppa is free software: you can redistribute it and/or modify it under   *
 * the terms of the GNU Lesser General Public License as published by the     *
 * Free Software Foundation, either version 3 of the License                  *
 * or (at your option) any later version.                                     *
 *                                                                            *
 * libcppa is distributed in the hope that it will be useful,                 *
 * but WITHOUT ANY WARRANTY; without even the implied warranty of             *
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                       *
 * See the GNU Lesser General Public License for more details.                *
 *                                                                            *
 * You should have received a copy of the GNU Lesser General Public License   *
 * along with libcppa. If not, see <http://www.gnu.org/licenses/>.            *
\******************************************************************************/


#include <set>
#include <stdexcept>

#include "cppa/any_tuple.hpp"
#include "cppa/serializer.hpp"
#include "cppa/deserializer.hpp"

#include "cppa/detail/middleman.hpp"
#include "cppa/detail/group_manager.hpp"
#include "cppa/detail/addressed_message.hpp"

#include "cppa/util/shared_spinlock.hpp"
#include "cppa/util/shared_lock_guard.hpp"
#include "cppa/util/upgrade_lock_guard.hpp"

namespace {

using namespace cppa;

typedef std::lock_guard<util::shared_spinlock> exclusive_guard;
typedef util::shared_lock_guard<util::shared_spinlock> shared_guard;
typedef util::upgrade_lock_guard<util::shared_spinlock> upgrade_guard;

class local_group_module;

class local_group : public group {

 public:

    void enqueue(actor* sender, any_tuple msg) {
        shared_guard guard(m_shared_mtx);
        for (auto& s : m_subscribers) {
            s->enqueue(sender, msg);
        }
    }

    group::subscription subscribe(const channel_ptr& who) {
        exclusive_guard guard(m_shared_mtx);
        if (m_subscribers.insert(who).second) {
            return {who, this};
        }
        return {};
    }

    void unsubscribe(const channel_ptr& who) {
        exclusive_guard guard(m_shared_mtx);
        m_subscribers.erase(who);
    }

    void serialize(serializer* sink);

    inline const process_information& process() const {
        return *m_process;
    }

    inline const process_information_ptr& process_ptr() const {
        return m_process;
    }

    local_group(local_group_module* mod, std::string id,
                process_information_ptr parent = process_information::get());

 private:

    util::shared_spinlock m_shared_mtx;
    std::set<channel_ptr> m_subscribers;
    process_information_ptr m_process;

};

class local_group_proxy : public local_group {

    typedef local_group super;

 public:

    template<typename... Args>
    local_group_proxy(Args&&... args) : super(std::forward<Args>(args)...) { }

    void enqueue(actor* sender, any_tuple msg) {
        detail::middleman_enqueue(process_ptr(), sender, this, std::move(msg));
    }

    void remote_enqueue(actor* sender, any_tuple msg) {
        super::enqueue(sender, std::move(msg));
    }

};

typedef intrusive_ptr<local_group> local_group_ptr;

class local_group_module : public group::module {

    typedef group::module super;

 public:

    local_group_module()
    : super("local"), m_process(process_information::get()) { }

    group_ptr get(const std::string& identifier) {
        shared_guard guard(m_instances_mtx);
        auto i = m_instances.find(identifier);
        if (i != m_instances.end()) {
            return i->second;
        }
        else {
            local_group_ptr tmp(new local_group(this, identifier));
            { // lifetime scope of uguard
                upgrade_guard uguard(guard);
                auto p = m_instances.insert(std::make_pair(identifier, tmp));
                // someone might preempt us
                return p.first->second;
            }
        }
    }

    intrusive_ptr<group> deserialize(deserializer* source) {
        primitive_variant ptup[3];
        primitive_type ptypes[] = {pt_u8string, pt_uint32, pt_u8string};
        source->read_tuple(3, ptypes, ptup);
        auto& identifier = cppa::get<std::string>(ptup[0]);
        auto  process_id = cppa::get<std::uint32_t>(ptup[1]);
        auto& node_id    = cppa::get<std::string>(ptup[2]);
        if (   process_id == process().process_id()
            && equal(node_id, process().node_id())) {
            return this->get(identifier);
        }
        else {
            process_information pinf(process_id, node_id);
            shared_guard guard(m_proxies_mtx);
            auto& node_map = m_proxies[pinf];
            auto i = node_map.find(identifier);
            if (i != node_map.end()) {
                return i->second;
            }
            else {
                local_group_ptr tmp(new local_group_proxy(this, identifier));
                process_information_ptr piptr;
                // re-use process_information_ptr from another proxy if possible
                if (node_map.empty()) {
                    piptr.reset(new process_information(pinf));
                } else {
                    piptr = node_map.begin()->second->process_ptr();
                }
                upgrade_guard uguard(guard);
                auto p = node_map.insert(std::make_pair(identifier, tmp));
                // someone might preempt us
                return p.first->second;
            }
        }
    }

    void serialize(local_group* ptr, serializer* sink) {
        primitive_variant ptup[3];
        ptup[0] = ptr->identifier();
        ptup[1] = ptr->process().process_id();
        ptup[2] = to_string(ptr->process().node_id());
        sink->write_tuple(3, ptup);
    }

    inline const process_information& process() const {
        return *m_process;
    }

 private:

    typedef std::map<std::string, local_group_ptr> local_group_map;

    process_information_ptr m_process;
    util::shared_spinlock m_instances_mtx;
    local_group_map m_instances;
    util::shared_spinlock m_proxies_mtx;
    std::map<process_information, local_group_map> m_proxies;

};

local_group::local_group(local_group_module* mod,
                         std::string id,
                         process_information_ptr parent)
: group(mod, std::move(id)), m_process(std::move(parent)) { }

void local_group::serialize(serializer* sink) {
    // this cast is safe, because the only available constructor accepts
    // local_group_module* as module pointer
    static_cast<local_group_module*>(m_module)->serialize(this, sink);
}

std::atomic<size_t> m_ad_hoc_id;

} // namespace <anonymous>

namespace cppa { namespace detail {

group_manager::group_manager() {
    group::unique_module_ptr ptr(new local_group_module);
{'fraction_non_alphanumeric': 0.13413630582778485, 'fraction_numerical': 0.0026183885688636195, 'mean_word_length': 1.7027901334411646, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 52, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5178496', 'n_tokens_mistral': 3741, 'n_tokens_neox': 3597, 'n_words': 765}