4
\$\begingroup\$

I just wrote a basic worker class for this person

https://stackoverflow.com/questions/35827459/assigning-a-new-task-to-a-thread-after-the-thread-completes-in-c

and I wanted to ask if the code has been written properly, it's supposed to have multi-thread support.

Any feedback would be nice.

Live Demo

class Worker
{
public:
    Worker(bool start) : m_Running(start) { if (start) private_start(); }
    Worker() : m_Running(false) { }
    ~Worker() { stop(); }

    template<typename... Args>
    void push_task(Args&&... args)
    {
        {
            std::lock_guard<std::mutex> lk(m_Mutex);
            m_Queue.push_back(std::bind(std::forward<Args>(args)...));
        }

        m_Condition.notify_all();
    }

    void start()
    {
        {
            std::lock_guard<std::mutex> lk(m_Mutex);
            if (m_Running == true) return;
            m_Running = true;
        }

        private_start();
    }

    void stop()
    {
        {
            std::lock_guard<std::mutex> lk(m_Mutex);
            if (m_Running == false) return;
            m_Running = false;
        }

        m_Condition.notify_all();
        m_Thread.join();
    }

private:
    void private_start()
    {
        m_Thread = std::thread([this]
        {
            for (;;)
            {
                decltype(m_Queue) local_queue;
                {
                    std::unique_lock<std::mutex> lk(m_Mutex);
                    m_Condition.wait(lk, [&] { return !m_Queue.empty() + !m_Running; });

                    if (!m_Running)
                    {
                        for (auto& func : m_Queue)
                            func();

                        m_Queue.clear();
                        return;
                    }

                    local_queue = std::swap(m_Queue, local_queue);
                }

                for (auto& func : local_queue)
                    func();
            }
        });
    }

private:
    std::condition_variable m_Condition;
    std::list<std::function<void()>> m_Queue;
    std::mutex m_Mutex;
    std::thread m_Thread;
    bool m_Running = false;
};

void do_work(unsigned id)
{
    //static std::mutex cout_mutex;
    //std::lock_guard<std::mutex> lk(cout_mutex);
    std::cout << id << std::endl;
}

int main()
{
    {
        Worker workers[3];
        int counter = 0;

        for (auto& worker : workers)
            worker.start();

        for (auto& worker : workers)
        {
            for (int i = 0; i < 5; i++)
                worker.push_task(do_work, ++counter + i);
        }
    }

    std::cout << "finish" << std::endl;
    getchar();

    return 0;
}
\$\endgroup\$
0

1 Answer 1

1
\$\begingroup\$

Does moving leave the object in a usable state?

According to this your m_Queue object is valid, yet in an unspecified state. The easiest way to make the state specified again might be calling m_Queue.clear() directly after local_queue = std::move(m_Queue); or even reinitializing m_Queue with new object.

I tried to find out whether just clearing m_Queue is sufficient, yet I couldn't find anything definate apart from comments. I think it is at least "more safe" to use than you current approach.


For readability reasons I would have also changed the structure of your private_start to something like this

void private_start()
{
    m_Thread = std::thread([this]
    {
        for (;;)
        {
            decltype(m_Queue) local_queue;
            {
                std::unique_lock<std::mutex> lk(m_Mutex);
                m_Condition.wait(lk, [&] { return !m_Queue.empty() + !m_Running; });

                if (!m_Running)
                {
                    for (auto& func : m_Queue)
                        func();

                    m_Queue.clear();
                    return;
                }

                //Move to local
                local_queue = std::move(m_Queue);
                m_Queue.clear();
            }

            //Process local
            for (auto& func : local_queue)
            {
                func();
            }
        }
    });
}

But that's more of a personnal preference.


As a last comment I would recommend updating your example to call stop on all workers so your application doesn't end prematurely.

\$\endgroup\$
1
  • \$\begingroup\$ I have updated my class with std::swap instead of std::move, but I could have used list<T> splice(whole list) instead, as I think the speed is almost the same. Both solutions are one liners and guaranty that m_Queue remains valid. From what I've read clearing m_Queue after moving it properly restores the state of the list, but not one liner tho, neither is reinitializing m_Queue with new object. Also, the destructor already calls stop, which waits (blocks) for the threads to finish. I've taken your advise to changed the structure of private_start \$\endgroup\$
    – José
    Commented Mar 7, 2016 at 11:03

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.