src/DiskIO/IpcIo/IpcIoFile.h

   1 #ifndef SQUID_IPC_IOFILE_H
   2 #define SQUID_IPC_IOFILE_H
   3
   4 #include "base/AsyncCall.h"
   5 #include "cbdata.h"
   6 #include "DiskIO/DiskFile.h"
   7 #include "DiskIO/IORequestor.h"
   8 #include "ipc/forward.h"
   9 #include "ipc/mem/Page.h"
  10 #include "SquidString.h"
  11 #include <list>
  12 #include <map>
  13 #include <memory>
  14
  15 namespace Ipc
  16 {
  17 class FewToFewBiQueue;
  18 } // Ipc
  19
  20 // TODO: expand to all classes
  21 namespace IpcIo
  22 {
  23
  24 /// what kind of I/O the disker needs to do or have done
  25 typedef enum { cmdNone, cmdOpen, cmdRead, cmdWrite } Command;
  26
  27 } // namespace IpcIo
  28
  29 /// converts DiskIO requests to IPC queue messages
  30 class IpcIoMsg
  31 {
  32 public:
  33     IpcIoMsg();
  34
  35 public:
  36     unsigned int requestId; ///< unique for requestor; matches request w/ response
  37
  38     off_t offset;
  39     size_t len;
  40     Ipc::Mem::PageId page;
  41
  42     IpcIo::Command command; ///< what disker is supposed to do or did
  43     struct timeval start; ///< when the I/O request was converted to IpcIoMsg
  44
  45     int xerrno; ///< I/O error code or zero
  46 };
  47
  48 class IpcIoPendingRequest;
  49
  50 class IpcIoFile: public DiskFile
  51 {
  52
  53 public:
  54     typedef RefCount<IpcIoFile> Pointer;
  55
  56     IpcIoFile(char const *aDb);
  57     virtual ~IpcIoFile();
  58
  59     /* DiskFile API */
  60     virtual void configure(const Config &cfg);
  61     virtual void open(int flags, mode_t mode, RefCount<IORequestor> callback);
  62     virtual void create(int flags, mode_t mode, RefCount<IORequestor> callback);
  63     virtual void read(ReadRequest *);
  64     virtual void write(WriteRequest *);
  65     virtual void close();
  66     virtual bool error() const;
  67     virtual int getFD() const;
  68     virtual bool canRead() const;
  69     virtual bool canWrite() const;
  70     virtual bool ioInProgress() const;
  71
  72     /// handle open response from coordinator
  73     static void HandleOpenResponse(const Ipc::StrandSearchResponse &response);
  74
  75     /// handle queue push notifications from worker or disker
  76     static void HandleNotification(const Ipc::TypedMsgHdr &msg);
  77
  78     DiskFile::Config config; ///< supported configuration options
  79
  80 protected:
  81     friend class IpcIoPendingRequest;
  82     void openCompleted(const Ipc::StrandSearchResponse *const response);
  83     void readCompleted(ReadRequest *readRequest, IpcIoMsg *const response);
  84     void writeCompleted(WriteRequest *writeRequest, const IpcIoMsg *const response);
  85     bool canWait() const;
  86
  87 private:
  88     void trackPendingRequest(const unsigned int id, IpcIoPendingRequest *const pending);
  89     void push(IpcIoPendingRequest *const pending);
  90     IpcIoPendingRequest *dequeueRequest(const unsigned int requestId);
  91
  92     static void Notify(const int peerId);
  93
  94     static void OpenTimeout(void *const param);
  95     static void CheckTimeouts(void *const param);
  96     void checkTimeouts();
  97     void scheduleTimeoutCheck();
  98
  99     static void HandleResponses(const char *const when);
 100     void handleResponse(IpcIoMsg &ipcIo);
 101
 102     static void DiskerHandleMoreRequests(void*);
 103     static void DiskerHandleRequests();
 104     static void DiskerHandleRequest(const int workerId, IpcIoMsg &ipcIo);
 105     static bool WaitBeforePop();
 106
 107 private:
 108     const String dbName; ///< the name of the file we are managing
 109     int diskId; ///< the process ID of the disker we talk to
 110     RefCount<IORequestor> ioRequestor;
 111
 112     bool error_; ///< whether we have seen at least one I/O error (XXX)
 113
 114     unsigned int lastRequestId; ///< last requestId used
 115
 116     /// maps requestId to the handleResponse callback
 117     typedef std::map<unsigned int, IpcIoPendingRequest*> RequestMap;
 118     RequestMap requestMap1; ///< older (or newer) pending requests
 119     RequestMap requestMap2; ///< newer (or older) pending requests
 120     RequestMap *olderRequests; ///< older requests (map1 or map2)
 121     RequestMap *newerRequests; ///< newer requests (map2 or map1)
 122     bool timeoutCheckScheduled; ///< we expect a CheckTimeouts() call
 123
 124     static const double Timeout; ///< timeout value in seconds
 125
 126     typedef std::list<Pointer> IpcIoFileList;
 127     static IpcIoFileList WaitingForOpen; ///< pending open requests
 128
 129     ///< maps diskerId to IpcIoFile, cleared in destructor
 130     typedef std::map<int, IpcIoFile*> IpcIoFilesMap;
 131     static IpcIoFilesMap IpcIoFiles;
 132
 133     typedef Ipc::FewToFewBiQueue Queue;
 134     static std::auto_ptr<Queue> queue; ///< IPC queue
 135
 136     /// whether we are waiting for an event to handle still queued I/O requests
 137     static bool DiskerHandleMoreRequestsScheduled;
 138
 139     CBDATA_CLASS2(IpcIoFile);
 140 };
 141
 142 /// keeps original I/O request parameters while disker is handling the request
 143 class IpcIoPendingRequest
 144 {
 145 public:
 146     IpcIoPendingRequest(const IpcIoFile::Pointer &aFile);
 147
 148     /// called when response is received and, with a nil response, on timeouts
 149     void completeIo(IpcIoMsg *const response);
 150
 151 public:
 152     const IpcIoFile::Pointer file; ///< the file object waiting for the response
 153     ReadRequest *readRequest; ///< set if this is a read requests
 154     WriteRequest *writeRequest; ///< set if this is a write request
 155
 156 private:
 157     IpcIoPendingRequest(const IpcIoPendingRequest &d); // not implemented
 158     IpcIoPendingRequest &operator =(const IpcIoPendingRequest &d); // ditto
 159 };
 160
 161 #endif /* SQUID_IPC_IOFILE_H */