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);
590 ssize_t retc = 0, bytesleft = Blen;
610 {
if (errno == EINTR)
continue;
613 bytesleft -= retc; Buff += retc;
619 if (retc >= 0)
return Blen;
654 int seglen, segcnt =
maxIOV, iolen = 0;
656 for (
int i = 0; i < segcnt; i++) seglen +=
iov[i].iov_len;
657 if ((retc =
SendIOV(
iov, segcnt, seglen)) < 0)
664 if (iocnt <=
maxIOV) segcnt = iocnt;
677 #if !defined(HAVE_SENDFILE)
681 #elif defined(__solaris__)
684 size_t xframt, totamt, bytes = 0;
690 for (i = 0; i < sfN; sfP++, i++)
692 {vecSF[i].sfv_fd = SFV_FD_SELF;
693 vecSF[i].sfv_off = (off_t)sfP->buffer;
695 vecSF[i].sfv_fd = sfP->
fdnum;
696 vecSF[i].sfv_off = sfP->offset;
698 vecSF[i].sfv_flag = 0;
699 vecSF[i].sfv_len = sfP->
sendsz;
710 do{retc = sendfilev(
LinkInfo.
FD, vecSFP, sfN, &xframt);
722 if (retc < 0 && errno != EINTR)
break;
728 while(xframt > 0 && sfN)
729 {
if ((ssize_t)xframt < (ssize_t)vecSFP->sfv_len)
730 {vecSFP->sfv_off += xframt; vecSFP->sfv_len -= xframt;
break;}
731 xframt -= vecSFP->sfv_len; vecSFP++; sfN--;
738 retc = (retc < 0 ? errno : ECANCELED);
740 Log.
Emsg(
"Link", retc,
"send file to",
ID);
743 #elif defined(__linux__) || defined(__GNU__)
745 static const int setON = 1, setOFF = 0;
746 ssize_t retc = 0, bytesleft;
748 int i, xfrbytes = 0, uncork = 1, xIntr = 0;
758 if (setsockopt(
PollInfo.
FD, SOL_TCP, TCP_CORK, &setON,
sizeof(setON)) < 0)
759 {
Log.
Emsg(
"Link", errno,
"cork socket for",
ID);
760 uncork = 0;
sfOK = 0;
765 for (i = 0; i < sfN; sfP++, i++)
767 else {myOffset = sfP->offset; bytesleft = sfP->
sendsz;
770 {bytesleft -= retc; xIntr++;}
772 if (retc < 0 && errno == EINTR)
continue;
773 if (retc <= 0)
break;
780 {
if (retc == 0) errno = ECANCELED;
782 Log.
Emsg(
"Link", errno,
"send file to",
ID);
789 && setsockopt(
PollInfo.
FD, SOL_TCP, TCP_CORK, &setOFF,
sizeof(setOFF)) < 0)
790 Log.
Emsg(
"Link", errno,
"uncork socket for",
ID);
794 if (xIntr > sfN)
SfIntr += (xIntr - sfN);
812 ssize_t retc = 0, bytesleft = Blen;
818 {
if (errno == EINTR)
continue;
821 bytesleft -= retc; Buff += retc;
835 ssize_t bytesleft, n, retc = 0;
845 bytesleft =
static_cast<ssize_t
>(bytes);
848 while(retc < 0 && errno == EINTR);
849 if (retc >= bytesleft || retc < 0)
break;
851 while(retc >= (n =
static_cast<ssize_t
>(
iov->iov_len)))
852 {retc -= n;
iov++; iocnt--;}
853 Buff = (
const char *)
iov->iov_base + retc; n -= retc;
iov++; iocnt--;
855 {
if (errno == EINTR)
continue;
858 n -= retc; Buff += retc; bytesleft -= retc;
860 if (retc < 0 || iocnt < 1)
break;
865 if (retc >= 0)
return bytes;
876 char buff[
sizeof(
Uname)], *bp, *sp;
879 snprintf(buff,
sizeof(buff),
"%s.%d:%d", userid, procid,
PollInfo.
FD);
881 sp = buff + ulen - 1;
883 if (ulen > (
int)
sizeof(
Uname)) ulen =
sizeof(
Uname);
885 while(ulen--) {*bp = *sp; bp--; sp--;}
902 #if !defined(__linux__)
969 if (
isTLS == enable)
return true;
988 snprintf(buff,
sizeof(buff),
"Unable to enable tls for %s;",
ID);
989 Log.
Emsg(
"LinkXeq", buff, eNote);
1001 else {
isTLS = enable;
1014 Log.
Emsg(
"TLS", rc,
"send file to",
ID);
1028 TRACEI(
DEBUG, (getLock ?
"Async" :
"Sync") <<
" link shutdown in progress");
1041 Log.
Emsg(
"Link", errno,
"shutdown FD for",
ID);
1056 static const char statfmt[] =
"<stats id=\"link\"><num>%d</num>"
1057 "<maxn>%d</maxn><tot>%lld</tot><in>%lld</in><out>%lld</out>"
1058 "<ctime>%lld</ctime><tmo>%d</tmo><stall>%d</stall>"
1059 "<sfps>%d</sfps></stats>";
1064 if (!buff)
return sizeof(statfmt)+17*6;
1152 snprintf(msg,
sizeof(msg),
"Unable to %s %s;", act,
ID);
1153 Log.
Emsg(
"TLS", msg, reason.c_str());
1176 if (rc < 1)
return rc;
1219 int pend, rlen, totlen = 0;
1232 {
if (pend < 0)
return -1;
1248 {
if (!totlen)
return -ENOMSG;
1252 if (rlen <= 0)
break;
1253 totlen += rlen; Blen -= rlen; Buff += rlen;
1266 int Blen, rlen, totlen = 0;
1275 for (
int i = 0; i < iocnt; i++)
1276 {Buff = (
char *)
iov[i].iov_base;
1277 Blen =
iov[i].iov_len;
1278 rlen =
TLS_Recv(Buff, Blen, timeout,
true);
1279 if (rlen <= 0)
break;
1281 if (rlen < Blen)
break;
1302 if (retc < 1)
return (retc ? -1 : -ETIMEDOUT);
1317 ssize_t bytesleft = Blen;
1333 {retc =
tlsIO.
Write(Buff, bytesleft, byteswritten);
1335 bytesleft -= byteswritten; Buff += byteswritten;
1363 for (
int i = 0; i < iocnt; i++)
1364 {ssize_t bytesleft =
iov[i].iov_len;
1365 char *Buff = (
char *)
iov[i].iov_base;
1367 {retc =
tlsIO.
Write(Buff, bytesleft, byteswritten);
1369 bytesleft -= byteswritten; Buff += byteswritten;
1383 int bytes, buffsz, fileFD, retc;
1392 for (
int i = 0; i < sfN; sfP++, i++)
1393 {
if (!(bytes = sfP->
sendsz))
continue;
1396 {
if (!
TLS_Write(sfP->buffer, bytes))
return -1;
1399 offset = sfP->offset;
1400 fileFD = sfP->
fdnum;
1401 buffsz = (bytes < (int)
sizeof(myBuff) ? bytes :
sizeof(myBuff));
1402 do {
do {retc =
pread(fileFD, myBuff, buffsz, offset);}
1403 while(retc < 0 && errno == EINTR);
1404 if (retc < 0)
return SFError(errno);
1406 if (!
TLS_Write(myBuff, buffsz))
return -1;
1407 offset += buffsz; bytes -= buffsz; totamt += retc;
1429 {retc =
tlsIO.
Write(Buff, Blen, byteswritten);
1434 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.