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
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;
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);
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);
382 do {rlen =
read(
LinkInfo.
FD, Buff, Blen);}
while(rlen < 0 && errno == EINTR);
386 if (rlen >= 0)
return int(rlen);
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);
423 if (!(polltab.revents & (POLLIN|POLLRDNORM)))
433 while(rlen < 0 && errno == EINTR);
435 {
if (!rlen)
return -ENOMSG;
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);
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;
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);
561 while(retc < 0 && errno == EINTR);
565 if (retc < 0)
Log.
Emsg(
"Link", errno,
"receive from",
ID);
594 ssize_t retc = 0, bytesleft = Blen;
614 {
if (errno == EINTR)
continue;
617 bytesleft -= retc; Buff += retc;
623 if (retc >= 0)
return Blen;
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;
714 do{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;
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;
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__)
973 if (
isTLS == enable)
return true;
992 snprintf(buff,
sizeof(buff),
"Unable to enable tls for %s;",
ID);
993 Log.
Emsg(
"LinkXeq", buff, eNote);
1005 else {
isTLS = enable;
1018 Log.
Emsg(
"TLS", rc,
"send file to",
ID);
1032 TRACEI(
DEBUG, (getLock ?
"Async" :
"Sync") <<
" link shutdown in progress");
1045 Log.
Emsg(
"Link", errno,
"shutdown FD for",
ID);
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;
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;
1223 int pend, rlen, totlen = 0;
1236 {
if (pend < 0)
return -1;
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;
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;
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)
bool Register(const char *hName)
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.