Loading...
Searching...
No Matches
XrdCmsClientMan.cc
Go to the documentation of this file.
/******************************************************************************/
/* */
/* X r d C m s C l i e n t M a n . c c */
/* */
/* (c) 2007 by the Board of Trustees of the Leland Stanford, Jr., University */
/* All Rights Reserved */
/* Produced by Andrew Hanushevsky for Stanford University under contract */
/* DE-AC02-76-SFO0515 with the Department of Energy */
/* */
/* This file is part of the XRootD software suite. */
/* */
/* XRootD is free software: you can redistribute it and/or modify it under */
/* the terms of the GNU Lesser General Public License as published by the */
/* Free Software Foundation, either version 3 of the License, or (at your */
/* option) any later version. */
/* */
/* XRootD is distributed in the hope that it will be useful, but WITHOUT */
/* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or */
/* FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public */
/* License for more details. */
/* */
/* You should have received a copy of the GNU Lesser General Public License */
/* along with XRootD in a file called COPYING.LESSER (LGPL license) and file */
/* COPYING (GPL license). If not, see <http://www.gnu.org/licenses/>. */
/* */
/* The copyright holder's institutional names and contributor's names may not */
/* be used to endorse or promote products derived from this software without */
/* specific prior written permission of the institution or contributor. */
/******************************************************************************/
#include <ctime>
#include "XrdCms/XrdCmsClientMan.hh"
#include "XrdCms/XrdCmsClientMsg.hh"
#include "XrdCms/XrdCmsLogin.hh"
#include "XrdCms/XrdCmsTrace.hh"
#include "XrdSfs/XrdSfsInterface.hh"
#include "XrdSys/XrdSysError.hh"
#include "XrdSys/XrdSysTimer.hh"
#include "Xrd/XrdInet.hh"
#include "Xrd/XrdLink.hh"
using namespace XrdCms;
/******************************************************************************/
/* G l o b a l s */
/******************************************************************************/
XrdOucBuffPool XrdCmsClientMan::BuffPool(XrdOucEI::Max_Error_Len, 65536, 1, 16);
XrdInet *XrdCmsClientMan::Network = 0;
char XrdCmsClientMan::doDebug = 0;
const char *XrdCmsClientMan::ConfigFN = 0;
XrdSysMutex XrdCmsClientMan::manMutex;
/******************************************************************************/
/* C o n s t r u c t o r */
/******************************************************************************/
XrdCmsClientMan::XrdCmsClientMan(char *host, int port,
int cw, int nr, int rw, int rd)
: syncResp(0)
{
static XrdSysMutex initMutex;
static int Instance = 0;
char *dot;
Host = strdup(host);
if ((dot = index(Host, '.')))
{*dot = '\0'; HPfx = strdup(Host); *dot = '.';}
else HPfx = strdup(Host);
Port = port;
Link = 0;
Active = 0;
Silent = 0;
Suspend = 1;
RecvCnt = 0;
nrMax = nr;
NetBuff = BuffPool.Alloc(XrdOucEI::Max_Error_Len);
repWMax = rw;
repWait = 0;
minDelay= rd;
maxDelay= rd*3;
chkCount= chkVal;
lastUpdt= lastTOut = time(0);
Next = 0;
manInst = 1;
// Compute dally value
//
dally = cw / 2 - 1;
if (dally < 3) dally = 3;
else if (dally > 10) dally = 10;
// Provide a unique mask number for this manager
//
initMutex.Lock();
manMask = 1<<Instance++;
initMutex.UnLock();
}
/******************************************************************************/
/* D e s t r u c t o r */
/******************************************************************************/
XrdCmsClientMan::~XrdCmsClientMan()
{
if (Link) Link->Close();
if (Host) free(Host);
if (HPfx) free(HPfx);
if (NetBuff) NetBuff->Recycle();
}
/******************************************************************************/
/* d e l a y R e s p */
/******************************************************************************/
int XrdCmsClientMan::delayResp(XrdOucErrInfo &Resp)
{
XrdCmsResp *rp;
int msgid;
// Obtain the message ID
//
if (!(msgid = Resp.getErrInfo()))
{Say.Emsg("Manager", Host, "supplied invalid waitr msgid");
Resp.setErrInfo(EILSEQ, "redirector protocol error");
syncResp.Post();
return SFS_ERROR;
}
// Allocate a delayed response object
//
if (!(rp = XrdCmsResp::Alloc(&Resp, msgid)))
{Say.Emsg("Manager",ENOMEM,"allocate resp object for",Resp.getErrUser());
Resp.setErrInfo(0, "0");
syncResp.Post();
return SFS_STALL;
}
// Add this object to our delayed response queue. If the manager bounced then
// purge all of the pending repsonses to avoid sending wrong ones.
//
if (msgid < maxMsgID) RespQ.Purge();
maxMsgID = msgid;
RespQ.Add(rp);
// Tell client to wait for response. The semaphore post allows the manager
// to get the next message from the cmsd. This prevents us from getting the
// delayed response before the response object is added to the queue.
//
Resp.setErrInfo(0, "");
syncResp.Post();
return SFS_STARTED;
}
/******************************************************************************/
/* S e n d */
/******************************************************************************/
int XrdCmsClientMan::Send(unsigned int &iMan, char *msg, int mlen)
{
int allok = 0;
// Determine message length
//
if (!mlen) mlen = strlen(msg);
// Send the request
//
myData.Lock();
iMan = manInst;
if (Active)
{if (Link)
{if (!(allok = Link->Send(msg, mlen) > 0))
{Active = 0;
Link->Close(1);
manInst++;
} else SendCnt++;
}
}
myData.UnLock();
// All done
//
return allok;
}
/******************************************************************************/
int XrdCmsClientMan::Send(unsigned int &iMan, const struct iovec *iov, int iovcnt, int iotot)
{
int allok = 0;
// Send the request
//
myData.Lock();
iMan = manInst;
if (Active)
{if (Link)
{if (!(allok = Link->Send(iov, iovcnt, iotot) > 0))
{Active = 0;
Link->Close(1);
manInst++;
} else SendCnt++;
}
}
myData.UnLock();
// All done
//
return allok;
}
/******************************************************************************/
/* S t a r t */
/******************************************************************************/
void *XrdCmsClientMan::Start()
{
// First step is to connect to the manager
//
do {Hookup();
// Now simply start receiving messages on the stream. When we get a
// respwait reply then we must be assured that the object representing
// the request is added to the queue before the actual reply arrives.
// We do this by waiting on syncResp which is posted once the request
// object is fully processed. The actual response associated with the
// respwait is synchronized during the callback phase since the client
// must receive the respwait before the subsequent response.
//
while(Receive())
if (Response.modifier & CmsResponse::kYR_async) relayResp();
else if (Response.rrCode == kYR_status) setStatus();
else if (XrdCmsClientMsg::Reply(HPfx, Response, NetBuff))
{if (Response.rrCode == kYR_waitresp) syncResp.Wait();}
// Tear down the connection
//
myData.Lock();
if (Link) {Link->Close(); Link = 0;}
Active = 0; Suspend = 1;
myData.UnLock();
// Indicate the problem
//
Say.Emsg("ClientMan", "Disconnected from", Host);
XrdSysTimer::Snooze(dally);
} while(1);
// We should never get here
//
return (void *)0;
}
/******************************************************************************/
/* w h a t s U p */
/******************************************************************************/
int XrdCmsClientMan::whatsUp(const char *user, const char *path,
unsigned int iMan)
{
EPNAME("whatsUp");
unsigned int xMan;
int theDelay, inQ;
bool lClose = false;
// The cmsd did not respond. Increase silent count and see if restart is needed
// Otherwise, increase the wait interval just in case things are just slow.
//
myData.Lock();
if (Active)
{if (Active == RecvCnt)
{if ((time(0)-lastTOut) >= repWait)
{Silent++;
if (Silent > nrMax)
{Active = 0; Silent = 0; Suspend = 1;
if (Link && iMan == manInst)
{Link->Close(1);
manInst++; lClose = true;
}
} else if (Silent & 0x02 && repWait < repWMax) repWait++;
}
} else {Active = RecvCnt; Silent = 0; lastTOut = time(0);}
}
// Calclulate how long to delay the client. This will be based on the number
// of outstanding requests bounded by the config delay value.
//
inQ = XrdCmsClientMsg::inQ();
theDelay = inQ * qTime;
xMan = manInst;
myData.UnLock();
theDelay = theDelay/1000 + (theDelay % 1000 ? 1 : 0);
if (theDelay < minDelay) theDelay = minDelay;
if (theDelay > maxDelay) theDelay = maxDelay;
// Do Some tracing here
//
TRACE(Redirect, user <<" no resp from inst " <<iMan <<" of "<<HPfx
<<" in " <<repWait
<<" inst " <<xMan <<(lClose ? " closed" : " steady")
<<"; inQ " <<inQ <<" delay " <<theDelay <<" path=" <<path);
return theDelay;
}
/******************************************************************************/
/* P r i v a t e M e t h o d s */
/******************************************************************************/
/******************************************************************************/
/* H o o k u p */
/******************************************************************************/
int XrdCmsClientMan::Hookup()
{
EPNAME("Hookup");
CmsLoginData Data;
XrdLink *lp;
char buff[256], hnBuff[264];
kXR_char *envData = 0;
int rc, oldWait, tries = 12, opts = 0;
// Turn off our debugging and version flags
//
manMutex.Lock();
doDebug &= ~manMask;
manMutex.UnLock();
// Report our hostname (there are better ways of doing this)
//
const char *hn = getenv("XRDHOST");
if (hn)
{snprintf(hnBuff, sizeof(hnBuff), "myHN=%s", hn);
envData = (kXR_char *)hnBuff;
}
// Keep trying to connect to the manager. Note that we bind the link to this
// thread to make sure we get notified should another thread close the socket.
//
do {while(!(lp = Network->Connect(Host, Port, opts)))
{XrdSysTimer::Snooze(dally);
if (tries--) opts = XRDNET_NOEMSG;
else {opts = 0; tries = 12;}
continue;
}
// lp->Bind(XrdSysThread::ID());
memset(&Data, 0, sizeof(Data));
Data.envCGI = envData;
Data.Mode = CmsLoginData::kYR_director;
Data.HoldTime = static_cast<int>(getpid());
if (!(rc = XrdCmsLogin::Login(lp, Data))) break;
lp->Close();
XrdSysTimer::Snooze(dally);
} while(1);
// Establish global state
//
manMutex.Lock();
doDebug |= (Data.Mode & CmsLoginData::kYR_debug ? manMask : 0);
manMutex.UnLock();
// All went well, finally
//
myData.Lock();
Link = lp;
Active = 1;
Silent = 0;
RecvCnt = 1;
SendCnt = 1;
Suspend = (Data.Mode & CmsLoginData::kYR_suspend);
// Calculate how long we will wait for replies before delaying the client.
// This is computed dynamically based on the expected response window.
//
if ((oldWait = (repWait*20/100)) < 2) oldWait = 2;
if (Data.HoldTime > repWMax*1000) repWait = repWMax;
else if (Data.HoldTime <= 0) repWait = repWMax;
else {repWait = Data.HoldTime*3;
repWait = (repWait/1000) + (repWait % 1000 ? 1 : 0);
if (repWait > repWMax) repWait = repWMax;
else if (repWait < oldWait) repWait = oldWait;
}
qTime = (Data.HoldTime < 100 ? 100 : Data.HoldTime);
lastTOut = time(0);
myData.UnLock();
// Tell the world
//
sprintf(buff, "v %d", Data.Version);
Say.Emsg("ClientMan", (Suspend ? "Connected to suspended" : "Connected to"),
Host, buff);
DEBUG(Host <<" qt=" <<qTime <<"ms rw=" <<repWait);
return 1;
}
/******************************************************************************/
/* R e c e i v e */
/******************************************************************************/
int XrdCmsClientMan::Receive()
{
// This method is always run out of the object's main thread. Other threads
// may call methods that initiate a link reset via a deferred close. We will
// notice that here because the file descriptor will be closed. This will
// cause us to return an error and precipitate a connection teardown.
//
EPNAME("Receive")
if (Link->RecvAll((char *)&Response, sizeof(Response)) > 0)
{int dlen = static_cast<int>(ntohs(Response.datalen));
RecvCnt++;
DEBUG(Link->Name() <<' ' <<dlen <<" bytes on " <<Response.streamid);
if (!dlen) return 1;
if ((dlen > NetBuff->BuffSize())
&& (Response.rrCode != kYR_data || !NetBuff->Resize(dlen)))
Say.Emsg("ClientMan", "Excessive msg length from", Host);
else {NetBuff->SetLen(dlen);
return Link->RecvAll(NetBuff->Buffer(), dlen);
}
}
return 0;
}
/******************************************************************************/
/* r e l a y R e s p */
/******************************************************************************/
void XrdCmsClientMan::relayResp()
{
EPNAME("relayResp");
XrdCmsResp *rp;
// Remove the response object from our queue.
//
if (!(rp = RespQ.Rem(Response.streamid)))
{DEBUG(Host <<" replied to non-existent request; id=" <<Response.streamid);
return;
}
// Queue the request for reply (this transfers the network buffer)
//
rp->Reply(HPfx, Response, NetBuff);
// Obtain a new network buffer
//
NetBuff = BuffPool.Alloc(XrdOucEI::Max_Error_Len);
}
/******************************************************************************/
/* Private: c h k S t a t u s */
/******************************************************************************/
int XrdCmsClientMan::chkStatus()
{
static CmsUpdateRequest Updt = {{0, kYR_update, 0, 0}};
XrdSysMutexHelper mdMon(myData);
time_t nowTime;
// Count down the query count and ask again every 30 seconds
//
if (!chkCount--)
{chkCount = chkVal;
nowTime = time(0);
if ((nowTime - lastUpdt) >= 30)
{lastUpdt = nowTime;
if (Active) Link->Send((char *)&Updt, sizeof(Updt));
}
}
return Suspend;
}
/******************************************************************************/
/* s e t S t a t u s */
/******************************************************************************/
void XrdCmsClientMan::setStatus()
{
EPNAME("setStatus");
const char *State = 0, *Event = "?";
myData.Lock();
if (Response.modifier & CmsStatusRequest::kYR_Suspend)
{Event = "suspend";
if (!Suspend) {Suspend = 1; State = "suspended";}
}
else if (Response.modifier & CmsStatusRequest::kYR_Resume)
{Event = "resume";
if (Suspend) {Suspend = 0; State = "resumed";}
}
myData.UnLock();
DEBUG(Host <<" sent " <<Event <<" event");
if (State) Say.Emsg("setStatus", "Manager", Host, State);
}
kXR_char
unsigned char kXR_char
Definition XPtypes.hh:65
DEBUG
#define DEBUG(x)
Definition XrdBwmTrace.hh:54
EPNAME
#define EPNAME(x)
Definition XrdBwmTrace.hh:56
XrdCmsClientMan.hh
XrdCmsClientMsg.hh
XrdCmsLogin.hh
XrdCmsTrace.hh
XrdInet.hh
XrdLink.hh
Updt
#define Updt(x)
Definition XrdMonitor.cc:102
XRDNET_NOEMSG
#define XRDNET_NOEMSG
Definition XrdNetOpts.hh:71
opts
struct myOpts opts
XrdSfsInterface.hh
SFS_ERROR
#define SFS_ERROR
Definition XrdSfsInterface.hh:109
SFS_STALL
#define SFS_STALL
Definition XrdSfsInterface.hh:107
SFS_STARTED
#define SFS_STARTED
Definition XrdSfsInterface.hh:111
XrdSysError.hh
if
if(ec< 0) ec
XrdSysTimer.hh
TRACE
#define TRACE(act, x)
Definition XrdTrace.hh:63
XrdCmsClientMan::Start
void * Start()
Definition XrdCmsClientMan.cc:225
XrdCmsClientMan::Send
int Send(unsigned int &iMan, char *msg, int mlen=0)
Definition XrdCmsClientMan.cc:167
XrdCmsClientMan::delayResp
int delayResp(XrdOucErrInfo &Resp)
Definition XrdCmsClientMan.cc:124
XrdCmsClientMan::~XrdCmsClientMan
~XrdCmsClientMan()
Definition XrdCmsClientMan.cc:112
XrdCmsClientMan::whatsUp
int whatsUp(const char *user, const char *path, unsigned int iMan)
Definition XrdCmsClientMan.cc:267
XrdCmsClientMan::doDebug
static char doDebug
Definition XrdCmsClientMan.hh:51
XrdCmsClientMan::XrdCmsClientMan
XrdCmsClientMan(char *host, int port, int cw, int nr, int rw, int rd)
Definition XrdCmsClientMan.cc:66
XrdCmsClientMsg::inQ
static int inQ()
Definition XrdCmsClientMsg.hh:52
XrdCmsClientMsg::Reply
static int Reply(const char *Man, XrdCms::CmsRRHdr &hdr, XrdOucBuffer *buff)
Definition XrdCmsClientMsg.cc:142
XrdCmsLogin::Login
static int Login(XrdLink *Link, XrdCms::CmsLoginData &Data, int timeout=-1)
Definition XrdCmsLogin.cc:125
XrdCmsResp
Definition XrdCmsResp.hh:70
XrdCmsResp::Reply
void Reply(const char *Man, XrdCms::CmsRRHdr &rrhdr, XrdOucBuffer *netbuff)
Definition XrdCmsResp.cc:138
XrdCmsResp::Alloc
static XrdCmsResp * Alloc(XrdOucErrInfo *erp, int msgid)
Definition XrdCmsResp.cc:64
XrdInet
Definition XrdInet.hh:48
XrdLink
Definition XrdLink.hh:52
XrdLink::Close
int Close(bool defer=false)
Definition XrdLink.cc:167
XrdOucBuffPool
Definition XrdOucBuffer.hh:55
XrdOucErrInfo
Definition XrdOucErrInfo.hh:101
XrdOucErrInfo::getErrUser
const char * getErrUser()
Definition XrdOucErrInfo.hh:297
XrdOucErrInfo::setErrInfo
int setErrInfo(int code, const char *emsg)
Definition XrdOucErrInfo.hh:147
XrdOucErrInfo::getErrInfo
int getErrInfo()
Definition XrdOucErrInfo.hh:243
XrdSysError::Emsg
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
Definition XrdSysError.cc:95
XrdSysMutex
Definition XrdSysPthread.hh:165
XrdSysMutex::Lock
void Lock()
Definition XrdSysPthread.hh:222
XrdSysMutex::UnLock
void UnLock()
Definition XrdSysPthread.hh:224
XrdSysTimer::Snooze
static void Snooze(int seconds)
Definition XrdSysTimer.cc:168
XrdCms
Definition YProtocol.hh:78
XrdCms::kYR_data
@ kYR_data
Definition YProtocol.hh:141
XrdCms::kYR_waitresp
@ kYR_waitresp
Definition YProtocol.hh:145
XrdCms::Say
XrdSysError Say
Definition XrdCmsManager.cc:62
XrdCms::kYR_update
@ kYR_update
Definition YProtocol.hh:115
XrdCms::kYR_status
@ kYR_status
Definition YProtocol.hh:112
XrdCms::CmsLoginData
Definition YProtocol.hh:277
XrdCms::CmsLoginData::Mode
kXR_unt32 Mode
Definition YProtocol.hh:279
XrdCms::CmsLoginData::HoldTime
kXR_int32 HoldTime
Definition YProtocol.hh:280
XrdCms::CmsLoginData::envCGI
kXR_char * envCGI
Definition YProtocol.hh:291
XrdCms::CmsLoginData::kYR_director
@ kYR_director
Definition YProtocol.hh:294
XrdCms::CmsLoginData::kYR_suspend
@ kYR_suspend
Definition YProtocol.hh:301
XrdCms::CmsLoginData::kYR_debug
@ kYR_debug
Definition YProtocol.hh:304
XrdCms::CmsLoginData::Version
kXR_unt16 Version
Definition YProtocol.hh:278
XrdCms::CmsResponse::kYR_async
@ kYR_async
Definition YProtocol.hh:170
XrdCms::CmsStatusRequest::kYR_Resume
@ kYR_Resume
Definition YProtocol.hh:597
XrdCms::CmsStatusRequest::kYR_Suspend
@ kYR_Suspend
Definition YProtocol.hh:597
XrdOucEI::Max_Error_Len
static const size_t Max_Error_Len
Definition XrdOucErrInfo.hh:51
iov
Definition XrdOucIOVec.hh:65