36 #include <sys/types.h>
39 #if defined(__linux__) || defined(__GNU__)
40 #include <netinet/tcp.h>
41 #if !defined(TCP_CORK)
48 #if defined(__solaris__) || defined(__linux__) || defined(__GNU__)
49 #include <sys/sendfile.h>
68 #define TRACE_IDENT ID
81 if ((maxiov = sysconf(_SC_IOV_MAX)) > 0)
return maxiov;
132 strcpy(
Lname,
"somewhere");
174 if (nbsz <= 0)
return 0;
176 if ((ulen +
HNlen) >= nbsz) ulen = 0;
177 else {strncpy(nbuf,
ID, ulen);
190 int csec, fd, rc = 0;
297 if (fd >= 2) {
if (
KeepFD) rc = 0;
298 else rc = (
close(fd) < 0 ? errno : 0);
300 if (rc)
Log.
Emsg(
"Link", rc,
"close",
ID);
321 else {
Log.
Emsg(
"Link",
"Dispatch on closed link",
ID);
330 else if (rc != -EINPROGRESS)
Close();
349 struct pollfd polltab = {
PollInfo.
FD, POLLIN|POLLRDNORM, 0};
360 do {retc = poll(&polltab, 1, timeout);}
while(retc < 0 && errno == EINTR);
362 {
if (retc == 0)
return 0;
363 return Log.
Emsg(
"Link", -errno,
"poll",
ID);
368 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
375 do {mlen = recv(
LinkInfo.
FD, Buff, Blen, MSG_PEEK);}
376 while(mlen < 0 && errno == EINTR);
380 if (mlen >= 0)
return int(mlen);
398 do {rlen =
read(
LinkInfo.
FD, Buff, Blen);}
while(rlen < 0 && errno == EINTR);
402 if (rlen >= 0)
return int(rlen);
412 struct pollfd polltab = {
PollInfo.
FD, POLLIN|POLLRDNORM, 0};
413 ssize_t rlen, totlen = 0;
424 {
do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
439 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
449 while(rlen < 0 && errno == EINTR);
451 {
if (!rlen)
return -ENOMSG;
455 totlen += rlen; Blen -= rlen; Buff += rlen;
467 struct pollfd polltab = {
PollInfo.
FD, POLLIN|POLLRDNORM, 0};
477 do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
488 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
504 int seglen, segcnt =
maxIOV, totlen = 0;
506 for (
int i = 0; i < segcnt; i++) seglen +=
iov[i].iov_len;
507 if ((rlen =
RecvIOV(
iov, segcnt)) < 0)
return rlen;
509 if (rlen < seglen)
break;
512 if (iocnt <=
maxIOV) segcnt = iocnt;
527 struct pollfd polltab = {
PollInfo.
FD, POLLIN|POLLRDNORM, 0};
535 {
do {retc = poll(&polltab,1,timeout);}
while(retc < 0 && errno == EINTR);
537 {
if (!retc)
return -ETIMEDOUT;
541 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
551 do {rlen = recv(
LinkInfo.
FD, Buff, Blen, MSG_WAITALL);}
552 while(rlen < 0 && errno == EINTR);
556 if (
int(rlen) == Blen)
return Blen;
557 if (!rlen) {
TRACEI(
DEBUG,
"No RecvAll() data; errno=" <<errno);}
558 else if (rlen > 0)
Log.
Emsg(
"RecvAll",
"Premature end from",
ID);
577 while(retc < 0 && errno == EINTR);
581 if (retc < 0)
Log.
Emsg(
"Link", errno,
"receive from",
ID);
606 ssize_t retc = 0, bytesleft = Blen;
626 {
if (errno == EINTR)
continue;
629 bytesleft -= retc; Buff += retc;
635 if (retc >= 0)
return Blen;
648 maxIOV = sysconf(_SC_IOV_MAX);
682 int seglen, segcnt =
maxIOV, iolen = 0;
684 for (
int i = 0; i < segcnt; i++) seglen +=
iov[i].iov_len;
685 if ((retc =
SendIOV(
iov, segcnt, seglen)) < 0)
692 if (iocnt <=
maxIOV) segcnt = iocnt;
705 #if !defined(HAVE_SENDFILE)
709 #elif defined(__solaris__)
712 size_t xframt, totamt, bytes = 0;
718 for (i = 0; i < sfN; sfP++, i++)
720 {vecSF[i].sfv_fd = SFV_FD_SELF;
721 vecSF[i].sfv_off = (off_t)sfP->buffer;
723 vecSF[i].sfv_fd = sfP->
fdnum;
724 vecSF[i].sfv_off = sfP->offset;
726 vecSF[i].sfv_flag = 0;
727 vecSF[i].sfv_len = sfP->
sendsz;
738 do{retc = sendfilev(
LinkInfo.
FD, vecSFP, sfN, &xframt);
750 if (retc < 0 && errno != EINTR)
break;
756 while(xframt > 0 && sfN)
757 {
if ((ssize_t)xframt < (ssize_t)vecSFP->sfv_len)
758 {vecSFP->sfv_off += xframt; vecSFP->sfv_len -= xframt;
break;}
759 xframt -= vecSFP->sfv_len; vecSFP++; sfN--;
766 retc = (retc < 0 ? errno : ECANCELED);
768 Log.
Emsg(
"Link", retc,
"send file to",
ID);
771 #elif defined(__linux__) || defined(__GNU__)
773 static const int setON = 1, setOFF = 0;
774 ssize_t retc = 0, bytesleft;
776 int i, xfrbytes = 0, uncork = 1, xIntr = 0;
786 if (setsockopt(
PollInfo.
FD, SOL_TCP, TCP_CORK, &setON,
sizeof(setON)) < 0)
787 {
Log.
Emsg(
"Link", errno,
"cork socket for",
ID);
788 uncork = 0;
sfOK = 0;
793 for (i = 0; i < sfN; sfP++, i++)
795 else {myOffset = sfP->offset; bytesleft = sfP->
sendsz;
798 {bytesleft -= retc; xIntr++;}
800 if (retc < 0 && errno == EINTR)
continue;
801 if (retc <= 0)
break;
808 {
if (retc == 0) errno = ECANCELED;
810 Log.
Emsg(
"Link", errno,
"send file to",
ID);
817 && setsockopt(
PollInfo.
FD, SOL_TCP, TCP_CORK, &setOFF,
sizeof(setOFF)) < 0)
818 Log.
Emsg(
"Link", errno,
"uncork socket for",
ID);
822 if (xIntr > sfN)
SfIntr += (xIntr - sfN);
840 ssize_t retc = 0, bytesleft = Blen;
846 {
if (errno == EINTR)
continue;
849 bytesleft -= retc; Buff += retc;
863 ssize_t bytesleft, n, retc = 0;
873 bytesleft =
static_cast<ssize_t
>(bytes);
876 while(retc < 0 && errno == EINTR);
877 if (retc >= bytesleft || retc < 0)
break;
879 while(retc >= (n =
static_cast<ssize_t
>(
iov->iov_len)))
880 {retc -= n;
iov++; iocnt--;}
881 Buff = (
const char *)
iov->iov_base + retc; n -= retc;
iov++; iocnt--;
883 {
if (errno == EINTR)
continue;
886 n -= retc; Buff += retc; bytesleft -= retc;
888 if (retc < 0 || iocnt < 1)
break;
893 if (retc >= 0)
return bytes;
904 char buff[
sizeof(
Uname)], *bp, *sp;
907 snprintf(buff,
sizeof(buff),
"%s.%d:%d", userid, procid,
PollInfo.
FD);
909 sp = buff + ulen - 1;
911 if (ulen > (
int)
sizeof(
Uname)) ulen =
sizeof(
Uname);
913 while(ulen--) {*bp = *sp; bp--; sp--;}
930 #if !defined(__linux__)
997 if (
isTLS == enable)
return true;
1016 snprintf(buff,
sizeof(buff),
"Unable to enable tls for %s;",
ID);
1017 Log.
Emsg(
"LinkXeq", buff, eNote);
1029 else {
isTLS = enable;
1042 Log.
Emsg(
"TLS", rc,
"send file to",
ID);
1056 TRACEI(
DEBUG, (getLock ?
"Async" :
"Sync") <<
" link shutdown in progress");
1069 Log.
Emsg(
"Link", errno,
"shutdown FD for",
ID);
1084 static const char statfmt[] =
"<stats id=\"link\"><num>%d</num>"
1085 "<maxn>%d</maxn><tot>%lld</tot><in>%lld</in><out>%lld</out>"
1086 "<ctime>%lld</ctime><tmo>%d</tmo><stall>%d</stall>"
1087 "<sfps>%d</sfps></stats>";
1092 if (!buff)
return sizeof(statfmt)+17*6;
1180 snprintf(msg,
sizeof(msg),
"Unable to %s %s;", act,
ID);
1181 Log.
Emsg(
"TLS", msg, reason.c_str());
1204 if (rc < 1)
return rc;
1247 int pend, rlen, totlen = 0;
1260 {
if (pend < 0)
return -1;
1276 {
if (!totlen)
return -ENOMSG;
1280 if (rlen <= 0)
break;
1281 totlen += rlen; Blen -= rlen; Buff += rlen;
1294 int Blen, rlen, totlen = 0;
1303 for (
int i = 0; i < iocnt; i++)
1304 {Buff = (
char *)
iov[i].iov_base;
1305 Blen =
iov[i].iov_len;
1306 rlen =
TLS_Recv(Buff, Blen, timeout,
true);
1307 if (rlen <= 0)
break;
1309 if (rlen < Blen)
break;
1330 if (retc < 1)
return (retc ? -1 : -ETIMEDOUT);
1345 ssize_t bytesleft = Blen;
1361 {retc =
tlsIO.
Write(Buff, bytesleft, byteswritten);
1363 bytesleft -= byteswritten; Buff += byteswritten;
1391 for (
int i = 0; i < iocnt; i++)
1392 {ssize_t bytesleft =
iov[i].iov_len;
1393 char *Buff = (
char *)
iov[i].iov_base;
1395 {retc =
tlsIO.
Write(Buff, bytesleft, byteswritten);
1397 bytesleft -= byteswritten; Buff += byteswritten;
1411 int bytes, buffsz, fileFD, retc;
1420 for (
int i = 0; i < sfN; sfP++, i++)
1421 {
if (!(bytes = sfP->
sendsz))
continue;
1424 {
if (!
TLS_Write(sfP->buffer, bytes))
return -1;
1427 offset = sfP->offset;
1428 fileFD = sfP->
fdnum;
1429 buffsz = (bytes < (int)
sizeof(myBuff) ? bytes :
sizeof(myBuff));
1430 do {
do {retc =
pread(fileFD, myBuff, buffsz, offset);}
1431 while(retc < 0 && errno == EINTR);
1432 if (retc < 0)
return SFError(errno);
1434 if (!
TLS_Write(myBuff, buffsz))
return -1;
1435 offset += buffsz; bytes -= buffsz; totamt += retc;
1457 {retc =
tlsIO.
Write(Buff, Blen, byteswritten);
1462 Blen -= byteswritten; Buff += byteswritten;
ssize_t pread(int fildes, void *buf, size_t nbyte, off_t offset)
ssize_t readv(int fildes, const struct iovec *iov, int iovcnt)
ssize_t write(int fildes, const void *buf, size_t nbyte)
ssize_t writev(int fildes, const struct iovec *iov, int iovcnt)
ssize_t read(int fildes, void *buf, size_t nbyte)
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.
void SetDialect(const char *dP)
virtual int Enable(XrdPollInfo &pInfo)=0
static char * Poll2Text(short events)
static void Detach(XrdPollInfo &pInfo)
virtual void Recycle(XrdLink *lp=0, int consec=0, const char *reason=0)=0
virtual int Stats(char *buff, int blen, int do_sync=0)=0
virtual int Process(XrdLink *lp)=0
void Terminate(XrdLink *lP=0)
int Send(const char *buff, int blen)
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
void Lock(XrdSysMutex *Mutex)
int fd
Socket file descriptor.
long long bytesOut
Bytes written to the socket.
int consec
Seconds connected.
virtual void Monitor(XrdNetAddrInfo &netInfo, LinkInfo &lnkInfo, int liLen)=0
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.
XrdTls::RC Accept(std::string *eMsg=0)
void Shutdown(SDType=sdImmed)
@ TLS_RBL_WBL
blocking read blocking write
XrdTls::RC Write(const char *buffer, size_t size, int &bytesOut)
XrdTls::RC Read(char *buffer, size_t size, int &bytesRead)
Read from the TLS connection. If necessary, a handshake will be done.
const char * Init(XrdTlsContext &ctx, int sfd, RW_Mode rwm, HS_Mode hsm, bool isClient, bool serial=true, const char *tid="")
void SetTraceID(const char *tid)
int Pending(bool any=true)
XrdTls::RC Peek(char *buffer, size_t size, int &bytesPeek)
XrdTlsPeerCerts * getCerts(bool ver=true)
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.