NetBSD/lib/libisns/isns_task.c

/*	$NetBSD: isns_task.c,v 1.1.1.1 2011/01/16 01:22:50 agc Exp $	*/

/*-
 * Copyright (c) 2004,2009 The NetBSD Foundation, Inc.
 * All rights reserved.
 *
 * This code is derived from software contributed to The NetBSD Foundation
 * by Wasabi Systems, Inc.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE NETBSD FOUNDATION, INC. AND CONTRIBUTORS
 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE FOUNDATION OR CONTRIBUTORS
 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */

#include <sys/cdefs.h>
__RCSID("$NetBSD: isns_task.c,v 1.1.1.1 2011/01/16 01:22:50 agc Exp $");

/*
 * isns_task.c
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>

#include "isns.h"
#include "isns_config.h"

static struct iovec write_buf[2 + (ISNS_MAX_PDU_PAYLOAD / ISNS_BUF_SIZE) +
    ((ISNS_MAX_PDU_PAYLOAD % ISNS_BUF_SIZE) != 0)];

static isns_task_handler isns_task_discover_server;
static isns_task_handler isns_task_reconnect_server;
static isns_task_handler isns_task_send_pdu;
static isns_task_handler isns_task_init_socket_io;
static isns_task_handler isns_task_init_refresh;


void
isns_run_task(struct isns_task_s *task_p)
{
	static isns_task_handler *task_dispatch_table[ISNS_NUM_TASKS] = {
		isns_task_discover_server,
		isns_task_reconnect_server,
		isns_task_send_pdu,
		isns_task_init_socket_io,
		isns_task_init_refresh
	};

	DBG("isns_run_task: task_type=%d\n", task_p->task_type);

	if (task_p->task_type < ARRAY_ELEMS(task_dispatch_table))
		task_dispatch_table[task_p->task_type](task_p);
	else
		DBG("isns_run_task: unknown task type=%d\n", task_p->task_type);
}


int
isns_wait_task(struct isns_task_s *task_p, const struct timespec *timeout_p)
{
	struct timeval tv_now;
	struct timespec ts_abstime;
	int rval;

	DBG("isns_wait_task: waitable=%d\n", task_p->waitable);

	if (!task_p->waitable)
		return EPERM;

	pthread_mutex_lock(&task_p->wait_mutex);

	if (timeout_p == NULL) {
		rval = pthread_cond_wait(&task_p->wait_condvar,
		    &task_p->wait_mutex);
	} else {
		gettimeofday(&tv_now, NULL);
		TIMEVAL_TO_TIMESPEC(&tv_now, &ts_abstime);
		timespecadd(&ts_abstime, timeout_p, &ts_abstime);

		rval = pthread_cond_timedwait(&task_p->wait_condvar,
		    &task_p->wait_mutex, &ts_abstime);
	}

	pthread_mutex_unlock(&task_p->wait_mutex);

	isns_free_task(task_p);

	DBG("isns_wait_task: wait done (rval=%d)\n", rval);

	return rval;
}


void
isns_end_task(struct isns_task_s *task_p)
{
	DBG("isns_end_task: %p\n", task_p);
	if (task_p == task_p->cfg_p->curtask_p)
		task_p->cfg_p->curtask_p = NULL;

	if (task_p->waitable)
		pthread_cond_signal(&task_p->wait_condvar);

	isns_free_task(task_p);
}


static void
isns_task_discover_server(struct isns_task_s *task_p)
{
	/* discover server here */
	DBG("isns_task_discover_server: entered\n");

	isns_end_task(task_p);
}


/*
 * isns_task_reconnect_server()
 */
static void
isns_task_reconnect_server(struct isns_task_s *task_p)
{
	struct addrinfo *ai_p;
	int rv;


	DBG("isns_task_reconnect_server: entered\n");
	
	ai_p = task_p->var.reconnect_server.ai_p;

	rv = isns_socket_create(&(task_p->cfg_p->sd), ai_p->ai_family,
	    ai_p->ai_socktype);
	if (rv != 0)
		return;

	rv = isns_socket_connect(task_p->cfg_p->sd, ai_p->ai_addr,
	    ai_p->ai_addrlen);
	if (rv != 0) {
		/* Add ISNS_EVT_TIMER_RECON to kqueue */
		rv = isns_change_kevent_list(task_p->cfg_p,
		    (uintptr_t)ISNS_EVT_TIMER_RECON, EVFILT_TIMER, EV_ADD,
		    (int64_t)ISNS_EVT_TIMER_RECON_PERIOD_MS,
		    (intptr_t)isns_kevent_timer_recon);
		if (rv == -1)
			DBG("isns_task_reconnect_server: error on "
			    "isns_change_kevent_list(1)\n");
	} else {
		task_p->cfg_p->sd_connected = 1;

		/* Add cfg_p->sd to kqueue */
		rv = isns_change_kevent_list(task_p->cfg_p,
		    (uintptr_t)(task_p->cfg_p->sd), EVFILT_READ,
		    EV_ADD | EV_CLEAR, (int64_t)0,
		    (intptr_t)isns_kevent_socket);
		if (rv == -1)
			DBG("isns_task_reconnect_server: error on "
			    "isns_change_kevent_lists(2)\n");

		isns_end_task(task_p);
	}
}

/*
 * isns_task_send_pdu()
 *
 * We send all of the pdu's associated with transaction task_p->trans_p here.
 *
 * Assumptions:
 *	(1) task_p->trans_p->pdu_req_list is an ordered (seq_id) list of
 *	    related (trans_id), appropriately sized pdus to be sent. The first
 *	    pdu has flag ISNS_FLAG_FIRST_PDU set and the last pdu has flag
 *	    ISNS_FLAG_LAST_PDU set.
 */
static void
isns_task_send_pdu(struct isns_task_s *task_p)
{
	struct iovec *iovp;
	struct isns_config_s *cfg_p;
	struct isns_pdu_s *pdu_p; /* points to first pdu in pdu_req_list */
	struct isns_buffer_s *buf_p;
	ssize_t bytes_written;
	ssize_t count;
	size_t bytes_to_write;
	int iovcnt, cur_iovec;
	char *ptr;


	DBG("isns_task_send_pdu: entered\n");

	cfg_p = task_p->cfg_p;
	pdu_p = task_p->var.send_pdu.pdu_p;

	while (pdu_p != NULL) {
		/* adjust byte order if necessary */
		if (pdu_p->byteorder_host) {
			pdu_p->hdr.isnsp_version = isns_htons(pdu_p->hdr.
			    isnsp_version);
			pdu_p->hdr.func_id = isns_htons(pdu_p->hdr.func_id);
			pdu_p->hdr.payload_len = isns_htons(pdu_p->hdr.
			    payload_len);
			pdu_p->hdr.flags = isns_htons(pdu_p->hdr.flags);
			pdu_p->hdr.trans_id = isns_htons(pdu_p->hdr.trans_id);
			pdu_p->hdr.seq_id = isns_htons(pdu_p->hdr.seq_id);

			pdu_p->byteorder_host = 0;
		}
		DUMP_PDU(pdu_p);

		/* send PDU via socket here */
		write_buf[0].iov_base = &(pdu_p->hdr);
		write_buf[0].iov_len = sizeof(pdu_p->hdr);
		bytes_to_write = write_buf[0].iov_len;
		iovcnt = 1;

		buf_p = pdu_p->payload_p;
		while (buf_p != NULL) {
			write_buf[iovcnt].iov_base = isns_buffer_data(buf_p,0);
			write_buf[iovcnt].iov_len = buf_p->cur_len;
			bytes_to_write += write_buf[iovcnt].iov_len;
			iovcnt++;
			buf_p = buf_p->next;
		}

		/* iovcnt and bytes_to_write are initialized */ 
		cur_iovec = 0;
		buf_p = ((struct isns_buffer_s *)(void *)pdu_p) - 1;
		do {
			iovp = &(write_buf[cur_iovec]);
			bytes_written = isns_socket_writev(cfg_p->sd, iovp,
			    iovcnt);
			if (bytes_written == -1) {
				DBG("isns_task_send_pdu: error on "
			    	"isns_socket_writev\n");
				isns_socket_close(cfg_p->sd);
				cfg_p->sd_connected = 0;

				isns_process_connection_loss(cfg_p);

				if (cfg_p->pdu_in_p != NULL) {
					isns_free_pdu(cfg_p->pdu_in_p);
					cfg_p->pdu_in_p = NULL;
				}
			
				break;
			}

			if (bytes_written < (ssize_t)bytes_to_write) {
				count = bytes_written;
				while (buf_p != NULL) { /* -OR- while (1) */
					if ((unsigned)count >= write_buf[
					    cur_iovec].iov_len) {
						count -= write_buf[cur_iovec].
						    iov_len;
						if (cur_iovec == 0)
							buf_p = pdu_p->
							    payload_p; 
						else
							buf_p = buf_p->next;
						cur_iovec++;
						iovcnt--;

						if (count == 0) {
							/* Do another write */
							break;
						} else {
							/* Look at new iovec */ 
							continue;
						}
					} else {
						write_buf[cur_iovec].iov_len -=
						    count;

						ptr = (char *) write_buf[cur_iovec].iov_base;
						ptr += count;
						write_buf[cur_iovec].iov_base = ptr;

						/* Do another write */
						break;
					}
				}
			}

			bytes_to_write -= bytes_written;
		} while (bytes_to_write);

		pdu_p = pdu_p->next;
	}

	if (!task_p->waitable) {
		isns_complete_trans(task_p->var.send_pdu.trans_p);
		isns_end_task(task_p);
	}
}

/*
 * isns_task_init_socket_io()
 */
static void
isns_task_init_socket_io(struct isns_task_s *task_p)
{
	struct isns_config_s *cfg_p;
	int rv;


	DBG("isns_task_init_socket_io: entered\n");

	cfg_p = task_p->cfg_p;

	if (cfg_p->sd_connected) {
		isns_socket_close(cfg_p->sd);
		cfg_p->sd_connected = 0;

		/* We may have received part of an unsolicited/duplicate pdu */
		if (cfg_p->pdu_in_p != NULL) {
			isns_free_pdu(cfg_p->pdu_in_p);
			cfg_p->pdu_in_p = NULL;
		}
	}

	/* May have an allocated 'struct addrinfo', whether connected or not */
	if (cfg_p->ai_p != NULL) {
		isns_free(cfg_p->ai_p);
		cfg_p->ai_p = NULL;
	}

	cfg_p->sd = task_p->var.init_socket_io.sd;
	cfg_p->ai_p = task_p->var.init_socket_io.ai_p;

	cfg_p->sd_connected = 1;

	/* Add cfg_p->sd to kqueue */
	rv = isns_change_kevent_list(cfg_p, (uintptr_t)cfg_p->sd,
	    EVFILT_READ, EV_ADD | EV_CLEAR, (int64_t)0,
	    (intptr_t)isns_kevent_socket);
	if (rv == -1)
		DBG("isns_task_init_socket_io: error on "
		    "isns_change_kevent_list\n");

	isns_end_task(task_p);
}


/*
 * isns_task_init_refresh(struct isns_task_s *task_p)
 */
static void
isns_task_init_refresh(struct isns_task_s *task_p)
{
	struct isns_config_s *cfg_p;
	int rval;

	DBG("isns_task_init_refresh: entered\n");

	/* Free any previous refresh info. */
	cfg_p = task_p->cfg_p;
	if (cfg_p->refresh_p != NULL) {
		if (cfg_p->refresh_p->trans_p != NULL)
			isns_free_trans(cfg_p->refresh_p->trans_p);
		isns_free(cfg_p->refresh_p);
	}

	/* Assign new refresh info into config struct. */
	cfg_p->refresh_p = task_p->var.init_refresh.ref_p;
	cfg_p->refresh_p->trans_p = NULL;

	/* Setup (or change) kevent timer for reg refresh. */
	rval = isns_change_kevent_list(cfg_p,
	    (uintptr_t)ISNS_EVT_TIMER_REFRESH, EVFILT_TIMER,
	    EV_ADD | EV_ENABLE, (int64_t)cfg_p->refresh_p->interval * 1000,
	    (intptr_t)isns_kevent_timer_refresh);
	if (rval == -1) {
		DBG("isns_task_init_refresh: "
		    "error on isns_change_kevent_list()\n");
	}

	isns_end_task(task_p);
}


struct isns_task_s *
isns_new_task(struct isns_config_s *cfg_p, uint8_t task_type, int waitable)
{
	struct isns_buffer_s *buf_p;
	struct isns_task_s *task_p;
	pthread_mutexattr_t mutexattr;
	pthread_condattr_t condattr;

	task_p = NULL;
	buf_p = isns_new_buffer((int)sizeof(struct isns_task_s));
	if (buf_p) {
		task_p = (struct isns_task_s *)isns_buffer_data(buf_p, 0);
		task_p->cfg_p = cfg_p;
		task_p->task_type = task_type;
		task_p->waitable = waitable;

		if (waitable) {
			pthread_mutexattr_init(&mutexattr);
			pthread_mutexattr_settype(&mutexattr,
			    ISNS_MUTEX_TYPE_NORMAL);
			pthread_mutex_init(&task_p->wait_mutex, &mutexattr);

			pthread_condattr_init(&condattr);
			pthread_cond_init(&task_p->wait_condvar, &condattr);
			task_p->wait_ref_count = 2;
		}
	}

	DBG("isns_new_task: %p, waitable=%d\n", task_p, waitable);

	return task_p;
}


void
isns_free_task(struct isns_task_s *task_p)
{
	struct isns_buffer_s *buf_p;
	int ref_count;

	DBG("isns_free_task: %p\n", task_p);
	if (task_p->waitable) {
		pthread_mutex_lock(&task_p->wait_mutex);
		ref_count = --task_p->wait_ref_count;
		pthread_mutex_unlock(&task_p->wait_mutex);

		if (ref_count > 0) {
			DBG("isns_free_task: ref_count > 0, no free done\n");
			return;
		}

		pthread_mutex_destroy(&task_p->wait_mutex);
		pthread_cond_destroy(&task_p->wait_condvar);
	}
	buf_p = ((struct isns_buffer_s *)(void *)(task_p))-1;
	isns_free_buffer(buf_p);
}


void
isns_taskq_insert_head(struct isns_config_s *cfg_p,
    struct isns_task_s *task_p)
{
	pthread_mutex_lock(&cfg_p->taskq_mutex);
	SIMPLEQ_INSERT_HEAD(&cfg_p->taskq_head, task_p, taskq_entry);
	pthread_mutex_unlock(&cfg_p->taskq_mutex);

	DBG("isns_taskq_insert_head: %p\n", task_p);
}


void
isns_taskq_insert_tail(struct isns_config_s *cfg_p,
    struct isns_task_s *task_p)
{
	pthread_mutex_lock(&cfg_p->taskq_mutex);
	SIMPLEQ_INSERT_TAIL(&cfg_p->taskq_head, task_p, taskq_entry);
	pthread_mutex_unlock(&cfg_p->taskq_mutex);

	DBG("isns_taskq_insert_tail: %p\n", task_p);
}


struct isns_task_s *
isns_taskq_remove(struct isns_config_s *cfg_p)
{
	struct isns_task_s *task_p = NULL;

	pthread_mutex_lock(&cfg_p->taskq_mutex);
	if ((task_p = SIMPLEQ_FIRST(&cfg_p->taskq_head)) != NULL)
		SIMPLEQ_REMOVE_HEAD(&cfg_p->taskq_head, taskq_entry);
	pthread_mutex_unlock(&cfg_p->taskq_mutex);

	DBG("isns_taskq_remove: %p\n", task_p);

	return task_p;
}


struct isns_task_s *
isns_taskq_remove_trans(struct isns_config_s *cfg_p, uint16_t trans_id)
{
	struct isns_task_s *task_p;
	int trans_found;

	trans_found = 0;
	pthread_mutex_lock(&cfg_p->taskq_mutex);
	SIMPLEQ_FOREACH(task_p, &cfg_p->taskq_head, taskq_entry) {
		if ((task_p->task_type == ISNS_TASK_SEND_PDU)
		    && (task_p->var.send_pdu.trans_p->id == trans_id)) {
			trans_found = 1;
			break;
		}
	}
	if (trans_found) {
		SIMPLEQ_REMOVE(&cfg_p->taskq_head, task_p, isns_task_s,
		    taskq_entry);
	}
	pthread_mutex_unlock(&cfg_p->taskq_mutex);

	return (trans_found ? task_p : NULL);
}