XRootD
XrdXrootdTransit.cc
Go to the documentation of this file.
1 /******************************************************************************/
2 /* */
3 /* X r d X r o o t d T r a n s i t . c c */
4 /* */
5 /* (c) 2012 by the Board of Trustees of the Leland Stanford, Jr., University */
6 /* All Rights Reserved */
7 /* Produced by Andrew Hanushevsky for Stanford University under contract */
8 /* DE-AC02-76-SFO0515 with the Department of Energy */
9 /* */
10 /* This file is part of the XRootD software suite. */
11 /* */
12 /* XRootD is free software: you can redistribute it and/or modify it under */
13 /* the terms of the GNU Lesser General Public License as published by the */
14 /* Free Software Foundation, either version 3 of the License, or (at your */
15 /* option) any later version. */
16 /* */
17 /* XRootD is distributed in the hope that it will be useful, but WITHOUT */
18 /* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
19 /* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
20 /* License for more details. */
21 /* */
22 /* You should have received a copy of the GNU Lesser General Public License */
23 /* along with XRootD in a file called COPYING.LESSER (LGPL license) and file */
24 /* COPYING (GPL license). If not, see <http://www.gnu.org/licenses/>. */
25 /* */
26 /* The copyright holder's institutional names and contributor's names may not */
27 /* be used to endorse or promote products derived from this software without */
28 /* specific prior written permission of the institution or contributor. */
29 /******************************************************************************/
30 
31 #include <cstring>
32 #include <unistd.h>
33 #include <sys/uio.h>
34 
35 #include "XProtocol/XProtocol.hh"
36 
37 #include "XrdSec/XrdSecEntity.hh"
38 
39 #include "Xrd/XrdBuffer.hh"
40 #include "Xrd/XrdLink.hh"
41 #include "XrdOuc/XrdOucErrInfo.hh"
42 #include "XrdOuc/XrdOucUtils.hh"
44 #include "XrdSys/XrdSysAtomics.hh"
50 
51 /******************************************************************************/
52 /* C l o b a l S y m b o l s */
53 /******************************************************************************/
54 
56 
57 #undef TRACELINK
58 #define TRACELINK Link
59 
60 #define XRD_GETNUM(x)\
61  ntohl(*(static_cast<unsigned int *>(static_cast<void *>(x))))
62 
63 /******************************************************************************/
64 /* S t a t i c M e m b e r s */
65 /******************************************************************************/
66 
67 const char *XrdXrootdTransit::reqTab = XrdXrootdTransit::ReqTable();
68 
70  XrdXrootdTransit::TranStack("TranStack",
71  "transit protocol anchor");
72 
73 /******************************************************************************/
74 /* A l l o c */
75 /******************************************************************************/
76 
78  XrdLink *linkP,
79  XrdSecEntity *seceP,
80  const char *nameP,
81  const char *protP
82  )
83 {
84  XrdXrootdTransit *xp;
85 
86 // Simply return a new transit object masquerading as a bridge
87 //
88  if (!(xp = TranStack.Pop())) xp = new XrdXrootdTransit();
89  xp->Init(rsltP, linkP, seceP, nameP, protP);
90  return xp;
91 }
92 
93 /******************************************************************************/
94 /* A t t n */
95 /******************************************************************************/
96 
97 int XrdXrootdTransit::Attn(XrdLink *lP, short *theSID, int rcode,
98  const struct iovec *ioV, int ioN, int ioL)
99 {
100  XrdXrootdTransPend *tP;
101 
102 // Find the request
103 //
104  if (!(tP = XrdXrootdTransPend::Remove(lP, *theSID)))
105  {TRACE(REQ, "Unable to find request for " <<lP->ID <<" sid=" <<*theSID);
106  return 0;
107  }
108 
109 // Resume the request as we have been waiting for the response.
110 //
111  return tP->bridge->AttnCont(tP, rcode, ioV, ioN, ioL);
112 }
113 
114 /******************************************************************************/
115 /* A t t n C o n t */
116 /******************************************************************************/
117 
118 int XrdXrootdTransit::AttnCont(XrdXrootdTransPend *tP, int rcode,
119  const struct iovec *ioV, int ioN, int ioL)
120 {
121  int rc;
122 
123 // Refresh the request structure
124 //
125  memcpy(&Request, &(tP->Pend.Request), sizeof(Request));
126  delete tP;
127  runWait = 0;
128 
129 // Reissue the request if it's a wait 0 response.
130 //
131  if (rcode==kXR_wait
132  && (!ioN || XRD_GETNUM(ioV[0].iov_base) == 0))
133  {Sched->Schedule((XrdJob *)&waitJob);
134  return 0;
135  }
136 
137 // Send off the deferred response
138 //
139  rc = Send(rcode, ioV, ioN, ioL);
140 
141 // Handle end based on current state
142 //
143  if (rc >= 0 && !runWait)
144  {if (runDone) runStatus.store(0, std::memory_order_release);
145  if (reInvoke) Sched->Schedule((XrdJob *)&respJob);
146  else Link->Enable();
147  }
148 
149 // All done
150 //
151  return rc;
152 }
153 
154 /******************************************************************************/
155 /* D i s c */
156 /******************************************************************************/
157 
159 {
160  char buff[128];
161 
162 // We do not allow disconnection while we are active
163 //
164  if (runStatus.fetch_add(1, std::memory_order_acq_rel)) return false;
165 
166 // Reconnect original protocol to the link
167 //
168  Link->setProtocol(realProt);
169 
170 // Now we need to recycle our xrootd part
171 //
172  sprintf(buff, "%s disconnection", pName);
173  XrdXrootdProtocol::Recycle(Link, time(0)-cTime, buff);
174 
175 // Now just free up our object.
176 //
177  TranStack.Push(&TranLink);
178  return true;
179 }
180 
181 /******************************************************************************/
182 /* Private: F a i l */
183 /******************************************************************************/
184 
185 bool XrdXrootdTransit::Fail(int ecode, const char *etext)
186 {
187  runError = ecode;
188  runEText = etext;
189  return true;
190 }
191 
192 /******************************************************************************/
193 /* F a t a l */
194 /******************************************************************************/
195 
196 int XrdXrootdTransit::Fatal(int rc)
197 {
200 
201  return (respObj->Error(rInfo, runError, runEText) ? rc : -1);
202 }
203 
204 /******************************************************************************/
205 /* I n i t */
206 /******************************************************************************/
207 
208 void XrdXrootdTransit::Init(XrdScheduler *schedP, int qMax, int qTTL)
209 {
210  TranStack.Set(schedP, &XrdXrootdTrace, TRACE_MEM);
211  TranStack.Set(qMax, qTTL);
212 }
213 
214 /******************************************************************************/
215 
217  XrdLink *linkP,
218  XrdSecEntity *seceP,
219  const char *nameP,
220  const char *protP
221  )
222 {
223  XrdNetAddrInfo *addrP;
224  const char *who;
225  char uname[sizeof(Request.login.username)+1];
226 
227 // Set standard stuff
228 //
229  runArgs = 0;
230  runALen = 0;
231  runABsz = 0;
232  runError = 0;
233  runStatus.store(0, std::memory_order_release);
234  runWait = 0;
235  runWTot = 0;
236  runWMax = 3600;
237  runWCall = false;
238  runDone = false;
239  reInvoke = false;
240  wBuff = 0;
241  wBLen = 0;
242  respObj = respP;
243  pName = protP;
244  mySID = getSID();
245 
246 // Bind the protocol to the link
247 //
248  SI->Bump(SI->Count);
249  Link = linkP;
250  Response.Set(linkP);
251  Response.Set(this);
252  strcpy(Entity.prot, "host");
253  Entity.host = (char *)linkP->Host();
254 
255 // Develop a trace identifier
256 //
257  strncpy(uname, nameP, sizeof(uname)-1);
258  uname[sizeof(uname)-1] = 0;
259  XrdOucUtils::Sanitize(uname);
260  linkP->setID(uname, mySID);
261 
262 // Place trace identifier everywhere is should be located
263 
264 // Indicate that this brige supports asynchronous responses
265 //
267 
268 // Mark the client as IPv4 if they came in as IPv4 or mapped IPv4. Note
269 // there is no way we can figure out if this is a dual-stack client.
270 //
271  addrP = Link->AddrInfo();
272  if (addrP->isIPType(XrdNetAddrInfo::IPv4) || addrP->isMapped())
274 
275 // Mark the client as being on a private net if the address is private
276 //
277  if (addrP->isPrivate()) {clientPV |= XrdOucEI::uPrip; rdType = 1;}
278  else rdType = 0;
279 
280 // Now tie the security information
281 //
282  Client = (seceP ? seceP : &Entity);
283  Client->ueid = mySID;
284  Client->tident = Client->pident = Link->ID;
285  Client->addrInfo = addrP;
286 
287 // Allocate a monitoring object, if needed for this connection and record login
288 //
289  if (Monitor.Ready())
290  {Monitor.Register(linkP->ID, linkP->Host(), protP);
291  if (Monitor.Logins())
292  {if (Monitor.Auths() && seceP) MonAuth();
293  else Monitor.Report(Monitor.Auths() ? "" : 0);
294  }
295  }
296 
297 // Complete the request ID object
298 //
299  ReqID.setID(Request.header.streamid, linkP->FDnum(), linkP->Inst());
300 
301 // Substitute our protocol for the existing one
302 //
303  realProt = linkP->setProtocol(this);
304  linkP->setProtName(protP);
305  linkP->armBridge();
306 
307 // Document this login
308 //
309  who = (seceP && seceP->name ? seceP->name : "nobody");
310  eDest.Log(SYS_LOG_01, "Bridge", Link->ID, "login as", who);
311 
312 // All done, indicate we are logged in
313 //
315  cTime = time(0);
316 
317 // Propogate a connect through the whole system
318 //
319  osFS->Connect(Client);
320 }
321 
322 /******************************************************************************/
323 /* P r o c e e d */
324 /******************************************************************************/
325 
327 {
328  int rc;
329 
330 // If we were interrupted in a reinvoke state, resume that state.
331 //
332  if (reInvoke) rc = Process(Link);
333  else rc = 0;
334 
335 // Handle ending status
336 //
337  if (rc >= 0) Link->Enable();
338  else if (rc != -EINPROGRESS) Link->Close();
339 }
340 
341 /******************************************************************************/
342 /* P r o c e s s */
343 /******************************************************************************/
344 
346 {
347  int rc;
348 
349 // This entry is serialized via link processing and data is now available.
350 // One of the following will be returned.
351 //
352 // < 0 -> Stop getting requests,
353 // -EINPROGRESS leave link disabled but otherwise all is well
354 // -n Error, disable and close the link
355 // = 0 -> OK, get next request, if allowed, o/w enable the link
356 // > 0 -> Slow link, stop getting requests and enable the link
357 //
358 
359 // Reflect data is present to the underlying protocol and if Run() has been
360 // called we need to dispatch that request. This may be iterative.
361 //
362 do{rc = realProt->Process((reInvoke ? 0 : lp));
363  if (rc >= 0 && runStatus.load(std::memory_order_acquire))
364  {reInvoke = (rc == 0);
365  if (runError) rc = Fatal(rc);
366  else {runDone = false;
367  rc = (Resume ? XrdXrootdProtocol::Process(lp) : Process2());
368  if (rc >= 0)
369  {if (runWait) rc = -EINPROGRESS;
370  if (!runDone) return rc;
371  runStatus.store(0, std::memory_order_release);
372  }
373  }
374  } else reInvoke = false;
375  } while(rc >= 0 && reInvoke);
376 
377 // Make sure that we indicate that we are no longer active
378 //
379  runStatus.store(0, std::memory_order_release);
380 
381 // All done
382 //
383  return (rc ? rc : 1);
384 }
385 
386 /******************************************************************************/
387 /* R e c y c l e */
388 /******************************************************************************/
389 
390 void XrdXrootdTransit::Recycle(XrdLink *lp, int consec, const char *reason)
391 {
392 
393 // Set ourselves as active so we can't get more requests
394 //
395  runStatus.fetch_add(1, std::memory_order_acq_rel);
396 
397 // If we were active then we will need to quiesce before dismantling ourselves.
398 // Note that Recycle() can only be called if the link is enabled. So, this bit
399 // of code is improbable but we check it anyway.
400 //
401  if (runWait > 0) {
402  TRACEP(EMSG, "WARNING: Recycle is canceling wait job; the wait job might already be running during recycle.");
403  Sched->Cancel(&waitJob);
404  }
405 
406 // First we need to recycle the real protocol
407 //
408  if (realProt) realProt->Recycle(lp, consec, reason);
409 
410 // Now we need to recycle our xrootd part
411 //
412  XrdXrootdProtocol::Recycle(lp, consec, reason);
413 
414 // Release the argument buffer
415 //
416  if (runArgs) {free(runArgs); runArgs = 0;}
417 
418 // Delete all pending requests
419 //
421 
422 // Now just free up our object.
423 //
424  TranStack.Push(&TranLink);
425 }
426 
427 /******************************************************************************/
428 /* R e d r i v e */
429 /******************************************************************************/
430 
432 {
433  static int eCode = htonl(kXR_NoMemory);
434  static char eText[] = "Insufficent memory to re-issue request";
435  static struct iovec ioV[] = {{(char *)&eCode,sizeof(eCode)},
436  {(char *)&eText,sizeof(eText)}};
437  int rc;
438 
439 // Do some tracing
440 //
441  TRACEP(REQ, "Bridge redrive runStatus="<<runStatus.load(std::memory_order_acquire)
442  <<" runError="<<runError
443  <<" runWait="<<runWait<<" runWTot="<<runWTot);
444 
445 // Update wait statistics
446 //
447  runWTot += runWait;
448  runWait = 0;
449 
450 // While we are running asynchronously, there is no way that this object can
451 // be deleted while a timer is outstanding as the link has been disabled. So,
452 // we can reissue the request with little worry.
453 //
454 // This is a bit tricky here as a redriven request may result in a wait. If
455 // this happens we cannot hand the result off to the real protocol until we
456 // wait and successfully redrive. The wait handling occurs asynchronously
457 // so all we need to do is honor it here.
458 //
459  if (!runALen || RunCopy(runArgs, runALen)) {
460  do{rc = Process2();
461  TRACEP(REQ, "Bridge redrive Process2 rc="<<rc
462  <<" runError="<<runError<<" runWait="<<runWait);
463  if (rc == 0 && !runWait && !runError) {
464  rc = realProt->Process(NULL);
465  TRACEP(REQ, "Bridge redrive callback rc="<<rc
466  <<" runStatus="<<runStatus.load(std::memory_order_acquire));
467  }
468  } while((rc == 0) && !runError && !runWait);
469  }
470  else rc = Send(kXR_error, ioV, 2, 0);
471 
472 // Defer the request if need be
473 //
474  if (rc >= 0 && runWait) return;
475  runWTot = 0;
476 
477 // Indicate we are no longer active
478 //
479  runStatus.store(0, std::memory_order_release);
480 
481 // If the link needs to be terminated, terminate the link. Otherwise, we can
482 // enable the link for new requests at this point.
483 //
484  if (rc < 0) Link->Close();
485  else Link->Enable();
486 }
487 
488 /******************************************************************************/
489 /* R e q T a b l e */
490 /******************************************************************************/
491 
492 #define KXR_INDEX(x) x-kXR_auth
493 
495 {
496  static char rTab[kXR_truncate-kXR_auth+1];
497 
498 // Initialize the table
499 //
500  memset(rTab, 0, sizeof(rTab));
501  rTab[KXR_INDEX(kXR_chmod)] = 1;
502  rTab[KXR_INDEX(kXR_close)] = 1;
503  rTab[KXR_INDEX(kXR_dirlist)] = 1;
504  rTab[KXR_INDEX(kXR_locate)] = 1;
505  rTab[KXR_INDEX(kXR_mkdir)] = 1;
506  rTab[KXR_INDEX(kXR_mv)] = 1;
507  rTab[KXR_INDEX(kXR_open)] = 1;
508  rTab[KXR_INDEX(kXR_prepare)] = 1;
509  rTab[KXR_INDEX(kXR_protocol)] = 1;
510  rTab[KXR_INDEX(kXR_query)] = 1;
511  rTab[KXR_INDEX(kXR_read)] = 2;
512  rTab[KXR_INDEX(kXR_readv)] = 2;
513  rTab[KXR_INDEX(kXR_rm)] = 1;
514  rTab[KXR_INDEX(kXR_rmdir)] = 1;
515  rTab[KXR_INDEX(kXR_set)] = 1;
516  rTab[KXR_INDEX(kXR_stat)] = 1;
517  rTab[KXR_INDEX(kXR_statx)] = 1;
518  rTab[KXR_INDEX(kXR_sync)] = 1;
519  rTab[KXR_INDEX(kXR_truncate)] = 1;
520  rTab[KXR_INDEX(kXR_write)] = 2;
521 
522 // Now return the address
523 //
524  return rTab;
525 }
526 
527 /******************************************************************************/
528 /* Private: R e q W r i t e */
529 /******************************************************************************/
530 
531 bool XrdXrootdTransit::ReqWrite(char *xdataP, int xdataL)
532 {
533 
534 // Make sure we always transit to the resume point
535 //
536  myBlen = 0;
537 
538 // If nothing was read, then this is a straight-up write
539 //
540  if (!xdataL || !xdataP || !Request.header.dlen)
541  {Resume = 0; wBuff = xdataP; wBLen = xdataL;
542  return true;
543  }
544 
545 // Partial data was read, we may have to split this between a direct write
546 // and a network read/write -- somewhat complicated.
547 //
548  myBuff = wBuff = xdataP;
549  myBlast = wBLen = xdataL;
551  return true;
552 }
553 
554 /******************************************************************************/
555 /* R u n */
556 /******************************************************************************/
557 
558 bool XrdXrootdTransit::Run(const char *xreqP, char *xdataP, int xdataL)
559 {
560  int movLen;
561 
562 // We do not allow re-entry if we are curently processing a request.
563 // It will be reset, as need, when a response is effected.
564 //
565 
566  if (runStatus.fetch_add(1, std::memory_order_acq_rel))
567  {TRACEP(REQ, "Bridge request failed due to re-entry");
568  return false;
569  }
570 
571 // Copy the request header
572 //
573  memcpy((void *)&Request, (void *)xreqP, sizeof(Request));
574 
575 // Validate that we can actually handle this request
576 //
578  if (Request.header.requestid & 0x8000
579  || Request.header.requestid > static_cast<kXR_unt16>(kXR_truncate)
580  || !reqTab[Request.header.requestid - kXR_auth])
581  {TRACEP(REQ, "Unsupported bridge request");
582  return Fail(kXR_Unsupported, "Unsupported bridge request");
583  }
584 
585 // Validate the data length
586 //
588  if (Request.header.dlen < 0)
589  {TRACEP(REQ, "Invalid request data length");
590  return Fail(kXR_ArgInvalid, "Invalid request data length");
591  }
592 
593 // Copy the stream id and trace this request
594 //
596  TRACEP(REQ, "Bridge req=" <<Request.header.requestid
597  <<" dlen=" <<Request.header.dlen <<" blen=" <<xdataL);
598 
599 // If this is a write request, we will need to do a lot more
600 //
601  if (Request.header.requestid == kXR_write) return ReqWrite(xdataP, xdataL);
602 
603 // Obtain any needed buffer and handle any existing data arguments. Also, we
604 // need to keep a shadow copy of the request arguments should we get a wait
605 // and will need to re-issue the request (the server mangles the args).
606 //
607  if (Request.header.dlen)
608  {movLen = (xdataL < Request.header.dlen ? xdataL : Request.header.dlen);
609  if (!RunCopy(xdataP, movLen)) return true;
610  if (!runArgs || movLen > runABsz)
611  {if (runArgs) free(runArgs);
612  if (!(runArgs = (char *)malloc(movLen)))
613  {TRACEP(REQ, "Failed to allocate memory");
614  return Fail(kXR_NoMemory, "Insufficient memory");
615  }
616  runABsz = movLen;
617  }
618  memcpy(runArgs, xdataP, movLen); runALen = movLen;
619  if ((myBlen = Request.header.dlen - movLen))
620  {myBuff = argp->buff + movLen;
622  return true;
623  }
624  } else runALen = 0;
625 
626 // If we have all the data, indicate request accepted.
627 //
628  runError = 0;
629  Resume = 0;
630  return true;
631 }
632 
633 /******************************************************************************/
634 /* Privae: R u n C o p y */
635 /******************************************************************************/
636 
637 bool XrdXrootdTransit::RunCopy(char *buffP, int buffL)
638 {
639 
640 // Allocate a buffer if we do not have one or it is too small
641 //
642  if (!argp || Request.header.dlen+1 > argp->bsize)
643  {if (argp) BPool->Release(argp);
644  if (!(argp = BPool->Obtain(Request.header.dlen+1)))
645  {Fail(kXR_ArgTooLong, "Request argument too long"); return false;}
646  hcNow = hcPrev; halfBSize = argp->bsize >> 1;
647  }
648 
649 // Copy the arguments to the buffer
650 //
651  memcpy(argp->buff, buffP, buffL);
652  argp->buff[buffL] = 0;
653  return true;
654 }
655 
656 /******************************************************************************/
657 /* S e n d */
658 /******************************************************************************/
659 
660 int XrdXrootdTransit::Send(int rcode, const struct iovec *ioV, int ioN, int ioL)
661 {
664  const char *eMsg;
665  int rc;
666  bool aOK;
667 
668 // Invoke the result object (we initially assume this is the final result)
669 //
670  runDone = true;
671  switch(rcode)
672  {case kXR_error:
673  rc = XRD_GETNUM(ioV[0].iov_base);
674  eMsg = (ioN < 2 ? "" : (const char *)ioV[1].iov_base);
675  if (wBuff) respObj->Free(rInfo, wBuff, wBLen);
676  aOK = respObj->Error(rInfo, rc, eMsg);
677  break;
678  case kXR_ok:
679  if (wBuff) respObj->Free(rInfo, wBuff, wBLen);
680  aOK = (ioN ? respObj->Data(rInfo, ioV, ioN, ioL, true)
681  : respObj->Done(rInfo));
682  break;
683  case kXR_oksofar:
684  aOK = respObj->Data(rInfo, ioV, ioN, ioL, false);
685  runDone = false;
686  break;
687  case kXR_redirect:
688  if (wBuff) respObj->Free(rInfo, wBuff, wBLen);
689  rc = XRD_GETNUM(ioV[0].iov_base);
690  aOK = respObj->Redir(rInfo,rc,(const char *)ioV[1].iov_base);
691  break;
692  case kXR_wait:
693  return Wait(rInfo, ioV, ioN, ioL);
694  break;
695  case kXR_waitresp:
696  runDone = false;
697  return WaitResp(rInfo, ioV, ioN, ioL);
698  break;
699  default: if (wBuff) respObj->Free(rInfo, wBuff, wBLen);
700  aOK = respObj->Error(rInfo, kXR_ServerError,
701  "internal logic error");
702  break;
703  };
704 
705 // All done
706 //
707  return (aOK ? 0 : -1);
708 }
709 
710 /******************************************************************************/
711 
712 int XrdXrootdTransit::Send(long long offset, int dlen, int fdnum)
713 {
716  offset, dlen, fdnum);
717 
718 // Effect callback (this is always a final result)
719 //
720  runDone = true;
721  return (respObj->File(sfInfo, dlen) ? 0 : -1);
722 }
723 
724 /******************************************************************************/
725 
726 int XrdXrootdTransit::Send(XrdOucSFVec *sfvec, int sfvnum, int dlen)
727 {
730  sfvec, sfvnum, dlen);
731 
732 // Effect callback (this is always a final result)
733 //
734  runDone = true;
735  return (respObj->File(sfInfo, dlen) ? 0 : -1);
736 }
737 
738 /******************************************************************************/
739 /* Private: W a i t */
740 /******************************************************************************/
741 
742 int XrdXrootdTransit::Wait(XrdXrootd::Bridge::Context &rInfo,
743  const struct iovec *ioV, int ioN, int ioL)
744 {
745  const char *eMsg;
746 
747 // Trace this request if need be
748 //
749  runWait = XRD_GETNUM(ioV[0].iov_base);
750  eMsg = (ioN < 2 ? "reason unknown" : (const char *)ioV[1].iov_base);
751 
752 // Check if the protocol wants to handle all waits
753 //
754  if (runWMax <= 0)
755  {int wtime = runWait;
756  runWait = 0;
757  return (respObj->Wait(rInfo, wtime, eMsg) ? 0 : -1);
758  }
759 
760 // Check if we have exceeded the maximum wait time
761 //
762  if (runWTot >= runWMax)
763  {runDone = true;
764  runWait = 0;
765  return (respObj->Error(rInfo, kXR_Cancelled, eMsg) ? 0 : -1);
766  }
767 
768 // Readjust wait time
769 //
770  if (runWait > runWMax) runWait = runWMax;
771 
772 // Check if the protocol wants a wait notification
773 //
774  if (runWCall && !(respObj->Wait(rInfo, runWait, eMsg))) return -1;
775 
776 // All done, schedule the wait
777 //
778  TRACEP(REQ, "Bridge delaying request " <<runWait <<" sec (" <<eMsg <<")");
779  Sched->Schedule((XrdJob *)&waitJob, time(0)+runWait);
780  return 0;
781 }
782 
783 /******************************************************************************/
784 /* Private: W a i t R e s p */
785 /******************************************************************************/
786 
787 int XrdXrootdTransit::WaitResp(XrdXrootd::Bridge::Context &rInfo,
788  const struct iovec *ioV, int ioN, int ioL)
789 {
790  XrdXrootdTransPend *trP;
791  const char *eMsg;
792  int wTime;
793 
794 // Trace this request if need be
795 //
796  wTime = XRD_GETNUM(ioV[0].iov_base);
797  eMsg = (ioN < 2 ? "reason unknown" : (const char *)ioV[1].iov_base);
798  TRACEP(REQ, "Bridge waiting for resp; sid=" <<rInfo.sID.num
799  <<" wt=" <<wTime <<" (" <<eMsg <<")");
800 
801 // We would issue callback to see how we should handle this. However, we can't
802 // predictably handle a waitresp. So that means we will just wait for a resp.
803 //
804 // XrdXrootd::Bridge::Result *newCBP = respObj->WaitResp(rInfo, runWait, eMsg);
805 
806 // Save the current state
807 //
808  trP = new XrdXrootdTransPend(Link, this, &Request);
809  trP->Queue();
810 
811 // Effect a wait
812 //
813  runWait = -1;
814  return 0;
815 }
@ kXR_ArgInvalid
Definition: XProtocol.hh:990
@ kXR_Unsupported
Definition: XProtocol.hh:1003
@ kXR_Cancelled
Definition: XProtocol.hh:1007
@ kXR_ServerError
Definition: XProtocol.hh:1002
@ kXR_ArgTooLong
Definition: XProtocol.hh:992
@ kXR_NoMemory
Definition: XProtocol.hh:998
kXR_char streamid[2]
Definition: XProtocol.hh:156
kXR_char username[8]
Definition: XProtocol.hh:396
@ kXR_waitresp
Definition: XProtocol.hh:906
@ kXR_redirect
Definition: XProtocol.hh:904
@ kXR_oksofar
Definition: XProtocol.hh:900
@ kXR_ok
Definition: XProtocol.hh:899
@ kXR_wait
Definition: XProtocol.hh:905
@ kXR_error
Definition: XProtocol.hh:903
struct ClientRequestHdr header
Definition: XProtocol.hh:846
struct ClientLoginRequest login
Definition: XProtocol.hh:857
kXR_unt16 requestid
Definition: XProtocol.hh:157
@ kXR_read
Definition: XProtocol.hh:125
@ kXR_open
Definition: XProtocol.hh:122
@ kXR_readv
Definition: XProtocol.hh:137
@ kXR_mkdir
Definition: XProtocol.hh:120
@ kXR_sync
Definition: XProtocol.hh:128
@ kXR_chmod
Definition: XProtocol.hh:114
@ kXR_dirlist
Definition: XProtocol.hh:116
@ kXR_rm
Definition: XProtocol.hh:126
@ kXR_query
Definition: XProtocol.hh:113
@ kXR_write
Definition: XProtocol.hh:131
@ kXR_auth
Definition: XProtocol.hh:112
@ kXR_set
Definition: XProtocol.hh:130
@ kXR_rmdir
Definition: XProtocol.hh:127
@ kXR_statx
Definition: XProtocol.hh:134
@ kXR_truncate
Definition: XProtocol.hh:140
@ kXR_protocol
Definition: XProtocol.hh:118
@ kXR_mv
Definition: XProtocol.hh:121
@ kXR_stat
Definition: XProtocol.hh:129
@ kXR_locate
Definition: XProtocol.hh:139
@ kXR_close
Definition: XProtocol.hh:115
@ kXR_prepare
Definition: XProtocol.hh:133
@ kXR_asyncap
Definition: XProtocol.hh:378
@ kXR_ver002
Definition: XProtocol.hh:386
kXR_int32 dlen
Definition: XProtocol.hh:159
unsigned short kXR_unt16
Definition: XPtypes.hh:67
unsigned char kXR_char
Definition: XPtypes.hh:65
#define EMSG(x)
Definition: XrdCpConfig.cc:55
#define eMsg(x)
const int SYS_LOG_01
Definition: XrdSysError.hh:72
#define TRACE_MEM
Definition: XrdTrace.hh:38
#define TRACE(act, x)
Definition: XrdTrace.hh:63
#define XRD_LOGGEDIN
#define TRACEP(act, x)
XrdSysTrace XrdXrootdTrace
#define XRD_GETNUM(x)
#define KXR_INDEX(x)
void Release(XrdBuffer *bp)
Definition: XrdBuffer.cc:221
XrdBuffer * Obtain(int bsz)
Definition: XrdBuffer.cc:140
int bsize
Definition: XrdBuffer.hh:46
char * buff
Definition: XrdBuffer.hh:45
Definition: XrdJob.hh:43
bool isMapped() const
bool isIPType(IPType ipType) const
void Set(int inQMax, time_t agemax=1800)
Definition: XrdObject.icc:90
void Push(XrdObject< T > *Node)
Definition: XrdObject.hh:101
T * Pop()
Definition: XrdObject.hh:93
void Bump(int &val)
Definition: XrdOucStats.hh:47
static void Sanitize(char *instr, char subc='_')
virtual void Recycle(XrdLink *lp=0, int consec=0, const char *reason=0)=0
virtual int Process(XrdLink *lp)=0
void Schedule(XrdJob *jp)
void Cancel(XrdJob *jp)
const char * pident
Trace identifier (originator)
Definition: XrdSecEntity.hh:82
XrdNetAddrInfo * addrInfo
Entity's connection details.
Definition: XrdSecEntity.hh:80
const char * tident
Trace identifier always preset.
Definition: XrdSecEntity.hh:81
char prot[XrdSecPROTOIDSIZE]
Auth protocol used (e.g. krb5)
Definition: XrdSecEntity.hh:67
char * name
Entity's name.
Definition: XrdSecEntity.hh:69
unsigned int ueid
Unique ID of entity instance.
Definition: XrdSecEntity.hh:79
char * host
Entity's host name dnr dependent.
Definition: XrdSecEntity.hh:70
virtual void Connect(const XrdSecEntity *client=0)
void Log(int mask, const char *esfx, const char *text1, const char *text2=0, const char *text3=0)
Definition: XrdSysError.hh:133
void Register(const char *Uname, const char *Hname, const char *Pname, unsigned int xSID=0)
void Report(const char *Info)
static XrdXrootdStats * SI
XrdSecEntity * Client
static XrdSysError & eDest
static unsigned int getSID()
XrdXrootdMonitor::User Monitor
int(XrdXrootdProtocol::* Resume)()
static XrdScheduler * Sched
int Process(XrdLink *lp) override
void Recycle(XrdLink *lp, int consec, const char *reason) override
XrdXrootdResponse Response
static XrdBuffManager * BPool
XrdXrootdReqID ReqID
static XrdSfsFileSystem * osFS
void setID(unsigned long long id)
void Set(XrdLink *lp)
static XrdXrootdTransPend * Remove(XrdLink *lP, short sid)
union XrdXrootdTransPend::@191 Pend
XrdXrootdTransit * bridge
static void Clear(XrdXrootdTransit *trP)
bool Run(const char *xreqP, char *xdataP=0, int xdataL=0)
Inject an xrootd request into the protocol stack.
static const char * ReqTable()
Initialize the valid request table.
void Redrive()
Redrive a request after a wait.
int Send(int rcode, const struct iovec *ioVec, int ioNum, int ioLen)
Handle request data response.
void Recycle(XrdLink *lp, int consec, const char *reason)
Handle link shutdown.
static void Init(XrdScheduler *schedP, int qMax, int qTTL)
Perform one-time initialization.
static XrdXrootdTransit * Alloc(XrdXrootd::Bridge::Result *respP, XrdLink *linkP, XrdSecEntity *seceP, const char *nameP, const char *protP)
Get a new transit object.
static int Attn(XrdLink *lP, short *theSID, int rcode, const struct iovec *ioVec, int ioNum, int ioLen)
Handle attention response (i.e. async response)
XrdXrootdTransit()
Constructor & Destructor.
void Proceed()
Resume processing after a waitresp completion.
bool Disc()
Handle dismantlement.
int Process(XrdLink *lp)
Handle link activation (replaces parent activation).
union XrdXrootd::Bridge::Context::@164 sID
associated request stream ID
virtual int File(Bridge::Context &info, int dlen)=0
virtual bool Data(Bridge::Context &info, const struct iovec *iovP, int iovN, int iovL, bool final)=0
virtual bool Error(Bridge::Context &info, int ecode, const char *etext)=0
virtual bool Done(Bridge::Context &info)=0
the result context
virtual bool Redir(Bridge::Context &info, int port, const char *hname)=0
virtual bool Wait(Bridge::Context &info, int wtime, const char *wtext)
virtual void Free(Bridge::Context &info, char *buffP, int buffL)
XrdScheduler * schedP
static const int uIPv4
ucap: Supports read redirects
static const int uPrip