localNode.h

/* Copyright (c) 2005-2016, Stefan Eilemann <eile@equalizergraphics.com>
 *                          Cedric Stalder <cedric.stalder@gmail.com>
 *                          Daniel Nachbaur <danielnachbaur@gmail.com>
 *
 * This file is part of Collage <https://github.com/Eyescale/Collage>
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License version 2.1 as published
 * by the Free Software Foundation.
 *
 * This library is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
 * details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#ifndef CO_LOCALNODE_H
#define CO_LOCALNODE_H

#include <co/node.h>                 // base class
#include <co/objectHandler.h>        // base class
#include <co/objectVersion.h>        // VERSION_FOO used inline
#include <lunchbox/requestHandler.h> // base class

#include <boost/function/function1.hpp>
#include <boost/function/function4.hpp>

namespace co
{
namespace detail
{
class LocalNode;
class ReceiverThread;
class CommandThread;
}

class LocalNode : public lunchbox::RequestHandler,
                  public Node,
                  public ObjectHandler
{
public:
    enum Counter
    {
        COUNTER_MAP_OBJECT_REMOTE, 
        COUNTER_ALL                // must be last
    };

    CO_API explicit LocalNode(const uint32_t type = co::NODETYPE_NODE);

    CO_API virtual bool initLocal(const int argc, char** argv);

    CO_API virtual bool listen();

    CO_API virtual bool close();

    virtual bool exitLocal() { return close(); }
    CO_API bool connect(NodePtr node);

    CO_API NodePtr connect(const NodeID& nodeID);

    CO_API NodePtr connectObjectMaster(const uint128_t& id);

    CO_API virtual bool disconnect(NodePtr node);

    CO_API bool launch(NodePtr node, const std::string& command);

    CO_API NodePtr syncLaunch(const uint128_t& nodeID, int64_t timeout);

    CO_API bool registerObject(Object* object) override;

    CO_API void deregisterObject(Object* object) override;

    CO_API f_bool_t mapObject(Object* object, const uint128_t& id,
                              NodePtr master,
                              const uint128_t& version = VERSION_OLDEST);

    f_bool_t mapObject(Object* object, const ObjectVersion& v)
    {
        return mapObject(object, v.identifier, 0, v.version);
    }

    f_bool_t mapObject(Object* object, const uint128_t& id,
                       const uint128_t& version = VERSION_OLDEST)
    {
        return mapObject(object, id, 0, version);
    }

    CO_API uint32_t mapObjectNB(Object* object, const uint128_t& id,
                                const uint128_t& version = VERSION_OLDEST);

    CO_API uint32_t mapObjectNB(Object* object, const uint128_t& id,
                                const uint128_t& version,
                                NodePtr master) override;

    CO_API bool mapObjectSync(const uint32_t requestID) override;

    CO_API f_bool_t
        syncObject(Object* object, const uint128_t& id, NodePtr master,
                   const uint32_t instanceID = CO_INSTANCE_ALL) override;
    CO_API void unmapObject(Object* object) override;

    CO_API void disableInstanceCache();

    CO_API void expireInstanceData(const int64_t age);

    CO_API void enableSendOnRegister();

    CO_API void disableSendOnRegister();

    CO_API virtual void objectPush(const uint128_t& groupID,
                                   const uint128_t& objectType,
                                   const uint128_t& objectID,
                                   DataIStream& istream);

    typedef boost::function<void(const uint128_t&, 
                                 const uint128_t&, 
                                 const uint128_t&, 
                                 DataIStream&)>
        PushHandler;
    CO_API void registerPushHandler(const uint128_t& groupID,
                                    const PushHandler& handler);

    typedef boost::function<bool(CustomICommand&)> CommandHandler;

    CO_API bool registerCommandHandler(const uint128_t& command,
                                       const CommandHandler& func,
                                       CommandQueue* queue);

    CO_API void swapObject(Object* oldObject, Object* newObject);

    CO_API NodePtr getNode(const NodeID& id) const;

    CO_API Nodes getNodes(const bool addSelf = true) const;

    CO_API CommandQueue* getCommandThreadQueue();

    CO_API bool inCommandThread() const;

    CO_API const Strings& getCommandLine() const;

    CO_API int64_t getTime64() const;                       
    CO_API ssize_t getCounter(const Counter counter) const; 


    CO_API ConnectionPtr addListener(ConnectionDescriptionPtr desc);

    CO_API void addListener(ConnectionPtr connection);

    CO_API void removeListeners(const Connections& connections);

    CO_API void flushCommands();

    CO_API BufferPtr allocBuffer(const uint64_t size);

    CO_API bool dispatchCommand(ICommand& command) override;

    typedef lunchbox::RefPtr<co::SendToken> SendToken;

    CO_API SendToken acquireSendToken(NodePtr toNode);

    CO_API void releaseSendToken(SendToken token);

    CO_API Zeroconf getZeroconf();

    CO_API void ackRequest(NodePtr node, const uint32_t requestID);

    CO_API void ping(NodePtr remoteNode);

    CO_API bool pingIdleNodes();

    CO_API void setAffinity(const int32_t affinity);

    CO_API void addConnection(ConnectionPtr connection);

protected:
    CO_API ~LocalNode() override;

    CO_API bool connect(NodePtr node, ConnectionPtr connection);

    virtual void notifyConnect(NodePtr) {}
    virtual void notifyDisconnect(NodePtr) {}
    CO_API virtual NodePtr createNode(const uint32_t type);

private:
    detail::LocalNode* const _impl;

    friend class detail::ReceiverThread;
    bool _startCommandThread(const int32_t threadID);
    void _runReceiverThread();

    friend class detail::CommandThread;
    bool _notifyCommandThreadIdle();

    void _cleanup();
    void _closeNode(NodePtr node);
    void _addConnection(ConnectionPtr connection);
    void _removeConnection(ConnectionPtr connection);

    lunchbox::Request<void> _removeListener(ConnectionPtr connection);

    uint32_t _connect(NodePtr node);
    NodePtr _connect(const NodeID& nodeID);
    uint32_t _connect(NodePtr node, ConnectionPtr connection);
    NodePtr _connect(const NodeID& nodeID, NodePtr peer);
    NodePtr _connectFromZeroconf(const NodeID& nodeID);
    bool _connectSelf();

    bool _setupPeer(const std::string& setupOpts);

    void _handleConnect();
    void _handleDisconnect();
    bool _handleData();
    BufferPtr _readHead(ConnectionPtr connection);
    ICommand _setupCommand(ConnectionPtr, ConstBufferPtr);
    bool _readTail(ICommand&, BufferPtr, ConnectionPtr);
    void _initService();
    void _exitService();

    friend class ObjectStore;
    template <typename T>
    void _registerCommand(const uint32_t command, const CommandFunc<T>& func,
                          CommandQueue* destinationQueue)
    {
        registerCommand(command, func, destinationQueue);
    }

    void _dispatchCommand(ICommand& command);
    void _redispatchCommands();

    bool _cmdAckRequest(ICommand& command);
    bool _cmdStopRcv(ICommand& command);
    bool _cmdStopCmd(ICommand& command);
    bool _cmdSetAffinity(ICommand& command);
    bool _cmdConnect(ICommand& command);
    bool _cmdConnectReply(ICommand& command);
    bool _cmdConnectAck(ICommand& command);
    bool _cmdID(ICommand& command);
    bool _cmdDisconnect(ICommand& command);
    bool _cmdGetNodeData(ICommand& command);
    bool _cmdGetNodeDataReply(ICommand& command);
    bool _cmdAcquireSendToken(ICommand& command);
    bool _cmdAcquireSendTokenReply(ICommand& command);
    bool _cmdReleaseSendToken(ICommand& command);
    bool _cmdAddListener(ICommand& command);
    bool _cmdRemoveListener(ICommand& command);
    bool _cmdPing(ICommand& command);
    bool _cmdCommand(ICommand& command);
    bool _cmdCommandAsync(ICommand& command);
    bool _cmdAddConnection(ICommand& command);
    bool _cmdDiscard(ICommand&) { return true; }

    LB_TS_VAR(_cmdThread)
    LB_TS_VAR(_rcvThread)
};

inline std::ostream& operator<<(std::ostream& os, const LocalNode& node)
{
    os << static_cast<const Node&>(node);
    return os;
}
}
#endif // CO_LOCALNODE_H