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