neighbors.c

-

neighborUp(const peer * p)
{
    if (!p->tcp_up)
	return 0;
    if (squid_curtime - p->stats.last_query > Config.Timeout.deadPeer)
	return 1;
    if (p->stats.last_query - p->stats.last_reply > Config.Timeout.deadPeer)
	return 0;
    return 1;
}

void
peerDestroy(void *data, int unused)
{
    peer *p = data;
    struct _domain_ping *l = NULL;
    struct _domain_ping *nl = NULL;
    if (p == NULL)
	return;
    for (l = p->peer_domain; l; l = nl) {
	nl = l->next;
	safe_free(l->domain);
	safe_free(l);
    }
    safe_free(p->host);
#if USE_CACHE_DIGESTS
    if (p->digest) {
	PeerDigest *pd = p->digest;
	p->digest = NULL;
	cbdataUnlock(pd);
    }
#endif
    xfree(p);
}

void
peerNoteDigestGone(peer * p)
{
#if USE_CACHE_DIGESTS
    if (p->digest) {
	PeerDigest *pd = p->digest;
	p->digest = NULL;
	cbdataUnlock(pd);
    }
#endif
}

static void
peerDNSConfigure(const ipcache_addrs * ia, void *data)
{
    peer *p = data;
    struct sockaddr_in *ap;
    int j;
    if (p->n_addresses == 0) {
	debug(15, 1) ("Configuring %s %s/%d/%d\n", neighborTypeStr(p),
	    p->host, p->http_port, p->icp.port);
	if (p->type == PEER_MULTICAST)
	    debug(15, 1) ("    Multicast TTL = %d\n", p->mcast.ttl);
    }
    p->n_addresses = 0;
    if (ia == NULL) {
	debug(0, 0) ("WARNING: DNS lookup for '%s' failed!\n", p->host);
	return;
    }
    if ((int) ia->count < 1) {
	debug(0, 0) ("WARNING: No IP address found for '%s'!\n", p->host);
	return;
    }
    for (j = 0; j < (int) ia->count && j < PEER_MAX_ADDRESSES; j++) {
	p->addresses[j] = ia->in_addrs[j];
	debug(15, 2) ("--> IP address #%d: %s\n", j, inet_ntoa(p->addresses[j]));
	p->n_addresses++;
    }
    ap = &p->in_addr;
    memset(ap, '\0', sizeof(struct sockaddr_in));
    ap->sin_family = AF_INET;
    ap->sin_addr = p->addresses[0];
    ap->sin_port = htons(p->icp.port);
    if (p->type == PEER_MULTICAST)
	peerCountMcastPeersSchedule(p, 10);
    if (p->type != PEER_MULTICAST)
	if (!p->options.no_netdb_exchange)
	    eventAddIsh("netdbExchangeStart", netdbExchangeStart, p, 30.0, 1);
}

static void
peerRefreshDNS(void *data)
{
    peer *p = NULL;
    if (eventFind(peerRefreshDNS, NULL))
	eventDelete(peerRefreshDNS, NULL);
    if (!data && 0 == stat5minClientRequests()) {
	/* no recent client traffic, wait a bit */
	eventAddIsh("peerRefreshDNS", peerRefreshDNS, NULL, 180.0, 1);
	return;
    }
    for (p = Config.peers; p; p = p->next)
	ipcache_nbgethostbyname(p->host, peerDNSConfigure, p);
    /* Reconfigure the peers every hour */
    eventAddIsh("peerRefreshDNS", peerRefreshDNS, NULL, 3600.0, 1);
}

/*
 * peerCheckConnect will NOT be called by eventRun if the peer/data
 * pointer becomes invalid.
 */
static void
peerCheckConnect(void *data)
{
    peer *p = data;
    int fd;
    fd = comm_open(SOCK_STREAM, 0, Config.Addrs.tcp_outgoing,
	0, COMM_NONBLOCKING, p->host);
    if (fd < 0)
	return;
    p->test_fd = fd;
    ipcache_nbgethostbyname(p->host, peerCheckConnect2, p);
}

static void
peerCheckConnect2(const ipcache_addrs * ianotused, void *data)
{
    peer *p = data;
    commConnectStart(p->test_fd,
	p->host,
	p->http_port,
	peerCheckConnectDone,
	p);
}

static void
peerCheckConnectDone(int fd, int status, void *data)
{
    peer *p = data;
    if (status == COMM_OK) {
	p->tcp_up = PEER_TCP_MAGIC_COUNT;
	debug(15, 1) ("TCP connection to %s/%d succeeded\n",
	    p->host, p->http_port);
    } else {
	eventAdd("peerCheckConnect", peerCheckConnect, p, 60.0, 1);
    }
    comm_close(fd);
    return;
}

void
peerCheckConnectStart(peer * p)
{
    if (!p->tcp_up)
	return;
    debug(15, 1) ("TCP connection to %s/%d failed\n", p->host, p->http_port);
    p->tcp_up--;
    if (p->tcp_up != (PEER_TCP_MAGIC_COUNT - 1))
	return;
    p->last_fail_time = squid_curtime;
    eventAdd("peerCheckConnect", peerCheckConnect, p, 30.0, 1);
}

static void
peerCountMcastPeersSchedule(peer * p, time_t when)
{
    if (p->mcast.flags.count_event_pending)
	return;
    eventAdd("peerCountMcastPeersStart",
	peerCountMcastPeersStart,
	p,
	(double) when, 1);
    p->mcast.flags.count_event_pending = 1;
}

static void
peerCountMcastPeersStart(void *data)
{
    peer *p = data;
    ps_state *psstate = xcalloc(1, sizeof(ps_state));
    StoreEntry *fake;
    MemObject *mem;
    icp_common_t *query;
    int reqnum;
    LOCAL_ARRAY(char, url, MAX_URL);
    assert(p->type == PEER_MULTICAST);
    p->mcast.flags.count_event_pending = 0;
    snprintf(url, MAX_URL, "http://%s/", inet_ntoa(p->in_addr.sin_addr));
    fake = storeCreateEntry(url, url, null_request_flags, METHOD_GET);
    psstate->request = requestLink(urlParse(METHOD_GET, url));
    psstate->entry = fake;
    psstate->callback = NULL;
    psstate->callback_data = p;
    psstate->ping.start = current_time;
    cbdataAdd(psstate, cbdataXfree, 0);
    mem = fake->mem_obj;
    mem->request = requestLink(psstate->request);
    mem->start_ping = current_time;
    mem->ping_reply_callback = peerCountHandleIcpReply;
    mem->ircb_data = psstate;
    mcastSetTtl(theOutIcpConnection, p->mcast.ttl);
    p->mcast.id = mem->id;
    reqnum = icpSetCacheKey(fake->key);
    query = icpCreateMessage(ICP_QUERY, 0, url, reqnum, 0);
    icpUdpSend(theOutIcpConnection,
	&p->in_addr,
	query,
	LOG_ICP_QUERY,
	0);
    fake->ping_status = PING_WAITING;
    eventAdd("peerCountMcastPeersDone",
	peerCountMcastPeersDone,
	psstate,
	(double) Config.Timeout.mcast_icp_query, 1);
    p->mcast.flags.counting = 1;
    peerCountMcastPeersSchedule(p, MCAST_COUNT_RATE);
}

static void
peerCountMcastPeersDone(void *data)
{
    ps_state *psstate = data;
    peer *p = psstate->callback_data;
    StoreEntry *fake = psstate->entry;
    p->mcast.flags.counting = 0;
    p->mcast.avg_n_members = doubleAverage(p->mcast.avg_n_members,
	(double) psstate->ping.n_recv,
	++p->mcast.n_times_counted,
	10);
    debug(15, 1) ("Group %s: %d replies, %4.1f average, RTT %d\n",
	p->host,
	psstate->ping.n_recv,
	p->mcast.avg_n_members,
	p->stats.rtt);
    p->mcast.n_replies_expected = (int) p->mcast.avg_n_members;
    EBIT_SET(fake->flags, ENTRY_ABORTED);
    requestUnlink(fake->mem_obj->request);
    fake->mem_obj->request = NULL;
    storeReleaseRequest(fake);
    storeUnlockObject(fake);
    requestUnlink(psstate->request);
    cbdataFree(psstate);
}

static void
peerCountHandleIcpReply(peer * p, peer_t type, protocol_t proto, void *hdrnotused, void *data)
{
    ps_state *psstate = data;
    StoreEntry *fake = psstate->entry;
    MemObject *mem = fake->mem_obj;
    int rtt = tvSubMsec(mem->start_ping, current_time);
    assert(proto == PROTO_ICP);
    assert(fake);
    assert(mem);
    psstate->ping.n_recv++;
    p->stats.rtt = intAverage(p->stats.rtt, rtt, psstate->ping.n_recv, RTT_AV_FACTOR);
}

static void
neighborDumpPeers(StoreEntry * sentry)
{
    dump_peers(sentry, Config.peers);
}

static void
neighborDumpNonPeers(StoreEntry * sentry)
{
    dump_peers(sentry, non_peers);
}

void
dump_peer_options(StoreEntry * sentry, peer * p)
{
    if (p->options.proxy_only)
	storeAppendPrintf(sentry, " proxy-only");
    if (p->options.no_query)
	storeAppendPrintf(sentry, " no-query");
    if (p->options.no_digest)
	storeAppendPrintf(sentry, " no-digest");
    if (p->options.default_parent)
	storeAppendPrintf(sentry, " default");
    if (p->options.roundrobin)
	storeAppendPrintf(sentry, " round-robin");
    if (p->options.mcast_responder)
	storeAppendPrintf(sentry, " multicast-responder");
    if (p->options.closest_only)
	storeAppendPrintf(sentry, " closest-only");
#if USE_HTCP
    if (p->options.htcp)
	storeAppendPrintf(sentry, " htcp");
#endif
    if (p->options.no_netdb_exchange)
	storeAppendPrintf(sentry, " no-netdb-exchange");
#if DELAY_POOLS
    if (p->options.no_delay)
	storeAppendPrintf(sentry, " no-delay");
#endif
    if (p->login)
	storeAppendPrintf(sentry, " login=%s", p->login);
    if (p->mcast.ttl > 0)
	storeAppendPrintf(sentry, " ttl=%d", p->mcast.ttl);
    storeAppendPrintf(sentry, "\n");
}

static void
dump_peers(StoreEntry * sentry, peer * peers)
{
    peer *e = NULL;
    struct _domain_ping *d = NULL;
    icp_opcode op;
    int i;
    if (peers == NULL)
	storeAppendPrintf(sentry, "There are no neighbors installed.\n");
    for (e = peers; e; e = e->next) {
	assert(e->host != NULL);
	storeAppendPrintf(sentry, "\n%-11.11s: %s/%d/%d\n",
	    neighborTypeStr(e),
	    e->host,
	    e->http_port,
	    e->icp.port);
	storeAppendPrintf(sentry, "Flags      :");
	dump_peer_options(sentry, e);
	for (i = 0; i < e->n_addresses; i++) {
	    storeAppendPrintf(sentry, "Address[%d] : %s\n", i,
		inet_ntoa(e->addresses[i]));
	}
	storeAppendPrintf(sentry, "Status     : %s\n",
	    neighborUp(e) ? "Up" : "Down");
	storeAppendPrintf(sentry, "AVG RTT    : %d msec\n", e->stats.rtt);
	storeAppendPrintf(sentry, "LAST QUERY : %8d seconds ago\n",
	    (int) (squid_curtime - e->stats.last_query));
	storeAppendPrintf(sentry, "LAST REPLY : %8d seconds ago\n",
	    (int) (squid_curtime - e->stats.last_reply));
	storeAppendPrintf(sentry, "PINGS SENT : %8d\n", e->stats.pings_sent);
	storeAppendPrintf(sentry, "PINGS ACKED: %8d %3d%%\n",
	    e->stats.pings_acked,
	    percent(e->stats.pings_acked, e->stats.pings_sent));
	storeAppendPrintf(sentry, "FETCHES    : %8d %3d%%\n",
	    e->stats.fetches,
	    percent(e->stats.fetches, e->stats.pings_acked));
	storeAppendPrintf(sentry, "IGNORED    : %8d %3d%%\n",
	    e->stats.ignored_replies,
	    percent(e->stats.ignored_replies, e->stats.pings_acked));
	storeAppendPrintf(sentry, "Histogram of PINGS ACKED:\n");
#if USE_HTCP
	if (e->options.htcp) {
	    storeAppendPrintf(sentry, "\tMisses\t%8d %3d%%\n",
		e->htcp.counts[0],
		percent(e->htcp.counts[0], e->stats.pings_acked));
	    storeAppendPrintf(sentry, "\tHits\t%8d %3d%%\n",
		e->htcp.counts[1],
		percent(e->htcp.counts[1], e->stats.pings_acked));
	} else {
#endif
	    for (op = ICP_INVALID; op < ICP_END; op++) {
		if (e->icp.counts[op] == 0)
		    continue;
		storeAppendPrintf(sentry, "    %12.12s : %8d %3d%%\n",
		    icp_opcode_str[op],
		    e->icp.counts[op],
		    percent(e->icp.counts[op], e->stats.pings_acked));
	    }
#if USE_HTCP
	}
#endif
	if (e->last_fail_time) {
	    storeAppendPrintf(sentry, "Last failed connect() at: %s\n",
		mkhttpdlogtime(&(e->last_fail_time)));
	}
	if (e->peer_domain != NULL) {
	    storeAppendPrintf(sentry, "DOMAIN LIST: ");
	    for (d = e->peer_domain; d; d = d->next) {
		storeAppendPrintf(sentry, "%s%s ",
		    d->do_ping ? null_string : "!", d->domain);
	    }
	    storeAppendPrintf(sentry, "\n");
	}
	storeAppendPrintf(sentry, "keep-alive ratio: %d%%\n",
	    percent(e->stats.n_keepalives_recv, e->stats.n_keepalives_sent));
    }
}

#if USE_HTCP
void
neighborsHtcpReply(const cache_key * key, htcpReplyData * htcp, const struct sockaddr_in *from)
{
    StoreEntry *e = storeGet(key);
    MemObject *mem = NULL;
    peer *p;
    peer_t ntype = PEER_NONE;
    debug(15, 6) ("neighborsHtcpReply: %s %s\n",
	htcp->hit ? "HIT" : "MISS", storeKeyText(key));
    if (NULL != (e = storeGet(key)))
	mem = e->mem_obj;
    if ((p = whichPeer(from)))
	neighborAliveHtcp(p, mem, htcp);
    /* Does the entry exist? */
    if (NULL == e) {
	debug(12, 3) ("neighyborsHtcpReply: Cache key '%s' not found\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    /* check if someone is already fetching it */
    if (EBIT_TEST(e->flags, ENTRY_DISPATCHED)) {
	debug(15, 3) ("neighborsUdpAck: '%s' already being fetched.\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (mem == NULL) {
	debug(15, 2) ("Ignoring reply for missing mem_obj: %s\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (e->ping_status != PING_WAITING) {
	debug(15, 2) ("neighborsUdpAck: Entry %s is not PING_WAITING\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (e->lock_count == 0) {
	debug(12, 1) ("neighborsUdpAck: '%s' has no locks\n",
	    storeKeyText(key));
	neighborCountIgnored(p);
	return;
    }
    if (p) {
	ntype = neighborType(p, mem->request);
	neighborUpdateRtt(p, mem);
    }
    if (ignoreMulticastReply(p, mem)) {
	neighborCountIgnored(p);
	return;
    }
    debug(15, 1) ("neighborsHtcpReply: e = %p\n", e);
    mem->ping_reply_callback(p, ntype, PROTO_HTCP, htcp, mem->ircb_data);
}
#endif