Line data Source code
1 :
2 : /* Copyright (c) 2007-2014, Stefan Eilemann <eile@equalizergraphics.com>
3 : * 2011-2012, Daniel Nachbaur <danielnachbaur@gmail.com>
4 : *
5 : * This file is part of Collage <https://github.com/Eyescale/Collage>
6 : *
7 : * This library is free software; you can redistribute it and/or modify it under
8 : * the terms of the GNU Lesser General Public License version 2.1 as published
9 : * by the Free Software Foundation.
10 : *
11 : * This library is distributed in the hope that it will be useful, but WITHOUT
12 : * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
13 : * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
14 : * details.
15 : *
16 : * You should have received a copy of the GNU Lesser General Public License
17 : * along with this library; if not, write to the Free Software Foundation, Inc.,
18 : * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 : */
20 :
21 : #include "objectCM.h"
22 :
23 : #include "nodeCommand.h"
24 : #include "nullCM.h"
25 : #include "node.h"
26 : #include "object.h"
27 : #include "objectCommand.h"
28 : #include "objectInstanceDataOStream.h"
29 : #include "objectDataOCommand.h"
30 :
31 : #include <lunchbox/scopedMutex.h>
32 :
33 21 : co::ObjectCMPtr co::ObjectCM::ZERO = new co::NullCM;
34 :
35 : #ifdef CO_INSTRUMENT_MULTICAST
36 : lunchbox::a_int32_t co::ObjectCM::_hit( 0 );
37 : lunchbox::a_int32_t co::ObjectCM::_miss( 0 );
38 : #endif
39 :
40 : namespace co
41 : {
42 82 : ObjectCM::ObjectCM( Object* object )
43 82 : : _object( object )
44 82 : {}
45 :
46 82 : ObjectCM::~ObjectCM()
47 81 : {}
48 :
49 89 : void ObjectCM::exit()
50 : {
51 89 : lunchbox::ScopedFastWrite mutex( _lock );
52 89 : _object = 0;
53 89 : }
54 :
55 2 : void ObjectCM::push( const uint128_t& groupID, const uint128_t& typeID,
56 : const Nodes& nodes )
57 : {
58 2 : LBASSERT( _object );
59 2 : LBASSERT( !nodes.empty( ));
60 2 : if( nodes.empty( ))
61 : {
62 0 : LBWARN << "Push to an empty set of nodes" << std::endl;
63 2 : return;
64 : }
65 :
66 2 : ObjectInstanceDataOStream os( this );
67 2 : os.enablePush( getVersion(), nodes );
68 2 : _object->getInstanceData( os );
69 :
70 : // Send push notification to remote cmd thread while connections are valid
71 2 : OCommand( os.getConnections(), CMD_NODE_OBJECT_PUSH )
72 4 : << _object->getID() << groupID << typeID;
73 :
74 2 : os.disable(); // handled by remote recv thread
75 : }
76 :
77 4 : bool ObjectCM::sendSync( const MasterCMCommand& command )
78 : {
79 4 : lunchbox::ScopedFastWrite mutex( _lock );
80 4 : if( !_object )
81 : {
82 0 : LBWARN << "Sync from detached object requested" << std::endl;
83 0 : return false;
84 : }
85 :
86 4 : const uint128_t& maxCachedVersion = command.getMaxCachedVersion();
87 : const bool useCache =
88 4 : command.useCache() &&
89 8 : command.getMasterInstanceID() == _object->getInstanceID() &&
90 4 : maxCachedVersion == getVersion();
91 :
92 4 : if( !useCache )
93 : {
94 4 : ObjectInstanceDataOStream os( this );
95 4 : os.enableSync( getVersion(), command );
96 4 : _object->getInstanceData( os );
97 4 : os.disable();
98 : }
99 8 : NodePtr node = command.getNode();
100 : node->send( CMD_NODE_SYNC_OBJECT_REPLY, useCache /*preferMulticast*/ )
101 8 : << node->getNodeID() << command.getObjectID() << command.getRequestID()
102 12 : << true << command.useCache() << useCache;
103 8 : return true;
104 : }
105 :
106 41 : bool ObjectCM::_addSlave( const MasterCMCommand& command,
107 : const uint128_t& version )
108 : {
109 41 : LBASSERT( version != VERSION_NONE );
110 41 : LBASSERT( command.getType() == COMMANDTYPE_NODE );
111 41 : LBASSERT( command.getCommand() == CMD_NODE_MAP_OBJECT );
112 :
113 : // process request
114 41 : if( command.getRequestedVersion() == VERSION_NONE )
115 : {
116 : // no data to send, send empty version
117 4 : _sendMapSuccess( command, false /* mc */ );
118 4 : _sendEmptyVersion( command, VERSION_NONE, false /* mc */ );
119 4 : _sendMapReply( command, VERSION_NONE, true, false, false /* mc */ );
120 4 : return true;
121 : }
122 :
123 37 : const bool replyUseCache = command.useCache() &&
124 37 : (command.getMasterInstanceID() == _object->getInstanceID( ));
125 37 : return _initSlave( command, version, replyUseCache );
126 : }
127 :
128 8 : bool ObjectCM::_initSlave( const MasterCMCommand& command,
129 : const uint128_t& replyVersion, bool replyUseCache )
130 : {
131 : #if 0
132 : LBLOG( LOG_OBJECTS ) << "Object id " << _object->_id << " v" << _version
133 : << ", instantiate on " << node->getNodeID()
134 : << std::endl;
135 : #endif
136 :
137 : #ifndef NDEBUG
138 8 : const uint128_t& version = command.getRequestedVersion();
139 8 : if( version != VERSION_OLDEST && version < replyVersion )
140 0 : LBINFO << "Mapping version " << replyVersion << " instead of "
141 0 : << version << std::endl;
142 : #endif
143 :
144 8 : if( replyUseCache &&
145 8 : command.getMinCachedVersion() <= replyVersion &&
146 0 : command.getMaxCachedVersion() >= replyVersion )
147 : {
148 : #ifdef CO_INSTRUMENT_MULTICAST
149 : ++_hit;
150 : #endif
151 0 : _sendMapSuccess( command, false );
152 0 : _sendMapReply( command, replyVersion, true, replyUseCache, false );
153 0 : return true;
154 : }
155 :
156 8 : lunchbox::ScopedFastWrite mutex( _lock );
157 8 : if( !_object )
158 : {
159 0 : LBWARN << "Map to detached object requested" << std::endl;
160 0 : return false;
161 : }
162 :
163 : #ifdef CO_INSTRUMENT_MULTICAST
164 : ++_miss;
165 : #endif
166 8 : replyUseCache = false;
167 :
168 8 : _sendMapSuccess( command, true );
169 :
170 : // send instance data
171 16 : ObjectInstanceDataOStream os( this );
172 :
173 8 : os.enableMap( replyVersion, command.getNode(), command.getInstanceID( ));
174 8 : _object->getInstanceData( os );
175 8 : os.disable();
176 8 : if( !os.hasSentData( ))
177 : // no data, send empty command to set version
178 0 : _sendEmptyVersion( command, replyVersion, true /* mc */ );
179 :
180 8 : _sendMapReply( command, replyVersion, true, replyUseCache, true );
181 16 : return true;
182 : }
183 :
184 41 : void ObjectCM::_sendMapSuccess( const MasterCMCommand& command,
185 : const bool multicast )
186 : {
187 : command.getNode()->send( CMD_NODE_MAP_OBJECT_SUCCESS, multicast )
188 82 : << command.getNode()->getNodeID() << command.getObjectID()
189 123 : << command.getRequestID() << command.getInstanceID()
190 123 : << _object->getChangeType() << _object->getInstanceID();
191 41 : }
192 :
193 41 : void ObjectCM::_sendMapReply( const MasterCMCommand& command,
194 : const uint128_t& version, const bool result,
195 : const bool useCache, const bool multicast )
196 : {
197 : command.getNode()->send( CMD_NODE_MAP_OBJECT_REPLY, multicast )
198 82 : << command.getNode()->getNodeID() << command.getObjectID()
199 123 : << version << command.getRequestID() << result
200 123 : << command.useCache() << useCache;
201 41 : }
202 :
203 4 : void ObjectCM::_sendEmptyVersion( const MasterCMCommand& command,
204 : const uint128_t& version,
205 : const bool multicast )
206 : {
207 4 : NodePtr node = command.getNode();
208 8 : ConnectionPtr connection = node->getConnection( multicast );
209 :
210 : ObjectDataOCommand( Connections( 1, connection ), CMD_OBJECT_INSTANCE,
211 4 : COMMANDTYPE_OBJECT, _object->getID(),
212 : command.getInstanceID(), version, 0, 0, 0, true, 0 )
213 12 : << NodeID() << _object->getInstanceID();
214 4 : }
215 :
216 63 : }
|
