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