doubango/trunk/tinyNET/src/tnet_transport_poll.c

/*
* Copyright (C) 2009-2010 Mamadou Diop.
*
* Contact: Mamadou Diop <diopmamadou(at)doubango.org>
*	
* This file is part of Open Source Doubango Framework.
*
* DOUBANGO 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 3 of the License, or
* (at your option) any later version.
*	
* DOUBANGO 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
* GNU General Public License for more details.
*	
* You should have received a copy of the GNU General Public License
* along with DOUBANGO.
*
*/

/**@file tnet_transport_poll.c
 * @brief Network transport layer using polling.
 *
 * @author Mamadou Diop <diopmamadou(at)doubango.org>
 *
 * @date Created: Sat Nov 8 16:54:58 2009 mdiop
 */
#include "tnet_transport.h"
#include "tsk_memory.h"
#include "tsk_string.h"
#include "tsk_debug.h"
#include "tsk_thread.h"
#include "tsk_buffer.h"
#include "tsk_safeobj.h"

#if TNET_USE_POLL

#include "tnet_poll.h"

#define TNET_MAX_FDS		64

/*== Socket description ==*/
typedef struct transport_socket_s
{
	tnet_fd_t fd;
	tsk_bool_t owner;
	tsk_bool_t connected;
	tsk_bool_t paused;

	tnet_socket_type_t type;
	tnet_tls_socket_handle_t* tlshandle;
}
transport_socket_t;

/*== Transport context structure definition ==*/
typedef struct transport_context_s
{
	TSK_DECLARE_OBJECT;
	
	tsk_size_t count;
	short events;
	tnet_fd_t pipeW;
	tnet_fd_t pipeR;
	tnet_pollfd_t ufds[TNET_MAX_FDS];
	transport_socket_t* sockets[TNET_MAX_FDS];

	TSK_DECLARE_SAFEOBJ;
}
transport_context_t;

static const transport_socket_t* getSocket(transport_context_t *context, tnet_fd_t fd);
static int addSocket(tnet_fd_t fd, tnet_socket_type_t type, tnet_transport_t *transport, tsk_bool_t take_ownership, tsk_bool_t is_client);
static int removeSocket(int index, transport_context_t *context);


int tnet_transport_add_socket(const tnet_transport_handle_t *handle, tnet_fd_t fd, tnet_socket_type_t type, tsk_bool_t take_ownership, tsk_bool_t isClient)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	transport_context_t* context;
	static char c = '\0';
	int ret = -1;

	if(!transport){
		TSK_DEBUG_ERROR("Invalid server handle.");
		return ret;
	}
	
	if(!(context = (transport_context_t*)transport->context)){
		TSK_DEBUG_ERROR("Invalid context.");
		return -2;
	}

	if(TNET_SOCKET_TYPE_IS_TLS(type)){
		transport->tls.have_tls = 1;
	}
	
	if((ret = addSocket(fd, type, transport, take_ownership, isClient))){
		TSK_DEBUG_ERROR("Failed to add new Socket.");
		return ret;
	}

	// signal
	if(context->pipeW){
		if((ret = write(context->pipeW, &c, 1)) > 0){
			TSK_DEBUG_INFO("Socket added (external call) %d", fd);
			return 0;
		}
		else{
			TSK_DEBUG_ERROR("Failed to add new Socket.");
			return ret;
		}
	}else{
		TSK_DEBUG_WARN("pipeW (write site) not initialized yet.");
		return 0; //Will be taken when mainthead start
	}
}

int tnet_transport_pause_socket(const tnet_transport_handle_t *handle, tnet_fd_t fd, tsk_bool_t pause)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	transport_context_t *context;
	transport_socket_t* socket;

	if(!transport || !(context = (transport_context_t*)transport->context)){
		TSK_DEBUG_ERROR("Invalid parameter");
		return -1;
	}

	if((socket = getSocket(context, fd))){
		socket->paused = pause;
	}
	else{
		TSK_DEBUG_WARN("Socket does not exist in this context");
	}
	return 0;
}

/* Remove socket */
int tnet_transport_remove_socket(const tnet_transport_handle_t *handle, tnet_fd_t *fd)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	transport_context_t *context;
	int ret = -1;
	tsk_size_t i;
	tsk_bool_t found = tsk_false;
	
	TSK_DEBUG_INFO("Removing socket %d", *fd);

	if(!transport){
		TSK_DEBUG_ERROR("Invalid server handle.");
		return ret;
	}
	
	if(!(context = (transport_context_t*)transport->context)){
		TSK_DEBUG_ERROR("Invalid context.");
		return -2;
	}
	
	for(i=0; i<context->count; i++){
		if(context->sockets[i]->fd == *fd){
			removeSocket(i, context);
			found = tsk_true;
			*fd = TNET_INVALID_FD;
			break;
		}
	}
	
	if(found){
		/* Signal */
		static char c = '\0';
		ret = write(context->pipeW, &c, 1);
		return (ret > 0 ? 0 : ret);
	}
	
	// ...
	
	return -1;
}


tsk_size_t tnet_transport_send(const tnet_transport_handle_t *handle, tnet_fd_t from, const void* buf, tsk_size_t size)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	int numberOfBytesSent = 0;

	if(!transport){
		TSK_DEBUG_ERROR("Invalid transport handle.");
		goto bail;
	}

	if(transport->tls.have_tls){
		const transport_socket_t* socket = getSocket(transport->context, from);
		if(socket && socket->tlshandle){
			if(!tnet_tls_socket_send(socket->tlshandle, buf, size)){
				numberOfBytesSent = size;
			}
			else{
				numberOfBytesSent = 0;
			}
			goto bail;
		}
	}
	else if((numberOfBytesSent = send(from, buf, size, 0)) <= 0){
		TNET_PRINT_LAST_ERROR("send have failed.");

		//tnet_sockfd_close(&from);
		goto bail;
	}

bail:
	return numberOfBytesSent;
}

tsk_size_t tnet_transport_sendto(const tnet_transport_handle_t *handle, tnet_fd_t from, const struct sockaddr *to, const void* buf, tsk_size_t size)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	int numberOfBytesSent = 0;
	
	if(!transport){
		TSK_DEBUG_ERROR("Invalid server handle.");
		goto bail;
	}
	
	if(!TNET_SOCKET_TYPE_IS_DGRAM(transport->master->type)){
		TSK_DEBUG_ERROR("In order to use WSASendTo you must use an udp transport.");
		goto bail;
	}
	
    if((numberOfBytesSent = sendto(from, buf, size, 0, to, sizeof(*to))) <= 0){
		TNET_PRINT_LAST_ERROR("sendto have failed.");
		goto bail;
	}
		
bail:
	return numberOfBytesSent;
}

int tnet_transport_have_socket(const tnet_transport_handle_t *handle, tnet_fd_t fd)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	
	if(!transport){
		TSK_DEBUG_ERROR("Invalid server handle.");
		return 0;
	}
	
	return (getSocket((transport_context_t*)transport->context, fd) != 0);
}

const tnet_tls_socket_handle_t* tnet_transport_get_tlshandle(const tnet_transport_handle_t *handle, tnet_fd_t fd)
{
	tnet_transport_t *transport = (tnet_transport_t*)handle;
	const transport_socket_t *socket;
	
	if(!transport){
		TSK_DEBUG_ERROR("Invalid server handle.");
		return 0;
	}
	
	if((socket = getSocket((transport_context_t*)transport->context, fd))){
		return socket->tlshandle;
	}
	return 0;
}


/*== Get socket ==*/
static const transport_socket_t* getSocket(transport_context_t *context, tnet_fd_t fd)
{
	tsk_size_t i;
	transport_socket_t* ret = 0;

	if(context){
		tsk_safeobj_lock(context);
		for(i=0; i<context->count; i++){
			if(context->sockets[i]->fd == fd){
				ret = context->sockets[i];
				break;
			}
		}
		tsk_safeobj_unlock(context);
	}
	
	return ret;
}

/*== Add new socket ==*/
int addSocket(tnet_fd_t fd, tnet_socket_type_t type, tnet_transport_t *transport, tsk_bool_t take_ownership, tsk_bool_t is_client)
{
	transport_context_t *context = transport?transport->context:0;
	if(context){
		transport_socket_t *sock = tsk_calloc(1, sizeof(transport_socket_t));
		sock->fd = fd;
		sock->type = type;
		sock->owner = take_ownership;

		if(TNET_SOCKET_TYPE_IS_TLS(sock->type)){
			sock->tlshandle = tnet_sockfd_set_tlsfiles(sock->fd, is_client, transport->tls.ca, transport->tls.pvk, transport->tls.pbk);
		}
		
		tsk_safeobj_lock(context);
		
		context->ufds[context->count].fd = fd;
		context->ufds[context->count].events = (fd == context->pipeR) ? TNET_POLLIN : context->events;
		context->ufds[context->count].revents = 0;
		context->sockets[context->count] = sock;
		
		context->count++;
		
		tsk_safeobj_unlock(context);
		
		TSK_DEBUG_INFO("Socket added %d", fd);
		
		return 0;
	}
	else{
		TSK_DEBUG_ERROR("Context is Null.");
		return -1;
	}
}

/*== change connection state ==*/
/*
static void setConnected(tnet_fd_t fd, transport_context_t *context, int connected)
{
	tsk_size_t i;

	for(i=0; i<context->count; i++)
	{
		if(context->sockets[i]->fd == fd){
			context->sockets[i]->connected = connected;
		}
	}
}
*/

/*== Remove socket ==*/
int removeSocket(int index, transport_context_t *context)
{
	int i;
	
	tsk_safeobj_lock(context);

	if(index < (int)context->count){

		/* Close the socket if we are the owner. */
		if(context->sockets[index]->owner){
			tnet_sockfd_close(&(context->sockets[index]->fd));
		}
		
		/* Free tls context */
		TSK_OBJECT_SAFE_FREE(context->sockets[index]->tlshandle);
		
		// Free socket
		TSK_FREE(context->sockets[index]);
		
		for(i=index ; i<context->count-1; i++){			
			context->sockets[i] = context->sockets[i+1];
			context->ufds[i] = context->ufds[i+1];
		}
		
		context->sockets[context->count-1] = 0;
		context->ufds[context->count-1].fd = 0;
		context->ufds[context->count-1].events = 0;
		context->ufds[context->count-1].revents = 0;
		
		context->count--;
		TSK_DEBUG_INFO("Socket removed");
	}

	tsk_safeobj_unlock(context);
	
	return 0;
}

int tnet_transport_stop(tnet_transport_t *transport)
{	
	int ret;
	transport_context_t *context;

	if(!transport){
		return -1;
	}
	
	context = transport->context;

	if((ret = tsk_runnable_stop(TSK_RUNNABLE(transport)))){
		return ret;
	}
	
	if(context){
		static char c = '\0';
		
		// signal
		tsk_safeobj_lock(context); // =>MUST
		write(context->pipeW, &c, 1);
		tsk_safeobj_unlock(context);
	}
	
	if(transport->mainThreadId[0]){
		return tsk_thread_join(transport->mainThreadId);
	}
	else{
		/* already soppped */
		return 0;
	}
}

int tnet_transport_prepare(tnet_transport_t *transport)
{
	int ret = -1;
	transport_context_t *context;
	tnet_fd_t pipes[2];
	
	if(!transport || !transport->context){
		TSK_DEBUG_ERROR("Invalid parameter.");
		return -1;
	}
	else{
		context = transport->context;
	}
	
	if(transport->prepared){
		TSK_DEBUG_ERROR("Transport already prepared.");
		return -2;
	}
	
	/* set events */
	context->events = TNET_POLLIN | TNET_POLLNVAL | TNET_POLLERR;
	if(TNET_SOCKET_TYPE_IS_STREAM(transport->master->type)){
//		context->events |= TNET_POLLPRI 
//#if !defined(ANDROID)
//			| TNET_POLLHUP /* FIXME: always present */
//#endif
//			;
	}
	
	/* Start listening */
	if(TNET_SOCKET_TYPE_IS_STREAM(transport->master->type)){
		if((ret = tnet_sockfd_listen(transport->master->fd, TNET_MAX_FDS))){
			TNET_PRINT_LAST_ERROR("listen have failed.");
			goto bail;
		}
	}
	
	/* Create and add pipes to the fd_set */
	if((ret = pipe(pipes))){
		TNET_PRINT_LAST_ERROR("Failed to create new pipes.");
		goto bail;
	}
	
	/* set both R and W sides */
	context->pipeR = pipes[0];
	context->pipeW = pipes[1];
	
	/* add R side */
	TSK_DEBUG_INFO("pipeR fd=%d", context->pipeR);
	if((ret = addSocket(context->pipeR, transport->master->type, transport, tsk_true, tsk_false))){
		goto bail;
	}
	
	/* Add the master socket to the context. */
	TSK_DEBUG_INFO("master fd=%d", transport->master->fd);
	if((ret = addSocket(transport->master->fd, transport->master->type, transport, tsk_true, tsk_false))){
		TSK_DEBUG_ERROR("Failed to add master socket");
		goto bail;
	}
	
	transport->prepared = tsk_true;
	
bail:
	return ret;
}

int tnet_transport_unprepare(tnet_transport_t *transport)
{
	int ret = -1;
	transport_context_t *context;
	
	if(!transport || !transport->context){
		TSK_DEBUG_ERROR("Invalid parameter.");
		return -1;
	}
	else{
		context = transport->context;
	}

	if(!transport->prepared){
		return 0;
	}

	transport->prepared = tsk_false;
	
	while(context->count){
		removeSocket(0, context); // safe
	}

	return 0;
}

/*=== Main thread */
void *tnet_transport_mainthread(void *param)
{
	tnet_transport_t *transport = param;
	transport_context_t *context = transport->context;
	int ret;
	tsk_size_t i;

	transport_socket_t* active_socket;

	/* check whether the transport is already prepared */
	if(!transport->prepared){
		TSK_DEBUG_ERROR("Transport must be prepared before strating.");
		goto bail;
	}
	
	TSK_DEBUG_INFO("Starting [%s] server with IP {%s} on port {%d}...", transport->description, transport->master->ip, transport->master->port);

	while(TSK_RUNNABLE(transport)->running || TSK_RUNNABLE(transport)->started){
		if((ret = tnet_poll(context->ufds, context->count, -1)) < 0){
			TNET_PRINT_LAST_ERROR("poll have failed.");
			goto bail;
		}

		if(!TSK_RUNNABLE(transport)->running && !TSK_RUNNABLE(transport)->started){
			goto bail;
		}
		
		/* lock context */
		tsk_safeobj_lock(context);

		/* == == */
		for(i=0; i<context->count; i++)
		{
			if(!context->ufds[i].revents){
				continue;
			}

			if(context->ufds[i].fd == context->pipeR){
				if(context->ufds[i].revents & TNET_POLLIN){
					static char __buffer[64];
					if(read(context->pipeR, __buffer, sizeof(__buffer))<0){
						TNET_PRINT_LAST_ERROR("Failed to read from the Pipe");
					}
				}
				TSK_DEBUG_INFO("PipeR event %d", context->ufds[i].revents);
				context->ufds[i].revents = 0;
				continue;
			}

			/* Get active event and socket */
			active_socket = context->sockets[i];
			
			/*================== POLLIN ==================*/
			if(context->ufds[i].revents & TNET_POLLIN)
			{
				tsk_size_t len = 0;
				void* buffer = tsk_null;
				tnet_transport_event_t* e;
				
				/* TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLIN", transport->description); */

				//
				// FIXME: check if accept() is needed or not
				//

				
				/* check whether the socket is paused or not */
				if(active_socket->paused){
					TSK_DEBUG_INFO("Socket is paused");
					break;
				}

				/* Retrieve the amount of pending data.
				 * IMPORTANT: If you are using Symbian please update your SDK to the latest build (August 2009) to have 'FIONREAD'.
				 * This apply whatever you are using the 3rd or 5th edition.
				 * Download link: http://wiki.forum.nokia.com/index.php/Open_C/C%2B%2B_Release_History
				 */
				if(tnet_ioctlt(active_socket->fd, FIONREAD, &len) < 0){
					TNET_PRINT_LAST_ERROR("IOCTLT FAILED.");
					continue;
				}
				
				if(!len){
					TSK_DEBUG_WARN("IOCTLT returned zero.");
#if defined(ANDROID) || 1 /* FIXME: On Android/MAC OS X this mean that the socket has been closed?  */
					TSK_RUNNABLE_ENQUEUE(transport, event_closed, transport->callback_data, active_socket->fd);
					removeSocket(i, context);
#else
					recv(active_socket->fd, 0, 0, 0);
#endif
					continue;

				}
				
				if(!(buffer = tsk_calloc(len, sizeof(uint8_t)))){
					TSK_DEBUG_ERROR("TSK_CALLOC FAILED.");
					continue;
				}
				

				/* Receive the waiting data. */
				if(active_socket->tlshandle){
					int isEncrypted;
					tsk_size_t tlslen = len;
					if((ret = tnet_tls_socket_recv(active_socket->tlshandle, &buffer, &tlslen, &isEncrypted)) == 0){
						if(isEncrypted){
							TSK_FREE(buffer);
							continue;
						}
						len = tlslen;
					}
				}
				else if((ret = tnet_sockfd_recv(active_socket->fd, buffer, len, 0)) < 0){
					TSK_FREE(buffer);
					
					removeSocket(i, context);
					TNET_PRINT_LAST_ERROR("recv have failed.");
					continue;
				}
				else if((len != (tsk_size_t)ret) && len){ /* useless test? */
					len = (tsk_size_t)ret;
					/* buffer = tsk_realloc(buffer, len); */
				}
					
				e = tnet_transport_event_create(event_data, transport->callback_data, active_socket->fd);
				e->data = buffer;
				e->size = len;
				
				TSK_RUNNABLE_ENQUEUE_OBJECT(TSK_RUNNABLE(transport), e);
			}


			/*================== TNET_POLLOUT ==================*/
			if(context->ufds[i].revents & TNET_POLLOUT)
			{
				TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLOUT", transport->description);
			}


			/*================== TNET_POLLPRI ==================*/
			if(context->ufds[i].revents & TNET_POLLPRI)
			{
				TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLPRI", transport->description);
			}
			
			/*================== TNET_POLLHUP ==================*/
			if(context->ufds[i].revents & (TNET_POLLHUP))
			{
				TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLHUP", transport->description);
#if defined(ANDROID)
				/* FIXME */
#else
				TSK_RUNNABLE_ENQUEUE(transport, event_closed, transport->callback_data, active_socket->fd);
				removeSocket(i, context);
#endif
			}

			/*================== TNET_POLLERR ==================*/
			if(context->ufds[i].revents & (TNET_POLLERR))
			{
				TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLERR", transport->description);

				TSK_RUNNABLE_ENQUEUE(transport, event_error, transport->callback_data, active_socket->fd);
				removeSocket(i, context);
			}
			
			/*================== TNET_POLLNVAL ==================*/
			if(context->ufds[i].revents & (TNET_POLLNVAL))
			{
				TSK_DEBUG_INFO("NETWORK EVENT FOR SERVER [%s] -- TNET_POLLNVAL", transport->description);
				
				TSK_RUNNABLE_ENQUEUE(transport, event_error, transport->callback_data, active_socket->fd);
				removeSocket(i, context);
			}


		}/* for */

done:
		/* unlock context */
		tsk_safeobj_unlock(context);

	} /* while */

bail:
	
	
	TSK_DEBUG_INFO("Stopped [%s] server with IP {%s} on port {%d}...", transport->description, transport->master->ip, transport->master->port);
	return 0;
}








void* tnet_transport_context_create()
{
	return tsk_object_new(tnet_transport_context_def_t);
}


//=================================================================================================
//	Transport context object definition
//
static tsk_object_t* transport_context_ctor(tsk_object_t * self, va_list * app)
{
	transport_context_t *context = self;
	if(context){
		tsk_safeobj_init(context);
	}
	return self;
}

static tsk_object_t* transport_context_dtor(tsk_object_t * self)
{ 
	transport_context_t *context = self;
	if(context){
		while(context->count){
			removeSocket(0, context);
		}
		tsk_safeobj_deinit(context);
	}
	return self;
}

static const tsk_object_def_t tnet_transport_context_def_s = 
{
sizeof(transport_context_t),
transport_context_ctor, 
transport_context_dtor,
tsk_null, 
};
const tsk_object_def_t *tnet_transport_context_def_t = &tnet_transport_context_def_s;

#endif /* HAVE_POLL_H */