src/PeerPoolMgr.cc

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