src/icp_v2.cc

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