src/icp_v2.cc

   1 /*
   2  * DEBUG: section 12    Internet Cache Protocol (ICP)
   3  * AUTHOR: Duane Wessels
   4  *
   5  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   6  * ----------------------------------------------------------
   7  *
   8  *  Squid is the result of efforts by numerous individuals from
   9  *  the Internet community; see the CONTRIBUTORS file for full
  10  *  details.   Many organizations have provided support for Squid's
  11  *  development; see the SPONSORS file for full details.  Squid is
  12  *  Copyrighted (C) 2001 by the Regents of the University of
  13  *  California; see the COPYRIGHT file for full details.  Squid
  14  *  incorporates software developed and/or copyrighted by other
  15  *  sources; see the CREDITS file for full details.
  16  *
  17  *  This program is free software; you can redistribute it and/or modify
  18  *  it under the terms of the GNU General Public License as published by
  19  *  the Free Software Foundation; either version 2 of the License, or
  20  *  (at your option) any later version.
  21  *
  22  *  This program is distributed in the hope that it will be useful,
  23  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  24  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  25  *  GNU General Public License for more details.
  26  *
  27  *  You should have received a copy of the GNU General Public License
  28  *  along with this program; if not, write to the Free Software
  29  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  30  *
  31  */
  32
  33 /**
  34  \defgroup ServerProtocolICPInternal2 ICPv2 Internals
  35  \ingroup ServerProtocolICPAPI
  36  */
  37
  38 #include "squid.h"
  39 #include "Store.h"
  40 #include "comm.h"
  41 #include "ICP.h"
  42 #include "HttpRequest.h"
  43 #include "acl/FilledChecklist.h"
  44 #include "acl/Acl.h"
  45 #include "AccessLogEntry.h"
  46 #include "wordlist.h"
  47 #include "SquidTime.h"
  48 #include "SwapDir.h"
  49 #include "icmp/net_db.h"
  50 #include "ip/Address.h"
  51 #include "ip/tools.h"
  52 #include "ipc/StartListening.h"
  53 #include "rfc1738.h"
  54
  55 /// dials icpIncomingConnectionOpened call
  56 class IcpListeningStartedDialer: public CallDialer,
  57         public Ipc::StartListeningCb
  58 {
  59 public:
  60     typedef void (*Handler)(int fd, int errNo, Ip::Address& addr);
  61     IcpListeningStartedDialer(Handler aHandler, Ip::Address& anAddr):
  62             handler(aHandler), addr(anAddr) {}
  63
  64     virtual void print(std::ostream &os) const {
  65         startPrint(os) <<
  66         ", address=" << addr << ')';
  67     }
  68
  69     virtual bool canDial(AsyncCall &) const { return true; }
  70     virtual void dial(AsyncCall &) { (handler)(fd, errNo, addr); }
  71
  72 public:
  73     Handler handler;
  74     Ip::Address addr;
  75 };
  76
  77 static void icpIncomingConnectionOpened(int fd, int errNo, Ip::Address& addr);
  78
  79 /// \ingroup ServerProtocolICPInternal2
  80 static void icpLogIcp(const Ip::Address &, log_type, int, const char *, int);
  81
  82 /// \ingroup ServerProtocolICPInternal2
  83 static void icpHandleIcpV2(int, Ip::Address &, char *, int);
  84
  85 /// \ingroup ServerProtocolICPInternal2
  86 static void icpCount(void *, int, size_t, int);
  87
  88 /// \ingroup ServerProtocolICPInternal2
  89 static void icpGetOutgoingIpAddress();
  90
  91 /**
  92  \ingroup ServerProtocolICPInternal2
  93  * IcpQueueHead is global so comm_incoming() knows whether or not
  94  * to call icpUdpSendQueue.
  95  */
  96 static icpUdpData *IcpQueueHead = NULL;
  97 /// \ingroup ServerProtocolICPInternal2
  98 static icpUdpData *IcpQueueTail = NULL;
  99
 100 /// \ingroup ServerProtocolICPInternal2
 101 Ip::Address theOutICPAddr;
 102
 103 /* icp_common_t */
 104 _icp_common_t::_icp_common_t() : opcode(ICP_INVALID), version(0), length(0), reqnum(0), flags(0), pad(0), shostid(0)
 105 {}
 106
 107 _icp_common_t::_icp_common_t(char *buf, unsigned int len)
 108 {
 109     if (len < sizeof(_icp_common_t)) {
 110         /* mark as invalid */
 111         length = len + 1;
 112         return;
 113     }
 114
 115     memcpy(this, buf, sizeof(icp_common_t));
 116     /*
 117      * Convert network order sensitive fields
 118      */
 119     length = ntohs(length);
 120     reqnum = ntohl(reqnum);
 121     flags = ntohl(flags);
 122     pad = ntohl(pad);
 123 }
 124
 125 icp_opcode
 126 _icp_common_t::getOpCode() const
 127 {
 128     if (opcode > (char)ICP_END)
 129         return ICP_INVALID;
 130
 131     return (icp_opcode)opcode;
 132 }
 133
 134 /* ICPState */
 135
 136 ICPState::ICPState(icp_common_t &aHeader, HttpRequest *aRequest):
 137         header(aHeader),
 138         request(HTTPMSGLOCK(aRequest)),
 139         fd(-1),
 140         url(NULL)
 141 {}
 142
 143 ICPState::~ICPState()
 144 {
 145     safe_free(url);
 146     HTTPMSGUNLOCK(request);
 147 }
 148
 149
 150 /* End ICPState */
 151
 152 /* ICP2State */
 153
 154 /// \ingroup ServerProtocolICPInternal2
 155 class ICP2State : public ICPState, public StoreClient
 156 {
 157
 158 public:
 159     ICP2State(icp_common_t & aHeader, HttpRequest *aRequest):
 160             ICPState(aHeader, aRequest),rtt(0),src_rtt(0),flags(0) {}
 161
 162     ~ICP2State();
 163     void created(StoreEntry * newEntry);
 164
 165     int rtt;
 166     int src_rtt;
 167     uint32_t flags;
 168 };
 169
 170 ICP2State::~ICP2State()
 171 {}
 172
 173 void
 174 ICP2State::created(StoreEntry *newEntry)
 175 {
 176     StoreEntry *entry = newEntry->isNull () ? NULL : newEntry;
 177     debugs(12, 5, "icpHandleIcpV2: OPCODE " << icp_opcode_str[header.opcode]);
 178     icp_opcode codeToSend;
 179
 180     if (icpCheckUdpHit(entry, request)) {
 181         codeToSend = ICP_HIT;
 182     } else {
 183 #if USE_ICMP
 184         if (Config.onoff.test_reachability && rtt == 0) {
 185             if ((rtt = netdbHostRtt(request->GetHost())) == 0)
 186                 netdbPingSite(request->GetHost());
 187         }
 188 #endif /* USE_ICMP */
 189
 190         if (icpGetCommonOpcode() != ICP_ERR)
 191             codeToSend = icpGetCommonOpcode();
 192         else if (Config.onoff.test_reachability && rtt == 0)
 193             codeToSend = ICP_MISS_NOFETCH;
 194         else
 195             codeToSend = ICP_MISS;
 196     }
 197
 198     icpCreateAndSend(codeToSend, flags, url, header.reqnum, src_rtt, fd, from);
 199     delete this;
 200 }
 201
 202 /* End ICP2State */
 203
 204 /// \ingroup ServerProtocolICPInternal2
 205 static void
 206 icpLogIcp(const Ip::Address &caddr, log_type logcode, int len, const char *url, int delay)
 207 {
 208     AccessLogEntry al;
 209
 210     if (LOG_TAG_NONE == logcode)
 211         return;
 212
 213     if (LOG_ICP_QUERY == logcode)
 214         return;
 215
 216     clientdbUpdate(caddr, logcode, PROTO_ICP, len);
 217
 218     if (!Config.onoff.log_udp)
 219         return;
 220
 221     al.icp.opcode = ICP_QUERY;
 222
 223     al.url = url;
 224
 225     al.cache.caddr = caddr;
 226
 227     al.cache.replySize = len;
 228
 229     al.cache.code = logcode;
 230
 231     al.cache.msec = delay;
 232
 233     accessLogLog(&al, NULL);
 234 }
 235
 236 /// \ingroup ServerProtocolICPInternal2
 237 void
 238 icpUdpSendQueue(int fd, void *unused)
 239 {
 240     icpUdpData *q;
 241
 242     while ((q = IcpQueueHead) != NULL) {
 243         int delay = tvSubUsec(q->queue_time, current_time);
 244         /* increment delay to prevent looping */
 245         const int x = icpUdpSend(fd, q->address, (icp_common_t *) q->msg, q->logcode, ++delay);
 246         IcpQueueHead = q->next;
 247         xfree(q);
 248
 249         if (x < 0)
 250             break;
 251     }
 252 }
 253
 254 _icp_common_t *
 255 _icp_common_t::createMessage(
 256     icp_opcode opcode,
 257     int flags,
 258     const char *url,
 259     int reqnum,
 260     int pad)
 261 {
 262     char *buf = NULL;
 263     icp_common_t *headerp = NULL;
 264     char *urloffset = NULL;
 265     int buf_len;
 266     buf_len = sizeof(icp_common_t) + strlen(url) + 1;
 267
 268     if (opcode == ICP_QUERY)
 269         buf_len += sizeof(uint32_t);
 270
 271     buf = (char *) xcalloc(buf_len, 1);
 272
 273     headerp = (icp_common_t *) (void *) buf;
 274
 275     headerp->opcode = (char) opcode;
 276
 277     headerp->version = ICP_VERSION_CURRENT;
 278
 279     headerp->length = (uint16_t) htons(buf_len);
 280
 281     headerp->reqnum = htonl(reqnum);
 282
 283     headerp->flags = htonl(flags);
 284
 285     headerp->pad = htonl(pad);
 286
 287     theOutICPAddr.GetInAddr( *((struct in_addr*)&headerp->shostid) );
 288
 289     urloffset = buf + sizeof(icp_common_t);
 290
 291     if (opcode == ICP_QUERY)
 292         urloffset += sizeof(uint32_t);
 293
 294     memcpy(urloffset, url, strlen(url));
 295
 296     return (icp_common_t *)buf;
 297 }
 298
 299 int
 300 icpUdpSend(int fd,
 301            const Ip::Address &to,
 302            icp_common_t * msg,
 303            log_type logcode,
 304            int delay)
 305 {
 306     icpUdpData *queue;
 307     int x;
 308     int len;
 309     len = (int) ntohs(msg->length);
 310     debugs(12, 5, "icpUdpSend: FD " << fd << " sending " <<
 311            icp_opcode_str[msg->opcode] << ", " << len << " bytes to " << to);
 312
 313     x = comm_udp_sendto(fd, to, msg, len);
 314
 315     if (x >= 0) {
 316         /* successfully written */
 317         icpLogIcp(to, logcode, len, (char *) (msg + 1), delay);
 318         icpCount(msg, SENT, (size_t) len, delay);
 319         safe_free(msg);
 320     } else if (0 == delay) {
 321         /* send failed, but queue it */
 322         queue = (icpUdpData *) xcalloc(1, sizeof(icpUdpData));
 323         queue->address = to;
 324         queue->msg = msg;
 325         queue->len = (int) ntohs(msg->length);
 326         queue->queue_time = current_time;
 327         queue->logcode = logcode;
 328
 329         if (IcpQueueHead == NULL) {
 330             IcpQueueHead = queue;
 331             IcpQueueTail = queue;
 332         } else if (IcpQueueTail == IcpQueueHead) {
 333             IcpQueueTail = queue;
 334             IcpQueueHead->next = queue;
 335         } else {
 336             IcpQueueTail->next = queue;
 337             IcpQueueTail = queue;
 338         }
 339
 340         commSetSelect(fd, COMM_SELECT_WRITE, icpUdpSendQueue, NULL, 0);
 341         statCounter.icp.replies_queued++;
 342     } else {
 343         /* don't queue it */
 344         statCounter.icp.replies_dropped++;
 345     }
 346
 347     return x;
 348 }
 349
 350 int
 351 icpCheckUdpHit(StoreEntry * e, HttpRequest * request)
 352 {
 353     if (e == NULL)
 354         return 0;
 355
 356     if (!e->validToSend())
 357         return 0;
 358
 359     if (Config.onoff.icp_hit_stale)
 360         return 1;
 361
 362     if (refreshCheckICP(e, request))
 363         return 0;
 364
 365     return 1;
 366 }
 367
 368 /**
 369  * This routine selects an ICP opcode for ICP misses.
 370  *
 371  \retval ICP_ERR            no opcode selected here
 372  \retval ICP_MISS_NOFETCH   store is rebuilding, no fetch is possible yet
 373  */
 374 icp_opcode
 375 icpGetCommonOpcode()
 376 {
 377     /* if store is rebuilding, return a UDP_MISS_NOFETCH */
 378
 379     if ((StoreController::store_dirs_rebuilding && opt_reload_hit_only) ||
 380             hit_only_mode_until > squid_curtime) {
 381         return ICP_MISS_NOFETCH;
 382     }
 383
 384     return ICP_ERR;
 385 }
 386
 387 log_type
 388 icpLogFromICPCode(icp_opcode opcode)
 389 {
 390     if (opcode == ICP_ERR)
 391         return LOG_UDP_INVALID;
 392
 393     if (opcode == ICP_DENIED)
 394         return LOG_UDP_DENIED;
 395
 396     if (opcode == ICP_HIT)
 397         return LOG_UDP_HIT;
 398
 399     if (opcode == ICP_MISS)
 400         return LOG_UDP_MISS;
 401
 402     if (opcode == ICP_MISS_NOFETCH)
 403         return LOG_UDP_MISS_NOFETCH;
 404
 405     fatal("expected ICP opcode\n");
 406
 407     return LOG_UDP_INVALID;
 408 }
 409
 410 void
 411 icpCreateAndSend(icp_opcode opcode, int flags, char const *url, int reqnum, int pad, int fd, const Ip::Address &from)
 412 {
 413     icp_common_t *reply = _icp_common_t::createMessage(opcode, flags, url, reqnum, pad);
 414     icpUdpSend(fd, from, reply, icpLogFromICPCode(opcode), 0);
 415 }
 416
 417 void
 418 icpDenyAccess(Ip::Address &from, char *url, int reqnum, int fd)
 419 {
 420     debugs(12, 2, "icpDenyAccess: Access Denied for " << from << " by " << AclMatchedName << ".");
 421
 422     if (clientdbCutoffDenied(from)) {
 423         /*
 424          * count this DENIED query in the clientdb, even though
 425          * we're not sending an ICP reply...
 426          */
 427         clientdbUpdate(from, LOG_UDP_DENIED, PROTO_ICP, 0);
 428     } else {
 429         icpCreateAndSend(ICP_DENIED, 0, url, reqnum, 0, fd, from);
 430     }
 431 }
 432
 433 int
 434 icpAccessAllowed(Ip::Address &from, HttpRequest * icp_request)
 435 {
 436     /* absent an explicit allow, we deny all */
 437     if (!Config.accessList.icp)
 438         return 0;
 439
 440     ACLFilledChecklist checklist(Config.accessList.icp, icp_request, NULL);
 441     checklist.src_addr = from;
 442     checklist.my_addr.SetNoAddr();
 443     int result = checklist.fastCheck();
 444     return result;
 445 }
 446
 447 char const *
 448 icpGetUrlToSend(char *url)
 449 {
 450     if (strpbrk(url, w_space))
 451         return rfc1738_escape(url);
 452     else
 453         return url;
 454 }
 455
 456 HttpRequest *
 457 icpGetRequest(char *url, int reqnum, int fd, Ip::Address &from)
 458 {
 459     if (strpbrk(url, w_space)) {
 460         url = rfc1738_escape(url);
 461         icpCreateAndSend(ICP_ERR, 0, rfc1738_escape(url), reqnum, 0, fd, from);
 462         return NULL;
 463     }
 464
 465     HttpRequest *result;
 466
 467     if ((result = HttpRequest::CreateFromUrl(url)) == NULL)
 468         icpCreateAndSend(ICP_ERR, 0, url, reqnum, 0, fd, from);
 469
 470     return result;
 471
 472 }
 473
 474 static void
 475 doV2Query(int fd, Ip::Address &from, char *buf, icp_common_t header)
 476 {
 477     int rtt = 0;
 478     int src_rtt = 0;
 479     uint32_t flags = 0;
 480     /* We have a valid packet */
 481     char *url = buf + sizeof(icp_common_t) + sizeof(uint32_t);
 482     HttpRequest *icp_request = icpGetRequest(url, header.reqnum, fd, from);
 483
 484     if (!icp_request)
 485         return;
 486
 487     HTTPMSGLOCK(icp_request);
 488
 489     if (!icpAccessAllowed(from, icp_request)) {
 490         icpDenyAccess(from, url, header.reqnum, fd);
 491         HTTPMSGUNLOCK(icp_request);
 492         return;
 493     }
 494 #if USE_ICMP
 495     if (header.flags & ICP_FLAG_SRC_RTT) {
 496         rtt = netdbHostRtt(icp_request->GetHost());
 497         int hops = netdbHostHops(icp_request->GetHost());
 498         src_rtt = ((hops & 0xFFFF) << 16) | (rtt & 0xFFFF);
 499
 500         if (rtt)
 501             flags |= ICP_FLAG_SRC_RTT;
 502     }
 503 #endif /* USE_ICMP */
 504
 505     /* The peer is allowed to use this cache */
 506     ICP2State *state = new ICP2State (header, icp_request);
 507
 508     state->fd = fd;
 509
 510     state->from = from;
 511
 512     state->url = xstrdup (url);
 513
 514     state->flags = flags;
 515
 516     state->rtt = rtt;
 517
 518     state->src_rtt = src_rtt;
 519
 520     StoreEntry::getPublic (state, url, METHOD_GET);
 521
 522     HTTPMSGUNLOCK(icp_request);
 523 }
 524
 525 void
 526 _icp_common_t::handleReply(char *buf, Ip::Address &from)
 527 {
 528     if (neighbors_do_private_keys && reqnum == 0) {
 529         debugs(12, 0, "icpHandleIcpV2: Neighbor " << from << " returned reqnum = 0");
 530         debugs(12, 0, "icpHandleIcpV2: Disabling use of private keys");
 531         neighbors_do_private_keys = 0;
 532     }
 533
 534     char *url = buf + sizeof(icp_common_t);
 535     debugs(12, 3, "icpHandleIcpV2: " << icp_opcode_str[opcode] << " from " << from << " for '" << url << "'");
 536
 537     const cache_key *key = icpGetCacheKey(url, (int) reqnum);
 538     /* call neighborsUdpAck even if ping_status != PING_WAITING */
 539     neighborsUdpAck(key, this, from);
 540 }
 541
 542 static void
 543 icpHandleIcpV2(int fd, Ip::Address &from, char *buf, int len)
 544 {
 545     if (len <= 0) {
 546         debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
 547         return;
 548     }
 549
 550     icp_common_t header(buf, len);
 551     /*
 552      * Length field should match the number of bytes read
 553      */
 554
 555     if (len != header.length) {
 556         debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
 557         return;
 558     }
 559
 560     switch (header.opcode) {
 561
 562     case ICP_QUERY:
 563         /* We have a valid packet */
 564         doV2Query(fd, from, buf, header);
 565         break;
 566
 567     case ICP_HIT:
 568
 569     case ICP_DECHO:
 570
 571     case ICP_MISS:
 572
 573     case ICP_DENIED:
 574
 575     case ICP_MISS_NOFETCH:
 576         header.handleReply(buf, from);
 577         break;
 578
 579     case ICP_INVALID:
 580
 581     case ICP_ERR:
 582         break;
 583
 584     default:
 585         debugs(12, 0, "icpHandleIcpV2: UNKNOWN OPCODE: " << header.opcode << " from " << from);
 586
 587         break;
 588     }
 589 }
 590
 591 #ifdef ICP_PKT_DUMP
 592 static void
 593 icpPktDump(icp_common_t * pkt)
 594 {
 595     Ip::Address a;
 596
 597     debugs(12, 9, "opcode:     " << std::setw(3) << pkt->opcode  << " " << icp_opcode_str[pkt->opcode]);
 598     debugs(12, 9, "version: "<< std::left << std::setw(8) << pkt->version);
 599     debugs(12, 9, "length:  "<< std::left << std::setw(8) << ntohs(pkt->length));
 600     debugs(12, 9, "reqnum:  "<< std::left << std::setw(8) << ntohl(pkt->reqnum));
 601     debugs(12, 9, "flags:   "<< std::left << std::hex << std::setw(8) << ntohl(pkt->flags));
 602     a = (struct in_addr)pkt->shostid;
 603     debugs(12, 9, "shostid: " << a );
 604     debugs(12, 9, "payload: " << (char *) pkt + sizeof(icp_common_t));
 605 }
 606
 607 #endif
 608
 609 void
 610 icpHandleUdp(int sock, void *data)
 611 {
 612     int *N = &incoming_sockets_accepted;
 613
 614     Ip::Address from;
 615     LOCAL_ARRAY(char, buf, SQUID_UDP_SO_RCVBUF);
 616     int len;
 617     int icp_version;
 618     int max = INCOMING_ICP_MAX;
 619     commSetSelect(sock, COMM_SELECT_READ, icpHandleUdp, NULL, 0);
 620
 621     while (max--) {
 622         len = comm_udp_recvfrom(sock,
 623                                 buf,
 624                                 SQUID_UDP_SO_RCVBUF - 1,
 625                                 0,
 626                                 from);
 627
 628         if (len == 0)
 629             break;
 630
 631         if (len < 0) {
 632             if (ignoreErrno(errno))
 633                 break;
 634
 635 #ifdef _SQUID_LINUX_
 636             /* Some Linux systems seem to set the FD for reading and then
 637              * return ECONNREFUSED when sendto() fails and generates an ICMP
 638              * port unreachable message. */
 639             /* or maybe an EHOSTUNREACH "No route to host" message */
 640             if (errno != ECONNREFUSED && errno != EHOSTUNREACH)
 641 #endif
 642
 643                 debugs(50, 1, "icpHandleUdp: FD " << sock << " recvfrom: " << xstrerror());
 644
 645             break;
 646         }
 647
 648         (*N)++;
 649         icpCount(buf, RECV, (size_t) len, 0);
 650         buf[len] = '\0';
 651         debugs(12, 4, "icpHandleUdp: FD " << sock << ": received " <<
 652                (unsigned long int)len << " bytes from " << from);
 653
 654 #ifdef ICP_PACKET_DUMP
 655
 656         icpPktDump(buf);
 657 #endif
 658
 659         if ((size_t) len < sizeof(icp_common_t)) {
 660             debugs(12, 4, "icpHandleUdp: Ignoring too-small UDP packet");
 661             break;
 662         }
 663
 664         icp_version = (int) buf[1];     /* cheat! */
 665
 666         if (icp_version == ICP_VERSION_2)
 667             icpHandleIcpV2(sock, from, buf, len);
 668         else if (icp_version == ICP_VERSION_3)
 669             icpHandleIcpV3(sock, from, buf, len);
 670         else
 671             debugs(12, 1, "WARNING: Unused ICP version " << icp_version <<
 672                    " received from " << from);
 673     }
 674 }
 675
 676 void
 677 icpConnectionsOpen(void)
 678 {
 679     uint16_t port;
 680     Ip::Address addr;
 681
 682     if ((port = Config.Port.icp) <= 0)
 683         return;
 684
 685     addr = Config.Addrs.udp_incoming;
 686     addr.SetPort(port);
 687
 688     if (!Ip::EnableIpv6 && !addr.SetIPv4()) {
 689         debugs(12, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << addr << " is not an IPv4 address.");
 690         fatal("ICP port cannot be opened.");
 691     }
 692     /* split-stack for now requires default IPv4-only ICP */
 693     if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && addr.IsAnyAddr()) {
 694         addr.SetIPv4();
 695     }
 696
 697     AsyncCall::Pointer call = asyncCall(12, 2,
 698                                         "icpIncomingConnectionOpened",
 699                                         IcpListeningStartedDialer(&icpIncomingConnectionOpened, addr));
 700
 701     Ipc::StartListening(SOCK_DGRAM,
 702                         IPPROTO_UDP,
 703                         addr,
 704                         COMM_NONBLOCKING,
 705                         Ipc::fdnInIcpSocket, call);
 706
 707     addr.SetEmpty(); // clear for next use.
 708     addr = Config.Addrs.udp_outgoing;
 709     if ( !addr.IsNoAddr() ) {
 710         enter_suid();
 711         addr.SetPort(port);
 712
 713         if (!Ip::EnableIpv6 && !addr.SetIPv4()) {
 714             debugs(49, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << addr << " is not an IPv4 address.");
 715             fatal("ICP port cannot be opened.");
 716         }
 717         /* split-stack for now requires default IPv4-only ICP */
 718         if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && addr.IsAnyAddr()) {
 719             addr.SetIPv4();
 720         }
 721
 722         theOutIcpConnection = comm_open_listener(SOCK_DGRAM,
 723                               IPPROTO_UDP,
 724                               addr,
 725                               COMM_NONBLOCKING,
 726                               "ICP Port");
 727         leave_suid();
 728
 729         if (theOutIcpConnection < 0)
 730             fatal("Cannot open Outgoing ICP Port");
 731
 732         commSetSelect(theOutIcpConnection,
 733                       COMM_SELECT_READ,
 734                       icpHandleUdp,
 735                       NULL,
 736                       0);
 737
 738         debugs(12, 1, "Outgoing ICP messages on port " << addr.GetPort() << ", FD " << theOutIcpConnection << ".");
 739
 740         fd_note(theOutIcpConnection, "Outgoing ICP socket");
 741         icpGetOutgoingIpAddress();
 742     }
 743 }
 744
 745 static void
 746 icpGetOutgoingIpAddress()
 747 {
 748     struct addrinfo *xai = NULL;
 749     theOutICPAddr.SetEmpty();
 750     theOutICPAddr.InitAddrInfo(xai);
 751
 752     if (getsockname(theOutIcpConnection, xai->ai_addr, &xai->ai_addrlen) < 0)
 753         debugs(50, 1, "theOutIcpConnection FD " << theOutIcpConnection << ": getsockname: " << xstrerror());
 754     else
 755         theOutICPAddr = *xai;
 756
 757     theOutICPAddr.FreeAddrInfo(xai);
 758 }
 759
 760 static void
 761 icpIncomingConnectionOpened(int fd, int errNo, Ip::Address& addr)
 762 {
 763     theInIcpConnection = fd;
 764
 765     if (theInIcpConnection < 0)
 766         fatal("Cannot open ICP Port");
 767
 768     commSetSelect(theInIcpConnection,
 769                   COMM_SELECT_READ,
 770                   icpHandleUdp,
 771                   NULL,
 772                   0);
 773
 774     for (const wordlist *s = Config.mcast_group_list; s; s = s->next)
 775         ipcache_nbgethostbyname(s->key, mcastJoinGroups, NULL);
 776
 777     debugs(12, 1, "Accepting ICP messages at " << addr << ", FD " << theInIcpConnection << ".");
 778
 779     fd_note(theInIcpConnection, "Incoming ICP socket");
 780
 781     if (Config.Addrs.udp_outgoing.IsNoAddr()) {
 782         theOutIcpConnection = theInIcpConnection;
 783         icpGetOutgoingIpAddress();
 784     }
 785 }
 786
 787 /**
 788  * icpConnectionShutdown only closes the 'in' socket if it is
 789  * different than the 'out' socket.
 790  */
 791 void
 792 icpConnectionShutdown(void)
 793 {
 794     if (theInIcpConnection < 0)
 795         return;
 796
 797     if (theInIcpConnection != theOutIcpConnection) {
 798         debugs(12, 1, "FD " << theInIcpConnection << " Closing ICP connection");
 799         comm_close(theInIcpConnection);
 800     }
 801
 802     /**
 803      * Here we set 'theInIcpConnection' to -1 even though the ICP 'in'
 804      * and 'out' sockets might be just one FD.  This prevents this
 805      * function from executing repeatedly.  When we are really ready to
 806      * exit or restart, main will comm_close the 'out' descriptor.
 807      */
 808     theInIcpConnection = -1;
 809
 810     /**
 811      * Normally we only write to the outgoing ICP socket, but
 812      * we also have a read handler there to catch messages sent
 813      * to that specific interface.  During shutdown, we must
 814      * disable reading on the outgoing socket.
 815      */
 816     assert(theOutIcpConnection > -1);
 817
 818     commSetSelect(theOutIcpConnection, COMM_SELECT_READ, NULL, NULL, 0);
 819 }
 820
 821 void
 822 icpConnectionClose(void)
 823 {
 824     icpConnectionShutdown();
 825
 826     if (theOutIcpConnection > -1) {
 827         debugs(12, 1, "FD " << theOutIcpConnection << " Closing ICP connection");
 828         comm_close(theOutIcpConnection);
 829         theOutIcpConnection = -1;
 830     }
 831 }
 832
 833 static void
 834 icpCount(void *buf, int which, size_t len, int delay)
 835 {
 836     icp_common_t *icp = (icp_common_t *) buf;
 837
 838     if (len < sizeof(*icp))
 839         return;
 840
 841     if (SENT == which) {
 842         statCounter.icp.pkts_sent++;
 843         kb_incr(&statCounter.icp.kbytes_sent, len);
 844
 845         if (ICP_QUERY == icp->opcode) {
 846             statCounter.icp.queries_sent++;
 847             kb_incr(&statCounter.icp.q_kbytes_sent, len);
 848         } else {
 849             statCounter.icp.replies_sent++;
 850             kb_incr(&statCounter.icp.r_kbytes_sent, len);
 851             /* this is the sent-reply service time */
 852             statHistCount(&statCounter.icp.reply_svc_time, delay);
 853         }
 854
 855         if (ICP_HIT == icp->opcode)
 856             statCounter.icp.hits_sent++;
 857     } else if (RECV == which) {
 858         statCounter.icp.pkts_recv++;
 859         kb_incr(&statCounter.icp.kbytes_recv, len);
 860
 861         if (ICP_QUERY == icp->opcode) {
 862             statCounter.icp.queries_recv++;
 863             kb_incr(&statCounter.icp.q_kbytes_recv, len);
 864         } else {
 865             statCounter.icp.replies_recv++;
 866             kb_incr(&statCounter.icp.r_kbytes_recv, len);
 867             /* statCounter.icp.query_svc_time set in clientUpdateCounters */
 868         }
 869
 870         if (ICP_HIT == icp->opcode)
 871             statCounter.icp.hits_recv++;
 872     }
 873 }
 874
 875 #define N_QUERIED_KEYS 8192
 876 #define N_QUERIED_KEYS_MASK 8191
 877 static cache_key queried_keys[N_QUERIED_KEYS][SQUID_MD5_DIGEST_LENGTH];
 878
 879 int
 880 icpSetCacheKey(const cache_key * key)
 881 {
 882     static int reqnum = 0;
 883
 884     if (++reqnum < 0)
 885         reqnum = 1;
 886
 887     storeKeyCopy(queried_keys[reqnum & N_QUERIED_KEYS_MASK], key);
 888
 889     return reqnum;
 890 }
 891
 892 const cache_key *
 893 icpGetCacheKey(const char *url, int reqnum)
 894 {
 895     if (neighbors_do_private_keys && reqnum)
 896         return queried_keys[reqnum & N_QUERIED_KEYS_MASK];
 897
 898     return storeKeyPublic(url, METHOD_GET);
 899 }