dox/Parallel/vtkMultiProcessController.h

Go to the documentation of this file.

/*=========================================================================

  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.

=========================================================================*/
#ifndef __vtkMultiProcessController_h
#define __vtkMultiProcessController_h

#include "vtkObject.h"

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

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

//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);

  virtual void Initialize(int* vtkNotUsed(argc), char*** vtkNotUsed(argv))=0;


  virtual void Initialize(int* vtkNotUsed(argc), char*** vtkNotUsed(argv),
                          int initializedExternally)=0;

  virtual void Finalize()=0;

  virtual void Finalize(int finalizedExternally)=0;


  void SetNumberOfProcesses(int num);
  int GetNumberOfProcesses();

  //BTX

  void SetSingleMethod(vtkProcessFunctionType, void *data);
  //ETX

  virtual void SingleMethodExecute() = 0;
  
  //BTX

  void SetMultipleMethod(int index, vtkProcessFunctionType, void *data); 
  //ETX

  virtual void MultipleMethodExecute() = 0;

  int GetLocalProcessId();

  static vtkMultiProcessController *GetGlobalController();

  virtual void CreateOutputWindow() = 0;


  virtual vtkMultiProcessController *CreateSubController(
                                                        vtkProcessGroup *group);
  
  //------------------ RMIs --------------------
  //BTX
  unsigned long AddRMI(vtkRMIFunctionType, void *localArg, int tag);
  
  int RemoveFirstRMI(int tag);

  int RemoveRMI(unsigned long id);


  void RemoveRMI(vtkRMIFunctionType f, void *arg, int tag)
    {f = f; arg = arg; tag = tag; vtkErrorMacro("RemoveRMI Not Implemented Yet");};
  //ETX
  
  void TriggerRMI(int remoteProcessId, void *arg, int argLength, int tag);

  void TriggerBreakRMIs();


  void TriggerRMI(int remoteProcessId, const char *arg, int tag) 
    { this->TriggerRMI(remoteProcessId, (void*)arg, 
                       static_cast<int>(strlen(arg))+1, tag); }


  void TriggerRMI(int remoteProcessId, int tag)
    { this->TriggerRMI(remoteProcessId, NULL, 0, tag); }


  void TriggerRMIOnAllChildren(void *arg, int argLength, int tag);
  void TriggerRMIOnAllChildren(const char *arg, int tag) 
    { 
    this->TriggerRMIOnAllChildren(
      (void*)arg, static_cast<int>(strlen(arg))+1, tag);
    }
  void TriggerRMIOnAllChildren(int tag)
    { 
    this->TriggerRMIOnAllChildren(NULL, 0, tag);
    }


  int ProcessRMIs(int reportErrors, int dont_loop = 0);
  int ProcessRMIs();
 

  vtkSetMacro(BreakFlag, int);
  vtkGetMacro(BreakFlag, int);


  vtkGetObjectMacro(Communicator, vtkCommunicator);


  static int GetBreakRMITag() { return BREAK_RMI_TAG; }
  static int GetRMITag() { return RMI_TAG; }
  static int GetRMIArgTag() { return RMI_ARG_TAG; }  
  
//BTX

  enum Errors 
  {
    RMI_NO_ERROR,
    RMI_TAG_ERROR,
    RMI_ARG_ERROR
  };

  enum Consts 
  {
    MAX_PROCESSES  = 8192,
    ANY_SOURCE     = -1,
    INVALID_SOURCE = -2
  };

  enum Tags 
  {
    RMI_TAG        = 1,
    RMI_ARG_TAG    = 2,
    BREAK_RMI_TAG  = 3,
    XML_WRITER_DATA_INFO = 4
  };

//ETX

  void Barrier();

  static void SetGlobalController(vtkMultiProcessController *controller);

  //------------------ Communication --------------------
  

  int Send(const int* data, vtkIdType length, int remoteProcessId, int tag);
  int Send(const unsigned int* data, vtkIdType length, int remoteProcessId, int tag);
  int Send(const unsigned long* data, vtkIdType length, int remoteProcessId, 
           int tag);
  int Send(const char* data, vtkIdType length, int remoteProcessId, int tag);
  int Send(const unsigned char* data, vtkIdType length, int remoteProcessId, int tag);
  int Send(const float* data, vtkIdType length, int remoteProcessId, int tag);
  int Send(const double* data, vtkIdType length, int remoteProcessId, int tag);
#ifdef VTK_USE_64BIT_IDS
  int Send(const vtkIdType* data, vtkIdType length, int remoteProcessId, int tag);
#endif
  int Send(vtkDataObject *data, int remoteId, int tag);
  int Send(vtkDataArray *data, int remoteId, int tag);

//BTX
  int Send(const vtkMultiProcessStream& stream, int remoteId, int tag);
//ETX


  int Receive(int* data, vtkIdType maxlength, int remoteProcessId, int tag);
  int Receive(unsigned int* data, vtkIdType maxlength, int remoteProcessId, int tag);
  int Receive(unsigned long* data, vtkIdType maxlength, int remoteProcessId, 
              int tag);
  int Receive(char* data, vtkIdType maxlength, int remoteProcessId, int tag);
  int Receive(unsigned char* data, vtkIdType maxlength, int remoteProcessId, int tag);
  int Receive(float* data, vtkIdType maxlength, int remoteProcessId, int tag);
  int Receive(double* data, vtkIdType maxlength, int remoteProcessId, int tag);
#ifdef VTK_USE_64BIT_IDS
  int Receive(vtkIdType* data, vtkIdType maxlength, int remoteProcessId, int tag);
#endif
  int Receive(vtkDataObject* data, int remoteId, int tag);
  int Receive(vtkDataArray* data, int remoteId, int tag);
//BTX
  int Receive(vtkMultiProcessStream& stream, int remoteId, int tag);
//ETX
  vtkDataObject *ReceiveDataObject(int remoteId, int tag);

  vtkIdType GetCount();


  //---------------------- Collective Operations ----------------------


  int Broadcast(int *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
  int Broadcast(unsigned long *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
  int Broadcast(unsigned char *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
  int Broadcast(char *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
  int Broadcast(float *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
  int Broadcast(double *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int Broadcast(vtkIdType *data, vtkIdType length, int srcProcessId) {
    return this->Communicator->Broadcast(data, length, srcProcessId);
  }
#endif
  int Broadcast(vtkDataObject *data, int srcProcessId) {
    return this->Communicator->Broadcast(data, srcProcessId);
  }
  int Broadcast(vtkDataArray *data, int srcProcessId) {
    return this->Communicator->Broadcast(data, srcProcessId);
  }
//BTX
  int Broadcast(vtkMultiProcessStream& stream, int srcProcessId) {
    return this->Communicator->Broadcast(stream, srcProcessId);
  }
//ETX


  int Gather(const int *sendBuffer, int *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
  int Gather(const unsigned long *sendBuffer, unsigned long *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
  int Gather(const unsigned char *sendBuffer, unsigned char *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
  int Gather(const char *sendBuffer, char *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
  int Gather(const float *sendBuffer, float *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
  int Gather(const double *sendBuffer, double *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int Gather(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
             vtkIdType length, int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, length,
                                      destProcessId);
  }
#endif
  int Gather(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
             int destProcessId) {
    return this->Communicator->Gather(sendBuffer, recvBuffer, destProcessId);
  }


  int GatherV(const int* sendBuffer, int* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
  int GatherV(const unsigned long* sendBuffer, unsigned long* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
  int GatherV(const unsigned char* sendBuffer, unsigned char* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
  int GatherV(const char* sendBuffer, char* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
  int GatherV(const float* sendBuffer, float* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
  int GatherV(const double* sendBuffer, double* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int GatherV(const vtkIdType* sendBuffer, vtkIdType* recvBuffer, 
              vtkIdType sendLength, vtkIdType* recvLengths, vtkIdType* offsets,
              int destProcessId) {
    return this->Communicator->GatherV(sendBuffer, recvBuffer,
                                       sendLength, recvLengths,
                                       offsets, destProcessId);
  }
#endif


  int Scatter(const int *sendBuffer, int *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
  int Scatter(const unsigned long *sendBuffer, unsigned long *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
  int Scatter(const unsigned char *sendBuffer, unsigned char *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
  int Scatter(const char *sendBuffer, char *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
  int Scatter(const float *sendBuffer, float *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
  int Scatter(const double *sendBuffer, double *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int Scatter(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
              vtkIdType length, int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, length,
                                       srcProcessId);
  }
#endif
  int Scatter(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
              int srcProcessId) {
    return this->Communicator->Scatter(sendBuffer, recvBuffer, srcProcessId);
  }


  int ScatterV(const int *sendBuffer, int *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
  int ScatterV(const unsigned long *sendBuffer, unsigned long *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
  int ScatterV(const unsigned char *sendBuffer, unsigned char *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
  int ScatterV(const char *sendBuffer, char *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
  int ScatterV(const float *sendBuffer, float *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
  int ScatterV(const double *sendBuffer, double *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int ScatterV(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
               vtkIdType *sendLengths, vtkIdType *offsets,
               vtkIdType recvLength, int srcProcessId) {
    return this->Communicator->ScatterV(sendBuffer, recvBuffer,
                                        sendLengths, offsets, recvLength,
                                        srcProcessId);
  }
#endif


  int AllGather(const int *sendBuffer, int *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
  int AllGather(const unsigned long *sendBuffer,
                unsigned long *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
  int AllGather(const unsigned char *sendBuffer,
                unsigned char *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
  int AllGather(const char *sendBuffer, char *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
  int AllGather(const float *sendBuffer, float *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
  int AllGather(const double *sendBuffer,
                double *recvBuffer, vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
#ifdef VTK_USE_64BIT_IDS
  int AllGather(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
                vtkIdType length) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer, length);
  }
#endif
  int AllGather(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer) {
    return this->Communicator->AllGather(sendBuffer, recvBuffer);
  }


  int AllGatherV(const int* sendBuffer, int* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
  int AllGatherV(const unsigned long* sendBuffer, unsigned long* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
  int AllGatherV(const unsigned char* sendBuffer, unsigned char* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
  int AllGatherV(const char* sendBuffer, char* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
  int AllGatherV(const float* sendBuffer, float* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
  int AllGatherV(const double* sendBuffer, double* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
#ifdef VTK_USE_64BIT_IDS
  int AllGatherV(const vtkIdType* sendBuffer, vtkIdType* recvBuffer, 
                 vtkIdType sendLength, vtkIdType* recvLengths,
                 vtkIdType* offsets) {
    return this->Communicator->AllGatherV(sendBuffer, recvBuffer,
                                          sendLength, recvLengths,
                                          offsets);
  }
#endif


  int Reduce(const int *sendBuffer, int *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const unsigned long *sendBuffer, unsigned long *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const unsigned char *sendBuffer, unsigned char *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const char *sendBuffer, char *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const float *sendBuffer, float *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const double *sendBuffer, double *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int Reduce(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
             vtkIdType length, int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
#endif
  int Reduce(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
             int operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer,
                                      operation, destProcessId);
  }

//BTX

  int Reduce(const int *sendBuffer, int *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const unsigned long *sendBuffer, unsigned long *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const unsigned char *sendBuffer, unsigned char *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const char *sendBuffer, char *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const float *sendBuffer, float *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
  int Reduce(const double *sendBuffer, double *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
#ifdef VTK_USE_64BIT_IDS
  int Reduce(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
             vtkIdType length, vtkCommunicator::Operation *operation,
             int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer, length,
                                      operation, destProcessId);
  }
#endif
  int Reduce(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
             vtkCommunicator::Operation *operation, int destProcessId) {
    return this->Communicator->Reduce(sendBuffer, recvBuffer,
                                      operation, destProcessId);
  }
//ETX


  int AllReduce(const int *sendBuffer, int *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const unsigned long *sendBuffer, unsigned long *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const unsigned char *sendBuffer, unsigned char *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const char *sendBuffer, char *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const float *sendBuffer, float *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const double *sendBuffer, double *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
#ifdef VTK_USE_64BIT_IDS
  int AllReduce(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
                vtkIdType length, int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
#endif
  int AllReduce(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
                int operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, operation);
  }
//BTX
  int AllReduce(const int *sendBuffer, int *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const unsigned long *sendBuffer, unsigned long *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const unsigned char *sendBuffer, unsigned char *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const char *sendBuffer, char *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const float *sendBuffer, float *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
  int AllReduce(const double *sendBuffer, double *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
#ifdef VTK_USE_64BIT_IDS
  int AllReduce(const vtkIdType *sendBuffer, vtkIdType *recvBuffer,
                vtkIdType length, vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, length,
                                         operation);
  }
#endif
  int AllReduce(vtkDataArray *sendBuffer, vtkDataArray *recvBuffer,
                vtkCommunicator::Operation *operation) {
    return this->Communicator->AllReduce(sendBuffer, recvBuffer, operation);
  }
//ETX

// Internally implemented RMI to break the process loop.

protected:
  vtkMultiProcessController();
  ~vtkMultiProcessController();


  virtual void TriggerRMIInternal(int remoteProcessId, 
    void* arg, int argLength, int rmiTag, bool propagate);
  
  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.

  unsigned long RMICount;
};


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(const int* data, vtkIdType length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}

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

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

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

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

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

inline int vtkMultiProcessController::Send(const double* data, vtkIdType 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(const vtkIdType* data,
                                           vtkIdType length, 
                                           int remoteProcessId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(data, length, remoteProcessId, tag);
    }
  else
    {
    return 0;
    }
}
#endif

inline int vtkMultiProcessController::Send(const vtkMultiProcessStream& stream,
  int remoteId, int tag)
{
  if (this->Communicator)
    {
    return this->Communicator->Send(stream, remoteId, tag);
    }
  return 0;
}

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

inline vtkDataObject* vtkMultiProcessController::ReceiveDataObject(