soa/service/message_loop.h

/* message_loop.h                                                  -*- C++ -*-
   Jeremy Barnes, 31 May 2012
   Copyright (c) 2012 Datacratic.  All rights reserved.

   Base class for loop that listens for various types of messages.
*/

#pragma once

#include "async_event_source.h"
#include <boost/thread/thread.hpp>
#include <mutex>
#include <functional>
#include <mutex>
#include "soa/service/epoller.h"


namespace Datacratic {



/*****************************************************************************/
/* MESSAGE LOOP                                                              */
/*****************************************************************************/



struct MessageLoop : public Epoller {
    
    MessageLoop(int numThreads = 1, double maxAddedLatency = 0.0005);
    ~MessageLoop();

    void init(int numThreads = 1, double maxAddedLatency = 0.0005);

    void start(std::function<void ()> onStop = std::function<void ()>());

    void startSync();
    
    //void sleepUntilIdle();

    void shutdown();

    void addSource(const std::string & name,
                   AsyncEventSource & source,
                   int priority = 0);

    void addSource(const std::string & name,
                   std::shared_ptr<AsyncEventSource> source,
                   int priority = 0);

    void addPeriodic(const std::string & name,
                     double timePeriodSeconds,
                     std::function<void (uint64_t)> toRun,
                     int priority = 0);

    void addSourceDeferred(const std::string & name,
                           AsyncEventSource & source,
                           int priority = 0);

    void addSourceDeferred(const std::string & name,
                           std::shared_ptr<AsyncEventSource> source,
                           int priority = 0);

    void addPeriodicDeferred(const std::string & name,
                             double timePeriodSeconds,
                             std::function<void (uint64_t)> toRun,
                             int priority = 0);
    
    typedef std::function<void (volatile int & shutdown_,
                                int64_t threadId)> SubordinateThreadFn;

    void startSubordinateThread(const SubordinateThreadFn & mainFn);

    virtual bool processOne();

    virtual bool poll() const;

    void debug(bool debugOn);

    void removeSource(AsyncEventSource * source);

    void checkNeedsPoll();

    double totalSleepSeconds() const { return totalSleepTime; }
    
private:
    void runWorkerThread();
    
    void wakeupMainThread();

    void addSourceImpl(const std::string & name,
                       std::shared_ptr<AsyncEventSource> source, int priority);
    
    //ML::Wakeup_Fd wakeup;
    
    std::vector<std::pair<std::string, std::shared_ptr<AsyncEventSource> > > sources;

    std::vector<std::tuple<std::string, std::shared_ptr<AsyncEventSource>, int > > deferredSources;

    int numThreadsCreated;
    boost::thread_group threads;
    
    // TODO: bad, bad, bad... make an API whereby we can ask this to happen later
    // so we don't need a recursive mutex
    //typedef std::recursive_mutex Lock;

    typedef std::mutex Lock;
    typedef std::unique_lock<Lock> Guard;
    mutable Lock lock;
    
    volatile int shutdown_;

    volatile int idle_;

    bool debug_;

    volatile double totalSleepTime;

    double maxAddedLatency_;

    bool handleEpollEvent(epoll_event & event);
};

} // namespace Datacratic