icp_v2.c

-

    case ICP_MISS_NOFETCH:
	if (neighbors_do_private_keys && header.reqnum == 0) {
	    debug(12, 0) ("icpHandleIcpV2: Neighbor %s returned reqnum = 0\n",
		inet_ntoa(from.sin_addr));
	    debug(12, 0) ("icpHandleIcpV2: Disabling use of private keys\n");
	    neighbors_do_private_keys = 0;
	}
	url = buf + sizeof(icp_common_t);
	debug(12, 3) ("icpHandleIcpV2: %s from %s for '%s'\n",
	    icp_opcode_str[header.opcode],
	    inet_ntoa(from.sin_addr),
	    url);
	key = icpGetCacheKey(url, (int) header.reqnum);
	/* call neighborsUdpAck even if ping_status != PING_WAITING */
	neighborsUdpAck(key, &header, &from);
	break;

    case ICP_INVALID:
    case ICP_ERR:
	break;

    default:
	debug(12, 0) ("icpHandleIcpV2: UNKNOWN OPCODE: %d from %s\n",
	    header.opcode, inet_ntoa(from.sin_addr));
	break;
    }
    if (icp_request)
	requestDestroy(icp_request);
}

#ifdef ICP_PKT_DUMP
static void
icpPktDump(icp_common_t * pkt)
{
    struct in_addr a;

    debug(12, 9) ("opcode:     %3d %s\n",
	(int) pkt->opcode,
	icp_opcode_str[pkt->opcode]);
    debug(12, 9) ("version: %-8d\n", (int) pkt->version);
    debug(12, 9) ("length:  %-8d\n", (int) ntohs(pkt->length));
    debug(12, 9) ("reqnum:  %-8d\n", ntohl(pkt->reqnum));
    debug(12, 9) ("flags:   %-8x\n", ntohl(pkt->flags));
    a.s_addr = pkt->shostid;
    debug(12, 9) ("shostid: %s\n", inet_ntoa(a));
    debug(12, 9) ("payload: %s\n", (char *) pkt + sizeof(icp_common_t));
}
#endif

void
icpHandleUdp(int sock, void *data)
{
    int *N = data;
    struct sockaddr_in from;
    socklen_t from_len;
    LOCAL_ARRAY(char, buf, SQUID_UDP_SO_RCVBUF);
    int len;
    int icp_version;
    int max = INCOMING_ICP_MAX;
    commSetSelect(sock, COMM_SELECT_READ, icpHandleUdp, NULL, 0);
    while (max--) {
	from_len = sizeof(from);
	memset(&from, '\0', from_len);
	Counter.syscalls.sock.recvfroms++;
	len = recvfrom(sock,
	    buf,
	    SQUID_UDP_SO_RCVBUF - 1,
	    0,
	    (struct sockaddr *) &from,
	    &from_len);
	if (len == 0)
	    break;
	if (len < 0) {
	    if (ignoreErrno(errno))
		break;
#ifdef _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 (errno != ECONNREFUSED && errno != EHOSTUNREACH)
#endif
		debug(50, 1) ("icpHandleUdp: FD %d recvfrom: %s\n",
		    sock, xstrerror());
	    break;
	}
	(*N)++;
	icpCount(buf, RECV, (size_t) len, 0);
	buf[len] = '\0';
	debug(12, 4) ("icpHandleUdp: FD %d: received %d bytes from %s.\n",
	    sock,
	    len,
	    inet_ntoa(from.sin_addr));
#ifdef ICP_PACKET_DUMP
	icpPktDump(buf);
#endif
	if (len < sizeof(icp_common_t)) {
	    debug(12, 4) ("icpHandleUdp: Ignoring too-small UDP packet\n");
	    break;
	}
	icp_version = (int) buf[1];	/* cheat! */
	if (icp_version == ICP_VERSION_2)
	    icpHandleIcpV2(sock, from, buf, len);
	else if (icp_version == ICP_VERSION_3)
	    icpHandleIcpV3(sock, from, buf, len);
	else
	    debug(12, 1) ("WARNING: Unused ICP version %d received from %s:%d\n",
		icp_version,
		inet_ntoa(from.sin_addr),
		ntohs(from.sin_port));
    }
}

void
icpConnectionsOpen(void)
{
    u_short port;
    struct in_addr addr;
    struct sockaddr_in xaddr;
    int x;
    socklen_t len;
    wordlist *s;
    if (Config2.Accel.on && !Config.onoff.accel_with_proxy)
	return;
    if ((port = Config.Port.icp) <= 0)
	return;
    enter_suid();
    theInIcpConnection = comm_open(SOCK_DGRAM,
	0,
	Config.Addrs.udp_incoming,
	port,
	COMM_NONBLOCKING,
	"ICP Socket");
    leave_suid();
    if (theInIcpConnection < 0)
	fatal("Cannot open ICP Port");
    commSetSelect(theInIcpConnection,
	COMM_SELECT_READ,
	icpHandleUdp,
	NULL,
	0);
    for (s = Config.mcast_group_list; s; s = s->next)
	ipcache_nbgethostbyname(s->key, mcastJoinGroups, NULL);
    debug(12, 1) ("Accepting ICP messages on port %d, FD %d.\n",
	(int) port, theInIcpConnection);
    if ((addr = Config.Addrs.udp_outgoing).s_addr != no_addr.s_addr) {
	enter_suid();
	theOutIcpConnection = comm_open(SOCK_DGRAM,
	    0,
	    addr,
	    port,
	    COMM_NONBLOCKING,
	    "ICP Port");
	leave_suid();
	if (theOutIcpConnection < 0)
	    fatal("Cannot open Outgoing ICP Port");
	commSetSelect(theOutIcpConnection,
	    COMM_SELECT_READ,
	    icpHandleUdp,
	    NULL,
	    0);
	debug(12, 1) ("Outgoing ICP messages on port %d, FD %d.\n",
	    (int) port, theOutIcpConnection);
	fd_note(theOutIcpConnection, "Outgoing ICP socket");
	fd_note(theInIcpConnection, "Incoming ICP socket");
    } else {
	theOutIcpConnection = theInIcpConnection;
    }
    memset(&theOutICPAddr, '\0', sizeof(struct in_addr));
    len = sizeof(struct sockaddr_in);
    memset(&xaddr, '\0', len);
    x = getsockname(theOutIcpConnection,
	(struct sockaddr *) &xaddr, &len);
    if (x < 0)
	debug(50, 1) ("theOutIcpConnection FD %d: getsockname: %s\n",
	    theOutIcpConnection, xstrerror());
    else
	theOutICPAddr = xaddr.sin_addr;
}

/*
 * icpConnectionShutdown only closes the 'in' socket if it is 
 * different than the 'out' socket.
 */
void
icpConnectionShutdown(void)
{
    if (theInIcpConnection < 0)
	return;
    if (theInIcpConnection != theOutIcpConnection) {
	debug(12, 1) ("FD %d Closing ICP connection\n", theInIcpConnection);
	comm_close(theInIcpConnection);
    }
    /*
     * Here we set 'theInIcpConnection' to -1 even though the ICP 'in'
     * and 'out' sockets might be just one FD.  This prevents this
     * function from executing repeatedly.  When we are really ready to
     * exit or restart, main will comm_close the 'out' descriptor.
     */
    theInIcpConnection = -1;
    /*
     * Normally we only write to the outgoing ICP socket, but
     * we also have a read handler there to catch messages sent
     * to that specific interface.  During shutdown, we must
     * disable reading on the outgoing socket.
     */
    assert(theOutIcpConnection > -1);
    commSetSelect(theOutIcpConnection, COMM_SELECT_READ, NULL, NULL, 0);
}

void
icpConnectionClose(void)
{
    icpConnectionShutdown();
    if (theOutIcpConnection > -1) {
	debug(12, 1) ("FD %d Closing ICP connection\n", theOutIcpConnection);
	comm_close(theOutIcpConnection);
	theOutIcpConnection = -1;
    }
}

static void
icpCount(void *buf, int which, size_t len, int delay)
{
    icp_common_t *icp = buf;
    if (len < sizeof(*icp))
	return;
    if (SENT == which) {
	Counter.icp.pkts_sent++;
	kb_incr(&Counter.icp.kbytes_sent, len);
	if (ICP_QUERY == icp->opcode) {
	    Counter.icp.queries_sent++;
	    kb_incr(&Counter.icp.q_kbytes_sent, len);
	} else {
	    Counter.icp.replies_sent++;
	    kb_incr(&Counter.icp.r_kbytes_sent, len);
	    /* this is the sent-reply service time */
	    statHistCount(&Counter.icp.reply_svc_time, delay);
	}
	if (ICP_HIT == icp->opcode)
	    Counter.icp.hits_sent++;
    } else if (RECV == which) {
	Counter.icp.pkts_recv++;
	kb_incr(&Counter.icp.kbytes_recv, len);
	if (ICP_QUERY == icp->opcode) {
	    Counter.icp.queries_recv++;
	    kb_incr(&Counter.icp.q_kbytes_recv, len);
	} else {
	    Counter.icp.replies_recv++;
	    kb_incr(&Counter.icp.r_kbytes_recv, len);
	    /* Counter.icp.query_svc_time set in clientUpdateCounters */
	}
	if (ICP_HIT == icp->opcode)
	    Counter.icp.hits_recv++;
    }
}

#define N_QUERIED_KEYS 8192
#define N_QUERIED_KEYS_MASK 8191
static cache_key queried_keys[N_QUERIED_KEYS][MD5_DIGEST_CHARS];

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, METHOD_GET);
}