src/PeerPoolMgr.cc

   1 /*
   2  * Copyright (C) 1996-2018 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 #include "squid.h"
  10 #include "AccessLogEntry.h"
  11 #include "base/AsyncJobCalls.h"
  12 #include "base/RunnersRegistry.h"
  13 #include "CachePeer.h"
  14 #include "comm/Connection.h"
  15 #include "comm/ConnOpener.h"
  16 #include "Debug.h"
  17 #include "fd.h"
  18 #include "FwdState.h"
  19 #include "globals.h"
  20 #include "HttpRequest.h"
  21 #include "MasterXaction.h"
  22 #include "neighbors.h"
  23 #include "pconn.h"
  24 #include "PeerPoolMgr.h"
  25 #include "security/BlindPeerConnector.h"
  26 #include "SquidConfig.h"
  27 #include "SquidTime.h"
  28
  29 CBDATA_CLASS_INIT(PeerPoolMgr);
  30
  31 /// Gives Security::PeerConnector access to Answer in the PeerPoolMgr callback dialer.
  32 class MyAnswerDialer: public UnaryMemFunT<PeerPoolMgr, Security::EncryptorAnswer, Security::EncryptorAnswer&>,
  33     public Security::PeerConnector::CbDialer
  34 {
  35 public:
  36     MyAnswerDialer(const JobPointer &aJob, Method aMethod):
  37         UnaryMemFunT<PeerPoolMgr, Security::EncryptorAnswer, Security::EncryptorAnswer&>(aJob, aMethod, Security::EncryptorAnswer()) {}
  38
  39     /* Security::PeerConnector::CbDialer API */
  40     virtual Security::EncryptorAnswer &answer() { return arg1; }
  41 };
  42
  43 PeerPoolMgr::PeerPoolMgr(CachePeer *aPeer): AsyncJob("PeerPoolMgr"),
  44     peer(cbdataReference(aPeer)),
  45     request(),
  46     opener(),
  47     securer(),
  48     closer(),
  49     addrUsed(0)
  50 {
  51 }
  52
  53 PeerPoolMgr::~PeerPoolMgr()
  54 {
  55     cbdataReferenceDone(peer);
  56 }
  57
  58 void
  59 PeerPoolMgr::start()
  60 {
  61     AsyncJob::start();
  62
  63     const MasterXaction::Pointer mx = new MasterXaction(XactionInitiator::initPeerPool);
  64     // ErrorState, getOutgoingAddress(), and other APIs may require a request.
  65     // We fake one. TODO: Optionally send this request to peers?
  66     request = new HttpRequest(Http::METHOD_OPTIONS, AnyP::PROTO_HTTP, "http", "*", mx);
  67     request->url.host(peer->host);
  68
  69     checkpoint("peer initialized");
  70 }
  71
  72 void
  73 PeerPoolMgr::swanSong()
  74 {
  75     AsyncJob::swanSong();
  76 }
  77
  78 bool
  79 PeerPoolMgr::validPeer() const
  80 {
  81     return peer && cbdataReferenceValid(peer) && peer->standby.pool;
  82 }
  83
  84 bool
  85 PeerPoolMgr::doneAll() const
  86 {
  87     return !(validPeer() && peer->standby.limit) && AsyncJob::doneAll();
  88 }
  89
  90 void
  91 PeerPoolMgr::handleOpenedConnection(const CommConnectCbParams &params)
  92 {
  93     opener = NULL;
  94
  95     if (!validPeer()) {
  96         debugs(48, 3, "peer gone");
  97         if (params.conn != NULL)
  98             params.conn->close();
  99         return;
 100     }
 101
 102     if (params.flag != Comm::OK) {
 103         /* it might have been a timeout with a partially open link */
 104         if (params.conn != NULL)
 105             params.conn->close();
 106         peerConnectFailed(peer);
 107         checkpoint("conn opening failure"); // may retry
 108         return;
 109     }
 110
 111     Must(params.conn != NULL);
 112
 113     // Handle TLS peers.
 114     if (peer->secure.encryptTransport) {
 115         typedef CommCbMemFunT<PeerPoolMgr, CommCloseCbParams> CloserDialer;
 116         closer = JobCallback(48, 3, CloserDialer, this,
 117                              PeerPoolMgr::handleSecureClosure);
 118         comm_add_close_handler(params.conn->fd, closer);
 119
 120         securer = asyncCall(48, 4, "PeerPoolMgr::handleSecuredPeer",
 121                             MyAnswerDialer(this, &PeerPoolMgr::handleSecuredPeer));
 122
 123         const int peerTimeout = peerConnectTimeout(peer);
 124         const int timeUsed = squid_curtime - params.conn->startTime();
 125         // Use positive timeout when less than one second is left for conn.
 126         const int timeLeft = positiveTimeout(peerTimeout - timeUsed);
 127         auto *connector = new Security::BlindPeerConnector(request, params.conn, securer, nullptr, timeLeft);
 128         AsyncJob::Start(connector); // will call our callback
 129         return;
 130     }
 131
 132     pushNewConnection(params.conn);
 133 }
 134
 135 void
 136 PeerPoolMgr::pushNewConnection(const Comm::ConnectionPointer &conn)
 137 {
 138     Must(validPeer());
 139     Must(Comm::IsConnOpen(conn));
 140     peer->standby.pool->push(conn, NULL /* domain */);
 141     // push() will trigger a checkpoint()
 142 }
 143
 144 void
 145 PeerPoolMgr::handleSecuredPeer(Security::EncryptorAnswer &answer)
 146 {
 147     Must(securer != NULL);
 148     securer = NULL;
 149
 150     if (closer != NULL) {
 151         if (answer.conn != NULL)
 152             comm_remove_close_handler(answer.conn->fd, closer);
 153         else
 154             closer->cancel("securing completed");
 155         closer = NULL;
 156     }
 157
 158     if (!validPeer()) {
 159         debugs(48, 3, "peer gone");
 160         if (answer.conn != NULL)
 161             answer.conn->close();
 162         return;
 163     }
 164
 165     if (answer.error.get()) {
 166         if (answer.conn != NULL)
 167             answer.conn->close();
 168         // PeerConnector calls peerConnectFailed() for us;
 169         checkpoint("conn securing failure"); // may retry
 170         return;
 171     }
 172
 173     pushNewConnection(answer.conn);
 174 }
 175
 176 void
 177 PeerPoolMgr::handleSecureClosure(const CommCloseCbParams &params)
 178 {
 179     Must(closer != NULL);
 180     Must(securer != NULL);
 181     securer->cancel("conn closed by a 3rd party");
 182     securer = NULL;
 183     closer = NULL;
 184     // allow the closing connection to fully close before we check again
 185     Checkpoint(this, "conn closure while securing");
 186 }
 187
 188 void
 189 PeerPoolMgr::openNewConnection()
 190 {
 191     // KISS: Do nothing else when we are already doing something.
 192     if (opener != NULL || securer != NULL || shutting_down) {
 193         debugs(48, 7, "busy: " << opener << '|' << securer << '|' << shutting_down);
 194         return; // there will be another checkpoint when we are done opening/securing
 195     }
 196
 197     // Do not talk to a peer until it is ready.
 198     if (!neighborUp(peer)) // provides debugging
 199         return; // there will be another checkpoint when peer is up
 200
 201     // Do not violate peer limits.
 202     if (!peerCanOpenMore(peer)) { // provides debugging
 203         peer->standby.waitingForClose = true; // may already be true
 204         return; // there will be another checkpoint when a peer conn closes
 205     }
 206
 207     // Do not violate global restrictions.
 208     if (fdUsageHigh()) {
 209         debugs(48, 7, "overwhelmed");
 210         peer->standby.waitingForClose = true; // may already be true
 211         // There will be another checkpoint when a peer conn closes OR when
 212         // a future pop() fails due to an empty pool. See PconnPool::pop().
 213         return;
 214     }
 215
 216     peer->standby.waitingForClose = false;
 217
 218     Comm::ConnectionPointer conn = new Comm::Connection;
 219     Must(peer->n_addresses); // guaranteed by neighborUp() above
 220     // cycle through all available IP addresses
 221     conn->remote = peer->addresses[addrUsed++ % peer->n_addresses];
 222     conn->remote.port(peer->http_port);
 223     conn->peerType = STANDBY_POOL; // should be reset by peerSelect()
 224     conn->setPeer(peer);
 225     getOutgoingAddress(request.getRaw(), conn);
 226     GetMarkingsToServer(request.getRaw(), *conn);
 227
 228     const int ctimeout = peerConnectTimeout(peer);
 229     typedef CommCbMemFunT<PeerPoolMgr, CommConnectCbParams> Dialer;
 230     opener = JobCallback(48, 5, Dialer, this, PeerPoolMgr::handleOpenedConnection);
 231     Comm::ConnOpener *cs = new Comm::ConnOpener(conn, opener, ctimeout);
 232     AsyncJob::Start(cs);
 233 }
 234
 235 void
 236 PeerPoolMgr::closeOldConnections(const int howMany)
 237 {
 238     debugs(48, 8, howMany);
 239     peer->standby.pool->closeN(howMany);
 240 }
 241
 242 void
 243 PeerPoolMgr::checkpoint(const char *reason)
 244 {
 245     if (!validPeer()) {
 246         debugs(48, 3, reason << " and peer gone");
 247         return; // nothing to do after our owner dies; the job will quit
 248     }
 249
 250     const int count = peer->standby.pool->count();
 251     const int limit = peer->standby.limit;
 252     debugs(48, 7, reason << " with " << count << " ? " << limit);
 253
 254     if (count < limit)
 255         openNewConnection();
 256     else if (count > limit)
 257         closeOldConnections(count - limit);
 258 }
 259
 260 void
 261 PeerPoolMgr::Checkpoint(const Pointer &mgr, const char *reason)
 262 {
 263     CallJobHere1(48, 5, mgr, PeerPoolMgr, checkpoint, reason);
 264 }
 265
 266 /// launches PeerPoolMgrs for peers configured with standby.limit
 267 class PeerPoolMgrsRr: public RegisteredRunner
 268 {
 269 public:
 270     /* RegisteredRunner API */
 271     virtual void useConfig() { syncConfig(); }
 272     virtual void syncConfig();
 273 };
 274
 275 RunnerRegistrationEntry(PeerPoolMgrsRr);
 276
 277 void
 278 PeerPoolMgrsRr::syncConfig()
 279 {
 280     for (CachePeer *p = Config.peers; p; p = p->next) {
 281         // On reconfigure, Squid deletes the old config (and old peers in it),
 282         // so should always be dealing with a brand new configuration.
 283         assert(!p->standby.mgr);
 284         assert(!p->standby.pool);
 285         if (p->standby.limit) {
 286             p->standby.mgr = new PeerPoolMgr(p);
 287             p->standby.pool = new PconnPool(p->name, p->standby.mgr);
 288             AsyncJob::Start(p->standby.mgr.get());
 289         }
 290     }
 291 }
 292