icp_v2.cc

Go to the documentation of this file.

/*
 * Copyright (C) 1996-2026 The Squid Software Foundation and contributors
 *
 * Squid software is distributed under GPLv2+ license and includes
 * contributions from numerous individuals and organizations.
 * Please see the COPYING and CONTRIBUTORS files for details.
 */
 
/* DEBUG: section 12    Internet Cache Protocol (ICP) */
 
#include "squid.h"
#include "AccessLogEntry.h"
#include "acl/Acl.h"
#include "acl/FilledChecklist.h"
#include "base/AsyncCallbacks.h"
#include "client_db.h"
#include "comm.h"
#include "comm/Connection.h"
#include "comm/Loops.h"
#include "fd.h"
#include "HttpRequest.h"
#include "icmp/net_db.h"
#include "ICP.h"
#include "ip/Address.h"
#include "ip/tools.h"
#include "ipc/StartListening.h"
#include "ipcache.h"
#include "md5.h"
#include "multicast.h"
#include "neighbors.h"
#include "refresh.h"
#include "rfc1738.h"
#include "SquidConfig.h"
#include "StatCounters.h"
#include "Store.h"
#include "store_key_md5.h"
#include "tools.h"
#include "wordlist.h"
 
#include <cerrno>
 
class DelayedUdpSend {
public:
    Ip::Address address; 
    icp_common_t *msg = nullptr; 
    DelayedUdpSend *next = nullptr; 
    AccessLogEntryPointer ale; 
    struct timeval queue_time = {}; 
};
 
static void icpIncomingConnectionOpened(Ipc::StartListeningAnswer &);
 
static void icpLogIcp(const Ip::Address &, const LogTags_ot, int, const char *, const int, AccessLogEntryPointer &);
 
static void icpHandleIcpV2(int, Ip::Address &, char *, int);
 
static void icpCount(void *, int, size_t, int);
 
static LogTags_ot icpLogFromICPCode(icp_opcode);
 
static int icpUdpSend(int fd, const Ip::Address &to, icp_common_t * msg, int delay, AccessLogEntryPointer al);
 
static void
icpSyncAle(AccessLogEntryPointer &al, const Ip::Address &caddr, const char *url, int len, int delay)
{
    if (!al)
        al = new AccessLogEntry();
    al->icp.opcode = ICP_QUERY;
    al->cache.caddr = caddr;
    al->url = url;
    al->setVirginUrlForMissingRequest(al->url);
    // XXX: move to use icp.clientReply instead
    al->http.clientReplySz.payloadData = len;
    al->cache.start_time = current_time;
    al->cache.start_time.tv_sec -= delay;
    al->cache.trTime.tv_sec = delay;
    al->cache.trTime.tv_usec = 0;
}
 
static DelayedUdpSend *IcpQueueHead = nullptr;
static DelayedUdpSend *IcpQueueTail = nullptr;
 
Comm::ConnectionPointer icpIncomingConn = nullptr;
Comm::ConnectionPointer icpOutgoingConn = nullptr;
 
/* icp_common_t */
icp_common_t::icp_common_t() :
    opcode(ICP_INVALID), version(0), length(0), reqnum(0),
    flags(0), pad(0), shostid(0)
{}
 
icp_common_t::icp_common_t(char *buf, unsigned int len) :
    opcode(ICP_INVALID), version(0), reqnum(0), flags(0), pad(0), shostid(0)
{
    if (len < sizeof(icp_common_t)) {
        /* mark as invalid */
        length = len + 1;
        return;
    }
 
    memcpy(this, buf, sizeof(icp_common_t));
    /*
     * Convert network order sensitive fields
     */
    length = ntohs(length);
    reqnum = ntohl(reqnum);
    flags = ntohl(flags);
    pad = ntohl(pad);
}
 
icp_opcode
icp_common_t::getOpCode() const
{
    if (opcode > static_cast<char>(icp_opcode::ICP_END))
        return ICP_INVALID;
 
    return static_cast<icp_opcode>(opcode);
}
 
/* ICPState */
 
ICPState::ICPState(icp_common_t &aHeader, HttpRequest *aRequest):
    header(aHeader),
    request(aRequest),
    fd(-1),
    url(nullptr)
{
    HTTPMSGLOCK(request);
}
 
ICPState::~ICPState()
{
    safe_free(url);
    HTTPMSGUNLOCK(request);
}
 
bool
ICPState::isHit() const
{
    const auto e = storeGetPublic(url, Http::METHOD_GET);
 
    const auto hit = e && confirmAndPrepHit(*e);
 
    if (e)
        e->abandon(__func__);
 
    return hit;
}
 
bool
ICPState::confirmAndPrepHit(const StoreEntry &e) const
{
    if (!e.validToSend())
        return false;
 
    if (e.hittingRequiresCollapsing() && !startCollapsingOn(e, false))
        return false;
 
    if (!Config.onoff.icp_hit_stale && !didCollapse && refreshCheckICP(&e, request))
        return false;
 
    return true;
}
 
LogTags *
ICPState::loggingTags() const
{
    // calling icpSyncAle(LOG_TAG_NONE) here would not change cache.code
    if (!al)
        al = new AccessLogEntry();
    return &al->cache.code;
}
 
void
ICPState::fillChecklist(ACLFilledChecklist &checklist) const
{
    checklist.setRequest(request);
    icpSyncAle(al, from, url, 0, 0);
    checklist.al = al;
}
 
/* End ICPState */
 
/* ICP2State */
 
class ICP2State: public ICPState
{
 
public:
    ICP2State(icp_common_t & aHeader, HttpRequest *aRequest):
        ICPState(aHeader, aRequest),rtt(0),src_rtt(0),flags(0) {}
 
    ~ICP2State() override;
 
    int rtt;
    int src_rtt;
    uint32_t flags;
};
 
ICP2State::~ICP2State()
{}
 
/* End ICP2State */
 
static void
icpLogIcp(const Ip::Address &caddr, const LogTags_ot logcode, const int len, const char *url, int delay, AccessLogEntry::Pointer &al)
{
    assert(logcode != LOG_TAG_NONE);
 
    // Optimization: No premature (ALE creation in) icpSyncAle().
    if (al) {
        icpSyncAle(al, caddr, url, len, delay);
        al->cache.code.update(logcode);
    }
 
    if (logcode == LOG_ICP_QUERY)
        return; // we never log queries
 
    if (!Config.onoff.log_udp) {
        clientdbUpdate(caddr, al ? al->cache.code : LogTags(logcode), AnyP::PROTO_ICP, len);
        return;
    }
 
    if (!al) {
        // The above attempt to optimize ALE creation has failed. We do need it.
        icpSyncAle(al, caddr, url, len, delay);
        al->cache.code.update(logcode);
    }
    clientdbUpdate(caddr, al->cache.code, AnyP::PROTO_ICP, len);
    accessLogLog(al, nullptr);
}
 
static void
icpUdpSendQueue(int fd, void *)
{
    DelayedUdpSend *q;
 
    while ((q = IcpQueueHead) != nullptr) {
        int delay = tvSubUsec(q->queue_time, current_time);
        /* increment delay to prevent looping */
        const int x = icpUdpSend(fd, q->address, q->msg, ++delay, q->ale);
        IcpQueueHead = q->next;
        delete q;
 
        if (x < 0)
            break;
    }
}
 
icp_common_t *
icp_common_t::CreateMessage(
    icp_opcode opcode,
    int flags,
    const char *url,
    int reqnum,
    int pad)
{
    char *buf = nullptr;
    icp_common_t *headerp = nullptr;
    char *urloffset = nullptr;
    int buf_len;
    buf_len = sizeof(icp_common_t) + strlen(url) + 1;
 
    if (opcode == ICP_QUERY)
        buf_len += sizeof(uint32_t);
 
    buf = (char *) xcalloc(buf_len, 1);
 
    headerp = (icp_common_t *) (void *) buf;
 
    headerp->opcode = (char) opcode;
 
    headerp->version = ICP_VERSION_CURRENT;
 
    headerp->length = (uint16_t) htons(buf_len);
 
    headerp->reqnum = htonl(reqnum);
 
    headerp->flags = htonl(flags);
 
    headerp->pad = htonl(pad);
 
    headerp->shostid = 0;
 
    urloffset = buf + sizeof(icp_common_t);
 
    if (opcode == ICP_QUERY)
        urloffset += sizeof(uint32_t);
 
    memcpy(urloffset, url, strlen(url));
 
    return (icp_common_t *)buf;
}
 
// TODO: Move retries to icpCreateAndSend(); the other caller does not retry.
static int
icpUdpSend(int fd,
           const Ip::Address &to,
           icp_common_t * msg,
           int delay,
           AccessLogEntryPointer al)
{
    int x;
    int len;
    len = (int) ntohs(msg->length);
    debugs(12, 5, "icpUdpSend: FD " << fd << " sending " <<
           icp_opcode_str[msg->opcode] << ", " << len << " bytes to " << to);
 
    x = comm_udp_sendto(fd, to, msg, len);
 
    if (x >= 0) {
        /* successfully written */
        const auto logcode = icpLogFromICPCode(static_cast<icp_opcode>(msg->opcode));
        icpLogIcp(to, logcode, len, (char *) (msg + 1), delay, al);
        icpCount(msg, SENT, (size_t) len, delay);
        safe_free(msg);
    } else if (0 == delay) {
        /* send failed, but queue it */
        const auto queue = new DelayedUdpSend();
        queue->address = to;
        queue->msg = msg;
        queue->queue_time = current_time;
        queue->ale = al;
 
        if (IcpQueueHead == nullptr) {
            IcpQueueHead = queue;
            IcpQueueTail = queue;
        } else if (IcpQueueTail == IcpQueueHead) {
            IcpQueueTail = queue;
            IcpQueueHead->next = queue;
        } else {
            IcpQueueTail->next = queue;
            IcpQueueTail = queue;
        }
 
        Comm::SetSelect(fd, COMM_SELECT_WRITE, icpUdpSendQueue, nullptr, 0);
        ++statCounter.icp.replies_queued;
    } else {
        /* don't queue it */
        // XXX: safe_free(msg)
        ++statCounter.icp.replies_dropped;
    }
 
    return x;
}
 
icp_opcode
icpGetCommonOpcode()
{
    /* if store is rebuilding, return a UDP_MISS_NOFETCH */
 
    if ((StoreController::store_dirs_rebuilding && opt_reload_hit_only) ||
            hit_only_mode_until > squid_curtime) {
        return ICP_MISS_NOFETCH;
    }
 
    return ICP_ERR;
}
 
static LogTags_ot
icpLogFromICPCode(icp_opcode opcode)
{
    if (opcode == ICP_ERR)
        return LOG_UDP_INVALID;
 
    if (opcode == ICP_DENIED)
        return LOG_UDP_DENIED;
 
    if (opcode == ICP_HIT)
        return LOG_UDP_HIT;
 
    if (opcode == ICP_MISS)
        return LOG_UDP_MISS;
 
    if (opcode == ICP_MISS_NOFETCH)
        return LOG_UDP_MISS_NOFETCH;
 
    if (opcode == ICP_DECHO)
        return LOG_ICP_QUERY;
 
    if (opcode == ICP_QUERY)
        return LOG_ICP_QUERY;
 
    fatal("expected ICP opcode\n");
 
    return LOG_UDP_INVALID;
}
 
void
icpCreateAndSend(icp_opcode opcode, int flags, char const *url, int reqnum, int pad, int fd, const Ip::Address &from, AccessLogEntry::Pointer al)
{
    // update potentially shared ALE ASAP; the ICP query itself may be delayed
    if (al)
        al->cache.code.update(icpLogFromICPCode(opcode));
    icp_common_t *reply = icp_common_t::CreateMessage(opcode, flags, url, reqnum, pad);
    icpUdpSend(fd, from, reply, 0, al);
}
 
void
icpDenyAccess(const Ip::Address &from, const char * const url, const int reqnum, const int fd)
{
    if (clientdbCutoffDenied(from)) {
        /*
         * count this DENIED query in the clientdb, even though
         * we're not sending an ICP reply...
         */
        clientdbUpdate(from, LogTags(LOG_UDP_DENIED), AnyP::PROTO_ICP, 0);
    } else {
        icpCreateAndSend(ICP_DENIED, 0, url, reqnum, 0, fd, from, nullptr);
    }
}
 
static bool
icpAccessAllowed(const Ip::Address &from, HttpRequest * icp_request)
{
    if (!Config.accessList.icp) {
        debugs(12, 2, "Access Denied due to lack of ICP access rules.");
        return false;
    }
 
    ACLFilledChecklist checklist(Config.accessList.icp, icp_request);
    checklist.src_addr = from;
    checklist.my_addr.setNoAddr();
    const auto &answer = checklist.fastCheck();
    if (answer.allowed())
        return true;
 
    debugs(12, 2, "Access Denied for " << from << " by " << answer.lastCheckDescription() << ".");
    return false;
}
 
const char *
icpGetUrl(const Ip::Address &from, const char * const buf, const icp_common_t &header)
{
    const auto receivedPacketSize = static_cast<size_t>(header.length);
    const auto payloadOffset = sizeof(header);
 
    // Query payload contains a "Requester Host Address" followed by a URL.
    // Payload of other ICP packets (with opcode that we recognize) is a URL.
    const auto urlOffset = payloadOffset + ((header.opcode == ICP_QUERY) ? sizeof(uint32_t) : 0);
 
    // A URL field cannot be empty because it includes a terminating NUL char.
    // Ensure that the packet has at least one URL field byte.
    if (urlOffset >= receivedPacketSize) {
        debugs(12, 3, "too small packet from " << from << ": " << urlOffset << " >= " << receivedPacketSize);
        return nullptr;
    }
 
    // All ICP packets (with opcode that we recognize) _end_ with a URL field.
    // RFC 2186 requires all URLs to be "Null-Terminated".
    if (buf[receivedPacketSize - 1] != '\0') {
        debugs(12, 3, "unterminated URL or trailing garbage from " << from);
        return nullptr;
    }
 
    const auto url = buf + urlOffset; // a possibly empty c-string
    if (urlOffset + strlen(url) + 1 != receivedPacketSize) {
        debugs(12, 3, "URL with an embedded NUL or trailing garbage from " << from);
        return nullptr;
    }
 
    return url;
}
 
HttpRequest::Pointer
icpGetRequest(const char * const url, const int reqnum, const int fd, const Ip::Address &from)
{
    if (strpbrk(url, w_space)) {
        icpCreateAndSend(ICP_ERR, 0, rfc1738_escape(url), reqnum, 0, fd, from, nullptr);
        return nullptr;
    }
 
    const auto mx = MasterXaction::MakePortless<XactionInitiator::initIcp>();
    if (const HttpRequest::Pointer request = HttpRequest::FromUrlXXX(url, mx)) {
        if (!icpAccessAllowed(from, request.getRaw())) {
            icpDenyAccess(from, url, reqnum, fd);
            return nullptr;
        }
 
        return request;
    }
 
    icpCreateAndSend(ICP_ERR, 0, url, reqnum, 0, fd, from, nullptr);
    return nullptr;
}
 
static void
doV2Query(const int fd, Ip::Address &from, const char * const buf, icp_common_t header)
{
    int rtt = 0;
    int src_rtt = 0;
    uint32_t flags = 0;
 
    const auto url = icpGetUrl(from, buf, header);
    if (!url) {
        icpCreateAndSend(ICP_ERR, 0, "", header.reqnum, 0, fd, from, nullptr);
        return;
    }
 
    const auto icp_request = icpGetRequest(url, header.reqnum, fd, from);
 
    if (!icp_request)
        return;
 
#if USE_ICMP
    if (header.flags & ICP_FLAG_SRC_RTT) {
        rtt = netdbHostRtt(icp_request->url.host());
        int hops = netdbHostHops(icp_request->url.host());
        src_rtt = ((hops & 0xFFFF) << 16) | (rtt & 0xFFFF);
 
        if (rtt)
            flags |= ICP_FLAG_SRC_RTT;
    }
#endif /* USE_ICMP */
 
    /* The peer is allowed to use this cache */
    ICP2State state(header, icp_request.getRaw());
    state.fd = fd;
    state.from = from;
    state.url = xstrdup(url);
    state.flags = flags;
    state.rtt = rtt;
    state.src_rtt = src_rtt;
 
    icp_opcode codeToSend;
 
    if (state.isHit()) {
        codeToSend = ICP_HIT;
    } else {
#if USE_ICMP
        if (Config.onoff.test_reachability && state.rtt == 0) {
            if ((state.rtt = netdbHostRtt(state.request->url.host())) == 0)
                netdbPingSite(state.request->url.host());
        }
#endif /* USE_ICMP */
 
        if (icpGetCommonOpcode() != ICP_ERR)
            codeToSend = icpGetCommonOpcode();
        else if (Config.onoff.test_reachability && rtt == 0)
            codeToSend = ICP_MISS_NOFETCH;
        else
            codeToSend = ICP_MISS;
    }
 
    icpCreateAndSend(codeToSend, flags, url, header.reqnum, src_rtt, fd, from, state.al);
}
 
void
icp_common_t::handleReply(char *buf, Ip::Address &from)
{
    if (neighbors_do_private_keys && reqnum == 0) {
        debugs(12, DBG_CRITICAL, "icpHandleIcpV2: Neighbor " << from << " returned reqnum = 0");
        debugs(12, DBG_CRITICAL, "icpHandleIcpV2: Disabling use of private keys");
        neighbors_do_private_keys = 0;
    }
 
    const auto url = icpGetUrl(from, buf, *this);
    if (!url)
        return;
    debugs(12, 3, "icpHandleIcpV2: " << icp_opcode_str[opcode] << " from " << from << " for '" << url << "'");
 
    const cache_key *key = icpGetCacheKey(url, (int) reqnum);
    /* call neighborsUdpAck even if ping_status != PING_WAITING */
    neighborsUdpAck(key, this, from);
}
 
static void
icpHandleIcpV2(int fd, Ip::Address &from, char *buf, int len)
{
    if (len <= 0) {
        debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
        return;
    }
 
    icp_common_t header(buf, len);
    /*
     * Length field should match the number of bytes read
     */
 
    if (len != header.length) {
        debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
        return;
    }
 
    debugs(12, 5, "OPCODE " << icp_opcode_str[header.getOpCode()] << '=' << uint8_t(header.opcode));
 
    switch (header.opcode) {
 
    case ICP_QUERY:
        /* We have a valid packet */
        doV2Query(fd, from, buf, header);
        break;
 
    case ICP_HIT:
 
    case ICP_DECHO:
 
    case ICP_MISS:
 
    case ICP_DENIED:
 
    case ICP_MISS_NOFETCH:
        header.handleReply(buf, from);
        break;
 
    case ICP_INVALID:
 
    case ICP_ERR:
        break;
 
    default:
        debugs(12, DBG_CRITICAL, "ERROR: icpHandleIcpV2: Unknown opcode: " << header.opcode << " from " << from);
 
        break;
    }
}
 
void
icpHandleUdp(int sock, void *)
{
    int *N = &incoming_sockets_accepted;
 
    Ip::Address from;
    LOCAL_ARRAY(char, buf, SQUID_UDP_SO_RCVBUF);
    int len;
    int icp_version;
    int max = INCOMING_UDP_MAX;
    Comm::SetSelect(sock, COMM_SELECT_READ, icpHandleUdp, nullptr, 0);
 
    while (max) {
        --max;
        len = comm_udp_recvfrom(sock,
                                buf,
                                SQUID_UDP_SO_RCVBUF - 1,
                                0,
                                from);
 
        if (len == 0)
            break;
 
        if (len < 0) {
            int xerrno = errno;
            if (ignoreErrno(xerrno))
                break;
 
#if _SQUID_LINUX_
            /* Some Linux systems seem to set the FD for reading and then
             * return ECONNREFUSED when sendto() fails and generates an ICMP
             * port unreachable message. */
            /* or maybe an EHOSTUNREACH "No route to host" message */
            if (xerrno != ECONNREFUSED && xerrno != EHOSTUNREACH)
#endif
                debugs(50, DBG_IMPORTANT, "icpHandleUdp: FD " << sock << " recvfrom: " << xstrerr(xerrno));
 
            break;
        }
 
        ++(*N);
        icpCount(buf, RECV, (size_t) len, 0);
        buf[len] = '\0';
        debugs(12, 4, "icpHandleUdp: FD " << sock << ": received " <<
               (unsigned long int)len << " bytes from " << from);
 
        if ((size_t) len < sizeof(icp_common_t)) {
            debugs(12, 4, "icpHandleUdp: Ignoring too-small UDP packet");
            break;
        }
 
        icp_version = (int) buf[1]; /* cheat! */
 
        // XXX: The IP equality comparison below ignores port differences but
        // should not. It also fails to detect loops when `local` is a wildcard
        // address (e.g., [::]:3130) because `from` address is never a wildcard.
        if (icpOutgoingConn && icpOutgoingConn->local == from)
            // ignore ICP packets which loop back (multicast usually)
            debugs(12, 4, "icpHandleUdp: Ignoring UDP packet sent by myself");
        else if (icp_version == ICP_VERSION_2)
            icpHandleIcpV2(sock, from, buf, len);
        else if (icp_version == ICP_VERSION_3)
            icpHandleIcpV3(sock, from, buf, len);
        else
            debugs(12, DBG_IMPORTANT, "WARNING: Unused ICP version " << icp_version <<
                   " received from " << from);
    }
}
 
void
icpOpenPorts(void)
{
    uint16_t port;
 
    if ((port = Config.Port.icp) <= 0)
        return;
 
    icpIncomingConn = new Comm::Connection;
    icpIncomingConn->local = Config.Addrs.udp_incoming;
    icpIncomingConn->local.port(port);
 
    if (!Ip::EnableIpv6 && !icpIncomingConn->local.setIPv4()) {
        debugs(12, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << icpIncomingConn->local << " is not an IPv4 address.");
        fatal("ICP port cannot be opened.");
    }
    /* split-stack for now requires default IPv4-only ICP */
    if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && icpIncomingConn->local.isAnyAddr()) {
        icpIncomingConn->local.setIPv4();
    }
 
    auto call = asyncCallbackFun(12, 2, icpIncomingConnectionOpened);
    Ipc::StartListening(SOCK_DGRAM,
                        IPPROTO_UDP,
                        icpIncomingConn,
                        Ipc::fdnInIcpSocket, call);
 
    if ( !Config.Addrs.udp_outgoing.isNoAddr() ) {
        icpOutgoingConn = new Comm::Connection;
        icpOutgoingConn->local = Config.Addrs.udp_outgoing;
        icpOutgoingConn->local.port(port);
 
        if (!Ip::EnableIpv6 && !icpOutgoingConn->local.setIPv4()) {
            debugs(49, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << icpOutgoingConn->local << " is not an IPv4 address.");
            fatal("ICP port cannot be opened.");
        }
        /* split-stack for now requires default IPv4-only ICP */
        if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && icpOutgoingConn->local.isAnyAddr()) {
            icpOutgoingConn->local.setIPv4();
        }
 
        enter_suid();
        comm_open_listener(SOCK_DGRAM, IPPROTO_UDP, icpOutgoingConn, "Outgoing ICP Port");
        leave_suid();
 
        if (!Comm::IsConnOpen(icpOutgoingConn))
            fatal("Cannot open Outgoing ICP Port");
 
        debugs(12, DBG_CRITICAL, "Sending ICP messages from " << icpOutgoingConn->local);
 
        Comm::SetSelect(icpOutgoingConn->fd, COMM_SELECT_READ, icpHandleUdp, nullptr, 0);
        fd_note(icpOutgoingConn->fd, "Outgoing ICP socket");
    }
}
 
static void
icpIncomingConnectionOpened(Ipc::StartListeningAnswer &answer)
{
    const auto &conn = answer.conn;
 
    if (!Comm::IsConnOpen(conn))
        fatal("Cannot open ICP Port");
 
    Comm::SetSelect(conn->fd, COMM_SELECT_READ, icpHandleUdp, nullptr, 0);
 
    for (const wordlist *s = Config.mcast_group_list; s; s = s->next)
        ipcache_nbgethostbyname(s->key, mcastJoinGroups, nullptr); // XXX: pass the conn for mcastJoinGroups usage.
 
    debugs(12, DBG_IMPORTANT, "Accepting ICP messages on " << conn->local);
 
    fd_note(conn->fd, "Incoming ICP port");
 
    if (Config.Addrs.udp_outgoing.isNoAddr()) {
        icpOutgoingConn = conn;
        debugs(12, DBG_IMPORTANT, "Sending ICP messages from " << icpOutgoingConn->local);
    }
}
 
void
icpConnectionShutdown(void)
{
    if (!Comm::IsConnOpen(icpIncomingConn))
        return;
 
    debugs(12, DBG_IMPORTANT, "Stop receiving ICP on " << icpIncomingConn->local);
 
    icpIncomingConn = nullptr;
 
    assert(Comm::IsConnOpen(icpOutgoingConn));
 
    Comm::SetSelect(icpOutgoingConn->fd, COMM_SELECT_READ, nullptr, nullptr, 0);
}
 
void
icpClosePorts(void)
{
    icpConnectionShutdown();
 
    if (icpOutgoingConn != nullptr) {
        debugs(12, DBG_IMPORTANT, "Stop sending ICP from " << icpOutgoingConn->local);
        icpOutgoingConn = nullptr;
    }
}
 
static void
icpCount(void *buf, int which, size_t len, int delay)
{
    icp_common_t *icp = (icp_common_t *) buf;
 
    if (len < sizeof(*icp))
        return;
 
    if (SENT == which) {
        ++statCounter.icp.pkts_sent;
        statCounter.icp.kbytes_sent += len;
 
        if (ICP_QUERY == icp->opcode) {
            ++statCounter.icp.queries_sent;
            statCounter.icp.q_kbytes_sent += len;
        } else {
            ++statCounter.icp.replies_sent;
            statCounter.icp.r_kbytes_sent += len;
            /* this is the sent-reply service time */
            statCounter.icp.replySvcTime.count(delay);
        }
 
        if (ICP_HIT == icp->opcode)
            ++statCounter.icp.hits_sent;
    } else if (RECV == which) {
        ++statCounter.icp.pkts_recv;
        statCounter.icp.kbytes_recv += len;
 
        if (ICP_QUERY == icp->opcode) {
            ++statCounter.icp.queries_recv;
            statCounter.icp.q_kbytes_recv += len;
        } else {
            ++statCounter.icp.replies_recv;
            statCounter.icp.r_kbytes_recv += len;
            /* statCounter.icp.querySvcTime set in clientUpdateCounters */
        }
 
        if (ICP_HIT == icp->opcode)
            ++statCounter.icp.hits_recv;
    }
}
 
#define N_QUERIED_KEYS 8192
#define N_QUERIED_KEYS_MASK 8191
static cache_key queried_keys[N_QUERIED_KEYS][SQUID_MD5_DIGEST_LENGTH];
 
int
icpSetCacheKey(const cache_key * key)
{
    static int reqnum = 0;
 
    if (++reqnum < 0)
        reqnum = 1;
 
    storeKeyCopy(queried_keys[reqnum & N_QUERIED_KEYS_MASK], key);
 
    return reqnum;
}
 
const cache_key *
icpGetCacheKey(const char *url, int reqnum)
{
    if (neighbors_do_private_keys && reqnum)
        return queried_keys[reqnum & N_QUERIED_KEYS_MASK];
 
    return storeKeyPublic(url, Http::METHOD_GET);
}