<commit_before>#pragma once

#include <string>
#include <thread>
#include "mmcore/AbstractNamedObject.h"
#include "mmcore/AbstractNamedObjectContainer.h"
#include "mmcore/AbstractService.h"
//#include "CommandFunctionPtr.h"
#include <atomic>
#include <map>
#include "mmcore/utility/ZMQContextUser.h"

namespace megamol {
namespace core {
namespace utility {

class LuaHostService : public core::AbstractService {
public:
    static unsigned int ID;

    virtual const char* Name() const { return "LuaRemote"; }

    LuaHostService(core::CoreInstance& core);
    virtual ~LuaHostService();

    virtual bool Initalize(bool& autoEnable);
    virtual bool Deinitialize();

    inline const std::string& GetAddress(void) const { return address; }
    void SetAddress(const std::string& ad);

protected:
    virtual bool enableImpl();
    virtual bool disableImpl();

private:
    void serve();
    void servePair();
    std::string makeAnswer(const std::string& req);
    std::string makePairAnswer(const std::string& req) const;
    std::atomic<int> lastPairPort;

    // ModuleGraphAccess mgAccess;
    ZMQContextUser::ptr context;

    std::thread serverThread;
    std::vector<std::thread> pairThreads;
    bool serverRunning;

    std::string address;
};


} /* namespace utility */
} /* namespace core */
} /* namespace megamol */

#include <atomic>
#include <future>
#include <queue>
#include <thread>
namespace megamol::core::utility {
template <typename Item> class threadsafe_queue {
#define guard std::lock_guard<std::mutex> lock(m_mutex);

public:
    void push(Item&& i) {
        guard;
        m_queue.emplace(std::move(i));
    }

    std::queue<Item> pop_queue() {
        guard;
        std::queue<Item> q = std::move(m_queue);
        m_queue = {};
        return std::move(q);
    }

    bool empty() {
        guard;
        return m_queue.empty();
    }

#undef guard
private:
    std::queue<Item> m_queue;
    mutable std::mutex m_mutex;
};

class LuaHostNetworkConnectionsBroker {
public:
    struct ThreadSignaler {
        std::atomic<bool> running = false;

        void start() { running.store(true, std::memory_order_release); }
        void stop() { running.store(false, std::memory_order_release); }
        bool is_running() { return running.load(std::memory_order_acquire); }
    };

    struct Worker {
        ThreadSignaler signal;
        std::thread thread;

        void join() {
            signal.stop();
            if (thread.joinable())
                thread.join();
        }
    };
    Worker broker_worker;
    std::list<Worker> lua_workers;

    void close() {
        broker_worker.join();
        for (auto& w : lua_workers) {
            w.join();
        }
    }

    struct LuaRequest {
        std::string request;
        std::reference_wrapper<std::promise<std::string>> answer_promise;
    };
    threadsafe_queue<LuaRequest> request_queue;

    std::queue<LuaRequest> get_request_queue() { return std::move(request_queue.pop_queue()); }

    zmq::context_t zmq_context;

    // connection broker starts threads that execute incoming lua commands
    bool spawn_connection_broker(std::string broker_address, bool retry_socket_port = false) {
        assert(broker_worker.signal.is_running() == false);

        int retries = 0;
        int max_retries = 100+1;

        auto delimiter_pos = broker_address.find_last_of(':');
        std::string base_address = broker_address.substr(0, delimiter_pos+1+1); // include delimiter
        std::string port_string = broker_address.substr(delimiter_pos+1); // ignore delimiter
        int port = std::stoi(port_string);

        // retry to start broker socket on next port until max_retries reached
        std::string address;<commit_msg>LuaHostService: fix string port extraction logic<commit_after>#pragma once

#include <string>
#include <thread>
#include "mmcore/AbstractNamedObject.h"
#include "mmcore/AbstractNamedObjectContainer.h"
#include "mmcore/AbstractService.h"
//#include "CommandFunctionPtr.h"
#include <atomic>
#include <map>
#include "mmcore/utility/ZMQContextUser.h"

namespace megamol {
namespace core {
namespace utility {

class LuaHostService : public core::AbstractService {
public:
    static unsigned int ID;

    virtual const char* Name() const { return "LuaRemote"; }

    LuaHostService(core::CoreInstance& core);
    virtual ~LuaHostService();

    virtual bool Initalize(bool& autoEnable);
    virtual bool Deinitialize();

    inline const std::string& GetAddress(void) const { return address; }
    void SetAddress(const std::string& ad);

protected:
    virtual bool enableImpl();
    virtual bool disableImpl();

private:
    void serve();
    void servePair();
    std::string makeAnswer(const std::string& req);
    std::string makePairAnswer(const std::string& req) const;
    std::atomic<int> lastPairPort;

    // ModuleGraphAccess mgAccess;
    ZMQContextUser::ptr context;

    std::thread serverThread;
    std::vector<std::thread> pairThreads;
    bool serverRunning;

    std::string address;
};


} /* namespace utility */
} /* namespace core */
} /* namespace megamol */

#include <atomic>
#include <future>
#include <queue>
#include <thread>
namespace megamol::core::utility {
template <typename Item> class threadsafe_queue {
#define guard std::lock_guard<std::mutex> lock(m_mutex);

public:
    void push(Item&& i) {
        guard;
        m_queue.emplace(std::move(i));
    }

    std::queue<Item> pop_queue() {
        guard;
        std::queue<Item> q = std::move(m_queue);
        m_queue = {};
        return std::move(q);
    }

    bool empty() {
        guard;
        return m_queue.empty();
    }

#undef guard
private:
    std::queue<Item> m_queue;
    mutable std::mutex m_mutex;
};

class LuaHostNetworkConnectionsBroker {
public:
    struct ThreadSignaler {
        std::atomic<bool> running = false;

        void start() { running.store(true, std::memory_order_release); }
        void stop() { running.store(false, std::memory_order_release); }
        bool is_running() { return running.load(std::memory_order_acquire); }
    };

    struct Worker {
        ThreadSignaler signal;
        std::thread thread;

        void join() {
            signal.stop();
            if (thread.joinable())
                thread.join();
        }
    };
    Worker broker_worker;
    std::list<Worker> lua_workers;

    void close() {
        broker_worker.join();
        for (auto& w : lua_workers) {
            w.join();
        }
    }

    struct LuaRequest {
        std::string request;
        std::reference_wrapper<std::promise<std::string>> answer_promise;
    };
    threadsafe_queue<LuaRequest> request_queue;

    std::queue<LuaRequest> get_request_queue() { return std::move(request_queue.pop_queue()); }

    zmq::context_t zmq_context;

    // connection broker starts threads that execute incoming lua commands
    bool spawn_connection_broker(std::string broker_address, bool retry_socket_port = false) {
        assert(broker_worker.signal.is_running() == false);

        int retries = 0;
        int max_retries = 100+1;

        auto delimiter_pos = broker_address.find_last_of(':');
        std::string base_address = broker_address.substr(0, delimiter_pos+1); // include delimiter
        std::string port_string = broker_address.substr(delimiter_pos+1); // ignore delimiter
        int port = std::stoi(port_string);

        // retry to start broker socket on next port until max_retries reached
        std::string address;
{'fraction_non_alphanumeric': 0.12042094045557479, 'fraction_numerical': 0.002264553083788464, 'mean_word_length': 3.2853881278538815, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 45, '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': '5385746', 'n_tokens_mistral': 2285, 'n_tokens_neox': 2216, 'n_words': 530}