NetP.h

Go to the documentation of this file.

#ifndef NETP_H
#define NETP_H

#include <string>
#include <cstring>
#include <cmath>
#include <sstream>
#include <map>
#include <set>
#include <pvm3.h>
#include <vector>
#include <cstdlib>
#include <unistd.h>
#include <deque>
#include <sys/timeb.h>
#include <sys/time.h>
#include <iostream>
#include <pthread.h>

#include <fcntl.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <errno.h>
#include <netinet/in.h>

#include "PData.h"

using namespace std;

class NetP {

public:


  
  NetP(string packettype);

  NetP();

  ~NetP();

  void cleanUp();

  int joinChannel(string channel);

  int leaveChannel(string channel);




  int send();

  string bReceive();

  string nReceive();

  string tReceive(double timeout_sec);

  string bReceiveNewest();

  string nReceiveNewest();

  string tReceiveNewest(double timeout_sec);

  int peek();


  
  PData& data();
  
  string getType();

  double getTimeStamp();

  static double currentTime();

  int getSize();

  int getTid();

  string getHostname();

  int getProcessID();

  string toString();

  friend ostream& operator<<(ostream& o, NetP& netp) {
    o << netp.toString();
    return o;
  }


  
  void setBufferSize(long bytes);

  static void setMaxPeriod(double seconds);

  static void setMinPeriod(double seconds);

  static void setUnthreaded();

  static void LockPvmMutex();
  
  static void UnlockPvmMutex();

  static void Exit();




  static void DirectOffer(string channel);

  static int DirectConnect(string channel, unsigned short int port = 51552);

  static void DirectClose(string channel);

  static void DirectClose(int socket);

  static void DirectCloseAll();

  static int DirectWrite(int socket, const char* buf, int len=-1);

  static int DirectWrite(int socket, string buf);

  static int DirectWrite(string channel, string buf);

  static int DirectRead(int socket, char* buf, int len);

  static int DirectConnectToServer(string hostname, string direct_id, 
                                   unsigned short int port=51552); // for direct connect

  static string DirectRead(int socket);

  static string DirectRead(string channel);

  static vector<string> ListOpenConnects();

  static vector<string> ListPendingConnects();

  static int GetSocketNum(string channel);
  
private:
  // static class functions
  static void Connect(); // establish PVM connection
  static void Spawn(); // create global processing thread
  static void* NetLoop(void* arg); // global PVM processing loop
  static void ReceiveMapAdd(string type, NetP* ptr); // add to receive list
  static void ReceiveMapDelete(string type); // remove from receive list
  static NetP* ReceiveMapLookup(string type); // mutex'd lookup in receive map
  static int ReceiveOnce(); // receive one pvm packet
  static int UnthreadedReceiveOnce(); // for unthreaded mode

  // instance functions
  void pvmInit(); // connect to pvm if necessary
  void threadInit(); // make thread if necessary
  void initPData(); // initialize a new pdata set, create pdata_mutex
  void clearPData(); // clear the dataRoot
  void destroyPData(); // destroy the dataRoot when we're about to destroy the NetPacket
  void receiveBufferToPData(); // mutex helper to move receive buffer to data
  int hasQueuedReceive(); // whether anything's waiting in receive queue
  void insertReceived(PData* pmeta, PData* pdata); // insert one element into receive queue
  void eatRecvBuffer(int remaining = 0); // eat the receive buffer until only this many remain

  PData& getMetaField(string index); // access a subnode of the Root structure
  void updateTimeStamp(); // update the timestamp on the send packet

  // static variables
  
  // metadata setup
  static const char* MetaDataSetup; // string that initializes each metadata pdata
  static int ProcessID; // the local PID of the running process
  static string HostName; // keep track of the host running the process

  // thread-related variables
  static int IsThreadCalled; // have we called the function to start the thread?
  static int IsThreadStarted; // was the thread started successfully?
  static int IsThreadRunning; // is the thread still running?
  static int RUN_THREAD; // set this to 0 if we want the thread to stop and exit
  static int NO_THREAD_MODE; // disable receive thread; must set before first constructor call
  static pthread_t ThreadID; // thread id

  // pvm-related variables
  static int MyTid; // pvm task id
  static pthread_mutex_t PvmOpMutex; // all pvm functions must be locked

  // timing variables
  static double MinReceivePeriod; // receive thread won't receive faster than this
  static double MaxReceivePeriod; // receive thread won't receive slower than this
  static double ReceivePeriod; // the current period being followed by NetPacket
  static long RECEIVE_TIMEOUT_US; // the timeout on the pvm call to trecv
  static timeval TimeSyncDiff; // difference between timesync clock and local clock
  static int TimeSyncDTID; // the host id of the timesync host

  // receive variables
  static map<string, NetP*> ReceiveMap; // the map from channels to receive np's
  static pthread_mutex_t ReceiveMapMutex; // lock the receive map so we don't corrupt it
  static char* ReceiveBuffer; // the buffer we use to receive stuff
  static int ReceiveBufferSize; // the current size of that receive buffer
  static pthread_mutex_t ReceiveBlockMutex; // condition variable mutex for blocked receives

  // direct connect variables
  static map<string, int> DirectSockets; // channel to socket map for direct connects
  static set<string> DirectChannels; // channels we're listening on with no connection yet

  // instance variables

  int cleaned_up; // mark that we've already cleaned up.
  
  int is_send_type; // 1 means send type, 0 means receive type
  string send_group; // the group we send to
  char* send_buffer; // our send buffer

  set<string> groups; // the groups this receive netp belongs to
  long recv_buffer_limit; // -1 means no limit
  long recv_buffer_size; // how much stuff in receive buffer
  pthread_mutex_t recv_buffer_mutex; // lock the receive buffer
  pthread_cond_t recv_block_condition; // for immediate blocking response
  deque<PData*> recv_buffer; // alternately holds a meta and then a data

  PData* _meta; // meta data
  PData* _data; // main data
  pthread_mutex_t pdata_mutex; // lock the data so they can't be left inconsistent
};


#endif