net_db.c

-

    rec_sz = 0;
    rec_sz += 1 + sizeof(addr.s_addr);
    rec_sz += 1 + sizeof(int);
    rec_sz += 1 + sizeof(int);
    ex->seen = ex->used + size;
    debug(38, 3) ("netdbExchangeHandleReply: %d bytes\n", (int) size);
    if (!cbdataValid(ex->p)) {
	debug(38, 3) ("netdbExchangeHandleReply: Peer became invalid\n");
	netdbExchangeDone(ex);
	return;
    }
    debug(38, 3) ("netdbExchangeHandleReply: for '%s:%d'\n", ex->p->host, ex->p->http_port);
    p = buf;
    if (0 == ex->used) {
	/* skip reply headers */
	if ((hdr_sz = headersEnd(p, size))) {
	    debug(38, 5) ("netdbExchangeHandleReply: hdr_sz = %d\n", hdr_sz);
	    rep = ex->e->mem_obj->reply;
	    if (0 == rep->sline.status)
		httpReplyParse(rep, buf);
	    debug(38, 3) ("netdbExchangeHandleReply: reply status %d\n",
		rep->sline.status);
	    if (HTTP_OK != rep->sline.status) {
		netdbExchangeDone(ex);
		return;
	    }
	    assert(size >= hdr_sz);
	    ex->used += hdr_sz;
	    size -= hdr_sz;
	    p += hdr_sz;
	} else {
	    size = 0;
	}
    }
    debug(38, 5) ("netdbExchangeHandleReply: start parsing loop, size = %d\n",
	size);
    while (size >= rec_sz) {
	debug(38, 5) ("netdbExchangeHandleReply: in parsing loop, size = %d\n",
	    size);
	addr.s_addr = any_addr.s_addr;
	hops = rtt = 0.0;
	for (o = 0; o < rec_sz;) {
	    switch ((int) *(p + o)) {
	    case NETDB_EX_NETWORK:
		o++;
		xmemcpy(&addr.s_addr, p + o, sizeof(addr.s_addr));
		o += sizeof(addr.s_addr);
		break;
	    case NETDB_EX_RTT:
		o++;
		xmemcpy(&j, p + o, sizeof(int));
		o += sizeof(int);
		rtt = (double) ntohl(j) / 1000.0;
		break;
	    case NETDB_EX_HOPS:
		o++;
		xmemcpy(&j, p + o, sizeof(int));
		o += sizeof(int);
		hops = (double) ntohl(j) / 1000.0;
		break;
	    }
	}
	if (addr.s_addr != any_addr.s_addr && rtt > 0)
	    netdbExchangeUpdatePeer(addr, ex->p, rtt, hops);
	assert(o == rec_sz);
	ex->used += rec_sz;
	size -= rec_sz;
	p += rec_sz;
	/*
	 * This is a fairly cpu-intensive loop, break after adding
	 * just a few
	 */
	if (++nused == 20)
	    break;
    }
    debug(38, 3) ("netdbExchangeHandleReply: used %d entries, (x %d bytes) == %d bytes total\n",
	nused, rec_sz, nused * rec_sz);
    debug(38, 3) ("netdbExchangeHandleReply: seen %d, used %d\n", ex->seen, ex->used);
    if (EBIT_TEST(ex->e->flags, ENTRY_ABORTED)) {
	debug(38, 3) ("netdbExchangeHandleReply: ENTRY_ABORTED\n");
	netdbExchangeDone(ex);
    } else if (ex->e->store_status == STORE_PENDING) {
	debug(38, 3) ("netdbExchangeHandleReply: STORE_PENDING\n");
	storeClientCopy(ex->e, ex->seen, ex->used, ex->buf_sz,
	    ex->buf, netdbExchangeHandleReply, ex);
    } else if (ex->seen < ex->e->mem_obj->inmem_hi) {
	debug(38, 3) ("netdbExchangeHandleReply: ex->e->mem_obj->inmem_hi\n");
	storeClientCopy(ex->e, ex->seen, ex->used, ex->buf_sz,
	    ex->buf, netdbExchangeHandleReply, ex);
    } else {
	debug(38, 3) ("netdbExchangeHandleReply: Done\n");
	netdbExchangeDone(ex);
    }
}

static void
netdbExchangeDone(void *data)
{
    netdbExchangeState *ex = data;
    debug(38, 3) ("netdbExchangeDone: %s\n", storeUrl(ex->e));
    memFree(ex->buf, MEM_4K_BUF);
    requestUnlink(ex->r);
    storeUnregister(ex->e, ex);
    storeUnlockObject(ex->e);
    cbdataUnlock(ex->p);
    cbdataFree(ex);
}

#endif /* USE_ICMP */

/* PUBLIC FUNCTIONS */

void
netdbInit(void)
{
#if USE_ICMP
    int n;
    if (addr_table)
	return;
    n = hashPrime(Config.Netdb.high / 4);
    addr_table = hash_create((HASHCMP *) strcmp, n, hash_string);
    n = hashPrime(3 * Config.Netdb.high / 4);
    host_table = hash_create((HASHCMP *) strcmp, n, hash_string);
    eventAddIsh("netdbSaveState", netdbSaveState, NULL, 3600.0, 1);
    netdbReloadState();
    cachemgrRegister("netdb",
	"Network Measurement Database",
	netdbDump, 0, 1);
#endif
}

void
netdbPingSite(const char *hostname)
{
#if USE_ICMP
    netdbEntry *n;
    char *h;
    if ((n = netdbLookupHost(hostname)) != NULL)
	if (n->next_ping_time > squid_curtime)
	    return;
    h = xstrdup(hostname);
    cbdataAdd(h, cbdataXfree, 0);
    cbdataLock(h);
    ipcache_nbgethostbyname(hostname, netdbSendPing, h);
#endif
}

void
netdbHandlePingReply(const struct sockaddr_in *from, int hops, int rtt)
{
#if USE_ICMP
    netdbEntry *n;
    int N;
    debug(38, 3) ("netdbHandlePingReply: from %s\n", inet_ntoa(from->sin_addr));
    if ((n = netdbLookupAddr(from->sin_addr)) == NULL)
	return;
    N = ++n->pings_recv;
    if (N > 5)
	N = 5;
    if (rtt < 1.0)
	rtt = 1.0;
    n->hops = ((n->hops * (N - 1)) + hops) / N;
    n->rtt = ((n->rtt * (N - 1)) + rtt) / N;
    debug(38, 3) ("netdbHandlePingReply: %s; rtt=%5.1f  hops=%4.1f\n",
	n->network,
	n->rtt,
	n->hops);
#endif
}

void
netdbFreeMemory(void)
{
#if USE_ICMP
    hashFreeItems(addr_table, netdbFreeNetdbEntry);
    hashFreeMemory(addr_table);
    addr_table = NULL;
    hashFreeItems(host_table, netdbFreeNameEntry);
    hashFreeMemory(host_table);
    host_table = NULL;
    wordlistDestroy(&peer_names);
    peer_names = NULL;
#endif
}

int
netdbHops(struct in_addr addr)
{
#if USE_ICMP
    netdbEntry *n = netdbLookupAddr(addr);
    if (n && n->pings_recv) {
	n->last_use_time = squid_curtime;
	return (int) (n->hops + 0.5);
    }
#endif
    return 256;
}

void
netdbDump(StoreEntry * sentry)
{
#if USE_ICMP
    netdbEntry *n;
    netdbEntry **list;
    net_db_name *x;
    int k;
    int i;
    int j;
    net_db_peer *p;
    storeAppendPrintf(sentry, "Network DB Statistics:\n");
    storeAppendPrintf(sentry, "%-16.16s %9s %7s %5s %s\n",
	"Network",
	"recv/sent",
	"RTT",
	"Hops",
	"Hostnames");
    list = xcalloc(memInUse(MEM_NETDBENTRY), sizeof(netdbEntry *));
    i = 0;
    hash_first(addr_table);
    while ((n = (netdbEntry *) hash_next(addr_table)))
	*(list + i++) = n;
    if (i != memInUse(MEM_NETDBENTRY))
	debug(38, 0) ("WARNING: netdb_addrs count off, found %d, expected %d\n",
	    i, memInUse(MEM_NETDBENTRY));
    qsort((char *) list,
	i,
	sizeof(netdbEntry *),
	sortByRtt);
    for (k = 0; k < i; k++) {
	n = *(list + k);
	storeAppendPrintf(sentry, "%-16.16s %4d/%4d %7.1f %5.1f",
	    n->network,
	    n->pings_recv,
	    n->pings_sent,
	    n->rtt,
	    n->hops);
	for (x = n->hosts; x; x = x->next)
	    storeAppendPrintf(sentry, " %s", x->name);
	storeAppendPrintf(sentry, "\n");
	p = n->peers;
	for (j = 0; j < n->n_peers; j++, p++) {
	    storeAppendPrintf(sentry, "    %-22.22s %7.1f %5.1f\n",
		p->peername,
		p->rtt,
		p->hops);
	}
    }
    xfree(list);
#else
    storeAppendPrintf(sentry,
	"NETDB support not compiled into this Squid cache.\n");
#endif
}

int
netdbHostHops(const char *host)
{
#if USE_ICMP
    netdbEntry *n = netdbLookupHost(host);
    if (n) {
	n->last_use_time = squid_curtime;
	return (int) (n->hops + 0.5);
    }
#endif
    return 0;
}

int
netdbHostRtt(const char *host)
{
#if USE_ICMP
    netdbEntry *n = netdbLookupHost(host);
    if (n) {
	n->last_use_time = squid_curtime;
	return (int) (n->rtt + 0.5);
    }
#endif
    return 0;
}

void
netdbHostData(const char *host, int *samp, int *rtt, int *hops)
{
#if USE_ICMP
    netdbEntry *n = netdbLookupHost(host);
    if (n == NULL)
	return;
    *samp = n->pings_recv;
    *rtt = (int) (n->rtt + 0.5);
    *hops = (int) (n->hops + 0.5);
#endif
}

int
netdbHostPeerRtt(const char *host, peer * p)
{
#if USE_ICMP
    const netdbEntry *n = netdbLookupHost(host);
    if (n) {
	const net_db_peer *np = netdbPeerByName(n, p->host);
	if (np && np->expires >= squid_curtime)
	    return (int) (np->rtt + 0.5);
    }
#endif
    return 0;
}

void
netdbUpdatePeer(request_t * r, peer * e, int irtt, int ihops)
{
#if USE_ICMP
    netdbEntry *n;
    double rtt = (double) irtt;
    double hops = (double) ihops;
    net_db_peer *p;
    debug(38, 3) ("netdbUpdatePeer: '%s', %d hops, %d rtt\n", r->host, ihops, irtt);
    n = netdbLookupHost(r->host);
    if (n == NULL) {
	debug(38, 3) ("netdbUpdatePeer: host '%s' not found\n", r->host);
	return;
    }
    if ((p = netdbPeerByName(n, e->host)) == NULL)
	p = netdbPeerAdd(n, e);
    p->rtt = rtt;
    p->hops = hops;
    p->expires = squid_curtime + 3600;
    if (n->n_peers < 2)
	return;
    qsort((char *) n->peers,
	n->n_peers,
	sizeof(net_db_peer),
	sortPeerByRtt);
#endif
}

void
netdbExchangeUpdatePeer(struct in_addr addr, peer * e, double rtt, double hops)
{
#if USE_ICMP
    netdbEntry *n;
    net_db_peer *p;
    debug(38, 5) ("netdbExchangeUpdatePeer: '%s', %0.1f hops, %0.1f rtt\n",
	inet_ntoa(addr), hops, rtt);
    n = netdbLookupAddr(addr);
    if (n == NULL)
	n = netdbAdd(addr);
    assert(NULL != n);
    if ((p = netdbPeerByName(n, e->host)) == NULL)
	p = netdbPeerAdd(n, e);
    p->rtt = rtt;
    p->hops = hops;
    p->expires = squid_curtime + 3600;	/* XXX ? */
    if (n->n_peers < 2)
	return;
    qsort((char *) n->peers,
	n->n_peers,
	sizeof(net_db_peer),
	sortPeerByRtt);
#endif
}

void
netdbDeleteAddrNetwork(struct in_addr addr)
{
#if USE_ICMP
    netdbEntry *n = netdbLookupAddr(addr);
    if (n == NULL)
	return;
    debug(38, 3) ("netdbDeleteAddrNetwork: %s\n", n->network);
    netdbRelease(n);
#endif
}

void
netdbBinaryExchange(StoreEntry * s)
{
    http_reply *reply = s->mem_obj->reply;
#if USE_ICMP
    netdbEntry *n;
    int i;
    int j;
    int rec_sz;
    char *buf;
    struct in_addr addr;
    storeBuffer(s);
    httpReplyReset(reply);
    httpReplySetHeaders(reply, 1.0, HTTP_OK, "OK",
	NULL, -1, squid_curtime, -2);
    httpReplySwapOut(reply, s);
    rec_sz = 0;
    rec_sz += 1 + sizeof(addr.s_addr);
    rec_sz += 1 + sizeof(int);
    rec_sz += 1 + sizeof(int);
    buf = memAllocate(MEM_4K_BUF);
    i = 0;
    hash_first(addr_table);
    while ((n = (netdbEntry *) hash_next(addr_table))) {
	if (0.0 == n->rtt)
	    continue;
	if (n->rtt > 60000)	/* RTT > 1 MIN probably bogus */
	    continue;
	if (!safe_inet_addr(n->network, &addr))
	    continue;
	buf[i++] = (char) NETDB_EX_NETWORK;
	xmemcpy(&buf[i], &addr.s_addr, sizeof(addr.s_addr));
	i += sizeof(addr.s_addr);
	buf[i++] = (char) NETDB_EX_RTT;
	j = htonl((int) (n->rtt * 1000));
	xmemcpy(&buf[i], &j, sizeof(int));
	i += sizeof(int);
	buf[i++] = (char) NETDB_EX_HOPS;
	j = htonl((int) (n->hops * 1000));
	xmemcpy(&buf[i], &j, sizeof(int));
	i += sizeof(int);
	if (i + rec_sz > 4096) {
	    storeAppend(s, buf, i);
	    i = 0;
	}
    }
    if (i > 0) {
	storeAppend(s, buf, i);
	i = 0;
    }
    assert(0 == i);
    storeBufferFlush(s);
    memFree(buf, MEM_4K_BUF);
#else
    httpReplyReset(reply);
    httpReplySetHeaders(reply, 1.0, HTTP_BAD_REQUEST, "Bad Request",
	NULL, -1, squid_curtime, -2);
    storeAppendPrintf(s, "NETDB support not compiled into this Squid cache.\n");
#endif
    storeComplete(s);
}

void
netdbExchangeStart(void *data)
{
#if USE_ICMP
    peer *p = data;
    char *uri;
    netdbExchangeState *ex = xcalloc(1, sizeof(*ex));
    cbdataAdd(ex, cbdataXfree, 0);
    cbdataLock(p);
    ex->p = p;
    uri = internalRemoteUri(p->host, p->http_port, "/squid-internal-dynamic/", "netdb");
    debug(38, 3) ("netdbExchangeStart: Requesting '%s'\n", uri);
    assert(NULL != uri);
    ex->r = urlParse(METHOD_GET, uri);
    if (NULL == ex->r) {
	debug(38, 1) ("netdbExchangeStart: Bad URI %s\n", uri);
	return;
    }
    requestLink(ex->r);
    assert(NULL != ex->r);
    ex->r->http_ver = 1.0;
    ex->e = storeCreateEntry(uri, uri, null_request_flags, METHOD_GET);
    ex->buf_sz = 4096;;
    ex->buf = memAllocate(MEM_4K_BUF);
    assert(NULL != ex->e);
    storeClientListAdd(ex->e, ex);
    storeClientCopy(ex->e, ex->seen, ex->used, ex->buf_sz,
	ex->buf, netdbExchangeHandleReply, ex);
    ex->r->flags.loopdetect = 1;	/* cheat! -- force direct */
    fwdStart(-1, ex->e, ex->r, no_addr, no_addr);
#endif
}

peer *
netdbClosestParent(request_t * request)
{
#if USE_ICMP
    peer *p = NULL;
    netdbEntry *n;
    const ipcache_addrs *ia;
    net_db_peer *h;
    int i;
    n = netdbLookupHost(request->host);
    if (NULL == n) {
	/* try IP addr */
	ia = ipcache_gethostbyname(request->host, 0);
	if (NULL != ia)
	    n = netdbLookupAddr(ia->in_addrs[ia->cur]);
    }
    if (NULL == n)
	return NULL;
    if (0 == n->n_peers)
	return NULL;
    /* 
     * Find the parent with the least RTT to the origin server.
     * Make sure we don't return a parent who is farther away than
     * we are.  Note, the n->peers list is pre-sorted by RTT.
     */
    for (i = 0; i < n->n_peers; i++) {
	h = &n->peers[i];
	if (n->rtt > 0)
	    if (n->rtt < h->rtt)
		break;
	p = peerFindByName(h->peername);
	if (NULL == p)		/* not found */
	    continue;
	if (neighborType(p, request) != PEER_PARENT)
	    continue;
	if (!peerHTTPOkay(p, request))	/* not allowed */
	    continue;
	return p;
    }
#endif
    return NULL;
}