path: root/src/dispatch.c

#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <poll.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <unistd.h>

#include <rpc/rpc.h>
#include <rpc/pmap_clnt.h>
#include <rpcsvc/yp_prot.h>

#include <nspr.h>
#include <secport.h>
#include <plarenas.h>

#include "dispatch.h"
#include "nis.h"
#include "plugin.h"
#include "portmap.h"

#define MAX_CLIENT_IDLE (60 * 1000)

struct dispatch_stream_client_params {
	int client;
	struct plugin_state *state;
};

/* Handle requests from one stream client. */
static void *
dispatch_stream_handler_thread(void *arg)
{
	int fraglen, flags, ret, frag_len, record_len, client;
	ssize_t i;
	int32_t len, nlen;
	char fragment[65540], record[65536];
	struct pollfd pollfd;
	struct dispatch_stream_client_params *params;
	struct plugin_state *state;
	bool_t done, last;

	/* Recover the full set of parameters. */
	params = arg;
	state = params->state;
	client = params->client;
	free(arg);

	slapi_log_error(SLAPI_LOG_PLUGIN,
			state->plugin_desc->spd_id,
			"opened client connection %d, thread started\n",
			client);

	/* Set the connection to be non-blocking. */
	flags = fcntl(client, F_GETFD);
	if ((flags == -1) ||
	    (fcntl(client, F_SETFD, flags | O_NONBLOCK) == -1)) {
		slapi_log_error(SLAPI_LOG_PLUGIN,
				state->plugin_desc->spd_id,
				"error setting new client connection to be "
				"non-blocking\n");
		close(client);
		return NULL;
	}
	slapi_log_error(SLAPI_LOG_PLUGIN,
			state->plugin_desc->spd_id,
			"set new client connection to be non-blocking\n");

	/* Assemble fragments and responses. */
	frag_len = 0;
	record_len = 0;
	for (;;) {
		/* Wait for incoming data. */
		pollfd.fd = client;
		pollfd.events = POLLIN | POLLHUP;
		switch (poll(&pollfd, 1, MAX_CLIENT_IDLE)) {
		case -1:
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"communication error\n");
			goto done;
			break;
		case 0:
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"client timeout\n");
			goto done;
			break;
		default:
			break;
		}
		/* Try to read as much data as we have space to hold. */
		i = read(client, fragment + frag_len,
			 sizeof(fragment) - frag_len);
		switch (i) {
		case -1:
			if (errno == EAGAIN) {
				continue;
			}
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"communication error\n");
			goto done;
			break;
		case 0:
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"client closed connection\n");
			goto done;
			break;
		default:
			break;
		}
		/* We managed to read some piece of a fragment. */
		slapi_log_error(SLAPI_LOG_PLUGIN,
				state->plugin_desc->spd_id,
				"read %d bytes\n", i);
		frag_len += i;
		/* If we can't know the fragment size yet, then wait for more
		 * data. */
		if (frag_len < 4) {
			continue;
		}
		/* Convert the fragment length information to local byte order,
		 * and determine if this is the last fragment in the record. */
		memcpy(&nlen, fragment, 4);
		len = ntohl(nlen);
		last = ((len & 0x80000000) != 0);
		len &= 0x7fffffff;
		/* If we don't have a complete fragment, then wait for more
		 * data. */
		if (frag_len < (len + 4)) {
			continue;
		}
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"expecting %d bytes\n", len);
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"in: frag_len=%d,record=%d\n",
				frag_len, record_len);
		/* Pull the fragment out of the buffer and append it to the
		 * record buffer. */
		slapi_log_error(SLAPI_LOG_PLUGIN,
				state->plugin_desc->spd_id,
				"got whole fragment (%d bytes)\n", len);
		memmove(record + record_len, fragment + 4, len);
		record_len += len;
		memmove(fragment, fragment + (len + 4), frag_len - (len + 4));
		frag_len -= (len + 4);
		/* If there are more fragments to come, then we need to wait
		 * for more data. */
		if (!last) {
			continue;
		}
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"got whole record (%d bytes)\n", record_len);
		nis_process_request(state, client, NULL, 0, record, record_len);
		record_len = 0;
		/* Note that we did it! */
		slapi_log_error(SLAPI_LOG_PLUGIN,
				state->plugin_desc->spd_id,
				"out: frag_len=%d,record=%d\n",
				frag_len, record_len);
	}
done:
	close(client);
	slapi_log_error(SLAPI_LOG_PLUGIN,
			state->plugin_desc->spd_id,
			"closed client connection %d, thread ending\n",
			client);
	return NULL;
}

/* Handle a stream client -- answer the connection and spawn a thread to handle
 * its actual requests. */
static void
dispatch_stream(struct plugin_state *state, int fd)
{
	struct sockaddr_storage client_addr;
	socklen_t client_addrlen;
	int client;
	pthread_t thread;
	struct dispatch_stream_client_params *params;

	/* Answer the connection request. */
	client = accept(fd, (struct sockaddr *) &client_addr, &client_addrlen);
	if (client == -1) {
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"failed to answer new stream request\n");
		return;
	} else {
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"new stream client on %d\n", client);
	}

	/* Bundle up enough info for the thread to do its work. */
	params = malloc(sizeof(*params));
	if (params == NULL) {
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"out of memory\n");
		close(client);
		return;
	}
	params->client = client;
	params->state = state;

	/* Kick off the thread. */
	if (pthread_create(&thread, NULL,
			   &dispatch_stream_handler_thread, params) != 0) {
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"error starting thread\n");
		close(client);
	}
	slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
			"started client-specific thread for client on %d\n",
			client);
	return;
}

/* Handle a datagram client -- read the request and handle it immediately. */
static void
dispatch_dgram(struct plugin_state *state, int fd)
{
	struct sockaddr_storage client_addr;
	socklen_t client_addrlen;
	char dgram[65536];
	int reqsize;

	/* Read the request. */
	client_addrlen = sizeof(client_addr);
	reqsize = recvfrom(fd, dgram, sizeof(dgram), 0,
			   (struct sockaddr *) &client_addr, &client_addrlen);
	slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
			"datagram request (%d bytes)\n", reqsize);

	/* Handle the request. */
	nis_process_request(state, fd,
			    (struct sockaddr *) &client_addr, client_addrlen,
			    dgram, reqsize);
}


/* Handle all incoming data. */
void *
dispatch_thread(void *p)
{
	struct plugin_state *state = p;
	struct pollfd fds[4];
	int i;
	for (;;) {
		/* Set up for polling. */
		memset(&fds, 0, sizeof(fds));
		for (i = 0; i < state->n_listeners; i++) {
			fds[i].fd = state->listener[i].fd;
			fds[i].events = POLLIN;
		}
		slapi_log_error(SLAPI_LOG_PLUGIN, state->plugin_desc->spd_id,
				"listening for next request\n");
		switch (poll(fds, state->n_listeners, -1)) {
		case -1:
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"done waiting\n");
			return NULL;
			break;
		case 0:
			slapi_log_error(SLAPI_LOG_PLUGIN,
					state->plugin_desc->spd_id,
					"no request(?)\n");
			continue;
		default:
			/* Iterate over listening sockets which have work for
			 * us to do. */
			for (i = 0; i < state->n_listeners; i++) {
				if ((fds[i].revents & POLLIN) == 0) {
					continue;
				}
				switch (state->listener[i].type) {
				case SOCK_DGRAM:
					dispatch_dgram(state, fds[i].fd);
					break;
				case SOCK_STREAM:
					dispatch_stream(state, fds[i].fd);
					break;
				default:
					/* never reached */
					assert(0);
					break;
				}
			}
		}
	}
	return state;
}