src/net/resolver.cc

/***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************
 *
 * net/resolver.cc
 * (c) 2005-2008 Murat Deligonul
 */

#include "autoconf.h"

#include <algorithm>
#include <iterator>
#include <csignal>
#include <cstdio>
#include <cstring>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include <fcntl.h>
#ifdef HAVE_SYS_FILIO_H
#   include <sys/filio.h>
#endif
#ifdef HAVE_SYS_IOCTL_H
#   include <sys/ioctl.h>
#endif
#include "util/strings.h"
#include "net/resolver.h"

#include "debug.h"

namespace net {

using namespace util::strings;

/* static */ const struct resolver::resolver_thread
	resolver::default_rt = { 0, 0 };


resolver::resolver(unsigned max_threads, unsigned max_lookups)
	: 	MAX_THREADS(max_threads),
		MAX_LOOKUPS_PER_THREAD(max_lookups),
		current_id(0),
		active_threads(0),
		first_thread(default_rt)
{
	assert(max_threads > 0);
	assert(max_lookups > 0);

	/**
	  * Set up:
	  * -- FIFO for relaying results
	  * -- mutex and condition variables
	  */
	if (pipe(fifo) < 0) {
		// FIXME: complain
		abort();
	}

	pthread_mutex_init(&queue_mutex, 0);
	pthread_mutex_init(&threads_mutex, 0);
	pthread_cond_init(&queue_cv, 0);

	DEBUG("resolver::resolver() : [%p] READY w/ fifo: [%d:%d] max threads: %d max lpt: %d\n",
				this, fifo[0], fifo[1], max_threads, max_lookups);
}

/* Destructor */
resolver::~resolver()
{
	if (first_thread.self) {
		pthread_kill(first_thread.thread, SIGKILL);
	}
	pthread_mutex_destroy(&queue_mutex);
	pthread_mutex_destroy(&threads_mutex);
	pthread_cond_destroy(&queue_cv);

	close(fifo_reader_fd());
	close(fifo_writer_fd());
}

/* static */ resolver::request * resolver::create_request(int family,
				const char * hostname,
				unsigned short port,
				int options,
				const resolver_callback * callback)
{
	struct request * req = new request;
	req->id 	= 0;		/* no ID is assigned until put into wait queue */
	req->family 	= family;
	req->options	= options;
	req->port	= port;
	req->ai		= 0;
	req->callback	= callback;
	my_strlcpy(req->name, hostname, sizeof(req->name));
	return req;
}


/**
 * Return ID of new async request on success.
 * Otherwise return error code
 **/
int resolver::async_lookup(resolver::request * req)
{
	req->id =  ++current_id;
	/* we reach this point with a lookup ready to go */
	pthread_mutex_lock(&queue_mutex);
	request_queue.push(req);
	unsigned int n = request_queue.size();
	pthread_mutex_unlock(&queue_mutex);

	/* determine if we have to dispatch a new thread */
	if (active_threads < MAX_THREADS
		&& n > (active_threads * MAX_LOOKUPS_PER_THREAD))
	{
		resolver_thread * rt;
		if (!first_thread.self)
			rt = &first_thread;
		else
			rt = new resolver_thread;
		rt->self = this;

		pthread_attr_t attr;
		pthread_attr_init(&attr);
		pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

		pthread_mutex_lock(&threads_mutex);
		++active_threads;
		pthread_mutex_unlock(&threads_mutex);

		/* race condition ?? */
		n = pthread_create(&rt->thread, &attr, resolver_thread_fn, rt);
		pthread_attr_destroy(&attr);
	}

	pthread_cond_broadcast(&queue_cv);

	return req->id;
}


/**
 * Put a request in the cancellation queue to ignore
 * the result of request 'i' when we come across it ..
 */
int resolver::cancel_async_lookup(int i)
{
	cancelled_requests.push_back(i);
	return cancelled_requests.size();
}

/* static */ void * resolver::resolver_thread_fn(void * param)
{
	assert(param);
	resolver_thread * rt = (resolver_thread *) param;
	resolver * self = rt->self;
	request * req = NULL;
	bool first = (rt == &self->first_thread);

	DEBUG("resolver::resolver_thread_fn() -- new thread\n");

	pthread_mutex_lock(&self->queue_mutex);

	while (first || self->request_queue.size() > self->MAX_LOOKUPS_PER_THREAD)
	{
		while (self->request_queue.empty()) {
			pthread_cond_wait(&self->queue_cv, &self->queue_mutex);
		}
		req = self->request_queue.front();
		self->request_queue.pop();
		pthread_mutex_unlock(&self->queue_mutex);

		DEBUG("resolver::resolver_thread_fn() -- processing request: %d\n", req->id);
		/** perform the blocking lookup **/
		self->process_request(req);
		self->write_result(req);

		pthread_mutex_lock(&self->queue_mutex);
	}

	pthread_mutex_unlock(&self->queue_mutex);

	/* No more to do. If we're not the first thread, delete
	 * our data parameter and kill thread */
	pthread_mutex_lock(&self->threads_mutex);
	--self->active_threads;
	pthread_mutex_unlock(&self->threads_mutex);

	DEBUG("resolver::resolver_thread_fn(): Thread exiting\n");
	if (!first) {
		delete rt;
	}
	pthread_exit(0);
	return 0; 	/* NOT REACHED */
}


int resolver::process_request(resolver::request * req)
{
	if (req->options & OPT_REVERSE_LOOKUP) {
		resolve_address(req->family, AI_NUMERICHOST, req->name, req->port, &req->ai);
		req->iresult = reverse_lookup(req->name, req->name, sizeof(req->name));
	}
	else {
		req->iresult = resolve_address(req->family, 0, req->name, req->port, &req->ai);
	}
	return req->iresult;
}

/**
 * Write the result to FIFO to be read by main thread.
 * Currently we are cheap and just write the pointer to the request
 * data structure.
 * This operation completes the work in the resolving thread.
 */
int resolver::write_result(const resolver::request * req)
{
	return write(fifo_writer_fd(), &req, sizeof(resolver::request *));
}


/** Read async look-up results **/
int resolver::process_results()
{
	int processed = 0;
	int i = 0;
	do {
		// FIXME: why is this in the loop?
		ioctl(fifo_reader_fd(), FIONREAD, &i);
		if (unsigned(i) >= sizeof(struct request *)) {
			struct request * req = read_result();
			process_result(req);
			++processed;
			/* at this point the request has resolved,
			 * the callback function notified, and any allocated
			 * resources for the request freed */
		}
		i -= sizeof(struct request *);
	} while (i > 0 && unsigned(i) >= sizeof(struct request *));

	return processed;
}

/** read an indiviual result from the FIFO */
resolver::request * resolver::read_result()
{
	request * req = NULL;
	read(fifo_reader_fd(), &req, sizeof(req));
	return req;
}

int resolver::process_result(request * req)
{
	assert(req);
	assert(req->id > 0);

	bool cancelled = false;

	do {
		/* First check if it's been cancelled */
		std::vector<int>::iterator i =
			std::find(cancelled_requests.begin(),
				cancelled_requests.end(),
				req->id);

		if (i != cancelled_requests.end()) {
			DEBUG("resolver::process_result(): CANCELLED: %d\n", req->id);
			/* Must cancel this one */
			cancelled_requests.erase(i);
			cancelled = true;
			continue;
		}
	} while (0);	/* we must get rid of potential duplicate cancellation requests */

	if (!cancelled) {
		if (req->iresult == 0) {
			req->callback->async_lookup_finished(req);
		}
		else {
			req->callback->async_lookup_failed(req);
		}
	}

	delete req;
	return 0;
}


/**
 * [BLOCKING] The unified address resolver.
 * Translates a network address or hostname into socket address structures.
 * Handles both IPv4 and IPv6 address families.
 *
 * @param family 	desired address family (or AF_UNSPEC)
 * @param hint_flags	hints or flags for the resolver (the addrinfo.ai_flags field)
 * @param hostname	host name to look up
 * @param port		port to fill address structures with
 * @param res		pointer to pointer to addrinfo data which will store the result
 *
 * @return [same as getaddrinfo()]
 * 		0:		success; res populated
 *		<> 0:		error; res untouched
 */
/* static */ int resolver::resolve_address(int family,
			int hint_flags,
			const char * hostname,
			unsigned short port,
			struct addrinfo ** res)
{
	struct addrinfo hints;
	char  portbuf[6] = "";

	memset(&hints, 0, sizeof(hints));
	hints.ai_flags = hint_flags;
	hints.ai_family = family;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_protocol = IPPROTO_TCP;

	sprintf(portbuf, "%u", port);

	return getaddrinfo(hostname, portbuf, &hints, res);
}

/**
 * [BLOCKING]
 * Simpler interface to resolve_address.
 * Translates hostname to IP address.
 *
 * @param hostname	name to lookup
 * @param buffer	buffer to store result in
 * @param len		length of buffer
 * @return 0 for success
 */
/* static */ int resolver::lookup(const char * hostname,
			char * buffer,
			size_t len)
{
	struct addrinfo * ai = NULL;
	int i = resolve_address(AF_UNSPEC, 0, hostname, 0, &ai);
	if (i != 0) {
		return i;
	}

	i = raw_to_ip(ai->ai_addr, ai->ai_addrlen, buffer, len);
	freeaddrinfo(ai);
	return i;
}


/**
 * [BLOCKING]
 * Attempt to translate IP -> hostname.
 * @param name		numeric address to translate
 * @param buffer	buffer to store result in
 * @param len		length of buffer
 *
 * @return 0 on success, otherwise getnameinfo() error code
 */
/* static */ int resolver::reverse_lookup(const char * name,
			char * buffer,
			size_t len)
{
	struct addrinfo * ai = NULL;
	int i = resolve_address(AF_UNSPEC, AI_NUMERICHOST,
				name, 0, &ai);
	if (i != 0) {
		return i;
	}

	i = getnameinfo(ai->ai_addr, ai->ai_addrlen, buffer, len,
				NULL, 0,
				0 /* NI_NAMEREQD */);	// XXX
	freeaddrinfo(ai);
	return i;
}


/**
 * Tests if binding to this interface will work.
 *
 * @param address	address to bind on.  May be NULL, in which case default interface will be used.
 * @param port		port to bind on.  May be zero.
 * @return 0 on success, -1 for name lookup related error, 1 for other errors
 */
/* static */ int resolver::test_bind(const char * address, unsigned short port)
{
	int hints = AI_ADDRCONFIG;
	if (address == NULL) {
		hints |= AI_PASSIVE;
	}
	struct addrinfo * ai = NULL;
	if (resolve_address(PF_UNSPEC, hints, address, port, &ai) != 0) {
		return -1;
	}

	// create socket and test bind
	int ret = 0;
	int s = socket(ai->ai_family, SOCK_STREAM, 0);
	if (s < 0 || bind(s, ai->ai_addr, ai->ai_addrlen) < 0) {
		ret = 1;
	}

	// result?
	close(s);
	freeaddrinfo(ai);
	return ret;
}

/**
 * Translate a raw network address structure into a text representation.
 *
 * @param addr		address structure
 * @param addrlen	size of address structure
 * @param buff		buffer to store result in
 * @param bufflen	length of buffer
 * @param port		pointer to unsigned short to store port; may be NULL
 * @return 0 on success
 */
/* static */ int resolver::raw_to_ip(const struct sockaddr * addr,
				size_t addrlen,
				char * buff,
				size_t bufflen,
				unsigned short * port)
{
	char temp[10];
	unsigned short dummy;
	if (port == NULL) {
		port = &dummy;
	}

	int i = getnameinfo(addr, addrlen, buff, bufflen,
				temp, sizeof temp,
				NI_NUMERICHOST | NI_NUMERICSERV);
	if (i != 0) {
		return i;
	}

	// assign port
	*port = static_cast<unsigned short>(atoi(temp));
	return i;
}


/**
 * Translate a numeric address string into a network address structure.
 * The address family is determined by a call to getaddrinfo().  The provided
 * buffer must be large enough to store the resulting address structure, or an
 * error code will be returned.
 *
 * @param name		address to string to translate
 * @param port		port to assign in address structure's field
 * @param out 		structure to fill
 * @param out_size	size of the structure
 * @return 0 on success, error code if address is invalid or buffer size is insufficient.
 */
int resolver::ip_to_raw(const char * name,
			unsigned short port,
			struct sockaddr * out,
			size_t out_size)
{
	struct addrinfo * ai = NULL;
	int i = resolve_address(AF_UNSPEC, AI_NUMERICHOST,
				name, port, &ai);
	if (i != 0) {
		return i;
	}

	// verify buffer space
	if (ai->ai_addrlen > out_size) {
#ifdef EAI_OVERFLOW
		// this might be Linux specific
		i = EAI_OVERFLOW;
#else
		i = -1;
#endif
	}
	else {
		memset(out, 0, out_size);
		memcpy(out, ai->ai_addr, ai->ai_addrlen);
	}
	freeaddrinfo(ai);
	return i;
}

/**
 * Determine whether a given address is a numeric IP address.
 *
 * @param af		address family to check (can be AF_UNSPEC)
 * @param name		name to check
 * @return true if ip address detected
 */
/* static */ bool resolver::is_ip_address(int af, const char * name)
{
	struct sockaddr_storage dummy;

	int ip4 = ip_to_raw(name, 0, (struct sockaddr *) &dummy, sizeof(struct sockaddr_in));
	int ip6 = ip_to_raw(name, 0, (struct sockaddr *) &dummy, sizeof(struct sockaddr_in6));

	if (af == AF_INET) {
		return ip4 == 0;
	}
	else if (af == AF_INET6) {
		return ip6 == 0;
	}
	return ip4 == 0 || ip6 == 0;
}

/* static */ char * resolver::result_to_string(const resolver::request * req,
						char * buf,
						size_t len)
{
	if (req->options & OPT_REVERSE_LOOKUP) {
		my_strlcpy(buf, req->name, len);
	}
	else {
		raw_to_ip(req->ai->ai_addr, req->ai->ai_addrlen, buf, len);
	}
	return buf;
}

} /* namespace net */