Collage  1.2.1
High-performance C++ library for developing object-oriented distributed applications.
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
localNode.h
1 
2 /* Copyright (c) 2005-2014, Stefan Eilemann <eile@equalizergraphics.com>
3  * 2010, Cedric Stalder <cedric.stalder@gmail.com>
4  * 2012-2014, Daniel Nachbaur <danielnachbaur@gmail.com>
5  *
6  * This file is part of Collage <https://github.com/Eyescale/Collage>
7  *
8  * This library is free software; you can redistribute it and/or modify it under
9  * the terms of the GNU Lesser General Public License version 2.1 as published
10  * by the Free Software Foundation.
11  *
12  * This library is distributed in the hope that it will be useful, but WITHOUT
13  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
14  * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
15  * details.
16  *
17  * You should have received a copy of the GNU Lesser General Public License
18  * along with this library; if not, write to the Free Software Foundation, Inc.,
19  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
20  */
21 
22 #ifndef CO_LOCALNODE_H
23 #define CO_LOCALNODE_H
24 
25 #include <co/node.h> // base class
26 #include <co/objectHandler.h> // base class
27 #include <co/objectVersion.h> // VERSION_FOO used inline
28 #include <lunchbox/requestHandler.h> // base class
29 
30 #include <boost/function/function1.hpp>
31 #include <boost/function/function4.hpp>
32 
33 namespace co
34 {
35 namespace detail { class LocalNode; class ReceiverThread; class CommandThread; }
36 
44 class LocalNode : public lunchbox::RequestHandler, public Node,
45  public ObjectHandler
46 {
47 public:
52  enum Counter
53  {
55  COUNTER_ALL // must be last
56  };
57 
59  CO_API explicit LocalNode( const uint32_t type = co::NODETYPE_NODE );
60 
91  CO_API virtual bool initLocal( const int argc, char** argv );
92 
105  CO_API virtual bool listen();
106 
116  CO_API virtual bool close();
117 
119  virtual bool exitLocal() { return close(); }
120 
137  CO_API bool connect( NodePtr node );
138 
151  CO_API NodePtr connect( const NodeID& nodeID );
152 
165  CO_API NodePtr connectObjectMaster( const uint128_t& id );
166 
174  CO_API virtual bool disconnect( NodePtr node );
176 
191  CO_API bool registerObject( Object* object ) override;
192 
202  CO_API void deregisterObject( Object* object ) override;
203 
248  CO_API f_bool_t mapObject( Object* object, const uint128_t& id,
249  NodePtr master,
250  const uint128_t& version = VERSION_OLDEST );
251 
253  f_bool_t mapObject( Object* object, const ObjectVersion& v )
254  { return mapObject( object, v.identifier, 0, v.version ); }
255 
257  f_bool_t mapObject( Object* object, const uint128_t& id,
258  const uint128_t& version = VERSION_OLDEST )
259  { return mapObject( object, id, 0, version ); }
260 
262  CO_API uint32_t mapObjectNB( Object* object, const uint128_t& id,
263  const uint128_t& version = VERSION_OLDEST );
264 
266  CO_API uint32_t mapObjectNB( Object* object, const uint128_t& id,
267  const uint128_t& version,
268  NodePtr master ) override;
269 
271  CO_API bool mapObjectSync( const uint32_t requestID ) override;
272 
293  CO_API f_bool_t syncObject( Object* object, NodePtr master,
294  const uint128_t& id,
295  const uint32_t instanceID = CO_INSTANCE_ALL ) override;
302  CO_API void unmapObject( Object* object ) override;
303 
305  CO_API void disableInstanceCache();
306 
308  CO_API void expireInstanceData( const int64_t age );
309 
327  CO_API void enableSendOnRegister();
328 
330  CO_API void disableSendOnRegister();
331 
352  CO_API virtual void objectPush( const uint128_t& groupID,
353  const uint128_t& objectType,
354  const uint128_t& objectID,
355  DataIStream& istream );
356 
358  typedef boost::function< void( const uint128_t&,
359  const uint128_t&,
360  const uint128_t&,
372  CO_API void registerPushHandler( const uint128_t& groupID,
373  const PushHandler& handler );
374 
375 
377  typedef boost::function< bool( CustomICommand& ) > CommandHandler;
378 
393  CO_API bool registerCommandHandler( const uint128_t& command,
394  const CommandHandler& func,
395  CommandQueue* queue );
396 
399  CO_API void swapObject( Object* oldObject, Object* newObject );
401 
413  CO_API NodePtr getNode( const NodeID& id ) const;
414 
416  CO_API void getNodes( Nodes& nodes, const bool addSelf = true ) const;
417 
420 
426  CO_API bool inCommandThread() const;
427 
428  CO_API int64_t getTime64() const;
429  CO_API ssize_t getCounter( const Counter counter ) const;
430 
431 
439 
441  CO_API void addListener( ConnectionPtr connection );
442 
444  CO_API void removeListeners( const Connections& connections );
445 
452  CO_API void flushCommands();
453 
455  CO_API BufferPtr allocBuffer( const uint64_t size );
456 
468  CO_API bool dispatchCommand( ICommand& command ) override;
469 
470 
472  typedef lunchbox::RefPtr< co::SendToken > SendToken;
473 
482  CO_API SendToken acquireSendToken( NodePtr toNode );
483 
485  CO_API void releaseSendToken( SendToken token );
486 
488  CO_API Zeroconf getZeroconf();
490 
492  CO_API void ackRequest( NodePtr node, const uint32_t requestID );
493 
495  CO_API void ping( NodePtr remoteNode );
496 
502  CO_API bool pingIdleNodes();
503 
508  CO_API void setAffinity( const int32_t affinity );
509 
511  CO_API void addConnection( ConnectionPtr connection );
512 
513 protected:
515  CO_API ~LocalNode() override;
516 
529  CO_API bool connect( NodePtr node, ConnectionPtr connection );
530 
532  virtual void notifyConnect( NodePtr ) {}
533 
535  virtual void notifyDisconnect( NodePtr ) {}
536 
545  CO_API virtual NodePtr createNode( const uint32_t type );
546 
547 private:
548  detail::LocalNode* const _impl;
549 
550  friend class detail::ReceiverThread;
551  bool _startCommandThread( const int32_t threadID );
552  void _runReceiverThread();
553 
554  friend class detail::CommandThread;
555  bool _notifyCommandThreadIdle();
556 
557  void _cleanup();
558  void _closeNode( NodePtr node );
559  void _addConnection( ConnectionPtr connection );
560  void _removeConnection( ConnectionPtr connection );
561 
562  lunchbox::Request< void > _removeListener( ConnectionPtr connection );
563 
564  uint32_t _connect( NodePtr node );
565  NodePtr _connect( const NodeID& nodeID );
566  uint32_t _connect( NodePtr node, ConnectionPtr connection );
567  NodePtr _connect( const NodeID& nodeID, NodePtr peer );
568  NodePtr _connectFromZeroconf( const NodeID& nodeID );
569  bool _connectSelf();
570 
571  void _handleConnect();
572  void _handleDisconnect();
573  bool _handleData();
574  BufferPtr _readHead( ConnectionPtr connection );
575  ICommand _setupCommand( ConnectionPtr, ConstBufferPtr );
576  bool _readTail( ICommand&, BufferPtr, ConnectionPtr );
577  void _initService();
578  void _exitService();
579 
580  friend class ObjectStore;
581  template< typename T >
582  void _registerCommand( const uint32_t command, const CommandFunc< T >& func,
583  CommandQueue* destinationQueue )
584  {
585  registerCommand( command, func, destinationQueue );
586  }
587 
588  void _dispatchCommand( ICommand& command );
589  void _redispatchCommands();
590 
592  bool _cmdAckRequest( ICommand& command );
593  bool _cmdStopRcv( ICommand& command );
594  bool _cmdStopCmd( ICommand& command );
595  bool _cmdSetAffinity( ICommand& command );
596  bool _cmdConnect( ICommand& command );
597  bool _cmdConnectReply( ICommand& command );
598  bool _cmdConnectAck( ICommand& command );
599  bool _cmdID( ICommand& command );
600  bool _cmdDisconnect( ICommand& command );
601  bool _cmdGetNodeData( ICommand& command );
602  bool _cmdGetNodeDataReply( ICommand& command );
603  bool _cmdAcquireSendToken( ICommand& command );
604  bool _cmdAcquireSendTokenReply( ICommand& command );
605  bool _cmdReleaseSendToken( ICommand& command );
606  bool _cmdAddListener( ICommand& command );
607  bool _cmdRemoveListener( ICommand& command );
608  bool _cmdPing( ICommand& command );
609  bool _cmdCommand( ICommand& command );
610  bool _cmdCommandAsync( ICommand& command );
611  bool _cmdAddConnection( ICommand& command );
612  bool _cmdDiscard( ICommand& ) { return true; }
614 
615  LB_TS_VAR( _cmdThread )
616  LB_TS_VAR( _rcvThread )
617 };
618 
619 inline std::ostream& operator << ( std::ostream& os, const LocalNode& node )
620 {
621  os << static_cast< const Node& >( node );
622  return os;
623 }
624 }
625 #endif // CO_LOCALNODE_H
CO_API void deregisterObject(Object *object) override
Deregister a distributed object.
Num of mapObjects served for other nodes.
Definition: localNode.h:54
std::vector< ConnectionPtr > Connections
A vector of ConnectionPtr's.
Definition: types.h:126
CO_API bool connect(NodePtr node)
Connect a remote node (proxy) to this listening node.
CO_API Zeroconf getZeroconf()
lunchbox::RefPtr< co::SendToken > SendToken
A handle for a send token acquired by acquireSendToken().
Definition: localNode.h:472
lunchbox::RefPtr< ConnectionDescription > ConnectionDescriptionPtr
A reference pointer for ConnectionDescription pointers.
Definition: types.h:97
CO_API void unmapObject(Object *object) override
Unmap a mapped object.
uint128_t NodeID
A unique identifier for nodes.
Definition: types.h:84
A distributed object.
Definition: object.h:45
virtual CO_API void objectPush(const uint128_t &groupID, const uint128_t &objectType, const uint128_t &objectID, DataIStream &istream)
Handler for an Object::push() operation.
virtual CO_API bool initLocal(const int argc, char **argv)
Initialize the node.
A helper struct bundling an object identifier and version.
Definition: objectVersion.h:47
CO_API void removeListeners(const Connections &connections)
Remove listening connections from this listening node.
A thread-safe, blocking queue for ICommand buffers.
Definition: commandQueue.h:33
CO_API bool dispatchCommand(ICommand &command) override
Dispatches a command to the registered command queue.
void registerCommand(const uint32_t command, const CommandFunc< T > &func, CommandQueue *queue)
Register a command member function for a command.
Definition: dispatcher.h:94
virtual CO_API NodePtr createNode(const uint32_t type)
Factory method to create a new node.
Proxy node representing a remote LocalNode.
Definition: node.h:42
f_bool_t mapObject(Object *object, const ObjectVersion &v)
Convenience wrapper for mapObject().
Definition: localNode.h:253
CO_API void registerPushHandler(const uint128_t &groupID, const PushHandler &handler)
Register a custom handler for Object::push operations.
CO_API CommandQueue * getCommandThreadQueue()
Return the command queue to the command thread.
uint128_t identifier
the object identifier
Definition: objectVersion.h:95
CO_API f_bool_t syncObject(Object *object, NodePtr master, const uint128_t &id, const uint32_t instanceID=CO_INSTANCE_ALL) override
Synchronize the local object with a remote object.
CO_API void enableSendOnRegister()
Enable sending instance data after registration.
lunchbox::RefPtr< Node > NodePtr
A reference pointer for Node pointers.
Definition: types.h:87
f_bool_t mapObject(Object *object, const uint128_t &id, const uint128_t &version=VERSION_OLDEST)
Definition: localNode.h:257
Counter
Counters are monotonically increasing performance variables for operations performed by a LocalNode i...
Definition: localNode.h:52
virtual bool exitLocal()
Close a listening node.
Definition: localNode.h:119
CO_API LocalNode(const uint32_t type=co::NODETYPE_NODE)
Construct a new local node of the given type.
Interface for entities which map and register objects.
Definition: objectHandler.h:29
CO_API bool registerObject(Object *object) override
Register a distributed object.
boost::function< bool(CustomICommand &) > CommandHandler
Function signature for custom command handlers.
Definition: localNode.h:377
CO_API SendToken acquireSendToken(NodePtr toNode)
Acquire a send token from the given node.
virtual CO_API bool disconnect(NodePtr node)
Disconnect a connected node.
virtual CO_API bool close()
Close a listening node.
CO_API bool inCommandThread() const
A zeroconf communicator.
Definition: zeroconf.h:46
A class managing received commands.
Definition: iCommand.h:43
Node specialization for a local node.
Definition: localNode.h:44
CO_API ~LocalNode() override
Destruct this local node.
uint128_t version
the object version
Definition: objectVersion.h:96
CO_API NodePtr connectObjectMaster(const uint128_t &id)
Find and connect the node where the given object is registered.
CO_API void getNodes(Nodes &nodes, const bool addSelf=true) const
Assemble a vector of the currently connected nodes.
A std::istream-like input data stream for binary data.
Definition: dataIStream.h:41
CO_API void disableInstanceCache()
Disable the instance cache of a stopped local node.
CO_API ConnectionPtr addListener(ConnectionDescriptionPtr desc)
Add a listening connection to this listening node.
CO_API void disableSendOnRegister()
Disable sending data of newly registered objects.
CO_API bool registerCommandHandler(const uint128_t &command, const CommandHandler &func, CommandQueue *queue)
Register a custom command handler handled by this node.
boost::function< void(const uint128_t &, const uint128_t &, const uint128_t &, DataIStream &) > PushHandler
Function signature for push handlers.
Definition: localNode.h:361
CO_API void setAffinity(const int32_t affinity)
Bind this, the receiver and the command thread to the given lunchbox::Thread affinity.
CO_API bool pingIdleNodes()
Request updates from all nodes above keep-alive timeout.
std::vector< NodePtr > Nodes
A vector of NodePtr's.
Definition: types.h:105
A plain co::Node.
Definition: nodeType.h:31
CO_API void releaseSendToken(SendToken token)
CO_API NodePtr getNode(const NodeID &id) const
Get a node by identifier.
virtual CO_API bool listen()
Open all connections and put this node into the listening state.
CO_API void ping(NodePtr remoteNode)
Request keep-alive update from the remote node.
CO_API f_bool_t mapObject(Object *object, const uint128_t &id, NodePtr master, const uint128_t &version=VERSION_OLDEST)
Map a distributed object.
CO_API bool mapObjectSync(const uint32_t requestID) override
lunchbox::RefPtr< Connection > ConnectionPtr
A reference pointer for Connection pointers.
Definition: types.h:95
CO_API uint32_t mapObjectNB(Object *object, const uint128_t &id, const uint128_t &version=VERSION_OLDEST)