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