/*
 * Copyright (c) 1999, 2000 University of Utah and the Flux Group.
 * All rights reserved.
 * 
 * This file is part of the Flux OSKit.  The OSKit is free software, also known
 * as "open source;" you can redistribute it and/or modify it under the terms
 * of the GNU General Public License (GPL), version 2, as published by the Free
 * Software Foundation (FSF).  To explore alternate licensing terms, contact
 * the University of Utah at csl-dist@cs.utah.edu or +1-801-585-3271.
 * 
 * The OSKit is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.  See the GPL for more details.  You should have
 * received a copy of the GPL along with the OSKit; see the file COPYING.  If
 * not, write to the FSF, 59 Temple Place #330, Boston, MA 02111-1307, USA.
 */

/*
 * This implements a threaded adaptor on an oskit_netio.  It is basically
 * the threaded listener code (see listener.c).
 */

#include <oskit/com.h>
#include <oskit/io/netio.h>
#include <oskit/c/stdio.h>
#include <oskit/c/stdlib.h>
#include <oskit/c/string.h>
#include <oskit/c/malloc.h>
#include "pthread_internal.h"
#include "pthread_mutex.h"

typedef struct pthread_netio_impl {
	oskit_netio_t	 iol;		/* COM interface */
        unsigned	 count;		/* reference count */
	pthread_t	 tid;		/* The thread */
	oskit_netio_t	 *netio;	/* Actual netio */
	oskit_bufio_t	 *bufio;	/* Bufio arg (used as a condition) */
	oskit_size_t	 size;		/* Size arg */
	pthread_mutex_t	 mutex;
	pthread_cond_t	 cond;
} pthread_netio_impl_t;

static OSKIT_COMDECL
netio_query(oskit_netio_t *io, const oskit_iid_t *iid, void **out_ihandle)
{
        pthread_netio_impl_t *nio = (pthread_netio_impl_t *)io;

        if (memcmp(iid, &oskit_iunknown_iid, sizeof(*iid)) == 0 ||
            memcmp(iid, &oskit_netio_iid, sizeof(*iid)) == 0) {
                *out_ihandle = &nio->iol;
                ++nio->count;
                return 0;
        }

        *out_ihandle = NULL;
        return OSKIT_E_NOINTERFACE;
}

static OSKIT_COMDECL_U
netio_addref(oskit_netio_t *io)
{
        pthread_netio_impl_t *nio = (pthread_netio_impl_t *)io;

        if (nio->count == 0)
		return OSKIT_E_INVALIDARG;

        return ++nio->count;
}

static OSKIT_COMDECL_U
netio_release(oskit_netio_t *io)
{
        pthread_netio_impl_t *nio = (pthread_netio_impl_t *)io;
        unsigned newcount;

        if (nio == NULL || nio->count == 0)
		return OSKIT_E_INVALIDARG;

        newcount = --nio->count;
        if (newcount == 0) {
		/*
		 * Cancel the thread, but do not free the netio yet, since
		 * we could end up deleting it right out from under the thread
		 * while it is using it. Let the cancel cleanup handler take
		 * care of it.
		 */
		pthread_cancel(nio->tid);
        }

        return newcount;
}

static OSKIT_COMDECL
netio_push(oskit_netio_t *io, oskit_bufio_t *b, oskit_size_t size)
{
        pthread_netio_impl_t *nio = (pthread_netio_impl_t *)io;

	/*
	 * Spin lock the mutex and set the bufio object. Signal the
	 * the thread to invoke the netio. This could happen in
	 * an interrupt context, hence the spinning mutex lock. The
	 * thread will not keep this lock very long, so its okay.
	 * It will also disable interrupts to prevent interrupt deadlock.
	 *
	 * XXX if the bufio object is already set, then the previous
	 * push hasn't completed.  Just return an error and, presumably,
	 * drop the packet.
	 */
	fast_mutex_spinlock(&nio->mutex);
	if (nio->bufio != 0) {
		fast_mutex_unlock(&nio->mutex);
		return ENOMEM;
	}
	oskit_bufio_addref(b);
	nio->bufio = b;
	nio->size = size;
	fast_mutex_unlock(&nio->mutex);
	pthread_cond_signal(&nio->cond);

	return 0;
}

static struct oskit_netio_ops oskit_pthread_netio_ops = {
	netio_query,
	netio_addref,
	netio_release,
	netio_push
};

static void
netio_canceled(void *arg)
{
	pthread_netio_impl_t	*nio = (pthread_netio_impl_t *) arg;

	DPRINTF("%p\n", arg);
	
	oskit_netio_release(nio->netio);
	sfree(nio, sizeof(*nio));
}

static void *
netio_run(void *arg)
{
	pthread_netio_impl_t	*nio = (pthread_netio_impl_t *) arg;
	oskit_bufio_t		*bufio;
	oskit_size_t		size;

	pthread_cleanup_push(netio_canceled, nio);

	/*
	 * First off, just signal the creator that we are ready.
	 */
	pthread_mutex_lock(&nio->mutex);
	pthread_mutex_unlock(&nio->mutex);
	pthread_cond_signal(&nio->cond);

	/*
	 * Now loop. The safe version of condwait allows interrupts to
	 * be disabled on entry, but does not check for cancelation.
	 */
	while (1) {
		assert_interrupts_enabled();
		disable_interrupts();
		
		pthread_mutex_lock(&nio->mutex);
		while (nio->bufio == 0) {
			pthread_cond_wait_safe(&nio->cond, &nio->mutex);
			pthread_testcancel();
		}
		bufio = nio->bufio;
		size = nio->size;
		nio->bufio = 0;
		pthread_mutex_unlock(&nio->mutex);
		
		enable_interrupts();

		oskit_netio_push(nio->netio, bufio, size);
		pthread_testcancel();
	}

	pthread_cleanup_pop(1);
}

oskit_netio_t *
oskit_threaded_netio_create(oskit_error_t (*func)(void *data, oskit_bufio_t *b,
						  oskit_size_t pkt_size),
			    void *data)
{
	pthread_netio_impl_t	*nio;
	oskit_error_t		rc;
	pthread_attr_t		attr;
	
	if ((nio = (pthread_netio_impl_t *) smalloc(sizeof(*nio))) == NULL)
		return 0;

	nio->iol.ops = &oskit_pthread_netio_ops;
	nio->count = 1;
	nio->bufio = 0;

	/*
	 * Create the underlying netio.
	 */
	nio->netio = oskit_netio_create(func, data);

	pthread_mutex_init(&nio->mutex, NULL);
	pthread_cond_init(&nio->cond, NULL);
	pthread_mutex_lock(&nio->mutex);

	/*
	 * Create the thread.
	 * XXX we create it to run at the highest priority so somewhat
	 * simulate delivery at interrupt time; i.e., it should get stuck
	 * behind "regular" threads.
	 */
	rc = pthread_attr_init(&attr);
	if (rc == 0)
		rc = oskit_pthread_attr_setprio(&attr, PRIORITY_MAX);
	if (rc == 0)
		rc = pthread_create(&nio->tid, &attr, netio_run, (void *) nio);
	if (rc != 0) {
		oskit_netio_release(nio->netio);
		sfree(nio, sizeof(nio));
		return 0;
	}
	pthread_detach(nio->tid);

	/*
	 * Wait until the thread runs and waits.
	 */
	pthread_cond_wait(&nio->cond, &nio->mutex);
	pthread_mutex_unlock(&nio->mutex);

	return &nio->iol;
}