invoking member functions from producer thread

marital_weeping 0
Viewed 33

Using a producer-consumer pattern I'd like to submit calls to the member functions of struct A i.e., func_1() and func_2() by passing the name of the function object (something like A::func_1), and a list of arguments taken by these functions in a queue-like buffer queue_in. I'd then like to collect these results (here a std::variant of all the possible return types (double and int)) in another buffer queue_out for later processing.

At the moment, I'm only able to hack my way by having q_in hold std::string and enumerating all the possibilities manually. Of course I'm also missing passing any possible arguments to func_1() and func_2().

#include <iostream>
#include "blockingconcurrentqueue.h" // https://github.com/cameron314/concurrentqueue
#include <variant>

using Result = std::variant<double, int>;
using queue_in = moodycamel::BlockingConcurrentQueue<std::string>;
using queue_out = moodycamel::BlockingConcurrentQueue<Result>;

struct A {

    explicit A(queue_in &q_in, queue_out &q_out) {
        std::thread t([&]() {
            for (;;) {
                std::string s;
                if (q_in.wait_dequeue_timed(s, -1)) {
                    if (s == "1")
                        q_out.enqueue(this->func_1());
                    else if (s == "2")
                        q_out.enqueue(this->func_2());
                }
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
        });
        t.join();
    }

    double func_1() const { // func_1 might have arguments!
        std::cout << "func_1() called.\n";
        return 13.0f;
    }

    int func_2() const { // func_2 might have arguments!
        std::cout << "func_2() called.\n";
        return 1;
    }
};

void producer(queue_in &q_in, const size_t N) {
    for (size_t i = 0; i != N; ++i) {
        q_in.enqueue("1" /*..., arguments for func_1*/); // prefer to call the function-object instead
        q_in.enqueue("2" /*..., arguments for func_2*/);
    }
}

void result_consumer(queue_out &q_out) {
    for (;;) {
        Result r;
        if (q_out.wait_dequeue_timed(r, -1)) {
            std::visit([](auto &&arg) {
                std::cout << "The result is: " <<  arg << std::endl;
            }, r);
        }
        std::this_thread::sleep_for(std::chrono::seconds(0));
    }
}

int main() {
    const size_t N = 2;
    queue_in q_in;
    queue_out q_out;

    std::thread producer_thread(producer, std::ref(q_in), N);
    std::thread result_consumer_thread(result_consumer, std::ref(q_out));
    A a(q_in, q_out);
    producer_thread.join();
    result_consumer_thread.join();
}

How can I do this instead via function objects? Second, how do I invoke any possible arguments for the member functions of A. And third, can I avoid the use of a std::variant like result type?

c++
1 Answers

Taking inspiration from @n.1.8e9-where's-my-sharem. 's comments here's how I got the above to work using std::function and std::bind

using Result = std::variant<double, int>;
using FO = std::function<Result(void)>;
using queue_in = moodycamel::BlockingConcurrentQueue<FO>;
using queue_out = moodycamel::BlockingConcurrentQueue<Result>;

struct A {

    explicit A(queue_in &q_in, queue_out &q_out) {
        std::thread t([&]() {
            for (;;) {
                FO o;
                if (q_in.wait_dequeue_timed(o, -1)) {
                    q_out.enqueue(o());
                }
                std::this_thread::sleep_for(std::chrono::seconds(1));
            }
        });
        t.detach();
    }

    void consumer() {
    }

    double func_1(double x, const size_t &s) const { // note the arguments
        std::cout << "func_1() called.\n";
        return x + s;
    }

    int func_2(int x) const { 
        std::cout << "func_2() called.\n";
        return x;
    }
};

void producer(const A& a, queue_in &q_in, const size_t N) { // note the first argument: the producer() must know about some instance of A
    for (size_t i = 0; i != N; ++i) {
        FO o;
        if (i % 2 == 0)
            o = std::bind(&A::func_1, a, double(i), size_t(10));
        else
            o = std::bind(&A::func_2, a, int(i));
        q_in.enqueue(o);
    }
}

void result_consumer(queue_out &q_out) {
    for (;;) {
        Result r;
        if (q_out.wait_dequeue_timed(r, -1)) {
            std::visit([&](auto &&arg) {
                std::cout << "The result is: " <<  arg << std::endl;
            }, r);
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
}

int main() {
    const size_t N = 2;
    queue_in q_in;
    queue_out q_out;

    A a(q_in, q_out);
    std::thread producer_thread(producer, std::ref(a), std::ref(q_in), N);
    std::thread result_consumer_thread(result_consumer, std::ref(q_out));
    producer_thread.join();
    result_consumer_thread.join();
}

Output:

func_1() called.
The result is: 10
func_2() called.
The result is: 1
marital_weeping0
Related