src/mgr/Inquirer.cc

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