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