Function object passing for a task scheduler

Question

Here is my full implementation of a generic Functor-like class:

    #ifndef FUNCTIONHPP
    #define FUNCTIONHPP

    #include "Types.hpp"
    #include <tuple>
    #include <utility>

    BEGIN_NAMESPACE //This is a define in "Types.hpp"

    namespace FN_HELPERS
    {
    template<int ...>
    struct seq
    {
    };
    template<int N,int ...S>
    struct gens : gens<N-1,N-1,S...>
    {
    };
    template<int ...S>
    struct gens<0, S...>
    {
        typedef seq<S...> type;
    };

    //Global Function Pointer Return Type Identification
    template <typename R>
    struct returnType;

    template <typename R, typename ...Args>
    struct returnType<R (*)(Args...)>
    {
        typedef R type;
    };

    template <typename R,typename ...Args>
    struct returnType<R (*)(Args...,...)>
    {
        typedef R type;
    };

};

//A generic functor to pass between threads
class Function
{
    public:
    Function() {}
    virtual ~Function() {}
    Function(const Function&) = delete;
    void operator=(const Function&) = delete;
    virtual void call()=0;
    virtual void* returnValue()=0;
};

template <typename F,typename R,typename ...Args>
class GlobalFunction : public Function
{
    public:
    GlobalFunction(F func,Args&&... args) : m_func(func),
                                        m_args(std::forward<Args>(args)...)
    {
    }
    virtual ~GlobalFunction()
    {
    }
    GlobalFunction(const GlobalFunction&) = delete;
    void operator=(const GlobalFunction&) = delete;
    virtual void call()
    {
        callFunction(typename FN_HELPERS::gens<sizeof...(Args)>::type());
    }
    virtual void* returnValue()
    {
        return static_cast<void*>(&m_returnValue);
    }
    template<int ...S>
    void callFunction(FN_HELPERS::seq<S...>)
    {
        m_returnValue = (*m_func)(std::get<S>(m_args) ...);
    }
    private:
    F m_func;
    R m_returnValue;
    std::tuple<Args...> m_args;

};

template <typename F,typename ...Args>
class GlobalFunction<F,void,Args...> : public Function
{
    public:
    GlobalFunction(F func,Args&&... args) : m_func(func),
                                        m_args(std::forward<Args>(args)...)
    {
    }
    virtual ~GlobalFunction()
    {
    }
    GlobalFunction(const GlobalFunction&) = delete;
    void operator=(const GlobalFunction&) = delete;
    virtual void call()
    {
        callFunction(typename FN_HELPERS::gens<sizeof...(Args)>::type());
    }
    virtual void* returnValue()
    {
        return (void*)0x0;
    }
    private:
    F m_func;
    std::tuple<Args...> m_args;
    template<int ...S>
    void callFunction(FN_HELPERS::seq<S...>)
    {
        (*m_func)(std::get<S>(m_args) ...);
    }
};

//Grab return value helper
//T - Return Type, PTR - Function* pointer
template <typename T>
T RET(Function* f)
{
    return *(static_cast<T*>(f->returnValue()));
}

template <typename F,typename ...Args>
Function* _wrapGFn(F func,Args&&... args)
{
    return new GlobalFunction<decltype(func),typename FN_HELPERS::returnType<F>::type,Args...>(func,std::forward<Args>(args)...);
}

//Create function object
//F - Function Pointer
#define S_FN(F, ... ) \
    _wrapGFn(F, ##__VA_ARGS__)

END_NAMESPACE //This is a #define in "Types.h"

#endif

I use this like so:

//This code wants to schedule a task (function) for execution in parallel on another thread.
Function* f = S_FN(&someFunction,someArg);
//Nanomsg send functions
nn_send(sockfd,(void*)f,sizeof(f),0); //Send it

The thread that receives the task, just runs it, and doesn't care about what it's calling, the return type, or the arguments.

//Recv
void* buf;
nn_recv(sockfd,buf,sizeof(Function*),0);
Function* f = (Function*)buf;
f->call();
//Send back "I'm done" message or the like.

This is for a simple thread pool, it works fine, but the efficiency of the code is what is bothering me. I use nanomsg to avoid mutexes and locks. It's like a mini zeroMQ I suppose. I like this approach because it's cheap on the message size (although there are no copies using nanomsg so it's kind of irrelevant, the zero copy code was omitted to keep this huge post shorter), and I do not need any locks or mutexes between threads.

The drawback is, these functions are called millions of times a second. The scheduler can handle the routing just fine, but at this rate, I'm worried that the construction cost of the function objects and the virtual call are adding too much overhead. (Most of the functions are relatively simple and are completed quickly)

I've looked into some methods (CRTP, static polymorphism, etc...) to try to eliminate the virtual call, but I do not see a way around this. The only reason I ask is because the code gets called so often and I'm wondering if there is a simpler way to do this that would avoid the overhead.

Is there a better way to do this? Is this design flawed in a way I haven't realized?

Answer 1

I think you have re-designed large parts of the standard library.

I would simply do this:

#include <utility>
#include <iostream>
#include <unistd.h>
#include <errno.h>

Structure to hold any function across a pipe.

This is the bit you put up for code review. Using the standard library it becomes trivial.

The current code does not allow you to pass back the result. But that functionality is provided by std::promise and std::future. I am not 100% conversant with this new technology so I will point you at another recent question that uses them well. Platform independant thread pool v2 from this you should be able to plug this in very easily.

struct FunctionOverPipe
{
    virtual ~FunctionOverPipe() {}
    virtual void call() = 0;
};

template<typename T>
struct FunctionOverPipeAction: public FunctionOverPipe
{
    T   action;
    FunctionOverPipeAction(T&& action)
        : action(std::move(action))
    {}
    virtual void call()
    {
        action();
    }
};

A JobQueue

This is the bit that is not up for code review. Its slightly longer but should be relatively trivial.

It uses a pipe as a lockless queue. If you assume that writes/reads are atomic it should work multi-processes. Of course writes/reads are not atomic but because we only read/write an object the size of a pointer it should work most of the time. If we need to we can add extra code to make sure they are atomic (or easy to spot if the atomic ness fails).

class JobQueue
{
    int mypipe[2];
    std::size_t const dataSize = sizeof(FunctionOverPipe*);
    public:
        JobQueue()
        {
            /* Create the pipe. */
            if (pipe (mypipe))
            {
                std::cerr << "Pipe failed.\n";
                exit(1);
            }
        }
        ~JobQueue()
        {
            close(mypipe[0]);
            close(mypipe[1]);
        }

        template<typename T>
        void send(T&& action)
        {
            std::unique_ptr<FunctionOverPipe> dataUniq(new FunctionOverPipeAction<T>(std::forward<T>(action)));
            FunctionOverPipe*       dataValue     = dataUniq.get();

            write(dataValue);
            dataUniq.release();
        }
        void executeNext()
        {
            FunctionOverPipe*   dataValue;
            read(dataValue);

            std::unique_ptr<FunctionOverPipe> dataUniq(dataValue);
            dataUniq->call();
        }

        void write(FunctionOverPipe*& dataValue)
        {
            char*       data    = reinterpret_cast<char*>(&dataValue);
            std::size_t sent    = 0;
            do
            {
                size_t sending = ::write(mypipe[1], data + sent, dataSize - sent);
                if (sending == -1 && (errno == EAGAIN || errno == EINTR))
                {   sending = 0;
                }
                if (sending == -1)
                {   std::cerr << "Send Failed\n";
                    exit(1);
                }
                sent += sending;
            } while(sent != dataSize);
        }

        void read(FunctionOverPipe*& dataValue)
        {
            char*       data    = reinterpret_cast<char*>(&dataValue);
            std::size_t recv    = 0;
            do
            {
                std::size_t recieved = ::read(mypipe[0], data + recv, dataSize - recv);
                if (recieved == -1 && (errno == EAGAIN || errno == EINTR))
                {   recieved = 0;
                }
                if (recieved == -1)
                {   std::cerr << "Read Failed\n";
                    exit(1);
                }
                recv += recieved;
            } while(recv != dataSize);
        }

};

Main.

Showing it working with a lambda. Which means it should work with nearly anything.

int main()
{
    JobQueue        jobQueue;

    jobQueue.send([](){std::cout << "Testing\n";});
    // Do this bit from any or multiple threads.
    jobQueue.executeNext();
}

Answer 2

Really don't like this.
In the old days (15 years ago) this was used as a trick to help support non conforming compilers that did not have namespace. That's not really relevant now and you use namespace on the next line.

BEGIN_NAMESPACE //This is a define in "Types.hpp"

namespace FN_HELPERS

These helper functions look like they are already supported by the standard std::integer_sequence.

template<int ...>
struct seq;
template<int N,int ...S>
struct gens : gens<N-1,N-1,S...>;
template<int ...S>
struct gens<0, S...>;

Return types from anonymous functions are covered by std::promise and std::future.

//Global Function Pointer Return Type Identification
template <typename R>
struct returnType;

template <typename R, typename ...Args>
struct returnType<R (*)(Args...)>;

template <typename R,typename ...Args>
struct returnType<R (*)(Args...,...)>;

Don't see the need to define the default constructor here.

class Function
{
    public:
    Function() {}
    virtual ~Function() {}
    Function(const Function&) = delete;
    void operator=(const Function&) = delete;
    virtual void call()=0;
    virtual void* returnValue()=0;
};

The compiler generated one should work just fine. But seems like you are re-building the std::function class.

After all that nice template code:

//Create function object
//F - Function Pointer
#define S_FN(F, ... ) \
    _wrapGFn(F, ##__VA_ARGS__)

You now resort to a macro!

Great. But nn_send() has error codes you need to check. You should never use functions like this directly. As they don't guarantee to send all the data first try. This is something that goes in a loop:

Also you need to pass the address of f. &f otherwise you are sending the bytes pointed to by f (after it is cast to void*). So that fails. Also your use of C-cast here is horrible. You were using C++ casts before (why change now). Its also un-necessary. All pointers auto convert when used as a void*.

Function* f = S_FN(&someFunction,someArg);
nn_send(sockfd,(void*)f,sizeof(f),0); //Send it

A lot of the same problems with the read.

You should check the return value. Use inside a loop to make sure you send it all. You should pass a pointer to buf (as you are writing over that pointer). No need for buf to be a void* that just means an extra cast (which again is a C-Cast).

void* buf;
nn_recv(sockfd,buf,sizeof(Function*),0);
Function* f = (Function*)buf;
f->call();