neighbors.c

-

	    /*
	     * fake a recent reply if its been a long time since our
	     * last query
	     */
	    p->stats.last_reply = squid_curtime;
	    /*
	     * We used to not expect a reply in this case; we assumed
	     * the peer was DEAD if we hadn't queried it in a long
	     * time.  However, the number of people whining to
	     * squid-users that ICP is broken became unbearable.  They
	     * tried a single request which, to their amazement, was
	     * forwarded directly to the origin server, even thought
	     * they KNEW it was in a neighbor cache.  Ok, I give up, you
	     * win!
	     */
	    (*exprep)++;
	    (*timeout) += 1000;
	} else if (neighborUp(p)) {
	    /* its alive, expect a reply from it */
	    (*exprep)++;
	    (*timeout) += p->stats.rtt;
	} else {
	    /* Neighbor is dead; ping it anyway, but don't expect a reply */
	    /* log it once at the threshold */
	    if (p->stats.logged_state == PEER_ALIVE) {
		debug(15, 1) ("Detected DEAD %s: %s/%d/%d\n",
		    neighborTypeStr(p),
		    p->host, p->http_port, p->icp.port);
		p->stats.logged_state = PEER_DEAD;
	    }
	}
	p->stats.last_query = squid_curtime;
    }
    if ((first_ping = first_ping->next) == NULL)
	first_ping = Config.peers;

#if ALLOW_SOURCE_PING
    /* only do source_ping if we have neighbors */
    if (Config.npeers) {
	const ipcache_addrs *ia = NULL;
	struct sockaddr_in to_addr;
	char *host = request->host;
	if (!Config.onoff.source_ping) {
	    debug(15, 6) ("neighborsUdpPing: Source Ping is disabled.\n");
	} else if ((ia = ipcache_gethostbyname(host, 0))) {
	    debug(15, 6) ("neighborsUdpPing: Source Ping: to %s for '%s'\n",
		host, url);
	    echo_hdr.reqnum = reqnum;
	    if (icmp_sock != -1) {
		icmpSourcePing(ia->in_addrs[ia->cur], &echo_hdr, url);
	    } else {
		to_addr.sin_family = AF_INET;
		to_addr.sin_addr = ia->in_addrs[ia->cur];
		to_addr.sin_port = htons(echo_port);
		query = icpCreateMessage(ICP_SECHO, 0, url, reqnum, 0);
		icpUdpSend(theOutIcpConnection,
		    &to_addr,
		    query,
		    LOG_ICP_QUERY,
		    0);
	    }
	} else {
	    debug(15, 6) ("neighborsUdpPing: Source Ping: unknown host: %s\n",
		host);
	}
    }
#endif
    /*
     * If there is a configured timeout, use it
     */
    if (Config.Timeout.icp_query)
	*timeout = Config.Timeout.icp_query;
    else if (*exprep > 0)
	(*timeout) = 2 * (*timeout) / (*exprep);
    else
	*timeout = 2000;	/* 2 seconds */
    return peers_pinged;
}

/* lookup the digest of a given peer */
lookup_t
peerDigestLookup(peer * p, request_t * request, StoreEntry * entry)
{
#if USE_CACHE_DIGESTS
    const cache_key *key = request ? storeKeyPublic(storeUrl(entry), request->method) : NULL;
    assert(p);
    assert(request);
    debug(15, 5) ("peerDigestLookup: peer %s\n", p->host);
    /* does the peeer have a valid digest? */
    if (!p->digest) {
	debug(15, 5) ("peerDigestLookup: gone!\n");
	return LOOKUP_NONE;
    } else if (!peerHTTPOkay(p, request)) {
	debug(15, 5) ("peerDigestLookup: !peerHTTPOkay\n");
	return LOOKUP_NONE;
    } else if (p->digest->flags.usable) {
	debug(15, 5) ("peerDigestLookup: usable\n");
	/* fall through; put here to have common case on top */ ;
    } else if (!p->digest->flags.needed) {
	debug(15, 5) ("peerDigestLookup: note need\n");
	peerDigestNeeded(p->digest);
	return LOOKUP_NONE;
    } else {
	debug(15, 5) ("peerDigestLookup: !ready && %srequested\n",
	    p->digest->flags.requested ? "" : "!");
	return LOOKUP_NONE;
    }
    debug(15, 5) ("peerDigestLookup: OK to lookup peer %s\n", p->host);
    assert(p->digest->cd);
    /* does digest predict a hit? */
    if (!cacheDigestTest(p->digest->cd, key))
	return LOOKUP_MISS;
    debug(15, 5) ("peerDigestLookup: peer %s says HIT!\n", p->host);
    return LOOKUP_HIT;
#endif
    return LOOKUP_NONE;
}

/* select best peer based on cache digests */
peer *
neighborsDigestSelect(request_t * request, StoreEntry * entry)
{
    peer *best_p = NULL;
#if USE_CACHE_DIGESTS
    const cache_key *key;
    int best_rtt = 0;
    int choice_count = 0;
    int ichoice_count = 0;
    peer *p;
    int p_rtt;
    int i;
    if (!request->flags.hierarchical)
	return NULL;
    key = storeKeyPublic(storeUrl(entry), request->method);
    for (i = 0, p = first_ping; i++ < Config.npeers; p = p->next) {
	lookup_t lookup;
	if (!p)
	    p = Config.peers;
	if (i == 1)
	    first_ping = p;
	lookup = peerDigestLookup(p, request, entry);
	if (lookup == LOOKUP_NONE)
	    continue;
	choice_count++;
	if (lookup == LOOKUP_MISS)
	    continue;
	p_rtt = netdbHostRtt(p->host);
	debug(15, 5) ("neighborsDigestSelect: peer %s rtt: %d\n",
	    p->host, p_rtt);
	/* is this peer better than others in terms of rtt ? */
	if (!best_p || (p_rtt && p_rtt < best_rtt)) {
	    best_p = p;
	    best_rtt = p_rtt;
	    if (p_rtt)		/* informative choice (aka educated guess) */
		ichoice_count++;
	    debug(15, 4) ("neighborsDigestSelect: peer %s leads with rtt %d\n",
		p->host, best_rtt);
	}
    }
    debug(15, 4) ("neighborsDigestSelect: choices: %d (%d)\n",
	choice_count, ichoice_count);
    peerNoteDigestLookup(request, best_p,
	best_p ? LOOKUP_HIT : (choice_count ? LOOKUP_MISS : LOOKUP_NONE));
    request->hier.n_choices = choice_count;
    request->hier.n_ichoices = ichoice_count;
#endif
    return best_p;
}

void
peerNoteDigestLookup(request_t * request, peer * p, lookup_t lookup)
{
#if USE_CACHE_DIGESTS
    if (p)
	strncpy(request->hier.cd_host, p->host, sizeof(request->hier.cd_host));
    else
	*request->hier.cd_host = '\0';
    request->hier.cd_lookup = lookup;
    debug(15, 4) ("peerNoteDigestLookup: peer %s, lookup: %s\n",
	p ? p->host : "<none>", lookup_t_str[lookup]);
#endif
}

static void
neighborAlive(peer * p, const MemObject * mem, const icp_common_t * header)
{
    if (p->stats.logged_state == PEER_DEAD && p->tcp_up) {
	debug(15, 1) ("Detected REVIVED %s: %s/%d/%d\n",
	    neighborTypeStr(p),
	    p->host, p->http_port, p->icp.port);
	p->stats.logged_state = PEER_ALIVE;
    }
    p->stats.last_reply = squid_curtime;
    p->stats.pings_acked++;
    if ((icp_opcode) header->opcode <= ICP_END)
	p->icp.counts[header->opcode]++;
    p->icp.version = (int) header->version;
}

static void
neighborUpdateRtt(peer * p, MemObject * mem)
{
    int rtt;
    if (!mem)
	return;
    if (!mem->start_ping.tv_sec)
	return;
    rtt = tvSubMsec(mem->start_ping, current_time);
    if (rtt < 1 || rtt > 10000)
	return;
    p->stats.rtt = intAverage(p->stats.rtt, rtt,
	p->stats.pings_acked, RTT_AV_FACTOR);
}

#if USE_HTCP
static void
neighborAliveHtcp(peer * p, const MemObject * mem, const htcpReplyData * htcp)
{
    if (p->stats.logged_state == PEER_DEAD && p->tcp_up) {
	debug(15, 1) ("Detected REVIVED %s: %s/%d/%d\n",
	    neighborTypeStr(p),
	    p->host, p->http_port, p->icp.port);
	p->stats.logged_state = PEER_ALIVE;
    }
    p->stats.last_reply = squid_curtime;
    p->stats.pings_acked++;
    p->htcp.counts[htcp->hit ? 1 : 0]++;
    p->htcp.version = htcp->version;
}
#endif

static void
neighborCountIgnored(peer * p)
{
    if (p == NULL)
	return;
    p->stats.ignored_replies++;
    NLateReplies++;
}

static peer *non_peers = NULL;

static void
neighborIgnoreNonPeer(const struct sockaddr_in *from, icp_opcode opcode)
{
    peer *np;
    double x;
    for (np = non_peers; np; np = np->next) {
	if (np->in_addr.sin_addr.s_addr != from->sin_addr.s_addr)
	    continue;
	if (np->in_addr.sin_port != from->sin_port)
	    continue;
	break;
    }
    if (np == NULL) {
	np = xcalloc(1, sizeof(peer));
	np->in_addr.sin_addr = from->sin_addr;
	np->in_addr.sin_port = from->sin_port;
	np->icp.port = ntohl(from->sin_port);
	np->type = PEER_NONE;
	np->host = xstrdup(inet_ntoa(from->sin_addr));
	np->next = non_peers;
	non_peers = np;
    }
    np->stats.ignored_replies++;
    np->icp.counts[opcode]++;
    x = log(np->stats.ignored_replies) / log(10.0);
    if (0.0 != x - (double) (int) x)
	return;
    debug(15, 1) ("WARNING: Ignored %d replies from non-peer %s\n",
	np->stats.ignored_replies, np->host);
}

/* ignoreMulticastReply
 * 
 * We want to ignore replies from multicast peers if the
 * cache_host_domain rules would normally prevent the peer
 * from being used
 */
static int
ignoreMulticastReply(peer * p, MemObject * mem)
{
    if (p == NULL)
	return 0;
    if (!p->options.mcast_responder)
	return 0;
    if (peerHTTPOkay(p, mem->request))
	return 0;
    return 1;
}

/* I should attach these records to the entry.  We take the first
 * hit we get our wait until everyone misses.  The timeout handler
 * call needs to nip this shopping list or call one of the misses.
 * 
 * If a hit process is already started, then sobeit
 */
void
neighborsUdpAck(const cache_key * key, icp_common_t * header, const struct sockaddr_in *from)
{
    peer *p = NULL;
    StoreEntry *entry;
    MemObject *mem = NULL;
    peer_t ntype = PEER_NONE;
    char *opcode_d;
    icp_opcode opcode = (icp_opcode) header->opcode;

    debug(15, 6) ("neighborsUdpAck: opcode %d '%s'\n",
	(int) opcode, storeKeyText(key));
    if (NULL != (entry = storeGet(key)))
	mem = entry->mem_obj;
    if ((p = whichPeer(from)))
	neighborAlive(p, mem, header);
    if (opcode > ICP_END)
	return;
    opcode_d = icp_opcode_str[opcode];
    if (p)
	neighborUpdateRtt(p, mem);
    /* Does the entry exist? */
    if (NULL == entry) {
	debug(12, 3) ("neighborsUdpAck: Cache key '%s' not found\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    /* check if someone is already fetching it */
    if (EBIT_TEST(entry->flags, ENTRY_DISPATCHED)) {
	debug(15, 3) ("neighborsUdpAck: '%s' already being fetched.\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (mem == NULL) {
	debug(15, 2) ("Ignoring %s for missing mem_obj: %s\n",
	    opcode_d, storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (entry->ping_status != PING_WAITING) {
	debug(15, 2) ("neighborsUdpAck: Late %s for %s\n",
	    opcode_d, storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (entry->lock_count == 0) {
	debug(12, 1) ("neighborsUdpAck: '%s' has no locks\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    debug(15, 3) ("neighborsUdpAck: %s for '%s' from %s \n",
	opcode_d, storeKeyText(key), p ? p->host : "source");
    if (p) {
	ntype = neighborType(p, mem->request);
    }
    if (ignoreMulticastReply(p, mem)) {
	neighborCountIgnored(p);
    } else if (opcode == ICP_MISS) {
	if (p == NULL) {
	    neighborIgnoreNonPeer(from, opcode);
	} else {
	    mem->ping_reply_callback(p, ntype, PROTO_ICP, header, mem->ircb_data);
	}
    } else if (opcode == ICP_HIT) {
	if (p == NULL) {
	    neighborIgnoreNonPeer(from, opcode);
	} else {
	    header->opcode = ICP_HIT;
	    mem->ping_reply_callback(p, ntype, PROTO_ICP, header, mem->ircb_data);
	}
    } else if (opcode == ICP_DECHO) {
	if (p == NULL) {
	    neighborIgnoreNonPeer(from, opcode);
	} else if (ntype == PEER_SIBLING) {
	    debug_trap("neighborsUdpAck: Found non-ICP cache as SIBLING\n");
	    debug_trap("neighborsUdpAck: non-ICP neighbors must be a PARENT\n");
	} else {
	    mem->ping_reply_callback(p, ntype, PROTO_ICP, header, mem->ircb_data);
	}
    } else if (opcode == ICP_SECHO) {
	if (p) {
	    debug(15, 1) ("Ignoring SECHO from neighbor %s\n", p->host);
	    neighborCountIgnored(p);
#if ALLOW_SOURCE_PING
	} else if (Config.onoff.source_ping) {
	    mem->ping_reply_callback(NULL, ntype, PROTO_ICP, header, mem->ircb_data);
#endif
	} else {
	    debug(15, 1) ("Unsolicited SECHO from %s\n", inet_ntoa(from->sin_addr));
	}
    } else if (opcode == ICP_DENIED) {
	if (p == NULL) {
	    neighborIgnoreNonPeer(from, opcode);
	} else if (p->stats.pings_acked > 100) {
	    if (100 * p->icp.counts[ICP_DENIED] / p->stats.pings_acked > 95) {
		debug(15, 0) ("95%% of replies from '%s' are UDP_DENIED\n", p->host);
		debug(15, 0) ("Disabling '%s', please check your configuration.\n", p->host);
		neighborRemove(p);
		p = NULL;
	    } else {
		neighborCountIgnored(p);
	    }
	}
    } else if (opcode == ICP_MISS_NOFETCH) {
	mem->ping_reply_callback(p, ntype, PROTO_ICP, header, mem->ircb_data);
    } else {
	debug(15, 0) ("neighborsUdpAck: Unexpected ICP reply: %s\n", opcode_d);
    }
}

peer *
peerFindByName(const char *name)
{
    peer *p = NULL;
    for (p = Config.peers; p; p = p->next) {
	if (!strcasecmp(name, p->host))
	    break;
    }
    return p;
}

peer *
peerFindByNameAndPort(const char *name, unsigned short port)
{
    peer *p = NULL;
    for (p = Config.peers; p; p = p->next) {
	if (strcasecmp(name, p->host))
	    continue;
	if (port != p->http_port)
	    continue;
	break;
    }
    return p;
}

int