Logo ROOT   6.10/09
Reference Guide
TXSocket.h
Go to the documentation of this file.
1 // @(#)root/proofx:$Id$
2 // Author: G. Ganis Oct 2005
3 
4 /*************************************************************************
5  * Copyright (C) 1995-2005, Rene Brun and Fons Rademakers. *
6  * All rights reserved. *
7  * *
8  * For the licensing terms see $ROOTSYS/LICENSE. *
9  * For the list of contributors see $ROOTSYS/README/CREDITS. *
10  *************************************************************************/
11 
12 #ifndef ROOT_TXSocket
13 #define ROOT_TXSocket
14 
15 //////////////////////////////////////////////////////////////////////////
16 // //
17 // TXSocket //
18 // //
19 // High level handler of connections to xproofd. //
20 // //
21 //////////////////////////////////////////////////////////////////////////
22 
23 #define DFLT_CONNECTMAXTRY 10
24 
25 #include "TSemaphore.h"
26 #include "TString.h"
27 #include "TList.h"
28 #include "TMessage.h"
29 #include "TUrl.h"
30 #include "TSocket.h"
31 #ifndef __XPTYPES_H
32 #include "XProtocol/XPtypes.hh"
33 #endif
35 
36 #include <list>
37 #include <mutex>
38 
39 class TObjString;
40 class TXSockBuf;
41 class TXSockPipe;
42 class TXHandler;
43 class TXSocketHandler;
44 class XrdClientMessage;
45 class XrdProofConn;
46 
47 // To transmit info to Handlers
48 typedef struct {
53 } XHandleIn_t;
54 typedef struct {
56  const char *fMsg;
57 } XHandleErr_t;
58 
60 
61 friend class TXProofMgr;
62 friend class TXProofServ;
63 friend class TXSlave;
64 friend class TXSocketHandler;
65 friend class TXSockPipe;
66 friend class TXUnixSocket;
67 
68 private:
69  char fMode; // 'e' (def) or 'i' (internal - proofsrv)
70  kXR_int32 fSendOpt; // Options for sending messages
71  Short_t fSessionID; // proofsrv: remote ID of connected session
72  TString fUser; // Username used for login
73  TString fHost; // Remote host
74  Int_t fPort; // Remote port
75 
76  Int_t fLogLevel; // Log level to be transmitted to servers
77 
78  TString fBuffer; // Container for exchanging information
79  TObject *fReference; // Generic object reference of this socket
80  TXHandler *fHandler; // Handler of asynchronous events (input, error)
81 
82  XrdProofConn *fConn; // instance of the underlying connection module
83 
84  // Asynchronous messages
85  TSemaphore fASem; // Control access to conn async msg queue
86  std::recursive_mutex fAMtx; // To protect async msg queue
87  Bool_t fAWait; // kTRUE if waiting at the async msg queue
88  std::list<TXSockBuf *> fAQue; // list of asynchronous messages
89  Int_t fByteLeft; // bytes left in the first buffer
90  Int_t fByteCur; // current position in the first buffer
91  TXSockBuf *fBufCur; // current read buffer
92 
93  TSemaphore fAsynProc; // Control actions while processing async messages
94 
95  // Interrupts
96  std::recursive_mutex fIMtx; // To protect interrupt queue
97  kXR_int32 fILev; // Highest received interrupt
98  Bool_t fIForward; // Whether the interrupt should be propagated
99 
100  // Process ID of the instatiating process (to signal interrupts)
102 
103  // Whether to timeout or not
104  Bool_t fDontTimeout; // If true wait forever for incoming messages
105  Bool_t fRDInterrupt; // To interrupt waiting for messages
106 
107  // Version of the remote XrdProofdProtocol
109 
110  // Static area for input handling
111  static TXSockPipe fgPipe; // Pipe for input monitoring
112  static TString fgLoc; // Location string
113  static Bool_t fgInitDone; // Avoid initializing more than once
114 
115  // List of spare buffers
116  static std::mutex fgSMtx; // To protect spare list
117  static std::list<TXSockBuf *> fgSQue; // list of spare buffers
118 
119  // Manage asynchronous message
120  Int_t PickUpReady();
122  void PushBackSpare();
123 
124  // Post a message into the queue for asynchronous processing
125  void PostMsg(Int_t type, const char *msg = 0);
126 
127  // Wake up all threads waiting for at the semaphore (used by TXSlave)
128  void PostSemAll();
129 
130  // Auxilliary
131  Int_t GetLowSocket() const;
132 
133  static void SetLocation(const char *loc = ""); // Set location string
134 
135  static void InitEnvs(); // Initialize environment variables
136 
137 public:
138  // Should be the same as in proofd/src/XrdProofdProtocol::Urgent
139  enum EUrgentMsgType { kStopProcess = 2000 };
140 
141  TXSocket(const char *url, Char_t mode = 'M', Int_t psid = -1, Char_t ver = -1,
142  const char *logbuf = 0, Int_t loglevel = -1, TXHandler *handler = 0);
143  virtual ~TXSocket();
144 
145  virtual void Close(Option_t *opt = "");
146  Bool_t Create(Bool_t attach = kFALSE);
147  void DisconnectSession(Int_t id, Option_t *opt = "");
148 
149  void DoError(int level,
150  const char *location, const char *fmt, va_list va) const;
151 
153  XrdClientMessage *msg);
154 
155  virtual Int_t GetClientID() const { return -1; }
156  virtual Int_t GetClientIDSize() const { return 1; }
157  Int_t GetLogConnID() const;
158  Int_t GetOpenError() const;
159  Int_t GetServType() const;
160  Int_t GetSessionID() const;
162 
163  Bool_t IsValid() const;
165  virtual void RemoveClientID() { }
166  virtual void SetClientID(Int_t) { }
167  void SetSendOpt(ESendRecvOptions o) { fSendOpt = o; }
168  void SetSessionID(Int_t id);
169 
170  // Send interfaces
171  Int_t Send(const TMessage &mess);
172  Int_t Send(Int_t kind) { return TSocket::Send(kind); }
173  Int_t Send(Int_t status, Int_t kind)
174  { return TSocket::Send(status, kind); }
175  Int_t Send(const char *mess, Int_t kind = kMESS_STRING)
176  { return TSocket::Send(mess, kind); }
177  Int_t SendRaw(const void *buf, Int_t len,
179 
180  TObjString *SendCoordinator(Int_t kind, const char *msg = 0, Int_t int2 = 0,
181  Long64_t l64 = 0, Int_t int3 = 0, const char *opt = 0);
182 
183  // Recv interfaces
184  Int_t Recv(TMessage *&mess);
185  Int_t Recv(Int_t &status, Int_t &kind)
186  { return TSocket::Recv(status, kind); }
187  Int_t Recv(char *mess, Int_t max)
188  { return TSocket::Recv(mess, max); }
189  Int_t Recv(char *mess, Int_t max, Int_t &kind)
190  { return TSocket::Recv(mess, max, kind); }
191  Int_t RecvRaw(void *buf, Int_t len,
192  ESendRecvOptions opt = kDefault);
193 
194  // Interrupts
197 
198  // Urgent message
199  void SendUrgent(Int_t type, Int_t int1, Int_t int2);
200 
201  // Interrupt the low level socket
202  void SetInterrupt(Bool_t i = kTRUE);
203  inline Bool_t IsInterrupt() { std::lock_guard<std::recursive_mutex> lock(fAMtx);
204  return fRDInterrupt; }
205 
206  // Set / Check async msg queue waiting status
207  inline void SetAWait(Bool_t w = kTRUE) {
208  std::lock_guard<std::recursive_mutex> lock(fAMtx);
209  fAWait = w; }
210  inline Bool_t IsAWait() { std::lock_guard<std::recursive_mutex> lock(fAMtx);
211  return fAWait; }
212  // Flush the asynchronous queue
213  Int_t Flush();
214 
215  // Ping the counterpart
216  Bool_t Ping(const char *ord = 0);
217 
218  // Request remote touch of the admin file associated with this connection
219  void RemoteTouch();
220  // Propagate a Ctrl-C
221  void CtrlC();
222 
223  // Standard options cannot be set
225 
226  // Disable / Enable read timeout
227  void DisableTimeout() { fDontTimeout = kTRUE; }
228  void EnableTimeout() { fDontTimeout = kFALSE; }
229 
230  // Try reconnection after error
231  virtual Int_t Reconnect();
232 
233  ClassDef(TXSocket, 0) //A high level connection class for PROOF
234 };
235 
236 
237 //
238 // The following structure is used to store buffers received asynchronously
239 //
240 class TXSockBuf {
241 public:
247 
248  TXSockBuf(Char_t *bp=0, Int_t sz=0, Bool_t own=1);
249  ~TXSockBuf();
250 
251  void Resize(Int_t sz);
252 
253  static Long64_t BuffMem();
254  static Long64_t GetMemMax();
255  static void SetMemMax(Long64_t memmax);
256 
257 private:
259  static Long64_t fgBuffMem; // Total allocated memory
260  static Long64_t fgMemMax; // Max allocated memory allowed
261 };
262 
263 //
264 // The following class describes internal pipes
265 //
266 class TXSockPipe {
267 public:
268 
269  TXSockPipe(const char *loc = "");
270  virtual ~TXSockPipe();
271 
272  Bool_t IsValid() const { return ((fPipe[0] >= 0 && fPipe[1] >= 0) ? kTRUE : kFALSE); }
273 
274  TXSocket *GetLastReady();
275 
276  Int_t GetRead() const { return fPipe[0]; }
277  Int_t Post(TSocket *s); // Notify socket ready via global pipe
278  Int_t Clean(TSocket *s); // Clean previous pipe notification
279  Int_t Flush(TSocket *s); // Remove any instance of 's' from the pipe
280  void DumpReadySock();
281 
282  void SetLoc(const char *loc = "") { fLoc = loc; }
283 
284 private:
285  std::recursive_mutex fMutex; // Protect access to the sockets-ready list
286  Int_t fPipe[2]; // Pipe for input monitoring
287  TString fLoc; // Location string
288  TList fReadySock; // List of sockets ready to be read
289 };
290 
291 //
292 // Guard for a semaphore
293 //
295 public:
296 
297  TXSemaphoreGuard(TSemaphore *sem) : fSem(sem), fValid(kTRUE) { if (!fSem || fSem->TryWait()) fValid = kFALSE; }
298  virtual ~TXSemaphoreGuard() { if (fValid && fSem) fSem->Post(); }
299 
300  Bool_t IsValid() const { return fValid; }
301 
302 private:
305 };
306 
307 #endif
void SetSendOpt(ESendRecvOptions o)
Definition: TXSocket.h:167
Int_t fCid
Definition: TXSocket.h:246
static void SetLocation(const char *loc="")
Set location string.
Definition: TXSocket.cxx:242
virtual void RemoveClientID()
Definition: TXSocket.h:165
virtual void SetClientID(Int_t)
Definition: TXSocket.h:166
Int_t Send(const char *mess, Int_t kind=kMESS_STRING)
Send a character string buffer.
Definition: TXSocket.h:175
TSemaphore fAsynProc
Definition: TXSocket.h:93
Int_t Recv(Int_t &status, Int_t &kind)
Receives a status and a message type.
Definition: TXSocket.h:185
long long Long64_t
Definition: RtypesCore.h:69
Int_t SendRaw(const void *buf, Int_t len, ESendRecvOptions opt=kDontBlock)
Send a raw buffer of specified length.
Definition: TXSocket.cxx:1218
TSemaphore fASem
Definition: TXSocket.h:85
void SetLoc(const char *loc="")
Definition: TXSocket.h:282
Int_t Recv(char *mess, Int_t max)
Receive a character string message of maximum max length.
Definition: TXSocket.h:187
int GetServType() const
Definition: XrdProofConn.h:133
Collectable string class.
Definition: TObjString.h:28
int GetLogConnID() const
Definition: XrdProofConn.h:130
Bool_t IsAWait()
Definition: TXSocket.h:210
const char Option_t
Definition: RtypesCore.h:62
ESockOptions
Definition: TSocket.h:36
Int_t fLogLevel
Definition: TXSocket.h:76
std::recursive_mutex fMutex
Definition: TXSocket.h:285
virtual Int_t Send(const TMessage &mess)
Send a TMessage object.
Definition: TSocket.cxx:520
XrdProofConn * fConn
Definition: TXSocket.h:82
char fMode
Definition: TXSocket.h:69
virtual Int_t Recv(TMessage *&mess)
Receive a TMessage object.
Definition: TSocket.cxx:818
EUrgentMsgType
Definition: TXSocket.h:139
std::recursive_mutex fIMtx
Definition: TXSocket.h:96
Basic string class.
Definition: TString.h:129
int Int_t
Definition: RtypesCore.h:41
bool Bool_t
Definition: RtypesCore.h:59
virtual ~TXSocket()
Destructor.
Definition: TXSocket.cxx:231
Int_t Recv(char *mess, Int_t max, Int_t &kind)
Receive a character string message of maximum max length.
Definition: TXSocket.h:189
void SetSessionID(Int_t id)
Set session ID to &#39;id&#39;. If id < 0, disable also the asynchronous handler.
Definition: TXSocket.cxx:256
This is the version of TSlave for workers servers based on XProofD.
Definition: TXSlave.h:32
static std::mutex fgSMtx
Definition: TXSocket.h:116
Int_t fInt3
Definition: TXSocket.h:51
void RemoteTouch()
Remote touch functionality: contact the server to proof our vitality.
Definition: TXSocket.cxx:1347
Int_t fOpt
Definition: TXSocket.h:55
ESendRecvOptions
Definition: TSocket.h:49
Int_t fInt2
Definition: TXSocket.h:50
static Long64_t fgBuffMem
Definition: TXSocket.h:259
virtual ~TXSemaphoreGuard()
Definition: TXSocket.h:298
int GetOpenError() const
Definition: XrdProofConn.h:132
void SetInterrupt()
Interrupt the underlying socket.
Implementation of TXSocket using PF_UNIX sockets.
Definition: TXUnixSocket.h:29
static Long64_t fgMemMax
Definition: TXSocket.h:260
Bool_t fAWait
Definition: TXSocket.h:87
Int_t fPort
Definition: TXSocket.h:74
Int_t RecvRaw(void *buf, Int_t len, ESendRecvOptions opt=kDefault)
Receive a raw buffer of specified length bytes.
Definition: TXSocket.cxx:1581
#define ClassDef(name, id)
Definition: Rtypes.h:297
static void InitEnvs()
Init environment variables for XrdClient.
Definition: TXSocket.cxx:1984
static std::list< TXSockBuf * > fgSQue
Definition: TXSocket.h:117
Int_t Send(const TMessage &mess)
Send a TMessage object.
Definition: TXSocket.cxx:1689
Int_t GetInterrupt(Bool_t &forward)
Get latest interrupt level and reset it; if the interrupt has to be propagated to lower stages forwar...
Definition: TXSocket.cxx:972
Bool_t fIForward
Definition: TXSocket.h:98
TXHandler * fHandler
Definition: TXSocket.h:80
TXSockBuf * PopUpSpare(Int_t sz)
Pop-up a buffer of at least size bytes from the spare list If none is found either one is reallocated...
Definition: TXSocket.cxx:1512
void DoError(int level, const char *location, const char *fmt, va_list va) const
Interface to ErrorHandler (protected).
Definition: TXSocket.cxx:71
Bool_t IsServProofd()
Return kTRUE if the remote server is a &#39;proofd&#39;.
Definition: TXSocket.cxx:959
TString fUser
Definition: TXSocket.h:72
Int_t fInt4
Definition: TXSocket.h:52
A doubly linked list.
Definition: TList.h:43
This class implements the XProofD version of TProofServ, with respect to which it differs only for th...
Definition: TXProofServ.h:30
Bool_t fOwn
Definition: TXSocket.h:245
Bool_t Create(Bool_t attach=kFALSE)
This method sends a request for creation of (or attachment to) a remote server application.
Definition: TXSocket.cxx:1053
Int_t fByteCur
Definition: TXSocket.h:90
Int_t Flush()
Flush the asynchronous queue.
Definition: TXSocket.cxx:1004
Bool_t IsInterrupt()
Definition: TXSocket.h:203
TString fHost
Definition: TXSocket.h:73
virtual Int_t GetClientID() const
Definition: TXSocket.h:155
UnsolRespProcResult
void SetAWait(Bool_t w=kTRUE)
Definition: TXSocket.h:207
Handler of asynchronous events for XProofD sockets.
Definition: TXHandler.h:28
Input handler for XProofD sockets.
Int_t fXrdProofdVersion
Definition: TXSocket.h:108
void PostMsg(Int_t type, const char *msg=0)
Post a message of type &#39;type&#39; into the read messages queue.
Definition: TXSocket.cxx:848
High level handler of connections to XProofD.
Definition: TXSocket.h:59
Bool_t Ping(const char *ord=0)
Ping functionality: contact the server to check its vitality.
Definition: TXSocket.cxx:1275
TString fBuffer
Definition: TXSocket.h:78
std::recursive_mutex fAMtx
Definition: TXSocket.h:86
short Short_t
Definition: RtypesCore.h:35
std::list< TXSockBuf * > fAQue
Definition: TXSocket.h:88
static TString fgLoc
Definition: TXSocket.h:112
Char_t * fBuf
Definition: TXSocket.h:244
Implementation of the functionality provided by TProofMgr in the case of a xproofd-based session...
Definition: TXProofMgr.h:40
TString fLoc
Definition: TXSocket.h:287
TList fReadySock
Definition: TXSocket.h:288
Int_t fSiz
Definition: TXSocket.h:242
Bool_t IsValid() const
Definition: TXSocket.h:272
TXSemaphoreGuard(TSemaphore *sem)
Definition: TXSocket.h:297
Int_t fLen
Definition: TXSocket.h:243
const Bool_t kFALSE
Definition: RtypesCore.h:92
Int_t Send(Int_t status, Int_t kind)
Send a status and a single message opcode.
Definition: TXSocket.h:173
void DisconnectSession(Int_t id, Option_t *opt="")
Disconnect a session.
Definition: TXSocket.cxx:268
TXSockBuf * fBufCur
Definition: TXSocket.h:91
int GetLowSocket()
Return the socket descriptor of the underlying connection.
Short_t fSessionID
Definition: TXSocket.h:71
Int_t Send(Int_t kind)
Send a single message opcode.
Definition: TXSocket.h:172
friend class TXSocket
Definition: XrdProofConn.h:50
Bool_t IsValid() const
Definition: TXSocket.h:300
int type
Definition: TGX11.cxx:120
virtual void Close(const char *opt="")
Close connection.
void forward(const LAYERDATA &prevLayerData, LAYERDATA &currLayerData)
apply the weights (and functions) in forward direction of the DNN
Definition: NeuralNet.icc:544
Bool_t fDontTimeout
Definition: TXSocket.h:104
void DisableTimeout()
Definition: TXSocket.h:227
Int_t Recv(TMessage *&mess)
Receive a TMessage object.
Definition: TXSocket.cxx:1774
Char_t * fMem
Definition: TXSocket.h:258
Int_t fInt1
Definition: TXSocket.h:49
kXR_int32 fSendOpt
Definition: TXSocket.h:70
Bool_t fRDInterrupt
Definition: TXSocket.h:105
TObject * fReference
Definition: TXSocket.h:79
Mother of all ROOT objects.
Definition: TObject.h:37
Int_t SetOption(ESockOptions, Int_t)
Set socket options.
Definition: TXSocket.h:224
char Char_t
Definition: RtypesCore.h:29
void PushBackSpare()
Release read buffer giving back to the spare list.
Definition: TXSocket.cxx:1560
const char * fMsg
Definition: TXSocket.h:56
kXR_int32 fILev
Definition: TXSocket.h:97
Int_t PickUpReady()
Wait and pick-up next buffer from the asynchronous queue.
Definition: TXSocket.cxx:1421
static Bool_t fgInitDone
Definition: TXSocket.h:113
Int_t Reconnect()
Try reconnection after failure.
TObjString * SendCoordinator(Int_t kind, const char *msg=0, Int_t int2=0, Long64_t l64=0, Int_t int3=0, const char *opt=0)
Send message to intermediate coordinator.
Definition: TXSocket.cxx:1825
void SendUrgent(Int_t type, Int_t int1, Int_t int2)
Send urgent message to counterpart; &#39;type&#39; specifies the type of the message (see TXSocket::EUrgentMs...
Definition: TXSocket.cxx:1942
static TXSockPipe fgPipe
Definition: TXSocket.h:111
short GetSessionID() const
Definition: XrdProofConn.h:134
bool IsValid() const
Test validity of this connection.
virtual Int_t GetClientIDSize() const
Definition: TXSocket.h:156
Int_t SendInterrupt(Int_t type)
Send urgent message (interrupt) to remote server Returns 0 or -1 in case of error.
Definition: TXSocket.cxx:1638
Int_t fPid
Definition: TXSocket.h:101
TSemaphore * fSem
Definition: TXSocket.h:303
Int_t fByteLeft
Definition: TXSocket.h:89
void PostSemAll()
Wake up all threads waiting for at the semaphore (used by TXSlave)
Definition: TXSocket.cxx:905
const Bool_t kTRUE
Definition: RtypesCore.h:91
Int_t GetRead() const
Definition: TXSocket.h:276
void EnableTimeout()
Definition: TXSocket.h:228
void CtrlC()
Interrupt the remote protocol instance.
Definition: TXSocket.cxx:1385
Int_t GetXrdProofdVersion() const
Definition: TXSocket.h:161
virtual UnsolRespProcResult ProcessUnsolicitedMsg(XrdClientUnsolMsgSender *s, XrdClientMessage *m)
We are here if an unsolicited response comes from a logical conn The response comes in the form of an...