File: vtkMultiProcessController.h

package info (click to toggle)
vtk 5.0.2-4
  • links: PTS
  • area: main
  • in suites: etch, etch-m68k
  • size: 51,080 kB
  • ctags: 67,442
  • sloc: cpp: 522,627; ansic: 221,292; tcl: 43,377; python: 14,072; perl: 3,102; java: 1,436; yacc: 1,033; sh: 469; lex: 248; makefile: 181; asm: 154
file content (560 lines) | stat: -rw-r--r-- 17,196 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
/*=========================================================================

  Program:   Visualization Toolkit
  Module:    $RCSfile: vtkMultiProcessController.h,v $

  Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
  All rights reserved.
  See Copyright.txt or http://www.kitware.com/Copyright.htm for details.

     This software is distributed WITHOUT ANY WARRANTY; without even
     the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
     PURPOSE.  See the above copyright notice for more information.

=========================================================================*/
// .NAME vtkMultiProcessController - Multiprocessing communication superclass
// .SECTION Description
// vtkMultiProcessController is used to control multiple processes
// in a distributed computing environment. It has
// methods for executing single/multiple method(s) on multiple processors,
// triggering registered callbacks (Remote Methods) (AddRMI(), TriggerRMI())
// and communication. Please note that the communication is done using
// the communicator which is accessible to the user. Therefore it is
// possible to get the communicator with GetCommunicator() and use
// it to send and receive data. This is the encoured communication method.
// The internal (RMI) communications are done using a second internal
// communicator (called RMICommunicator).
//
// .SECTION see also
// vtkMPIController
// vtkCommunicator vtkMPICommunicator

#ifndef __vtkMultiProcessController_h
#define __vtkMultiProcessController_h

#include "vtkObject.h"

#include "vtkCommunicator.h" // Needed for direct access to communicator

class vtkDataSet;
class vtkImageData;
class vtkCollection;
class vtkOutputWindow;
class vtkDataObject;
class vtkMultiProcessController;

//BTX
// The type of function that gets called when new processes are initiated.
typedef void (*vtkProcessFunctionType)(vtkMultiProcessController *controller, 
                                       void *userData);

// The type of function that gets called when an RMI is triggered.
typedef void (*vtkRMIFunctionType)(void *localArg, 
                                   void *remoteArg, int remoteArgLength, 
                                   int remoteProcessId);
//ETX


class VTK_PARALLEL_EXPORT vtkMultiProcessController : public vtkObject
{
public:
  vtkTypeRevisionMacro(vtkMultiProcessController,vtkObject);
  void PrintSelf(ostream& os, vtkIndent indent);

  // Description:
  // This method is for setting up the processes.
  // If a subclass needs to initialize process communication (i.e. MPI)
  // it would over ride this method.
  virtual void Initialize(int* vtkNotUsed(argc), char*** vtkNotUsed(argv))=0;

  // Description:
  // This method is for setting up the processes.
  // If a subclass needs to initialize process communication (i.e. MPI)
  // it would over ride this method.  Provided for initialization outside vtk.
  virtual void Initialize(int* vtkNotUsed(argc), char*** vtkNotUsed(argv),
                          int initializedExternally)=0;

  // Description:
  // This method is for cleaning up.
  // If a subclass needs to clean up process communication (i.e. MPI)
  // it would over ride this method.
  virtual void Finalize()=0;

  // Description:
  // This method is for cleaning up.
  // If a subclass needs to clean up process communication (i.e. MPI)
  // it would over ride this method.  Provided for finalization outside vtk.
  virtual void Finalize(int finalizedExternally)=0;

  // Description:
  // Set the number of processes you will be using.  This defaults
  // to the maximum number available.  If you set this to a value
  // higher than the default, you will get an error.
  virtual void SetNumberOfProcesses(int num);
  vtkGetMacro( NumberOfProcesses, int );

  //BTX
  // Description:
  // Set the SingleMethod to f() and the UserData of the
  // for the method to be executed by all of the processes
  // when SingleMethodExecute is called.  All the processes will
  // start by calling this function.
  void SetSingleMethod(vtkProcessFunctionType, void *data);
  //ETX

  // Description:
  // Execute the SingleMethod (as define by SetSingleMethod) using
  // this->NumberOfProcesses processes.  This will only return when
  // all the processes finish executing their methods.
  virtual void SingleMethodExecute() = 0;
  
  //BTX
  // Description:
  // Set the MultipleMethod to f() and the UserData of the
  // for the method to be executed by the process index
  // when MultipleMethodExecute is called.  This is for having each 
  // process start with a different function and data argument.
  void SetMultipleMethod(int index, vtkProcessFunctionType, void *data); 
  //ETX

  // Description:
  // Execute the MultipleMethods (as define by calling SetMultipleMethod
  // for each of the required this->NumberOfProcesses methods) using
  // this->NumberOfProcesses processes.
  virtual void MultipleMethodExecute() = 0;

  // Description:
  // Tells you which process [0, NumProcess) you are in.
  vtkGetMacro(LocalProcessId, int);

  // Description:
  // This convenience method returns the controller associated with the 
  // local process.  It returns NULL until the processes are spawned.
  // It is better if you hang on to the controller passed as an argument to the
  // SingleMethod or MultipleMethod functions.
  static vtkMultiProcessController *GetGlobalController();

  // Description:
  // This method can be used to tell the controller to create
  // a special output window in which all messages are preceded
  // by the process id.
  virtual void CreateOutputWindow() = 0;
  
  //------------------ RMIs --------------------
  //BTX
  // Description:
  // Register remote method invocation in the receiving process
  // which makes the call.  It must have a unique tag as an RMI id.
  // The vtkRMIFunctionType has several arguments: localArg (same as passed in),
  // remoteArg, remoteArgLength (memory passed by process triggering the RMI),
  // remoteProcessId.
  void AddRMI(vtkRMIFunctionType, void *localArg, int tag);
  
  // Description:
  // Remove the first RMI matching the tag.
  int RemoveFirstRMI(int tag);

  // Description:
  // Take an RMI away.
  void RemoveRMI(vtkRMIFunctionType f, void *arg, int tag)
    {f = f; arg = arg; tag = tag; vtkErrorMacro("RemoveRMI Not Implemented Yet");};
  //ETX
  
  // Description:
  // A method to trigger a method invocation in another process.
  void TriggerRMI(int remoteProcessId, void *arg, int argLength, int tag);

  // Description:
  // A conveniance method.  Called on process 0 to break "ProcessRMIs" loop
  // on all other processes.
  void TriggerBreakRMIs();

  // Description:
  // Convenience method when the arg is a string. 
  void TriggerRMI(int remoteProcessId, const char *arg, int tag) 
    { this->TriggerRMI(remoteProcessId, (void*)arg, 
                       static_cast<int>(strlen(arg))+1, tag); }

  // Description:
  // Convenience method when there is no argument.
  void TriggerRMI(int remoteProcessId, int tag)
    { this->TriggerRMI(remoteProcessId, NULL, 0, tag); }

  // Description:
  // Calling this method gives control to the controller to start
  // processing RMIs. Possible return values are:
  // RMI_NO_ERROR,
  // RMI_TAG_ERROR : rmi tag could not be received,
  // RMI_ARG_ERROR : rmi arg could not be received.
  // If reportErrors is false, no vtkErrorMacro is called.
  // ProcessRMIs() calls ProcessRMIs(int) with reportErrors = 0.
  int ProcessRMIs(int reportErrors);
  int ProcessRMIs();
 
  // Description:
  // Setting this flag to 1 will cause the ProcessRMIs loop to return.
  // This also causes vtkUpStreamPorts to return from
  // their WaitForUpdate loops.
  vtkSetMacro(BreakFlag, int);
  vtkGetMacro(BreakFlag, int);

  // Description:
  vtkGetObjectMacro(Communicator, vtkCommunicator);
  
//BTX

  enum Errors 
  {
    RMI_NO_ERROR,
    RMI_TAG_ERROR,
    RMI_ARG_ERROR
  };

  enum Consts 
  {
    MAX_PROCESSES  = 8192,
    ANY_SOURCE     = -1,
    INVALID_SOURCE = -2,
    RMI_TAG        = 315167,
    RMI_ARG_TAG    = 315168,
    BREAK_RMI_TAG  = 239954
  };

//ETX

  // Description:
  // This method can be used to synchronize processes.
  virtual void Barrier() = 0;

  static void SetGlobalController(vtkMultiProcessController *controller);

  //------------------ Communication --------------------
  
  // Description:
  // This method sends data to another process.  Tag eliminates ambiguity
  // when multiple sends or receives exist in the same process.
  int Send(int* data, int length, int remoteProcessId, int tag);
  int Send(unsigned long* data, int length, int remoteProcessId, 
           int tag);
  int Send(char* data, int length, int remoteProcessId, int tag);
  int Send(unsigned char* data, int length, int remoteProcessId, int tag);
  int Send(float* data, int length, int remoteProcessId, int tag);
  int Send(double* data, int length, int remoteProcessId, int tag);
#ifdef VTK_USE_64BIT_IDS
  int Send(vtkIdType* data, int length, int remoteProcessId, int tag);
#endif
  int Send(vtkDataObject *data, int remoteId, int tag);
  int Send(vtkDataArray *data, int remoteId, int tag);

  // Description:
  // This method receives data from a corresponding send. It blocks
  // until the receive is finished.  It calls methods in "data"
  // to communicate the sending data.
  int Receive(int* data, int length, int remoteProcessId, int tag);
  int Receive(unsigned long* data, int length, int remoteProcessId, 
              int tag);
  int Receive(char* data, int length, int remoteProcessId, int tag);
  int Receive(unsigned char* data, int length, int remoteProcessId, int tag);
  int Receive(float* data, int length, int remoteProcessId, int tag);
  int Receive(double* data, int length, int remoteProcessId, int tag);
#ifdef VTK_USE_64BIT_IDS
  int Receive(vtkIdType* data, int length, int remoteProcessId, int tag);
#endif
  int Receive(vtkDataObject* data, int remoteId, int tag);
  int Receive(vtkDataArray* data, int remoteId, int tag);

// Internally implemented RMI to break the process loop.

protected:
  vtkMultiProcessController();
  ~vtkMultiProcessController();
  
  int MaximumNumberOfProcesses;
  int NumberOfProcesses;

  int LocalProcessId;
  
  vtkProcessFunctionType      SingleMethod;
  void                       *SingleData;
  vtkProcessFunctionType      MultipleMethod[MAX_PROCESSES];
  void                       *MultipleData[MAX_PROCESSES];  
  
  vtkCollection *RMIs;
  
  // This is a flag that can be used by the ports to break
  // their update loop. (same as ProcessRMIs)
  int BreakFlag;

  void ProcessRMI(int remoteProcessId, void *arg, int argLength, int rmiTag);

  // This method implements "GetGlobalController".  
  // It needs to be virtual and static.
  virtual vtkMultiProcessController *GetLocalController();

  
  // This flag can force deep copies during send.
  int ForceDeepCopy;

  vtkOutputWindow* OutputWindow;

  // Note that since the communicators can be created differently
  // depending on the type of controller, the subclasses are
  // responsible of deleting them.
  vtkCommunicator* Communicator;

  // Communicator which is a copy of the current user
  // level communicator except the context; i.e. even if the tags 
  // are the same, the RMI messages will not interfere with user 
  // level messages. 
  // Note that since the communicators can be created differently
  // depending on the type of controller, the subclasses are
  // responsible of deleting them.
  vtkCommunicator* RMICommunicator;

private:
  vtkMultiProcessController(const vtkMultiProcessController&);  // Not implemented.
  void operator=(const vtkMultiProcessController&);  // Not implemented.
};


inline int vtkMultiProcessController::Send(vtkDataObject *data, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(vtkDataArray *data, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(int* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(unsigned long* data, 
                                           int length, int remoteProcessId, 
                                           int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(char* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(unsigned char* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(float* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Send(double* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

#ifdef VTK_USE_64BIT_IDS
inline int vtkMultiProcessController::Send(vtkIdType* data, int length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}
#endif

inline int vtkMultiProcessController::Receive(vtkDataObject* data, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(vtkDataArray* data, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(int* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(unsigned long* data, 
                                              int length,int remoteProcessId, 
                                              int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(char* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(unsigned char* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(float* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

inline int vtkMultiProcessController::Receive(double* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

#ifdef VTK_USE_64BIT_IDS
inline int vtkMultiProcessController::Receive(vtkIdType* data, int length, 
                                              int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Receive(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}
#endif

#endif