vtkSocketCommunicator.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
// SPDX-License-Identifier: BSD-3-Clause
#ifndef vtkSocketCommunicator_h
#define vtkSocketCommunicator_h
 
#include "vtkCommunicator.h"
#include "vtkEndian.h"             // for VTK_WORDS_BIGENDIAN
#include "vtkParallelCoreModule.h" // For export macro
 
#include "vtkByteSwap.h" // Needed for vtkSwap macros
 
#ifdef VTK_WORDS_BIGENDIAN
#define vtkSwap4 vtkByteSwap::Swap4LE
#define vtkSwap4Range vtkByteSwap::Swap4LERange
#define vtkSwap8 vtkByteSwap::Swap8LE
#define vtkSwap8Range vtkByteSwap::Swap8LERange
#else
#define vtkSwap4 vtkByteSwap::Swap4BE
#define vtkSwap4Range vtkByteSwap::Swap4BERange
#define vtkSwap8 vtkByteSwap::Swap8BE
#define vtkSwap8Range vtkByteSwap::Swap8BERange
#endif
 
VTK_ABI_NAMESPACE_BEGIN
class vtkClientSocket;
class vtkServerSocket;
 
class VTKPARALLELCORE_EXPORT vtkSocketCommunicator : public vtkCommunicator
{
public:
  static vtkSocketCommunicator* New();
  vtkTypeMacro(vtkSocketCommunicator, vtkCommunicator);
  void PrintSelf(ostream& os, vtkIndent indent) override;
 
 
  virtual int WaitForConnection(int port);
  virtual int WaitForConnection(vtkServerSocket* socket, unsigned long msec = 0);
 
  virtual void CloseConnection();
 
  virtual int ConnectTo(const char* hostName, int port);
 
 
  vtkGetMacro(SwapBytesInReceivedData, int);
 
  int GetIsConnected();
 
  void SetNumberOfProcesses(int num) override;
 
  //------------------ Communication --------------------
 
 
  int SendVoidArray(
    const void* data, vtkIdType length, int type, int remoteHandle, int tag) override;
  int ReceiveVoidArray(void* data, vtkIdType length, int type, int remoteHandle, int tag) override;
 
  void Barrier() override;
 
 
  int BroadcastVoidArray(void* data, vtkIdType length, int type, int srcProcessId) override;
  int GatherVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType length, int type,
    int destProcessId) override;
  int GatherVVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType sendLength,
    vtkIdType* recvLengths, vtkIdType* offsets, int type, int destProcessId) override;
  int ScatterVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType length, int type,
    int srcProcessId) override;
  int ScatterVVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType* sendLengths,
    vtkIdType* offsets, vtkIdType recvLength, int type, int srcProcessId) override;
  int AllGatherVoidArray(
    const void* sendBuffer, void* recvBuffer, vtkIdType length, int type) override;
  int AllGatherVVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType sendLength,
    vtkIdType* recvLengths, vtkIdType* offsets, int type) override;
  int ReduceVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType length, int type,
    int operation, int destProcessId) override;
  int ReduceVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType length, int type,
    Operation* operation, int destProcessId) override;
  int AllReduceVoidArray(
    const void* sendBuffer, void* recvBuffer, vtkIdType length, int type, int operation) override;
  int AllReduceVoidArray(const void* sendBuffer, void* recvBuffer, vtkIdType length, int type,
    Operation* operation) override;
 
 
  vtkSetClampMacro(PerformHandshake, vtkTypeBool, 0, 1);
  vtkBooleanMacro(PerformHandshake, vtkTypeBool);
  vtkGetMacro(PerformHandshake, vtkTypeBool);
 
 
  virtual void SetLogStream(ostream* stream);
  virtual ostream* GetLogStream();
 
 
  virtual int LogToFile(const char* name);
  virtual int LogToFile(const char* name, int append);
 
 
  vtkSetMacro(ReportErrors, int);
  vtkGetMacro(ReportErrors, int);
 
 
  vtkGetObjectMacro(Socket, vtkClientSocket);
  void SetSocket(vtkClientSocket*);
 
  int Handshake();
 
  int ServerSideHandshake();
 
  int ClientSideHandshake();
 
 
  vtkGetMacro(IsServer, int);
 
  static int GetVersion();
 
  void BufferCurrentMessage() { this->BufferMessage = true; }
 
  bool HasBufferredMessages();
 
protected:
  vtkClientSocket* Socket;
  int SwapBytesInReceivedData;
  int RemoteHas64BitIds;
  vtkTypeBool PerformHandshake;
  int IsServer;
 
  int ReportErrors;
 
  ostream* LogFile;
  ostream* LogStream;
 
  vtkSocketCommunicator();
  ~vtkSocketCommunicator() override;
 
  // Wrappers around send/recv calls to implement loops.  Return 1 for
  // success, and 0 for failure.
  int SendTagged(const void* data, int wordSize, int numWords, int tag, const char* logName);
  int ReceiveTagged(void* data, int wordSize, int numWords, int tag, const char* logName);
  int ReceivePartialTagged(void* data, int wordSize, int numWords, int tag, const char* logName);
 
  int ReceivedTaggedFromBuffer(
    void* data, int wordSize, int numWords, int tag, const char* logName);
 
  void FixByteOrder(void* data, int wordSize, int numWords);
 
  // Internal utility methods.
  void LogTagged(
    const char* name, const void* data, int wordSize, int numWords, int tag, const char* logName);
  int CheckForErrorInternal(int id);
  bool BufferMessage;
 
private:
  vtkSocketCommunicator(const vtkSocketCommunicator&) = delete;
  void operator=(const vtkSocketCommunicator&) = delete;
 
  int SelectSocket(int socket, unsigned long msec);
 
  // SwapBytesInReceiveData needs an invalid / not set.
  // This avoids checking length of endian handshake.
  enum ErrorIds
  {
    SwapOff = 0,
    SwapOn,
    SwapNotSet
  };
 
  // One may be tempted to change this to a vtkIdType, but really an int is
  // enough since we split messages > VTK_INT_MAX.
  int TagMessageLength;
 
  //  Buffer to save messages received with different tag than requested.
  class vtkMessageBuffer;
  vtkMessageBuffer* ReceivedMessageBuffer;
};
 
VTK_ABI_NAMESPACE_END
#endif