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

  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;


  virtual void SetNumberOfProcesses(int num);
  vtkGetMacro( NumberOfProcesses, int );

  //BTX

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

  virtual void SingleMethodExecute() = 0;
  
  //BTX

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

  virtual void MultipleMethodExecute() = 0;


  vtkGetMacro(LocalProcessId, int);

  static vtkMultiProcessController *GetGlobalController();

  virtual void CreateOutputWindow() = 0;
  
  //------------------ RMIs --------------------
  //BTX
  void AddRMI(vtkRMIFunctionType, void *localArg, int tag);
  
  int RemoveFirstRMI(int tag);


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


  int ProcessRMIs(int reportErrors);
  int ProcessRMIs();
 

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

  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

  virtual void Barrier() = 0;

  static void SetGlobalController(vtkMultiProcessController *controller);

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

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


  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

---