src/mgr/Inquirer.cc

   1 /*
   2  * DEBUG: section 16    Cache Manager API
   3  *
   4  */
   5
   6 #include "squid.h"
   7 #include "base/TextException.h"
   8 #include "comm.h"
   9 #include "comm/Connection.h"
  10 #include "comm/Write.h"
  11 #include "CommCalls.h"
  12 #include "HttpReply.h"
  13 #include "HttpRequest.h"
  14 #include "ipc/UdsOp.h"
  15 #include "mgr/ActionWriter.h"
  16 #include "mgr/IntParam.h"
  17 #include "mgr/Inquirer.h"
  18 #include "mgr/Command.h"
  19 #include "mgr/Request.h"
  20 #include "mgr/Response.h"
  21 #include "SquidTime.h"
  22 #include "errorpage.h"
  23 #include <memory>
  24 #include <algorithm>
  25
  26 CBDATA_NAMESPACED_CLASS_INIT(Mgr, Inquirer);
  27
  28 Mgr::Inquirer::Inquirer(Action::Pointer anAction,
  29                         const Request &aCause, const Ipc::StrandCoords &coords):
  30         Ipc::Inquirer(aCause.clone(), applyQueryParams(coords, aCause.params.queryParams), anAction->atomic() ? 10 : 100),
  31         aggrAction(anAction)
  32 {
  33     conn = aCause.conn;
  34     Ipc::ImportFdIntoComm(conn, SOCK_STREAM, IPPROTO_TCP, Ipc::fdnHttpSocket);
  35
  36     debugs(16, 5, HERE << conn << " action: " << aggrAction);
  37
  38     closer = asyncCall(16, 5, "Mgr::Inquirer::noteCommClosed",
  39                        CommCbMemFunT<Inquirer, CommCloseCbParams>(this, &Inquirer::noteCommClosed));
  40     comm_add_close_handler(conn->fd, closer);
  41 }
  42
  43 /// closes our copy of the client HTTP connection socket
  44 void
  45 Mgr::Inquirer::cleanup()
  46 {
  47     if (Comm::IsConnOpen(conn)) {
  48         removeCloseHandler();
  49         conn->close();
  50     }
  51 }
  52
  53 void
  54 Mgr::Inquirer::removeCloseHandler()
  55 {
  56     if (closer != NULL) {
  57         comm_remove_close_handler(conn->fd, closer);
  58         closer = NULL;
  59     }
  60 }
  61
  62 void
  63 Mgr::Inquirer::start()
  64 {
  65     debugs(16, 5, HERE);
  66     Ipc::Inquirer::start();
  67     Must(Comm::IsConnOpen(conn));
  68     Must(aggrAction != NULL);
  69
  70 #if HAVE_UNIQUE_PTR
  71     std::unique_ptr<MemBuf> replyBuf;
  72 #else
  73     std::auto_ptr<MemBuf> replyBuf;
  74 #endif
  75     if (strands.empty()) {
  76         LOCAL_ARRAY(char, url, MAX_URL);
  77         snprintf(url, MAX_URL, "%s", aggrAction->command().params.httpUri.termedBuf());
  78         HttpRequest *req = HttpRequest::CreateFromUrl(url);
  79         ErrorState err(ERR_INVALID_URL, Http::scNotFound, req);
  80 #if HAVE_UNIQUE_PTR
  81         std::unique_ptr<HttpReply> reply(err.BuildHttpReply());
  82 #else
  83         std::auto_ptr<HttpReply> reply(err.BuildHttpReply());
  84 #endif
  85         replyBuf.reset(reply->pack());
  86     } else {
  87 #if HAVE_UNIQUE_PTR
  88         std::unique_ptr<HttpReply> reply(new HttpReply);
  89 #else
  90         std::auto_ptr<HttpReply> reply(new HttpReply);
  91 #endif
  92         reply->setHeaders(Http::scOkay, NULL, "text/plain", -1, squid_curtime, squid_curtime);
  93         reply->header.putStr(HDR_CONNECTION, "close"); // until we chunk response
  94         replyBuf.reset(reply->pack());
  95     }
  96     writer = asyncCall(16, 5, "Mgr::Inquirer::noteWroteHeader",
  97                        CommCbMemFunT<Inquirer, CommIoCbParams>(this, &Inquirer::noteWroteHeader));
  98     Comm::Write(conn, replyBuf.get(), writer);
  99 }
 100
 101 /// called when we wrote the response header
 102 void
 103 Mgr::Inquirer::noteWroteHeader(const CommIoCbParams& params)
 104 {
 105     debugs(16, 5, HERE);
 106     writer = NULL;
 107     Must(params.flag == COMM_OK);
 108     Must(params.conn.getRaw() == conn.getRaw());
 109     Must(params.size != 0);
 110     // start inquiries at the initial pos
 111     inquire();
 112 }
 113
 114 /// called when the HTTP client or some external force closed our socket
 115 void
 116 Mgr::Inquirer::noteCommClosed(const CommCloseCbParams& params)
 117 {
 118     debugs(16, 5, HERE);
 119     Must(!Comm::IsConnOpen(conn) && params.conn.getRaw() == conn.getRaw());
 120     conn = NULL;
 121     mustStop("commClosed");
 122 }
 123
 124 bool
 125 Mgr::Inquirer::aggregate(Ipc::Response::Pointer aResponse)
 126 {
 127     Mgr::Response& response = static_cast<Response&>(*aResponse);
 128     if (response.hasAction())
 129         aggrAction->add(response.getAction());
 130     return true;
 131 }
 132
 133 void
 134 Mgr::Inquirer::sendResponse()
 135 {
 136     if (!strands.empty() && aggrAction->aggregatable()) {
 137         removeCloseHandler();
 138         AsyncJob::Start(new ActionWriter(aggrAction, conn));
 139         conn = NULL; // should not close because we passed it to ActionWriter
 140     }
 141 }
 142
 143 bool
 144 Mgr::Inquirer::doneAll() const
 145 {
 146     return !writer && Ipc::Inquirer::doneAll();
 147 }
 148
 149 Ipc::StrandCoords
 150 Mgr::Inquirer::applyQueryParams(const Ipc::StrandCoords& aStrands, const QueryParams& aParams)
 151 {
 152     Ipc::StrandCoords sc;
 153
 154     QueryParam::Pointer processesParam = aParams.get("processes");
 155     QueryParam::Pointer workersParam = aParams.get("workers");
 156
 157     if (processesParam == NULL || workersParam == NULL) {
 158         if (processesParam != NULL) {
 159             IntParam* param = dynamic_cast<IntParam*>(processesParam.getRaw());
 160             if (param != NULL && param->type == QueryParam::ptInt) {
 161                 const std::vector<int>& processes = param->value();
 162                 for (Ipc::StrandCoords::const_iterator iter = aStrands.begin();
 163                         iter != aStrands.end(); ++iter) {
 164                     if (std::find(processes.begin(), processes.end(), iter->kidId) != processes.end())
 165                         sc.push_back(*iter);
 166                 }
 167             }
 168         } else if (workersParam != NULL) {
 169             IntParam* param = dynamic_cast<IntParam*>(workersParam.getRaw());
 170             if (param != NULL && param->type == QueryParam::ptInt) {
 171                 const std::vector<int>& workers = param->value();
 172                 for (int i = 0; i < (int)aStrands.size(); ++i) {
 173                     if (std::find(workers.begin(), workers.end(), i + 1) != workers.end())
 174                         sc.push_back(aStrands[i]);
 175                 }
 176             }
 177         } else {
 178             sc = aStrands;
 179         }
 180     }
 181
 182     debugs(16, 4, HERE << "strands kid IDs = ");
 183     for (Ipc::StrandCoords::const_iterator iter = sc.begin(); iter != sc.end(); ++iter) {
 184         debugs(16, 4, HERE << iter->kidId);
 185     }
 186
 187     return sc;
 188 }