soa/service/endpoint.h

/* endpoint.h                                                      -*- C++ -*-
   Jeremy Barnes, 21 February 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

   Generic endpoint; will be subclassed for a particular connection type.
*/

#ifndef __rtb__endpoint_h__
#define __rtb__endpoint_h__

#include <ace/Synch.h>
#include <ace/Guard_T.h>
#include <set>
#include <boost/function.hpp>
#include <boost/thread/thread.hpp>
#include <iostream>
#include "jml/arch/exception.h"
#include "jml/arch/backtrace.h"
#include "jml/utils/smart_ptr_utils.h"
#include "jml/arch/wakeup_fd.h"
#include "transport.h"
#include "connection_handler.h"
#include "soa/service/epoller.h"
#include <map>


namespace Datacratic {


struct ConnectionHandler;
struct EndpointBase;
struct PassiveEndpoint;

/*****************************************************************************/
/* ENDPOINT BASE                                                             */
/*****************************************************************************/

struct EndpointBase : public Epoller {

    EndpointBase(const std::string & name);

    virtual ~EndpointBase();

    void useThisThread();

    void shutdown();

    virtual std::string hostname() const = 0;

    virtual int port() const = 0;

    boost::function<bool ()> onCheckFinished;

    void sleepUntilIdle() const;

    int threadsActive() const { return threadsActive_; }

    virtual void dumpState() const;
    
    virtual int numConnections() const;

    virtual std::map<std::string, int> numConnectionsByHost() const;

    typedef boost::function<void (TransportBase *)> OnTransportEvent;
    OnTransportEvent onTransportOpen, onTransportClose;

    const std::string & name() const { return name_; }

    void makeRealTime(int priority = 1);

    virtual void spinup(int num_threads, bool synchronous);

protected:

    bool checkFinished() const
    {
        if (onCheckFinished) return onCheckFinished();
        return false;
    }

    virtual void
    associateHandler(const std::shared_ptr<TransportBase> & transport)
    {
        if (!transport->hasSlave())
            throw ML::Exception("either makeNewTransport or associateHandler"
                                "need to be overridden to make a handler");
    }
    
    virtual void closePeer() = 0;

    struct SPLess {
        template<typename SP>
        bool operator () (const SP & sp1, const SP & sp2) const
        {
            return sp1.get() < sp2.get();
        }
    };

    typedef std::set<std::shared_ptr<TransportBase>, SPLess> Connections;

    Connections alive;

    virtual void
    notifyNewTransport(const std::shared_ptr<TransportBase> & transport);

    virtual void
    notifyCloseTransport(const std::shared_ptr<TransportBase> & transport);

    virtual void
    notifyRecycleTransport(const std::shared_ptr<TransportBase> & transport);

    virtual void restartPolling(TransportBase * transport);

    virtual void startPolling(TransportBase * transport);

    virtual void stopPolling(TransportBase * transport);

    void doAsync(const std::shared_ptr<TransportBase> & transport,
                 const boost::function<void ()> & callback,
                 const char * nameOfCallback);

    typedef ACE_Recursive_Thread_Mutex Lock;
    typedef ACE_Guard<Lock> Guard;
    
    mutable Lock lock;

    mutable ACE_Semaphore idle;
    
    mutable bool modifyIdle;

private:
    std::string name_;
    std::unique_ptr<boost::thread_group> eventThreads;
    std::vector<boost::thread *> eventThreadList;
    int threadsActive_;

    friend class TransportBase;
    friend class ConnectionHandler;
    template<typename Transport> friend class ConnectorT;

    /* Number of active FDs in items */
    int numTransports;

    /* FD we can use to wake up the event loop */
    ML::Wakeup_Fd wakeup;

    /* Are we shutting down? */
    bool shutdown_;

    std::map<std::string, int> numTransportsByHost;

    void runEventThread(int threadNum, int numThreads);

    bool handleEpollEvent(epoll_event & event);
};

} // namespace Datacratic

#endif /* __rtb__endpoint_h__ */