SocketThread.h

Go to the documentation of this file.

// $Id: SocketThread.h 4469 2020-04-15 12:53:59Z linev $
 
/************************************************************
 * The Data Acquisition Backbone Core (DABC)                *
 ************************************************************
 * Copyright (C) 2009 -                                     *
 * GSI Helmholtzzentrum fuer Schwerionenforschung GmbH      *
 * Planckstr. 1, 64291 Darmstadt, Germany                   *
 * Contact:  http://dabc.gsi.de                             *
 ************************************************************
 * This software can be used under the GPL license          *
 * agreements as stated in LICENSE.txt file                 *
 * which is part of the distribution.                       *
 ************************************************************/
 
#ifndef DABC_SocketThread
#define DABC_SocketThread
 
#ifndef DABC_Thread
#include "dabc/Thread.h"
#endif
 
#ifndef DABC_Worker
#include "dabc/Worker.h"
#endif
 
#include <netdb.h>
 
struct pollfd;
 
// #define SOCKET_PROFILING
 
 
namespace dabc {
 
   class SocketThread;
 
   class SocketAddon : public WorkerAddon {
 
      friend class SocketThread;
 
      protected:
 
         int           fSocket;                 
         bool          fDoingInput;             
         bool          fDoingOutput;            
         int           fIOPriority;             
         bool          fDeliverEventsToWorker;  
         bool          fDeleteWorkerOnClose;    
 
         virtual void ProcessEvent(const EventId&);
 
         inline void SetDoingInput(bool on = true) { fDoingInput = on; }
 
         inline void SetDoingOutput(bool on = true) { fDoingOutput = on; }
 
         virtual void OnSocketError(int msg, const std::string &info);
 
         ssize_t DoRecvBuffer(void* buf, ssize_t len);
         ssize_t DoRecvBufferHdr(void* hdr, ssize_t hdrlen, void* buf, ssize_t len, void* srcaddr = 0, unsigned srcaddrlen = 0);
         ssize_t DoSendBuffer(void* buf, ssize_t len);
         ssize_t DoSendBufferHdr(void* hdr, ssize_t hdrlen, void* buf, ssize_t len, void* tgtaddr = 0, unsigned tgtaddrlen = 0);
 
         bool IsDeleteWorkerOnClose() const { return fDeleteWorkerOnClose; }
         void SetDeleteWorkerOnClose(bool on = true) { fDeleteWorkerOnClose = on; }
 
 
      public:
 
         enum ESocketEvents {
            evntSocketRead = Worker::evntFirstAddOn,
            evntSocketWrite,
            evntSocketError,
            evntSocketStartConnect,
            evntSocketLast,        
            evntSocketRecvInfo  = Worker::evntFirstSystem,   
            evntSocketSendInfo,                     
            evntSocketErrorInfo,                    
            evntSocketCloseInfo,                    
            evntSocketLastInfo                      
         };
 
         SocketAddon(int fd = -1);
         virtual ~SocketAddon();
 
         virtual const char* ClassName() const { return "SocketAddon"; }
         virtual std::string RequiredThrdClass() const { return typeSocketThread; }
 
         inline int Socket() const { return fSocket; }
         inline bool IsSocket() const { return Socket() >= 0; }
 
         inline bool IsDoingInput() const { return fDoingInput; }
         inline bool IsDoingOutput() const { return fDoingOutput; }
 
         void CloseSocket();
         void SetSocket(int fd);
 
         int TakeSocket();
         int TakeSocketError();
 
         bool IsDeliverEventsToWorker() const { return fDeliverEventsToWorker; }
         void SetDeliverEventsToWorker(bool on = true) { fDeliverEventsToWorker = on; }
 
         inline void SetIOPriority(int prior = 1) { fIOPriority = prior; }
   };
 
 
   class SocketAddonRef : public WorkerAddonRef {
      DABC_REFERENCE(SocketAddonRef, WorkerAddonRef, SocketAddon)
 
      bool IsSocket() const
         { return GetObject() ? GetObject()->IsSocket() : false; }
   };
 
 
   // ______________________________________________________________________
 
   class SocketIOAddon : public SocketAddon {
      protected:
         bool          fDatagramSocket; 
         bool          fUseMsgOper;     
 
         // sending data
         bool          fSendUseMsg;     
         struct iovec* fSendIOV;        
         unsigned      fSendIOVSize;    
         unsigned      fSendIOVFirst;   
         unsigned      fSendIOVNumber;  
         struct sockaddr_in fSendAddr;  
         bool          fSendUseAddr;    
 
         // receiving data
         bool          fRecvUseMsg;     
         struct iovec* fRecvIOV;        
         unsigned      fRecvIOVSize;    
         unsigned      fRecvIOVFirst;   
         unsigned      fRecvIOVNumber;  
         struct sockaddr_in fRecvAddr;  
         unsigned      fLastRecvSize;   
 
#ifdef SOCKET_PROFILING
         long           fSendOper;
         double         fSendTime;
         long           fSendSize;
         long           fRecvOper;
         double         fRecvTime;
         long           fRecvSize;
#endif
 
         virtual const char* ClassName() const { return "SocketIOAddon"; }
 
         bool IsDatagramSocket() const { return fDatagramSocket; }
 
         virtual void ProcessEvent(const EventId&);
 
         void AllocateSendIOV(unsigned size);
         void AllocateRecvIOV(unsigned size);
 
         inline bool IsDoingSend() const { return fSendIOVNumber>0; }
         inline bool IsDoingRecv() const { return fRecvIOVNumber>0; }
 
         virtual void OnSendCompleted()
         {
            if (IsDeliverEventsToWorker()) FireWorkerEvent(evntSocketSendInfo);
         }
 
         virtual void OnRecvCompleted()
         {
            if (IsDeliverEventsToWorker()) FireWorkerEvent(evntSocketRecvInfo);
         }
 
         struct sockaddr_in& GetRecvAddr() { return fRecvAddr; }
 
         unsigned GetRecvSize() const { return fLastRecvSize; }
 
      public:
 
         SocketIOAddon(int fd = 0, bool isdatagram = false, bool usemsg = true);
 
         virtual ~SocketIOAddon();
 
         void SetSendAddr(const std::string &host = "", int port = 0);
 
         bool StartSend(const void* buf, unsigned size,
                        const void* buf2 = 0, unsigned size2 = 0,
                        const void* buf3 = 0, unsigned size3 = 0);
         bool StartRecv(void* buf, size_t size);
 
         bool StartRecvHdr(void* hdr, unsigned hdrsize, void* buf, size_t size);
 
         bool StartSend(const Buffer& buf);
         bool StartRecv(Buffer& buf, BufferSize_t datasize);
 
         bool StartNetRecv(void* hdr, unsigned hdrsize, Buffer& buf, BufferSize_t datasize);
         bool StartNetSend(void* hdr, unsigned hdrsize, const Buffer& buf);
 
         void CancelIOOperations();
   };
 
   // ______________________________________________________________________
 
   class SocketConnectAddon : public SocketAddon {
      protected:
         WorkerRef fConnRcv;
         std::string fConnId;
 
      public:
         SocketConnectAddon(int fd) :
            SocketAddon(fd),
            fConnRcv(),
            fConnId()
         {}
 
         virtual void ObjectCleanup()
         {
            fConnRcv.Release();
            SocketAddon::ObjectCleanup();
         }
 
         virtual ~SocketConnectAddon() {}
 
         void SetConnHandler(const WorkerRef& rcv, const std::string &connid)
         {
            fConnRcv = rcv;
            fConnId = connid;
         }
 
         virtual const char* ClassName() const { return "SocketConnectAddon"; }
 
   };
 
   // ________________________________________________________________
 
   class SocketServerAddon : public SocketConnectAddon {
      protected:
         std::string fServerHostName;
         int  fServerPortNumber{0};
         int  fAcceptErrors{0};
         int  ai_family{0};
         int  ai_socktype{0};
         int  ai_protocol{0};
         socklen_t ai_addrlen{0};
         struct sockaddr_storage ai_addr;
 
         virtual void OnClientConnected(int fd);
 
         bool RecreateSocket();
 
      public:
         SocketServerAddon(int serversocket, const char* hostname = nullptr, int portnum = -1, struct addrinfo *info = nullptr);
         virtual ~SocketServerAddon() {}
 
         virtual void ProcessEvent(const EventId&);
 
         std::string ServerHostName() { return fServerHostName; }
         int ServerPortNumber() const { return fServerPortNumber; }
         std::string ServerId() { return dabc::format("%s:%d", fServerHostName.c_str(), fServerPortNumber); }
 
         virtual const char* ClassName() const { return "SocketServerAddon"; }
 
   };
 
   // ______________________________________________________________
 
   class SocketClientAddon : public SocketConnectAddon {
      public:
         SocketClientAddon(const struct addrinfo* serv_addr, int fd = -1);
         virtual ~SocketClientAddon();
 
         void SetRetryOpt(int nretry, double tmout = 1.);
 
         virtual void ProcessEvent(const EventId&);
 
         virtual double ProcessTimeout(double);
 
         virtual const char* ClassName() const { return "SocketClientAddon"; }
 
      protected:
         virtual void OnConnectionEstablished(int fd);
         virtual void OnConnectionFailed();
 
         virtual void OnThreadAssigned();
 
         struct addrinfo fServAddr; // own copy of server address
         int             fRetry;
         double          fRetryTmout;
   };
 
   // ______________________________________________________________
 
   class SocketThread : public Thread {
      protected:
         enum ESocketEvents {
            evntEnableCheck = evntLastThrd+1,  
            evntLastSocketThrdEvent            
         };
 
         struct ProcRec {
             bool      use;  
             uint32_t  indx; 
         };
 
         int            fPipe[2];         
         long           fPipeFired;       
         bool           fWaitFire;        
         int            fScalerCounter;   
         unsigned       f_sizeufds;       
         pollfd        *f_ufds;           
         ProcRec       *f_recs;           
         bool           fIsAnySocket;     
         bool           fCheckNewEvents;  
         int            fBalanceCnt;      
 
#ifdef SOCKET_PROFILING
         long           fWaitCalls;
         long           fWaitDone;
         long           fPipeCalled;
         double         fWaitTime;
         double         fFillTime;
#endif
 
         virtual bool WaitEvent(EventId&, double tmout);
 
         virtual void _Fire(const EventId& evnt, int nq);
 
         virtual void WorkersSetChanged();
 
         virtual void ProcessExtraThreadEvent(const EventId& evid);
 
      public:
 
         // list of all events for all kind of socket processors
 
         SocketThread(Reference parent, const std::string &name, Command cmd);
         virtual ~SocketThread();
 
         virtual const char* ClassName() const { return typeSocketThread; }
         virtual bool CompatibleClass(const std::string &clname) const;
 
         static bool SetNonBlockSocket(int fd);
         static bool SetNoDelaySocket(int fd);
 
         static int StartClient(const std::string &host, int nport, bool nonblocking = true);
 
         static int SendBuffer(int fd, void* buf, int len);
 
         static int RecvBuffer(int fd, void* buf, int len);
 
 
         static int CreateUdp();
 
         static int BindUdp(int fd, int nport, int portmin=-1, int portmax=-1);
 
         static void CloseUdp(int fd);
 
         static bool AttachMulticast(int handle, const std::string &addr);
 
         static void DettachMulticast(int handle, const std::string &addr);
 
         static std::string DefineHostName(bool force = true);
 
         static int ConnectUdp(int fd, const std::string &remhost, int remport);
 
         static SocketServerAddon* CreateServerAddon(const std::string &host, int nport, int portmin=-1, int portmax=-1);
 
         static SocketClientAddon* CreateClientAddon(const std::string &servid, int dflt_port = -1);
   };
}
 
#endif