dispatcher.h

Go to the documentation of this file.

/*
 * Copyright (c) 2013-2014 ADTECH GmbH
 * Licensed under MIT (https://github.com/adtechlabs/libtasks/blob/master/COPYING)
 *
 * Author: Andreas Pohl
 */

#ifndef _TASKS_DISPATCHER_H_
#define _TASKS_DISPATCHER_H_

#include <vector>
#include <list>
#include <condition_variable>
#include <mutex>
#include <memory>
#include <atomic>
#include <unistd.h>

#include <tasks/ev_wrapper.h>
#include <tasks/tools/bitset.h>
#include <tasks/logging.h>

namespace tasks {

class worker;
class executor;
class task;
class event_task;
class exec_task;
class disposable;

struct signal_data;

typedef std::function<void(int)> signal_func_t;

class dispatcher {
    friend class test_exec;

  public:
    enum class mode { SINGLE_LOOP, MULTI_LOOP };

    dispatcher(uint8_t num_workers);

    static void init_workers(uint8_t num_workers) {
        if (nullptr == m_instance) {
            m_instance = std::make_shared<dispatcher>(num_workers);
        }
    }

    static void init_run_mode(mode m) {
        if (nullptr != m_instance) {
            terr("ERROR: dispatcher::init_run_mode must be called before anything else!" << std::endl);
            assert(false);
        }
        m_run_mode = m;
    }

    static mode run_mode() { return m_run_mode; }

    static std::shared_ptr<dispatcher> instance() {
        if (nullptr == m_instance) {
            // Create as many workers as we have CPU's per default
            m_instance = std::make_shared<dispatcher>(sysconf(_SC_NPROCESSORS_ONLN));
        }
        return m_instance;
    }

    static void destroy() {
        if (nullptr != m_instance) {
            if (!m_instance->m_term) {
                m_instance->terminate();
                m_instance->join();
            }
            m_instance.reset();
            m_instance = nullptr;
        }
    }

    static void add_signal_handler(int sig, signal_func_t func);

    std::shared_ptr<worker> free_worker();

    std::shared_ptr<executor> free_executor();

    void add_free_worker(uint8_t id);

    inline worker* last_worker() {
        return m_workers[m_last_worker_id].get();
    }

    worker* get_worker_by_task(event_task* task);

    void add_task(task* task);
    void add_event_task(event_task* task);
    void add_exec_task(exec_task* task);

    void remove_task(task* task);
    void remove_event_task(event_task* task);
    void remove_exec_task(exec_task* task);

    [[deprecated]] void run(int num, ...);

    void run(std::vector<tasks::task*>& tasks);

    void start();

    void join();

    inline void terminate() {
        m_term = true;
        m_finish_cond.notify_one();
    }

    void print_worker_stats() const;

  private:
    static std::shared_ptr<dispatcher> m_instance;
    std::atomic<bool> m_term;

    std::vector<std::shared_ptr<worker> > m_workers;
    uint8_t m_num_workers = 0;

    std::list<std::shared_ptr<executor> > m_executors;
    std::mutex m_executor_mutex;

    static mode m_run_mode;

    tools::bitset m_workers_busy;
    tools::bitset::int_type m_last_worker_id = 0;

    std::atomic<uint8_t> m_rr_worker_id;

    std::condition_variable m_finish_cond;
    std::mutex m_finish_mutex;

    bool m_started = false;
};

}  // tasks

#endif  // _TASKS_DISPATCHER_H_