39#if defined(__linux__) || defined(__GNU__)
40#include <netinet/tcp.h>
48#if defined(__solaris__) || defined(__linux__) || defined(__GNU__)
49#include <sys/sendfile.h>
116 strcpy(
Lname,
"somewhere");
158 if (nbsz <= 0)
return 0;
160 if ((ulen +
HNlen) >= nbsz) ulen = 0;
161 else {strncpy(nbuf,
ID, ulen);
174 int csec, fd, rc = 0;
196 sendQ->Terminate(
this);
281 if (fd >= 2) {
if (
KeepFD) rc = 0;
282 else rc = (
close(fd) < 0 ? errno : 0);
284 if (rc)
Log.Emsg(
"Link", rc,
"close",
ID);
304 do {rc =
Protocol->Process(
this);}
while (!rc &&
Sched.canStick());
305 else {
Log.Emsg(
"Link",
"Dispatch on closed link",
ID);
314 else if (rc != -EINPROGRESS)
Close();
333 struct pollfd polltab = {
PollInfo.FD, POLLIN|POLLRDNORM, 0};
344 do {retc = poll(&polltab, 1, timeout);}
while(retc < 0 && errno == EINTR);
346 {
if (retc == 0)
return 0;
347 return Log.Emsg(
"Link", -errno,
"poll",
ID);
352 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
359 do {mlen = recv(
LinkInfo.FD, Buff, Blen, MSG_PEEK);}
360 while(mlen < 0 && errno == EINTR);
364 if (mlen >= 0)
return int(mlen);
365 Log.Emsg(
"Link", errno,
"peek on",
ID);
382 do {rlen =
read(
LinkInfo.FD, Buff, Blen);}
while(rlen < 0 && errno == EINTR);
386 if (rlen >= 0)
return int(rlen);
387 if (
LinkInfo.FD >= 0)
Log.Emsg(
"Link", errno,
"receive from",
ID);
396 struct pollfd polltab = {
PollInfo.FD, POLLIN|POLLRDNORM, 0};
397 ssize_t rlen, totlen = 0;
408 {
do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
418 return (
LinkInfo.FD >= 0 ?
Log.Emsg(
"Link",-errno,
"poll",
ID) : -1);
423 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
432 do {rlen = recv(
LinkInfo.FD, Buff, Blen, 0);}
433 while(rlen < 0 && errno == EINTR);
435 {
if (!rlen)
return -ENOMSG;
436 if (
LinkInfo.FD > 0)
Log.Emsg(
"Link", -errno,
"receive from",
ID);
439 totlen += rlen; Blen -= rlen; Buff += rlen;
451 struct pollfd polltab = {
PollInfo.FD, POLLIN|POLLRDNORM, 0};
461 do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
467 return (
LinkInfo.FD >= 0 ?
Log.Emsg(
"Link",-errno,
"poll",
ID) : -1);
472 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
488 int seglen, segcnt =
maxIOV, totlen = 0;
490 for (
int i = 0; i < segcnt; i++) seglen +=
iov[i].iov_len;
491 if ((rlen =
RecvIOV(
iov, segcnt)) < 0)
return rlen;
493 if (rlen < seglen)
break;
496 if (iocnt <=
maxIOV) segcnt = iocnt;
511 struct pollfd polltab = {
PollInfo.FD, POLLIN|POLLRDNORM, 0};
519 {
do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
521 {
if (!retc)
return -ETIMEDOUT;
522 Log.Emsg(
"Link",errno,
"poll",
ID);
525 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
535 do {rlen = recv(
LinkInfo.FD, Buff, Blen, MSG_WAITALL);}
536 while(rlen < 0 && errno == EINTR);
540 if (
int(rlen) == Blen)
return Blen;
541 if (!rlen) {
TRACEI(
DEBUG,
"No RecvAll() data; errno=" <<errno);}
542 else if (rlen > 0)
Log.Emsg(
"RecvAll",
"Premature end from",
ID);
543 else if (
LinkInfo.FD >= 0)
Log.Emsg(
"Link", errno,
"receive from",
ID);
561 while(retc < 0 && errno == EINTR);
565 if (retc < 0)
Log.Emsg(
"Link", errno,
"receive from",
ID);
578 if (!
Addr.Register(hName))
return false;
594 ssize_t retc = 0, bytesleft = Blen;
605 {retc =
sendQ->Send(Buff, Blen);
614 {
if (errno == EINTR)
continue;
617 bytesleft -= retc; Buff += retc;
623 if (retc >= 0)
return Blen;
624 Log.Emsg(
"Link", errno,
"send to",
ID);
643 {retc =
sendQ->Send(
iov, iocnt, bytes);
658 int seglen, segcnt =
maxIOV, iolen = 0;
660 for (
int i = 0; i < segcnt; i++) seglen +=
iov[i].iov_len;
661 if ((retc =
SendIOV(
iov, segcnt, seglen)) < 0)
668 if (iocnt <=
maxIOV) segcnt = iocnt;
681#if !defined(HAVE_SENDFILE)
685#elif defined(__solaris__)
688 size_t xframt, totamt, bytes = 0;
694 for (i = 0; i < sfN; sfP++, i++)
696 {vecSF[i].sfv_fd = SFV_FD_SELF;
697 vecSF[i].sfv_off = (off_t)sfP->buffer;
699 vecSF[i].sfv_fd = sfP->
fdnum;
700 vecSF[i].sfv_off = sfP->offset;
702 vecSF[i].sfv_flag = 0;
703 vecSF[i].sfv_len = sfP->
sendsz;
714do{retc = sendfilev(
LinkInfo.FD, vecSFP, sfN, &xframt);
726 if (retc < 0 && errno != EINTR)
break;
732 while(xframt > 0 && sfN)
733 {
if ((ssize_t)xframt < (ssize_t)vecSFP->sfv_len)
734 {vecSFP->sfv_off += xframt; vecSFP->sfv_len -= xframt;
break;}
735 xframt -= vecSFP->sfv_len; vecSFP++; sfN--;
742 retc = (retc < 0 ? errno : ECANCELED);
744 Log.Emsg(
"Link", retc,
"send file to",
ID);
747#elif defined(__linux__) || defined(__GNU__)
749 static const int setON = 1, setOFF = 0;
750 ssize_t retc = 0, bytesleft;
752 int i, xfrbytes = 0, uncork = 1, xIntr = 0;
762 if (setsockopt(
PollInfo.FD, SOL_TCP, TCP_CORK, &setON,
sizeof(setON)) < 0)
763 {
Log.Emsg(
"Link", errno,
"cork socket for",
ID);
764 uncork = 0;
sfOK = 0;
769 for (i = 0; i < sfN; sfP++, i++)
771 else {myOffset = sfP->offset; bytesleft = sfP->
sendsz;
773 && (retc=sendfile(
LinkInfo.FD,sfP->
fdnum,&myOffset,bytesleft)) > 0)
774 {bytesleft -= retc; xIntr++;}
776 if (retc < 0 && errno == EINTR)
continue;
777 if (retc <= 0)
break;
784 {
if (retc == 0) errno = ECANCELED;
786 Log.Emsg(
"Link", errno,
"send file to",
ID);
793 && setsockopt(
PollInfo.FD, SOL_TCP, TCP_CORK, &setOFF,
sizeof(setOFF)) < 0)
794 Log.Emsg(
"Link", errno,
"uncork socket for",
ID);
798 if (xIntr > sfN)
SfIntr += (xIntr - sfN);
816 ssize_t retc = 0, bytesleft = Blen;
822 {
if (errno == EINTR)
continue;
825 bytesleft -= retc; Buff += retc;
839 ssize_t bytesleft, n, retc = 0;
849 bytesleft =
static_cast<ssize_t
>(bytes);
852 while(retc < 0 && errno == EINTR);
853 if (retc >= bytesleft || retc < 0)
break;
855 while(retc >= (n =
static_cast<ssize_t
>(
iov->iov_len)))
856 {retc -= n;
iov++; iocnt--;}
857 Buff = (
const char *)
iov->iov_base + retc; n -= retc;
iov++; iocnt--;
859 {
if (errno == EINTR)
continue;
862 n -= retc; Buff += retc; bytesleft -= retc;
864 if (retc < 0 || iocnt < 1)
break;
869 if (retc >= 0)
return bytes;
870 Log.Emsg(
"Link", errno,
"send to",
ID);
880 char buff[
sizeof(
Uname)], *bp, *sp;
883 snprintf(buff,
sizeof(buff),
"%s.%d:%d", userid, procid,
PollInfo.FD);
885 sp = buff + ulen - 1;
887 if (ulen > (
int)
sizeof(
Uname)) ulen =
sizeof(
Uname);
889 while(ulen--) {*bp = *sp; bp--; sp--;}
906#if !defined(__linux__)
954 Addr.SetDialect(name);
973 if (
isTLS == enable)
return true;
986 eNote =
tlsIO.Init(*ctx,
PollInfo.FD, rwMode, hsMode,
false,
false,
ID);
992 snprintf(buff,
sizeof(buff),
"Unable to enable tls for %s;",
ID);
993 Log.Emsg(
"LinkXeq", buff, eNote);
1005 else {
isTLS = enable;
1006 Addr.SetTLS(enable);
1007 Log.Emsg(
"LinkXeq",
ID,
"connection upgraded to",
verTLS());
1018 Log.Emsg(
"TLS", rc,
"send file to",
ID);
1032 TRACEI(
DEBUG, (getLock ?
"Async" :
"Sync") <<
" link shutdown in progress");
1036 if (getLock)
LinkInfo.opMutex.Lock();
1045 Log.Emsg(
"Link", errno,
"shutdown FD for",
ID);
1051 if (getLock)
LinkInfo.opMutex.UnLock();
1060 static const char statfmt[] =
"<stats id=\"link\"><num>%d</num>"
1061 "<maxn>%d</maxn><tot>%lld</tot><in>%lld</in><out>%lld</out>"
1062 "<ctime>%lld</ctime><tmo>%d</tmo><stall>%d</stall>"
1063 "<sfps>%d</sfps></stats>";
1068 if (!buff)
return sizeof(statfmt)+17*6;
1101 if (!ctime)
LinkInfo.opMutex.Lock();
1114 {*ctime = time(0) -
LinkInfo.conTime;
1144 if (!ctime)
LinkInfo.opMutex.UnLock();
1156 snprintf(msg,
sizeof(msg),
"Unable to %s %s;", act,
ID);
1157 Log.Emsg(
"TLS", msg, reason.c_str());
1180 if (rc < 1)
return rc;
1185 retc =
tlsIO.Peek(Buff, Blen, rlen);
1211 retc =
tlsIO.Read(Buff, Blen, rlen);
1223 int pend, rlen, totlen = 0;
1233 {pend =
tlsIO.Pending(
true);
1236 {
if (pend < 0)
return -1;
1250 retc =
tlsIO.Read(Buff, Blen, rlen);
1252 {
if (!totlen)
return -ENOMSG;
1256 if (rlen <= 0)
break;
1257 totlen += rlen; Blen -= rlen; Buff += rlen;
1270 int Blen, rlen, totlen = 0;
1279 for (
int i = 0; i < iocnt; i++)
1280 {Buff = (
char *)
iov[i].iov_base;
1281 Blen =
iov[i].iov_len;
1282 rlen =
TLS_Recv(Buff, Blen, timeout,
true);
1283 if (rlen <= 0)
break;
1285 if (rlen < Blen)
break;
1304 {retc =
tlsIO.Pending(
true);
1306 if (retc < 1)
return (retc ? -1 : -ETIMEDOUT);
1321 ssize_t bytesleft = Blen;
1337 {retc =
tlsIO.Write(Buff, bytesleft, byteswritten);
1339 bytesleft -= byteswritten; Buff += byteswritten;
1367 for (
int i = 0; i < iocnt; i++)
1368 {ssize_t bytesleft =
iov[i].iov_len;
1369 char *Buff = (
char *)
iov[i].iov_base;
1371 {retc =
tlsIO.Write(Buff, bytesleft, byteswritten);
1373 bytesleft -= byteswritten; Buff += byteswritten;
1387 int bytes, buffsz, fileFD, retc;
1396 for (
int i = 0; i < sfN; sfP++, i++)
1397 {
if (!(bytes = sfP->
sendsz))
continue;
1400 {
if (!
TLS_Write(sfP->buffer, bytes))
return -1;
1403 offset = sfP->offset;
1404 fileFD = sfP->
fdnum;
1405 buffsz = (bytes < (int)
sizeof(myBuff) ? bytes :
sizeof(myBuff));
1406 do {
do {retc =
pread(fileFD, myBuff, buffsz, offset);}
1407 while(retc < 0 && errno == EINTR);
1408 if (retc < 0)
return SFError(errno);
1410 if (!
TLS_Write(myBuff, buffsz))
return -1;
1411 offset += buffsz; bytes -= buffsz; totamt += retc;
1433 {retc =
tlsIO.Write(Buff, Blen, byteswritten);
1438 Blen -= byteswritten; Buff += byteswritten;
1452 return tlsIO.Version();
#define pread(a, b, c, d)
static void SyncAll()
Synchronize statustics for ll links.
static void Unhook(int fd)
Unhook a link from the active table of links.
static const char * TraceID
int TLS_Send(const char *Buff, int Blen)
int TLS_Error(const char *act, XrdTls::RC rc)
int TLS_Peek(char *Buff, int Blen, int timeout)
int Client(char *buff, int blen)
XrdTlsPeerCerts * getPeerCerts()
int Close(bool defer=false)
int TLS_Recv(char *Buff, int Blen)
int sendData(const char *Buff, int Blen)
bool TLS_Write(const char *Buff, int Blen)
int SendIOV(const struct iovec *iov, int iocnt, int bytes)
XrdProtocol * setProtocol(XrdProtocol *pp, bool push)
static long long LinkCountTot
void Shutdown(bool getLock)
int Peek(char *buff, int blen, int timeout=-1)
static int Stats(char *buff, int blen, bool do_sync=false)
void setID(const char *userid, int procid)
int Recv(char *buff, int blen)
static long long LinkBytesIn
int TLS_RecvAll(char *Buff, int Blen, int timeout)
int Send(const char *buff, int blen)
int RecvIOV(const struct iovec *iov, int iocnt)
static long long LinkConTime
int RecvAll(char *buff, int blen, int timeout=-1)
bool Register(const char *hName)
static XrdSysMutex statsMutex
void setProtName(const char *name)
static long long LinkBytesOut
void syncStats(int *ctime=0)
bool setTLS(bool enable, XrdTlsContext *ctx=0)
void Serialize()
Wait for all outstanding requests to be completed on the link.
int Wait4Data(int timeout)
char * ID
Pointer to the client's link identity.
static char * Poll2Text(short events)
static void Detach(XrdPollInfo &pInfo)
void Lock(XrdSysMutex *Mutex)
int fd
Socket file descriptor.
long long bytesOut
Bytes written to the socket.
int consec
Seconds connected.
long long bytesIn
Bytes read from the socket.
const char * tident
Pointer to the client's trace identifier.
@ TLS_HS_BLOCK
Always block during handshake.
@ TLS_RBL_WBL
blocking read blocking write
static std::string RC2Text(XrdTls::RC rc, bool dbg=false)
@ TLS_AOK
All went well, will always be zero.
int fdnum
File descriptor for data.
int sendsz
Length of data at offset.