path: root/server/query.c

/*
 * query.c - request/reply handling for the server
 * Copyright (C) 2006 Thorsten Zitterell <thorsten@zitterell.de>
 */
 
#include <stdio.h>		
#include <pthread.h>		
#include <stdlib.h>	
#include <string.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#include <linux/types.h>
#include <unistd.h>
#include <time.h>

#define DNBD_USERSPACE		1
#include "../common/dnbd-cliserv.h"

#include "query.h"

/* number of threads used to service requests */
#define NUM_HANDLER_THREADS	1		/* default */
#define MAX_BLOCK_SIZE		4096

struct query_thread {
	query_info_t *query_info;
	int id;
	pthread_t p_thread;
};

struct query_thread query_thread[NUM_HANDLER_THREADS];

/* recursive global mutex for our program. */
pthread_mutex_t query_mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP;
/* mutex to avoid concurrent file access */
pthread_mutex_t handler_mutex = PTHREAD_MUTEX_INITIALIZER;

/* global condition variable for our program. */
pthread_cond_t got_query = PTHREAD_COND_INITIALIZER;

int num_queries = 0;		/* number of pending requests, initially none */
int max_queries = 100;		/* this value should be high enough */

query_t *queries = NULL;	/* head of linked list of requests. */
int last_query = 0;		/* initial position in circular buffer */
int next_query = 0;		


void query_handle(struct query_info *query_info, query_t * query);

/* 
 * function query_add_loop(): add incoming requests to circular buffer 
 */
void *query_add_loop(void *data)
{
	int rc;
	query_t *query;
	query_info_t *query_info = (query_info_t *) data;

	int tmp_query;

	while (1) {

		rc = pthread_mutex_lock(&query_mutex);
		tmp_query = (next_query + 1) % max_queries;
		rc = pthread_mutex_unlock(&query_mutex);

		if (tmp_query == last_query)
			continue;

		query = &queries[next_query];

		/* loop until a proper request arrives */
		while (!net_rx(query_info->net_info, &query->request)) {}

		rc = pthread_mutex_lock(&query_mutex);

		next_query = tmp_query;

		/* increase total number of pending requests */
		num_queries++;

		rc = pthread_mutex_unlock(&query_mutex);

		/* signal that there's a new request to handle */
		rc = pthread_cond_signal(&got_query);
	}
}

/*
 * function: query_get(): fetch request from circular buffer
 * returns: pointer to request
 */
query_t *query_get(pthread_mutex_t * p_mutex)
{
	int rc;		
	query_t *query;		/* pointer to request */

	rc = pthread_mutex_lock(p_mutex);

	if (last_query == next_query)
		return NULL;

	query = &queries[last_query];

	last_query = (last_query + 1) % max_queries;
	num_queries--;

	rc = pthread_mutex_unlock(p_mutex);
	/* return the request to the caller */
	return query;
}

/*
 * function query_handle(): handle a single request.
 */
void query_handle(struct query_info *query_info, query_t * query)
{
	int i, rc;
	dnbd_request_t *dnbd_request;
	dnbd_request_t *dnbd_old_request;
	dnbd_reply_t *dnbd_reply = NULL;
	struct dnbd_reply_init *dnbd_reply_init;
	int tmp_query;
	int recent = 0;
	time_t timestamp;

	dnbd_request = (dnbd_request_t *) & query->request.data;

	query->reply.len = 0;

	/* convert data from network to host byte order */
	dnbd_request->magic = ntohl(dnbd_request->magic);
	dnbd_request->time = ntohs(dnbd_request->time);
	dnbd_request->id = ntohs(dnbd_request->id);
	dnbd_request->cmd = ntohs(dnbd_request->cmd);
	dnbd_request->pos = ntohll(dnbd_request->pos);
	dnbd_request->len = ntohs(dnbd_request->len);

	if (dnbd_request->magic != DNBD_MAGIC)
		return;

	/* we ususally only respond to a client */
	if (!(dnbd_request->cmd & DNBD_CMD_CLI))
		return;

	/* does the client ask for our id? */
	if (dnbd_request->id && (dnbd_request->id != query_info->id))
		return;

	switch (dnbd_request->cmd & DNBD_CMD_MASK) {
	/* handle init request */
	case DNBD_CMD_INIT:
	/* handle heartbeat request */
	case DNBD_CMD_HB:
		dnbd_reply_init =
		    (struct dnbd_reply_init *) query->reply.data;
		dnbd_reply_init->magic = htonl(DNBD_MAGIC);

		dnbd_reply_init->capacity =
		    htonll(filer_getcapacity(query_info->filer_info));

		dnbd_reply_init->cmd =
		    htons((dnbd_request->cmd
			   & ~DNBD_CMD_CLI) | DNBD_CMD_SRV);

		dnbd_reply_init->blksize = htons(MAX_BLOCK_SIZE);
		dnbd_reply_init->id = htons(query_info->id);

		query->reply.len = sizeof(struct dnbd_reply_init);

		net_tx(query_info->net_info, &query->reply);
		break;
	/* handle read request */
	case DNBD_CMD_READ:
		timestamp = time(NULL);
	
		/* burst avoidance */
		rc = pthread_mutex_lock(&query_mutex);
		for (i = 2; i < max_queries; i++) {

			tmp_query =
			    (last_query + (max_queries - i)) % max_queries;
				
			if (tmp_query == last_query)
				break;

			/* check only up to one second */
			if (!tmp_query
			    || queries[tmp_query].time - timestamp > 1) {
				break;
			}
			dnbd_old_request =
			    (dnbd_request_t *) & queries[tmp_query].
			    request.data;

			/* someone requested the same block before? */
			if (dnbd_request->pos == dnbd_old_request->pos) {
				/* was it the same client, then retransmit
				as the packet was probably lost, otherwise
				drop the request */
				if (!((query->request.clientlen ==
				     queries[tmp_query].request.clientlen)
				    &&
				    (!memcmp
				     (&query->request.client,
				      &queries[tmp_query].request.client,
				      query->request.clientlen)))) {
					recent = 1;
					break;
				}
				else
					break;
			}
		} 
		rc = pthread_mutex_unlock(&query_mutex);

		if (recent)
			break;

		/* size of request block too high? */
		if (dnbd_request->len > MAX_BLOCK_SIZE)
			break;

		/* create a DNBD reply packet */
		dnbd_reply = (dnbd_reply_t *) query->reply.data;

		dnbd_reply->magic = htonl(DNBD_MAGIC);
		dnbd_reply->time = htons(dnbd_request->time);
		dnbd_reply->id = htons(query_info->id);
		dnbd_reply->pos = htonll(dnbd_request->pos);

		dnbd_reply->cmd =
		    htons((dnbd_request->cmd
			   & ~DNBD_CMD_CLI) | DNBD_CMD_SRV);

		/* read from underlying device/file */
		pthread_mutex_lock(&handler_mutex);
		filer_readblock(query_info->filer_info,
				(void *) dnbd_reply +
				sizeof(struct dnbd_reply),
				dnbd_request->len, dnbd_request->pos);

		pthread_mutex_unlock(&handler_mutex);

		query->reply.len =
		    dnbd_request->len + sizeof(dnbd_reply_t);

		query->time = time(NULL);

		/* send reply */
		net_tx(query_info->net_info, &query->reply);
		break;
	}


}

/*
 * function query_handle_loop(): get queries and handle them in a loop
 */
void *query_handle_loop(void *data)
{
	int rc;			
	query_t *query;				/* pointer to a request */
	int thread_id = *((int *) data);	/* thread id */

	printf("Starting thread '%d'\n", thread_id);
	fflush(stdout);

	rc = pthread_mutex_lock(&query_mutex);

	/* do forever.... */
	while (1) {

		if (num_queries > 0) {	
			/* a request is pending */
			query = query_get(&query_mutex);

			/* got a request? */
			if (query) {	
				
				rc = pthread_mutex_unlock(&query_mutex);
				/* handle request */
				query_handle(query_thread[thread_id].
					     query_info, query);

				rc = pthread_mutex_lock(&query_mutex);
			}
		} else {
			/* wait for a request to arrive */
			rc = pthread_cond_wait(&got_query, &query_mutex);
		}
	}
}

/*
 * function query_init(): initialize request handling
 * returns: pointer to data structure query_info (see header file)
 */
query_info_t *query_init(net_info_t * net_info, filer_info_t * filer_info,
			 int id, int threads)
{
	int i;
	query_info_t *query_info = NULL;

	query_info = (query_info_t *) malloc(sizeof(query_info_t));
	if (!query_info)
		return NULL;

	/* fill query_info structure */
	query_info->net_info = net_info;
	query_info->filer_info = filer_info;
	query_info->id = id;

	if (!(queries = (query_t *) malloc(sizeof(query_t) * max_queries))) {
		free(query_info);
		return NULL;
	}

	last_query = 0;
	next_query = 0;

	/* reserve memory for circular buffer */
	for (i = 0; i < max_queries; i++) {
		queries[i].reply.data =
		    malloc(MAX_BLOCK_SIZE + sizeof(dnbd_reply_t));
	}

	/* create the request-handling threads */
	for (i = 0; i < threads; i++) {

		query_thread[i].id = i;
		query_thread[i].query_info = query_info;

		pthread_create(&query_thread[i].p_thread, NULL,
			       query_handle_loop,
			       (void *) &query_thread[i].id);
	}

	/* create thread for receiving network requests */
	pthread_create(&query_info->p_thread, NULL,
		       query_add_loop, (void *) query_info);

	return query_info;
}