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