XrdSysIOEvents.hh

Go to the documentation of this file.

#ifndef __XRDSYSIOEVENTS_HH__
#define __XRDSYSIOEVENTS_HH__
/******************************************************************************/
/*                                                                            */
/*                     X r d S y s I O E v e n t s . h h                      */
/*                                                                            */
/* (c) 2012 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 <poll.h>
#include <time.h>
#include <sys/types.h>

#include "XrdSys/XrdSysPthread.hh"
#include "XrdSys/XrdSysAtomics.hh"

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

namespace XrdSys
{
namespace IOEvents
{

/******************************************************************************/
/*                        C l a s s   C a l l B a c k                         */
/******************************************************************************/

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
  
class Channel;
class CallBack
{
public:

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

      enum EventType
      {
        ReadyToRead  = 0x01,  
        ReadTimeOut  = 0x02,  
        ReadyToWrite = 0x04,  
        WriteTimeOut = 0x08,  
        ValidEvents  = 0x0f   
      };

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

virtual bool Event(Channel *chP, void *cbArg, int evFlags) = 0;

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

virtual void Fatal(Channel *chP, void *cbArg, int eNum, const char *eTxt)
{
  (void)chP; (void)cbArg; (void)eNum; (void)eTxt;
};

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

virtual void Stop(Channel *chP, void *cbArg) { (void)chP; (void)cbArg;}

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

            CallBack() {}

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

virtual    ~CallBack() {}
};

/******************************************************************************/
/*                         C l a s s   C h a n n e l                          */
/******************************************************************************/

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------
  
class ChannelWait;
class Poller;
class Channel
{
friend class Poller;
public:

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        void Delete();

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

enum EventCode {readEvents  = 0x01, 
                writeEvents = 0x04, 
                rwEvents    = 0x05, 
                errorEvents = 0x10, 
                stopEvent   = 0x20, 
                allEvents   = 0x35  
               };

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        bool Disable(int events, const char **eText=0);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        bool Enable(int events, int timeout=0, const char **eText=0);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        void GetCallBack(CallBack **cbP, void **cbArg);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

inline  int  GetEvents() {return (chPoller ? static_cast<int>(chEvents) : -1);}

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

inline  int  GetFD() {return chFD;}

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        void SetCallBack(CallBack *cbP, void *cbArg=0);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

        void SetFD(int fd);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

      Channel(Poller *pollP, int fd, CallBack *cbP=0, void *cbArg=0);

private:

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

     ~Channel() {}

struct dlQ {Channel *next; Channel *prev;};

XrdSysRecMutex chMutex;

dlQ            attList;     // List of attached channels
dlQ            tmoList;     // List of channels in the timeout queue

Poller        *chPoller;    // The effective poller
Poller        *chPollXQ;    // The real      poller
CallBack      *chCB;        // CallBack function
void          *chCBA;       // CallBack argument
int            chFD;        // Associated file descriptor
int            pollEnt;     // Used only for poll() type pollers
int            chRTO;       // Read  timeout value (0 means none)
int            chWTO;       // Write timeout value (0 means none)
time_t         rdDL;        // Read  deadline
time_t         wrDL;        // Write deadline
time_t         deadLine;    // The deadline in effect (read or write)
char           dlType;      // The deadline type in deadLine as CallBack type
char           chEvents;    // Enabled events as Channel type
char           chStat;      // Channel status below (!0 -> in callback mode)
enum Status   {isClear = 0, isCBMode, isDead};
char           inTOQ;       // True if the channel is in the timeout queue
char           inPSet;      // FD is in the actual poll set
char           reMod;       // Modify issued while defered, re-issue needed
short          chFault;     // Defered error, 0 if all is well

void           Reset(Poller *thePoller, int fd, int eNum=0);
};

/******************************************************************************/
/*                          C l a s s   P o l l e r                           */
/******************************************************************************/
  
//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

class Poller
{
friend class BootStrap;
friend class Channel;
public:

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

enum   CreateOpts  {optTOM};

static Poller     *Create(int &eNum, const char **eTxt=0, int crOpts=0);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

       void        Stop();

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

         Poller(int cFD, int rFD);

//-----------------------------------------------------------------------------
//-----------------------------------------------------------------------------

virtual ~Poller() {}

protected:
struct  PipeData;

        void  CbkTMO();
        bool  CbkXeq(Channel *cP, int events, int eNum, const char *eTxt);
inline  int   GetFault(Channel *cP)   {return cP->chFault;}
inline  int   GetPollEnt(Channel *cP) {return cP->pollEnt;}
        int   GetRequest();
        bool  Init(Channel *cP, int &eNum, const char **eTxt, bool &isLockd);
inline  void  LockChannel(Channel *cP) {cP->chMutex.Lock();}
        int   Poll2Enum(short events);
        int   SendCmd(PipeData &cmd);
        void  SetPollEnt(Channel *cP, int ptEnt);
        bool  TmoAdd(Channel *cP, int tmoSet);
        void  TmoDel(Channel *cP);
        int   TmoGet();
inline  void  UnLockChannel(Channel *cP) {cP->chMutex.UnLock();}

virtual void Begin(XrdSysSemaphore *syncp, int &rc, const char **eTxt) = 0;

virtual void Exclude(Channel *cP, bool &isLocked, bool dover=1) = 0;

virtual bool Include(Channel     *cP,
                     int         &eNum,
                     const char **eTxt,
                     bool        &isLocked) = 0;

virtual bool Modify (Channel     *cP,
                     int         &eNum,
                     const char **eTxt,
                     bool        &isLocked) = 0;

//
virtual void Shutdown() = 0;

// The following is common to all implementations
//
Channel        *attBase;    // -> First channel in attach  queue or 0
Channel        *tmoBase;    // -> First channel in timeout queue or 0

pthread_t       pollTid;    // Poller's thread ID

struct pollfd   pipePoll;   // Stucture to wait for pipe events
int             cmdFD;      // FD to send PipeData commands
int             reqFD;      // FD to recv PipeData requests
struct          PipeData {char req; char evt; short ent; int fd;
                          XrdSysSemaphore *theSem;
                          enum cmd {NoOp = 0, MdFD = 1, Post = 2,
                                    MiFD = 3, RmFD = 4, Stop = 5};
                          PipeData(char reQ=0, char evT=0, short enT=0,
                                   int  fD =0, XrdSysSemaphore *sP=0)
                                  : req(reQ), evt(evT), ent(enT), fd(fD),
                                    theSem(sP) {}
                         ~PipeData() {}
                         };
PipeData        reqBuff;    // Buffer used by poller thread to recv data
char           *pipeBuff;   // Read resumption point in buffer
int             pipeBlen;   // Number of outstanding bytes
unsigned char   tmoMask;    // Timeout mask
CPP_ATOMIC_TYPE(bool) wakePend;   // Wakeup is effectively pending (don't send)
bool            chDead;     // True if channel deleted by callback

static time_t   maxTime;    // Maximum time allowed

private:

void Attach(Channel *cP);
void Detach(Channel *cP, bool &isLocked, bool keep=true);
void WakeUp();

// newPoller() called to get a specialized new poll object at in response to
//             Create(). A specialized implementation must be supplied.
//
static Poller *newPoller(int pFD[2], int &eNum, const char **eTxt);

XrdSysMutex    adMutex; // Mutex for adding & detaching channels
XrdSysMutex    toMutex; // Mutex for handling the timeout list
};
};
};
#endif