diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/common/evdp.h | 168 | ||||
| -rw-r--r-- | src/common/evdp_epoll.c | 233 | ||||
| -rw-r--r-- | src/common/netbuffer.c | 221 | ||||
| -rw-r--r-- | src/common/netbuffer.h | 83 | ||||
| -rw-r--r-- | src/common/network.c | 1062 | ||||
| -rw-r--r-- | src/common/network.h | 189 |
6 files changed, 1956 insertions, 0 deletions
diff --git a/src/common/evdp.h b/src/common/evdp.h new file mode 100644 index 000000000..bc3454686 --- /dev/null +++ b/src/common/evdp.h @@ -0,0 +1,168 @@ +#ifndef _rA_EVDP_H_ +#define _rA_EVDP_H_ + +#include "../common/cbasetypes.h" + +typedef struct EVDP_DATA EVDP_DATA; + + +//#idef EVDP_EPOLL +#include <sys/epoll.h> +struct EVDP_DATA{ + struct epoll_event ev_data; + bool ev_added; +}; +//#endif + + +enum EVDP_EVENTFLAGS{ + EVDP_EVENT_IN = 1, // Incomming data + EVDP_EVENT_OUT = 2, // Connection accepts writing. + EVDP_EVENT_HUP = 4 // Connection Closed. +}; + +typedef struct EVDP_EVENT{ + int32 events; // due to performance reasons, this should be the first member. + int32 fd; // Connection Identifier +} EVDP_EVENT; + + + +/** + * Network Event Dispatcher Initialization / Finalization routines + */ +void evdp_init(); +void evdp_final(); + + +/** + * Will Wait for events. + * + * @param *out_ev pointer to array in size at least of max_events. + * @param max_events max no of events to report with this call (coalesc) + * @param timeout_ticks max time to wait in ticks (milliseconds) + * + * @Note: + * The function will block until an event has occured on one of the monitored connections + * or the timeout of timeout_ticks has passed by. + * Upon successfull call (changed connections) this function will write the connection + * Identifier & event to the out_fds array. + * + * @return 0 -> Timeout, > 0 no of changed connections. + */ +int32 evdp_wait(EVDP_EVENT *out_fds, int32 max_events, int32 timeout_ticks); + + +/** + * Applys the given mask on the given connection. + * + * @param fd connection identifier + * @param *ep event data pointer for the connection + * @param mask new event mask we're monitoring for. + */ +//void evdp_apply(int32 fd, EVDP_DATA *ep, int32 mask); + + +/** + * Adds a connection (listner) to the event notification system. + * + * @param fd connection identifier + * @param *ep event data pointer for the connection + * + * @note: + * Listener type sockets are edge triggered, (see epoll manual for more information) + * - This basicaly means that youll receive one event, adn you have to accept until accept returns an error (nothing to accept) + * + * MONITORS by default: IN + * + * @return success indicator. + */ +bool evdp_addlistener(int32 fd, EVDP_DATA *ep); + +/** + * Adds a connection (client connectioN) to the event notification system + * + * @param fd connection identifier + * @param *ep event data pointr for the connection + * + * @note: + * + * MONITORS by default: IN, HUP + * + * @return success indicator. + */ +bool evdp_addclient(int32 fd, EVDP_DATA *ep); + +/** + * Adds a connection (pending / outgoing connection!) to the event notification system. + * + * @param fd connection identifier + * @param *ep event data pointer for the conneciton. + * + * @note: + * Outgoing connection type sockets are getting monitored for connection established + * successfull + * - if the connection has been established - we're generitng a writable notification .. (send) + * this is typical for BSD / posix conform network stacks. + * - Additinionally its edge triggered. + * + * @see evdp_outgoingconnection_established + * + * + * @return success indicator + */ +bool evdp_addconnecting(int32 fd, EVDP_DATA *ep); + +/** + * Adds an outgoing connection to the normal event notification system after it has been successfully established. + * + * @param fd connection identifier + * @param *ep event data pointer for the conneciton. + + * @note + * after this call, its handled like a normal "client" connection (incomming) + * + * @rturn success indicator + */ +bool evdp_outgoingconnection_established(int32 fd, EVDP_DATA *ep); + +/** + * Marks a connection to be monitored for writable. + * + * @param fd connection identifier + * @param *ep event data pointer for the connection + * + * @note: + * the connection must be already added (as client or listener) + * + * + * @return sucess indicator + */ +bool evdp_writable_add(int32 fd, EVDP_DATA *ep); + +/** + * Removes the connection from writable notification monitoring + * + * @param fd connection identifier + * @param *ep event data pointr for the connection + * + */ +void evdp_writable_remove(int32 fd, EVDP_DATA *ep); + +/** + * Removes an connectio from the event notification system. + * + * @param fd connection iditentfir + * @param *ep event data pointer for th connection + * + * + * @note: + * this will also clear the given EVENT_DATA block + * so the connection slot is in an "initial" blank status / ready to get reused. + * + */ +void evdp_remove(int32 fd, EVDP_DATA *ep); + + + +#endif diff --git a/src/common/evdp_epoll.c b/src/common/evdp_epoll.c new file mode 100644 index 000000000..d71c8d4d7 --- /dev/null +++ b/src/common/evdp_epoll.c @@ -0,0 +1,233 @@ +// +// Event Dispatcher Abstraction for EPOLL +// +// Author: Florian Wilkemeyer <fw@f-ws.de> +// +// Copyright (c) rAthena Project (www.rathena.org) - Licensed under GNU GPL +// For more information, see LICENCE in the main folder +// +// + +#include <stdio.h> +#include <errno.h> +#include <string.h> +#include <stdlib.h> +#include <unistd.h> + +#include <sys/epoll.h> +#include <sys/fcntl.h> +#include <sys/socket.h> + +#include "../common/cbasetypes.h" +#include "../common/showmsg.h" +#include "../common/evdp.h" + + +#define EPOLL_MAX_PER_CYCLE 10 // Max Events to coalesc. per cycle. + + +static int epoll_fd = -1; + + +void evdp_init(){ + + epoll_fd = epoll_create( EPOLL_MAX_PER_CYCLE ); + if(epoll_fd == -1){ + ShowFatalError("evdp [EPOLL]: Cannot create event dispatcher (errno: %u / %s)\n", errno, strerror(errno) ); + exit(1); + } + +}//end: evdp_init() + + +void evdp_final(){ + + if(epoll_fd != -1){ + close(epoll_fd); + epoll_fd = -1; + } + +}//end: evdp_final() + + +int32 evdp_wait(EVDP_EVENT *out_fds, int32 max_events, int32 timeout_ticks){ + struct epoll_event l_events[EPOLL_MAX_PER_CYCLE]; + register struct epoll_event *ev; + register int nfds, n; + + if(max_events > EPOLL_MAX_PER_CYCLE) + max_events = EPOLL_MAX_PER_CYCLE; + + nfds = epoll_wait( epoll_fd, l_events, max_events, timeout_ticks); + if(nfds == -1){ + // @TODO: check if core is in shutdown mode. if - ignroe error. + + ShowFatalError("evdp [EPOLL]: epoll_wait returned bad / unexpected status (errno: %u / %s)\n", errno, strerror(errno)); + exit(1); //.. + } + + // Loop thru all events and copy it to the local ra evdp_event.. struct. + for(n = 0; n < nfds; n++){ + ev = &l_events[n]; + + out_fds->fd = ev->data.fd; + out_fds->events = 0; // clear + + if(ev->events & EPOLLHUP) + out_fds->events |= EVDP_EVENT_HUP; + + if(ev->events & EPOLLIN) + out_fds->events |= EVDP_EVENT_IN; + + if(ev->events & EPOLLOUT) + out_fds->events |= EVDP_EVENT_OUT; + + out_fds++; + } + + return nfds; // 0 on timeout or > 0 .. +}//end: evdp_wait() + + +void evdp_remove(int32 fd, EVDP_DATA *ep){ + + if(ep->ev_added == true){ + + if( epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, &ep->ev_data) != 0){ + ShowError("evdp [EPOLL]: evdp_remove - epoll_ctl (EPOLL_CTL_DEL) failed! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + } + + ep->ev_data.events = 0; // clear struct. + ep->ev_data.data.fd = -1; // .. clear struct .. + + ep->ev_added = false; // not added! + } + + +}//end: evdp_remove() + + +bool evdp_addlistener(int32 fd, EVDP_DATA *ep){ + + ep->ev_data.events = EPOLLET|EPOLLIN; + ep->ev_data.data.fd = fd; + + // No check here for 'added ?' + // listeners cannot be added twice. + // + if( epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ep->ev_data) != 0 ){ + ShowError("evdp [EPOLL]: evdp_addlistener - epoll_ctl (EPOLL_CTL_ADD) faield! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + ep->ev_data.events = 0; + ep->ev_data.data.fd = -1; + return false; + } + + ep->ev_added = true; + + return true; +}//end: evdp_addlistener() + + +bool evdp_addclient(int32 fd, EVDP_DATA *ep){ + + ep->ev_data.events = EPOLLIN | EPOLLHUP; + ep->ev_data.data.fd = fd; + + // No check for "added?" here, + // this function only gets called upon accpept. + // + + if( epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ep->ev_data) != 0){ + ShowError("evdp [EPOLL]: evdp_addclient - epoll_ctl (EPOLL_CTL_ADD) failed! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + ep->ev_data.events = 0; + ep->ev_data.data.fd = -1; + return false; + } + + ep->ev_added = true; + + return true; +}//end: evdp_addclient() + + +bool evdp_addconnecting(int32 fd, EVDP_DATA *ep){ + + ep->ev_data.events = EPOLLET | EPOLLOUT | EPOLLHUP; + ep->ev_data.data.fd = fd; + + if( epoll_ctl(epoll_fd, EPOLL_CTL_ADD, fd, &ep->ev_data) != 0){ + ShowError("evdp [EPOLL]: evdp_addconnecting - epoll_ctl (EPOLL_CTL_ADD) failed! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + ep->ev_data.events = 0; + ep->ev_data.data.fd = -1; + } + + ep->ev_added = true; + + return true; +}//end: evdp_addconnecting() + + +bool evdp_outgoingconnection_established(int32 fd, EVDP_DATA *ep){ + int32 saved_mask; + + if(ep->ev_added != true){ + // ! + ShowError("evdp [EPOLL]: evdp_outgoingconnection_established fd #%u is not added to event dispatcher! invalid call.\n", fd); + return false; + } + + saved_mask = ep->ev_data.events; + + ep->ev_data.events = EPOLLIN | EPOLLHUP; + + if( epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ep->ev_data) != 0){ + ep->ev_data.events = saved_mask; // restore old mask. + ShowError("evdp [EPOLL]: evdp_outgoingconnection_established - epoll_ctl (EPOLL_CTL_MOD) failed! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + return false; + } + + return true; +}//end: evdp_outgoingconnection_established() + + +bool evdp_writable_add(int32 fd, EVDP_DATA *ep){ + + if(ep->ev_added != true){ + ShowError("evdp [EPOLL]: evdp_writable_add - tried to add not added fd #%u\n",fd); + return false; + } + + if(! (ep->ev_data.events & EPOLLOUT) ){ // + + ep->ev_data.events |= EPOLLOUT; + if( epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ep->ev_data) != 0 ){ + ShowError("evdp [EPOLL]: evdp_writable_add - epoll_ctl (EPOLL_CTL_MOD) failed! fd #%u (errno: %u / %s)\n", fd, errno, strerror(errno)); + ep->ev_data.events &= ~EPOLLOUT; // remove from local flagmask due to failed syscall. + return false; + } + } + + return true; +}//end: evdp_writable_add() + + +void evdp_writable_remove(int32 fd, EVDP_DATA *ep){ + + if(ep->ev_added != true){ + ShowError("evdp [EPOLL]: evdp_writable_remove - tried to remove not added fd #%u\n", fd); + return; + } + + if( ep->ev_data.events & EPOLLOUT ){ + + ep->ev_data.events &= ~EPOLLOUT; + if( epoll_ctl(epoll_fd, EPOLL_CTL_MOD, fd, &ep->ev_data) != 0){ + ShowError("evdp [EPOLL]: evdp_writable_remove - epoll_ctl (EPOLL_CTL_MOD) failed! fd #%u (errno %u / %s)\n", fd, errno, strerror(errno)); + ep->ev_data.events |= EPOLLOUT; // add back to local flagmask because of failed syscall. + return; + } + } + + return; +}//end: evdp_writable_remove() + diff --git a/src/common/netbuffer.c b/src/common/netbuffer.c new file mode 100644 index 000000000..57742d612 --- /dev/null +++ b/src/common/netbuffer.c @@ -0,0 +1,221 @@ + +// +// Network Buffer Subsystem (iobuffer) +// +// +// Author: Florian Wilkemeyer <fw@f-ws.de> +// +// Copyright (c) rAthena Project (www.rathena.org) - Licensed under GNU GPL +// For more information, see LICENCE in the main folder +// +// + +#include <stdlib.h> +#include <string.h> +#include <stdio.h> + +#include "../common/cbasetypes.h" +#include "../common/atomic.h" +#include "../common/mempool.h" +#include "../common/showmsg.h" +#include "../common/raconf.h" +#include "../common/thread.h" +#include "../common/malloc.h" +#include "../common/core.h" + +#include "../common/netbuffer.h" + + +// +// Buffers are available in the following sizes: +// 48, 192, 2048, 8192 +// 65536 (inter server connects may use it for charstatus struct..) +// + + +/// +// Implementation: +// +static volatile int32 l_nEmergencyAllocations = 0; // stats. +static sysint l_nPools = 0; +static sysint *l_poolElemSize = NULL; +static mempool *l_pool = NULL; + + +void netbuffer_init(){ + char localsection[32]; + raconf conf; + sysint i; + + // Initialize Statistic counters: + l_nEmergencyAllocations = 0; + + // Set localsection name according to running serverype. + switch(SERVER_TYPE){ + case ATHENA_SERVER_LOGIN: strcpy(localsection, "login-netbuffer"); break; + case ATHENA_SERVER_CHAR: strcpy(localsection, "char-netbuffer"); break; + case ATHENA_SERVER_INTER: strcpy(localsection, "inter-netbuffer"); break; + case ATHENA_SERVER_MAP: strcpy(localsection, "map-netbuffer"); break; + default: strcpy(localsection, "unsupported_type"); break; + } + + + conf = raconf_parse("conf/network.conf"); + if(conf == NULL){ + ShowFatalError("Failed to Parse required Configuration (conf/network.conf)"); + exit(EXIT_FAILURE); + } + + // Get Values from config file + l_nPools = (sysint)raconf_getintEx(conf, localsection, "netbuffer", "num", 0); + if(l_nPools == 0){ + ShowFatalError("Netbuffer (network.conf) failure - requires at least 1 Pool.\n"); + exit(EXIT_FAILURE); + } + + // Allocate arrays. + l_poolElemSize = (sysint*)aCalloc( l_nPools, sizeof(sysint) ); + l_pool = (mempool*)aCalloc( l_nPools, sizeof(mempool) ); + + + for(i = 0; i < l_nPools; i++){ + int64 num_prealloc, num_realloc; + char key[32]; + + sprintf(key, "pool_%u_size", (uint32)i+1); + l_poolElemSize[i] = (sysint)raconf_getintEx(conf, localsection, "netbuffer", key, 4096); + if(l_poolElemSize[i] < 32){ + ShowWarning("Netbuffer (network.conf) failure - minimum allowed buffer size is 32 byte) - fixed.\n"); + l_poolElemSize[i] = 32; + } + + sprintf(key, "pool_%u_prealloc", (uint32)i+1); + num_prealloc = raconf_getintEx(conf, localsection, "netbuffer", key, 150); + + sprintf(key, "pool_%u_realloc_step", (uint32)i+1); + num_realloc = raconf_getintEx(conf, localsection, "netbuffer", key, 100); + + // Create Pool! + sprintf(key, "Netbuffer %u", (uint32)l_poolElemSize[i]); // name. + + // Info + ShowInfo("NetBuffer: Creating Pool %u (Prealloc: %u, Realloc Step: %u) - %0.2f MiB\n", l_poolElemSize[i], num_prealloc, num_realloc, (float)((sizeof(struct netbuf) + l_poolElemSize[i] - 32)* num_prealloc)/1024.0f/1024.0f); + + // + // Size Calculation: + // struct netbuf + requested buffer size - 32 (because the struct already contains 32 byte buffer space at the end of struct) + l_pool[i] = mempool_create(key, (sizeof(struct netbuf) + l_poolElemSize[i] - 32), num_prealloc, num_realloc, NULL, NULL); + if(l_pool[i] == NULL){ + ShowFatalError("Netbuffer: cannot create Pool for %u byte buffers.\n", l_poolElemSize[i]); + // @leak: clean everything :D + exit(EXIT_FAILURE); + } + + }// + + + raconf_destroy(conf); + +}//end: netbuffer_init() + + +void netbuffer_final(){ + sysint i; + + if(l_nPools > 0){ + /// .. finalize mempools + for(i = 0; i < l_nPools; i++){ + mempool_stats stats = mempool_get_stats(l_pool[i]); + + ShowInfo("Netbuffer: Freeing Pool %u (Peak Usage: %u, Realloc Events: %u)\n", l_poolElemSize[i], stats.peak_nodes_used, stats.num_realloc_events); + + mempool_destroy(l_pool[i]); + } + + if(l_nEmergencyAllocations > 0){ + ShowWarning("Netbuffer: did %u Emergency Allocations, please tune your network.conf!\n", l_nEmergencyAllocations); + l_nEmergencyAllocations = 0; + } + + aFree(l_poolElemSize); l_poolElemSize = NULL; + aFree(l_pool); l_pool = NULL; + l_nPools = 0; + } + + +}//end: netbuffer_final() + + +netbuf netbuffer_get( sysint sz ){ + sysint i; + netbuf nb = NULL; + + // Search an appropriate pool + for(i = 0; i < l_nPools; i++){ + if(sz <= l_poolElemSize[i]){ + // match + + nb = (netbuf)mempool_node_get(l_pool[i]); + nb->pool = i; + + break; + } + } + + // No Bufferpool found that mets there quirements?.. (thats bad..) + if(nb == NULL){ + ShowWarning("Netbuffer: get(%u): => no appropriate pool found - emergency allocation required.\n", sz); + ShowWarning("Please reconfigure your network.conf!"); + + InterlockedIncrement(&l_nEmergencyAllocations); + + // .. better to check (netbuf struct provides 32 byte bufferspace itself. + if(sz < 32) sz = 32; + + // allocate memory using malloc .. + while(1){ + nb = (netbuf) aMalloc( (sizeof(struct netbuf) + sz - 32) ); + if(nb != NULL){ + memset(nb, 0x00, (sizeof(struct netbuf) + sz - 32) ); // zero memory! (to enforce commit @ os.) + nb->pool = -1; // emergency alloc. + break; + } + + rathread_yield(); + }// spin allocation. + + } + + + nb->refcnt = 1; // Initial refcount is 1 + + return nb; +}//end: netbuffer_get() + + +void netbuffer_put( netbuf nb ){ + + // Decrement reference counter, if > 0 do nothing :) + if( InterlockedDecrement(&nb->refcnt) > 0 ) + return; + + // Is this buffer an emergency allocated buffer? + if(nb->pool == -1){ + aFree(nb); + return; + } + + + // Otherwise its a normal mempool based buffer + // return it to the according mempool: + mempool_node_put( l_pool[nb->pool], nb); + + +}//end: netbuffer_put() + + +void netbuffer_incref( netbuf nb ){ + + InterlockedIncrement(&nb->refcnt); + +}//end: netbuf_incref() diff --git a/src/common/netbuffer.h b/src/common/netbuffer.h new file mode 100644 index 000000000..844241226 --- /dev/null +++ b/src/common/netbuffer.h @@ -0,0 +1,83 @@ +// Copyright (c) rAthena Project (www.rathena.org) - Licensed under GNU GPL +// For more information, see LICENCE in the main folder + +#ifndef _rA_NETBUFFER_H_ +#define _rA_NETBUFFER_H_ + +#include "../common/cbasetypes.h" + +typedef struct netbuf{ + sysint pool; // The pool ID this buffer belongs to, + // is set to -1 if its an emergency allocated buffer + + struct netbuf *next; // Used by Network system. + + volatile int32 refcnt; // Internal Refcount, it gets lowered every call to netbuffer_put, + // if its getting zero, the buffer will returned back to the pool + // and can be reused. + + int32 dataPos; // Current Offset + // Used only for Reading (recv job) + // write cases are using the sessions local datapos member due to + // shared write buffer support. + + int32 dataLen; // read buffer case: + // The length expected to read to. + // when this->dataPos == dateLen, read job has been completed. + // write buffer case: + // The lngth of data in te buffer + // when s->dataPos == dataLen, write job has been completed + // + // Note: + // leftBytes = (dateLen - dataPos) + // + // Due to shared buffer support + // dataPos gets not used in write case (each connection has its local offset) + // + + // The Bufferspace itself. + char buf[32]; +} *netbuf; + + +void netbuffer_init(); +void netbuffer_final(); + +/** + * Gets a netbuffer that has atleast (sz) byes space. + * + * @note: The netbuffer system guarantees that youll always recevie a buffer. + * no check for null is required! + * + * @param sz - minimum size needed. + * + * @return pointer to netbuf struct + */ +netbuf netbuffer_get( sysint sz ); + + +/** + * Returns the given netbuffer (decreases refcount, if its 0 - the buffer will get returned to the pool) + * + * @param buf - the buffer to return + */ +void netbuffer_put( netbuf buf ); + + +/** + * Increases the Refcount on the given buffer + * (used for areasends .. etc) + * + */ +void netbuffer_incref( netbuf buf ); + + +// Some Useful macros +#define NBUFP(netbuf,pos) (((uint8*)(netbuf->buf)) + (pos)) +#define NBUFB(netbuf,pos) (*(uint8*)((netbuf->buf) + (pos))) +#define NBUFW(netbuf,pos) (*(uint16*)((netbuf->buf) + (pos))) +#define NBUFL(netbuf,pos) (*(uint32*)((netbuf->buf) + (pos))) + + + +#endif diff --git a/src/common/network.c b/src/common/network.c new file mode 100644 index 000000000..2574f3cc1 --- /dev/null +++ b/src/common/network.c @@ -0,0 +1,1062 @@ + // +// Network Subsystem (previously known as socket system) +// +// Author: Florian Wilkemeyer <fw@f-ws.de> +// +// Copyright (c) rAthena Project (www.rathena.org) - Licensed under GNU GPL +// For more information, see LICENCE in the main folder +// +// +//#ifdef HAVE_ACCETP4 +#define _GNU_SOURCE +//#endif + +#include <stdio.h> +#include <string.h> +#include <unistd.h> +#include <errno.h> + +#include <sys/types.h> +#include <sys/fcntl.h> +#include <sys/socket.h> +#include <netinet/in.h> +#include <netinet/tcp.h> +#include <arpa/inet.h> + + +#include "../common/cbasetypes.h" +#include "../common/showmsg.h" +#include "../common/timer.h" +#include "../common/evdp.h" +#include "../common/netbuffer.h" + +#include "../common/network.h" + +#define ENABLE_IPV6 +#define HAVE_ACCEPT4 +#define EVENTS_PER_CYCLE 10 +#define PARANOID_CHECKS + +// Local Vars (settings..) +static int l_ListenBacklog = 64; + +// +// Global Session Array (previously exported as session[] +// +SESSION g_Session[MAXCONN]; + + +// +static bool onSend(int32 fd); + + +#define _network_free_netbuf_async( buf ) add_timer( 0, _network_async_free_netbuf_proc, 0, (intptr_t) buf) +static int _network_async_free_netbuf_proc(int tid, unsigned int tick, int id, intptr_t data){ + // netbuf is in data + netbuffer_put( (netbuf)data ); + + return 0; +}//end: _network_async_free_netbuf_proc() + + + +void network_init(){ + SESSION *s; + int32 i; + + memset(g_Session, 0x00, (sizeof(SESSION) * MAXCONN) ); + + for(i = 0; i < MAXCONN; i++){ + s = &g_Session[i]; + + s->type = NST_FREE; + s->disconnect_in_progress = false; + + } + + // Initialize the correspondig event dispatcher + evdp_init(); + + // + add_timer_func_list(_network_async_free_netbuf_proc, "_network_async_free_netbuf_proc"); + +}//end: network_init() + + +void network_final(){ + + // @TODO: + // .. disconnect and cleanup everything! + + evdp_final(); + +}//end: network_final() + + +void network_do(){ + struct EVDP_EVENT l_events[EVENTS_PER_CYCLE]; + register struct EVDP_EVENT *ev; + register int n, nfds; + register SESSION *s; + + nfds = evdp_wait( l_events, EVENTS_PER_CYCLE, 1000); // @TODO: timer_getnext() + + for(n = 0; n < nfds; n++){ + ev = &l_events[n]; + s = &g_Session[ ev->fd ]; + + if(ev->events & EVDP_EVENT_HUP){ + network_disconnect( ev->fd ); + continue; // no further event processing. + }// endif vent is HUP (disconnect) + + + if(ev->events & EVDP_EVENT_IN){ + + if(s->onRecv != NULL){ + if( false == s->onRecv(ev->fd) ){ + network_disconnect(ev->fd); + continue; // .. + } + }else{ + ShowError("network_do: fd #%u has no onRecv proc set. - disconnecting\n", ev->fd); + network_disconnect(ev->fd); + continue; + } + + }// endif event is IN (recv) + + + if(ev->events & EVDP_EVENT_OUT){ + if(s->onSend != NULL){ + if( false == s->onSend(ev->fd) ){ + network_disconnect(ev->fd); + continue; + } + }else{ + ShowError("network_do: fd #%u has no onSend proc set. - disconnecting\n", ev->fd); + network_disconnect(ev->fd); + continue; + } + }// endif event is OUT (send) + + }//endfor + +}//end: network_do() + + +static bool _setnonblock(int32 fd){ + int flags = fcntl(fd, F_GETFL, 0); + if(fcntl(fd, F_SETFL, flags | O_NONBLOCK) != 0) + return false; + + return true; +}//end: _setnonblock() + + +static bool _network_accept(int32 fd){ + SESSION *listener = &g_Session[fd]; + SESSION *s; + union{ + struct sockaddr_in v4; +#ifdef ENABLE_IPV6 + struct sockaddr_in6 v6; +#endif + } _addr; + int newfd; + socklen_t addrlen; + struct sockaddr *addr; + + // Accept until OS returns - nothing to accept anymore + // - this is required due to our EVDP abstraction. (which handles on listening sockets similar to epoll's EPOLLET flag.) + while(1){ +#ifdef ENABLE_IPV6 + if(listener->v6 == true){ + addrlen = sizeof(_addr.v6); + addr = (struct sockaddr*)&_addr.v6; + }else{ +#endif + addrlen = sizeof(_addr.v4); + addr = (struct sockaddr*)&_addr.v4; +#ifdef ENABLE_IPV6 + } +#endif + +#ifdef HAVE_ACCEPT4 + newfd = accept4(fd, addr, &addrlen, SOCK_NONBLOCK); +#else + newfd = accept(fd, addr, &addrlen); +#endif + + if(newfd == -1){ + if(errno == EAGAIN || errno == EWOULDBLOCK) + break; // this is fully valid & whished., se explaination on top of while(1) + + // Otherwis .. we have serious problems :( seems tahat our listner has gone away.. + // @TODO handle this .. + ShowError("_network_accept: accept() returned error. closing listener. (errno: %u / %s)\n", errno, strerror(errno)); + + return false; // will call disconnect after return. + //break; + } + +#ifndef HAVE_ACCEPT4 // no accept4 means, we have to set nonblock by ourself. .. + if(_setnonblock(newfd) == false){ + ShowError("_network_accept: failed to set newly accepted connection nonblocking (errno: %u / %s). - disconnecting.\n", errno, strerror(errno)); + close(newfd); + continue; + } +#endif + + // Check connection limits. + if(newfd >= MAXCONN){ + ShowError("_network_accept: failed to accept connection - MAXCONN (%u) exceeded.\n", MAXCONN); + close(newfd); + continue; // we have to loop over the events (and disconnect them too ..) but otherwise we would leak event notifications. + } + + + // Create new Session. + s = &g_Session[newfd]; + s->type = NST_CLIENT; + + // The new connection inherits listenr's handlers. + s->onDisconnect = listener->onDisconnect; + s->onConnect = listener->onConnect; // maybe useless but .. fear the future .. :~ + + // Register the new connection @ EVDP + if( evdp_addclient(newfd, &s->evdp_data) == false){ + ShowError("_network_accept: failed to accept connection - event subsystem returned an error.\n"); + close(newfd); + s->type = NST_FREE; + } + + // Call the onConnect handler on the listener. + if( listener->onConnect(newfd) == false ){ + // Resfused by onConnect handler.. + evdp_remove(newfd, &s->evdp_data); + + close(newfd); + s->type = NST_FREE; + + s->data = NULL; // be on the safe side ~ ! + continue; + } + + + } + + return true; +}//end: _network_accept() + + +void network_disconnect(int32 fd){ + SESSION *s = &g_Session[fd]; + netbuf b, bn; + + // Prevent recursive calls + // by wrong implemented on disconnect handlers.. and such.. + if(s->disconnect_in_progress == true) + return; + + s->disconnect_in_progress = true; + + + // Disconnect Todo: + // - Call onDisconnect Handler + // - Release all Assigned buffers. + // - remove from event system (notifications) + // - cleanup session structure + // - close connection. + // + + if(s->onDisconnect != NULL && + s->type != NST_LISTENER){ + + s->onDisconnect( fd ); + } + + // Read Buffer + if(s->read.buf != NULL){ + netbuffer_put(s->read.buf); + s->read.buf = NULL; + } + + // Write Buffer(s) + b = s->write.buf; + while(1){ + if(b == NULL) break; + + bn = b->next; + + netbuffer_put(b); + + b = bn; + } + s->write.buf = NULL; + s->write.buf_last = NULL; + + s->write.n_outstanding = 0; + s->write.max_outstanding = 0; + + + // Remove from event system. + evdp_remove(fd, &s->evdp_data); + + // Cleanup Session Structure. + s->type = NST_FREE; + s->data = NULL; // no application level data assigned + s->disconnect_in_progress = false; + + + // Close connection + close(fd); + +}//end: network_disconnect() + + +int32 network_addlistener(bool v6, const char *addr, uint16 port){ + SESSION *s; + int optval, fd; + +#if !defined(ENABLE_IPV6) + if(v6 == true){ + ShowError("network_addlistener(%c, '%s', %u): this release has no IPV6 support.\n", (v6==true?'t':'f'), addr, port); + return -1; + } +#endif + + +#ifdef ENABLE_IPV6 + if(v6 == true) + fd = socket(AF_INET6, SOCK_STREAM, 0); + else +#endif + fd = socket(AF_INET, SOCK_STREAM, 0); + + // Error? + if(fd == -1){ + ShowError("network_addlistener(%c, '%s', %u): socket() failed (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + return -1; + } + + // Too many connections? + if(fd >= MAXCONN){ + ShowError("network_addlistener(%c, '%s', %u): cannot create listener, exceeds more than supported connections (%u).\n", (v6==true?'t':'f'), addr, port, MAXCONN); + close(fd); + return -1; + } + + + s = &g_Session[fd]; + if(s->type != NST_FREE){ // additional checks.. :) + ShowError("network_addlistener(%c, '%s', %u): failed, got fd #%u which is already in use in local session table?!\n", (v6==true?'t':'f'), addr, port, fd); + close(fd); + return -1; + } + + + // Fill ip addr structs +#ifdef ENABLE_IPV6 + if(v6 == true){ + memset(&s->addr.v6, 0x00, sizeof(s->addr.v6)); + s->addr.v6.sin6_family = AF_INET6; + s->addr.v6.sin6_port = htons(port); + if(inet_pton(AF_INET6, addr, &s->addr.v6.sin6_addr) != 1){ + ShowError("network_addlistener(%c, '%s', %u): failed to parse the given IPV6 address.\n", (v6==true?'t':'f'), addr, port); + close(fd); + return -1; + } + + }else{ +#endif + memset(&s->addr.v4, 0x00, sizeof(s->addr.v4)); + s->addr.v4.sin_family = AF_INET; + s->addr.v4.sin_port = htons(port); + s->addr.v4.sin_addr.s_addr = inet_addr(addr); +#ifdef ENABLE_IPV6 + } +#endif + + + // if OS has support for SO_REUSEADDR, apply the flag + // so the address could be used when there're still time_wait sockets outstanding from previous application run. +#ifdef SO_REUSEADDR + optval=1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)); +#endif + + // Bind +#ifdef ENABLE_IPV6 + if(v6 == true){ + if( bind(fd, (struct sockaddr*)&s->addr.v6, sizeof(s->addr.v6)) == -1) { + ShowError("network_addlistener(%c, '%s', %u): bind failed (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + }else{ +#endif + if( bind(fd, (struct sockaddr*)&s->addr.v4, sizeof(s->addr.v4)) == -1) { + ShowError("network_addlistener(%c, '%s', %u): bind failed (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } +#ifdef ENABLE_IPV6 + } +#endif + + if( listen(fd, l_ListenBacklog) == -1){ + ShowError("network_addlistener(%c, '%s', %u): listen failed (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + + + // Set to nonblock! + if(_setnonblock(fd) == false){ + ShowError("network_addlistener(%c, '%s', %u): cannot set to nonblock (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + + + // Rgister @ evdp. + if( evdp_addlistener(fd, &s->evdp_data) != true){ + ShowError("network_addlistener(%c, '%s', %u): eventdispatcher subsystem returned an error.\n", (v6==true?'t':'f'), addr, port); + close(fd); + return -1; + } + + + // Apply flags on Session array for this conneciton. + if(v6 == true) s->v6 = true; + else s->v6 = false; + + s->type = NST_LISTENER; + s->onRecv = _network_accept; + + ShowStatus("Added Listener on '%s':%u\n", addr, port, (v6==true ? "(ipv6)":"(ipv4)") ); + + return fd; +}//end: network_addlistener() + + +static bool _network_connect_establishedHandler(int32 fd){ + register SESSION *s = &g_Session[fd]; + int val; + socklen_t val_len; + + if(s->type == NST_FREE) + return true; // due to multiple non coalesced event notifications + // this can happen .. when a previous handled event has already disconnected the connection + // within the same cycle.. + + val = -1; + val_len = sizeof(val); + getsockopt(fd, SOL_SOCKET, SO_ERROR, &val, &val_len); + + if(val != 0){ + // :( .. cleanup session.. + s->type = NST_FREE; + s->onSend = NULL; + s->onConnect = NULL; + s->onDisconnect = NULL; + + evdp_remove(fd, &s->evdp_data); + close(fd); + + return true; // we CANT return false, + // becuase the normal disconnect procedure would execute the ondisconnect handler, which we dont want .. in this case. + }else{ + // ok + if(s->onConnect(fd) == false) { + // onConnect handler has refused the connection .. + // cleanup .. and ok + s->type = NST_FREE; + s->onSend = NULL; + s->onConnect = NULL; + s->onDisconnect = NULL; + + evdp_remove(fd, &s->evdp_data); + close(fd); + + return true; // we dnot want the ondisconnect handler to be executed, so its okay to handle this by ourself. + } + + // connection established ! + // + if( evdp_outgoingconnection_established(fd, &s->evdp_data) == false ){ + return false; // we want the normal disconnect procedure.. with call to ondisconnect handler. + } + + s->onSend = NULL; + + ShowStatus("#%u connection successfull!\n", fd); + } + + return true; +}//end: _network_connect_establishedHandler() + + +int32 network_connect(bool v6, + const char *addr, + uint16 port, + const char *from_addr, + uint16 from_port, + bool (*onConnectionEstablishedHandler)(int32 fd), + void (*onConnectionLooseHandler)(int32 fd) +){ + register SESSION *s; + int32 fd, optval, ret; + struct sockaddr_in ip4; +#ifdef ENABLE_IPV6 + struct sockaddr_in6 ip6; +#endif + +#ifdef ENABLE_IPV6 + if(v6 == true) + fd = socket(AF_INET6, SOCK_STREAM, 0); + else +#endif + fd = socket(AF_INET, SOCK_STREAM, 0); + +#ifndef ENABLE_IPV6 + // check.. + if(v6 == true){ + ShowError("network_connect(%c, '%s', %u...): tried to create an ipv6 connection, IPV6 is not supported in this release.\n", (v6==true?'t':'f'), addr, port); + return -1; + } +#endif + + // check connection limits. + if(fd >= MAXCONN){ + ShowError("network_connect(%c, '%s', %u...): cannot create new connection, exceeeds more than supported connections (%u)\n", (v6==true?'t':'f'), addr, port ); + close(fd); + return -1; + } + + + // Originating IP/Port pair given ? + if(from_addr != NULL && *from_addr != 0){ + //.. + #ifdef SO_REUSEADDR + optval=1; + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)); + #endif + + #ifdef ENABLE_IPV6 + if(v6 == true){ + memset(&ip6, 0x00, sizeof(ip6)); + ip6.sin6_family = AF_INET6; + ip6.sin6_port = htons(from_port); + + if(inet_pton(AF_INET6, from_addr, &ip6.sin6_addr) != 1){ + ShowError("network_connect(%c, '%s', %u...): cannot parse originating (from) IPV6 address (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + + ret = bind(fd, (struct sockaddr*)&ip6, sizeof(ip6)); + }else{ + #endif + memset(&ip4, 0x00, sizeof(ip4)); + + ip4.sin_family = AF_INET; + ip4.sin_port = htons(from_port); + ip4.sin_addr.s_addr = inet_addr(from_addr); + ret = bind(fd, (struct sockaddr*)&ip4, sizeof(ip4)); + #ifdef ENABLE_IPV6 + } + #endif + + } + + + // Set non block + if(_setnonblock(fd) == false){ + ShowError("network_connect(%c, '%s', %u...): cannot set socket to nonblocking (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + + + // Create ip addr block to connect to .. +#ifdef ENABLE_IPV6 + if(v6 == true){ + memset(&ip6, 0x00, sizeof(ip6)); + ip6.sin6_family = AF_INET6; + ip6.sin6_port = htons(port); + + if(inet_pton(AF_INET6, addr, &ip6.sin6_addr) != 1){ + ShowError("network_connect(%c, '%s', %u...): cannot parse destination IPV6 address (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + close(fd); + return -1; + } + + }else{ +#endif + memset(&ip4, 0x00, sizeof(ip4)); + + ip4.sin_family = AF_INET; + ip4.sin_port = htons(port); + ip4.sin_addr.s_addr = inet_addr(addr); +#ifdef ENABLE_IPV6 + } +#endif + + + // Assign Session.. + s = &g_Session[fd]; + s->type = NST_OUTGOING; + s->v6 = v6; + s->onConnect = onConnectionEstablishedHandler; + s->onDisconnect = onConnectionLooseHandler; + s->onRecv = NULL; + s->onSend = _network_connect_establishedHandler; +#ifdef ENABLE_IPV6 + if(v6 == true) + memcpy(&s->addr.v6, &ip6, sizeof(ip6)); + else +#endif + memcpy(&s->addr.v4, &ip4, sizeof(ip4)); + + // Register @ EVDP. as outgoing (see doc of the function) + if(evdp_addconnecting(fd, &s->evdp_data) == false){ + ShowError("network_connect(%c, '%s', %u...): eventdispatcher subsystem returned an error.\n", (v6==true?'t':'f'), addr, port); + + // cleanup session x.x.. + s->type = NST_FREE; + s->onConnect = NULL; + s->onDisconnect = NULL; + s->onSend = NULL; + + // close, return error code. + close(fd); + return -1; + } + + +#ifdef ENABLE_IPV6 + if(v6 == true) + ret = connect(fd, (struct sockaddr*)&ip6, sizeof(ip6)); + else +#endif + ret = connect(fd, (struct sockaddr*)&ip4, sizeof(ip4)); + + + // + if(ret != 0 && errno != EINPROGRESS){ + ShowWarning("network_connect(%c, '%s', %u...): connection failed (errno: %u / %s)\n", (v6==true?'t':'f'), addr, port, errno, strerror(errno)); + + // Cleanup session .. + s->type = NST_FREE; + s->onConnect = NULL; + s->onDisconnect = NULL; + s->onSend = NULL; + + // .. remove from evdp and close fd. + evdp_remove(fd, &s->evdp_data); + close(fd); + return -1; + } + + + // ! The Info Message :~D + ShowStatus("network_connect fd#%u (%s:%u) in progress.. \n", fd, addr, port); + +return fd; +}//end: network_connect() + + +static bool _onSend(int32 fd){ + register SESSION *s = &g_Session[fd]; + register netbuf buf, buf_next; + register uint32 szNeeded; + register int wLen; + + if(s->type == NST_FREE) + return true; // Possible due to multipl non coalsced event notifications + // so onSend gets called after disconnect caused by an previous vent. + // we can ignore the call to onSend, then. + + buf = s->write.buf; + while(1){ + if(buf == NULL) + break; + + buf_next = buf->next; + + + szNeeded = (buf->dataLen - s->write.dataPos); // using th session-local .dataPos member, due to shared write buffer support. + + // try to write. + wLen = write(fd, &buf->buf[s->write.dataPos], szNeeded); + if(wLen == 0){ + return false; // eof. + }else if(wLen == -1){ + if(errno == EAGAIN || errno == EWOULDBLOCK) + return true; // dont disconnect / try again later. + + // all other errors. . + return false; + } + + // Wrote data.. => + szNeeded -= wLen; + if(szNeeded > 0){ + // still data left .. + // + s->write.dataPos += wLen; // fix offset. + return true; + }else{ + // this buffer has been written successfully + // could be returned to pool. + netbuffer_put(buf); + s->write.n_outstanding--; // When threadsafe -> Interlocked here. + s->write.dataPos = 0; + } + + + buf = buf_next; + } + + // okay, + // reaching this part means: + // while interrupted by break - + // which means all buffers are written, nothing left + // + + s->write.buf_last = NULL; + s->write.buf = NULL; + s->write.n_outstanding = 0; + s->write.dataPos = 0; + + // Remove from event dispatcher (write notification) + // + evdp_writable_remove(fd, &s->evdp_data); + + return true; +}//end: _onSend() + + +static bool _onRORecv(int32 fd){ + register SESSION *s = &g_Session[fd]; + register uint32 szNeeded; + register char *p; + register int rLen; + + if(s->type == NST_FREE) + return true; // Possible due to multiple non coalesced events by evdp. + // simply ignore this call returning positive result. + + // Initialize p and szNeeded depending on change + // + switch(s->read.state){ + case NRS_WAITOP: + szNeeded = s->read.head_left; + p = ((char*)&s->read.head[0]) + (2-szNeeded); + break; + + case NRS_WAITLEN: + szNeeded = s->read.head_left; + p = ((char*)&s->read.head[1]) + (2-szNeeded); + break; + + case NRS_WAITDATA:{ + register netbuf buf = s->read.buf; + + szNeeded = (buf->dataLen - buf->dataPos); + p = (char*)&buf->buf[ buf->dataPos ]; + } + break; + + default: + // .. the impossible gets possible .. + ShowError("_onRORecv: fd #%u has unknown read.state (%d) - disconnecting\n", fd, s->read.state); + return false; + break; + } + + + // + + rLen = read(fd, p, szNeeded); + if(rLen == 0){ + // eof.. + return false; + }else if(rLen == -1){ + + if(errno == EAGAIN || errno == EWOULDBLOCK){ + // try again later .. (this case shouldnt happen, because we're event trigered.. but .. sometimes it happens :) + return true; + } + + // an additional interesting case would be + // EINTR, this 'could' be handled .. but: + // posix says that its possible that data gets currupted during irq + // or data gor read and not reported.., so we'd have a data loss.. + // (which shouldnt happen with stream based protocols such as tcp) + // its better to disonnect the client in that case. + + return false; + } + + // + // Got Data: + // next action also depends on current state .. + // + szNeeded -= rLen; + switch(s->read.state){ + case NRS_WAITOP: + + if(szNeeded > 0){ + // still data missing .. + s->read.head_left = szNeeded; + return true; // wait for completion. + }else{ + // complete .. + // next state depends on packet type. + + s->read.head[1] = ((uint16*)s->netparser_data)[ s->read.head[0] ]; // store lenght of packet by opcode head[0] to head[1] + + if(s->read.head[1] == ROPACKET_UNKNOWN){ + // unknown packet - disconnect + ShowWarning("_onRORecv: fd #%u got unlnown packet 0x%04x - disconnecting.\n", fd, s->read.head[0]); + return false; + } + else if(s->read.head[1] == ROPACKET_DYNLEN){ + // dynamic length + // next state: requrie len. + s->read.state = NRS_WAITLEN; + s->read.head_left = 2; + return true; // + } + else if(s->read.head[1] == 2){ + // packet has no data (only opcode) + register netbuf buf = netbuffer_get(2); // :D whoohoo its giant! + + NBUFW(buf, 0) = s->read.head[0]; // store opcode @ packet begin. + buf->dataPos = 2; + buf->dataLen = 2; + buf->next = NULL; + + // Back to initial state -> Need opcode. + s->read.state = NRS_WAITOP; + s->read.head_left = 2; + s->read.buf = NULL; + + // Call completion routine here. + s->onPacketComplete(fd, s->read.head[0], 2, buf); + + return true; // done :) + } + else{ + // paket needs .. data .. + register netbuf buf = netbuffer_get( s->read.head[1] ); + + NBUFW(buf, 0) = s->read.head[0]; // store opcode @ packet begin. + buf->dataPos = 2; + buf->dataLen = s->read.head[1]; + buf->next = NULL; + + // attach buffer. + s->read.buf = buf; + + // set state: + s->read.state = NRS_WAITDATA; + + return true; + } + + }//endif: szNeeded > 0 (opcode read completed?) + + break; + + + case NRS_WAITLEN: + + if(szNeeded > 0){ + // incomplete .. + s->read.head_left = szNeeded; + return true; + }else{ + + if(s->read.head[1] == 4){ + // packet has no data (only opcode + length) + register netbuf buf = netbuffer_get( 4 ); + + NBUFL(buf, 0) = *((uint32*)&s->read.head[0]); // copy Opcode + length to netbuffer using MOVL + buf->dataPos = 4; + buf->dataLen = 4; + buf->next = NULL; + + // set initial state (need opcode) + s->read.state = NRS_WAITOP; + s->read.head_left = 2; + s->read.buf = NULL; + + // call completion routine. + s->onPacketComplete(fd, s->read.head[0], 4, buf); + + return true; + } + else if(s->read.head[1] < 4){ + // invalid header. + ShowWarning("_onRORecv: fd #%u invalid header - got packet 0x%04x, reported length < 4 - INVALID - disconnecting\n", fd, s->read.head[0]); + return false; + } + else{ + // Data needed + // next state -> waitdata! + register netbuf buf = netbuffer_get( s->read.head[1] ); + + NBUFL(buf, 0) = *((uint32*)&s->read.head[0]); // copy Opcode + length to netbuffer using MOVL + buf->dataPos = 4; + buf->dataLen = s->read.head[1]; + buf->next = NULL; + + // attach to session: + s->read.buf = buf; + s->read.state = NRS_WAITDATA; + + return true; + } + + }//endif: szNeeded > 0 (length read complete?) + + break; + + + case NRS_WAITDATA: + + if(szNeeded == 0){ + // Packet finished! + // compltion. + register netbuf buf = s->read.buf; + + // set initial state. + s->read.state = NRS_WAITOP; + s->read.head_left = 2; + s->read.buf = NULL; + + // Call completion routine. + s->onPacketComplete(fd, NBUFW(buf, 0), buf->dataLen, buf); + + return true; + }else{ + // still data needed + s->read.buf->dataPos += rLen; + + return true; + } + break; + + + // + default: + ShowError("_onRORecv: fd #%u has unknown read.state (%d) [2] - disconnecting\n", fd, s->read.state); + return false; + break; + } + + + return false; +}//end: _onRORecv() + + +void network_send(int32 fd, netbuf buf){ + register SESSION *s = &g_Session[fd]; + +#ifdef PARANOID_CHECKS + if(fd >= MAXCONN){ + ShowError("network_send: tried to attach buffer to connection idientifer #%u which is out of bounds.\n", fd); + _network_free_netbuf_async(buf); + return; + } +#endif + + + if(s->type == NST_FREE) + return; + + // Check Max Outstanding buffers limit. + if( (s->write.max_outstanding > 0) && + (s->write.n_outstanding >= s->write.max_outstanding) ){ + + ShowWarning("network_send: fd #%u max Outstanding buffers exceeded. - disconnecting.\n", fd); + network_disconnect(fd); + // + _network_free_netbuf_async(buf); + return; + } + + + // Attach to the end: + buf->next = NULL; + if(s->write.buf_last != NULL){ + s->write.buf_last->next = buf; + s->write.buf_last = buf; + + }else{ + // currently no buffer attached. + s->write.buf = s->write.buf_last = buf; + + // register @ evdp for writable notification. + evdp_writable_add(fd, &s->evdp_data); // + } + + + // + s->write.n_outstanding++; + +}//end: network_send() + + +void network_parser_set_ro(int32 fd, + int16 *packetlentable, + void (*onPacketCompleteProc)(int32 fd, uint16 op, uint16 len, netbuf buf) + ){ + register SESSION *s = &g_Session[fd]; + register netbuf b, nb; // used for potential free attached buffers. + + if(s->type == NST_FREE) + return; + + s->onPacketComplete = onPacketCompleteProc; + + s->onRecv = _onRORecv; // .. + s->onSend = _onSend; // Using the normal generic netbuf based send function. + + s->netparser_data = packetlentable; + + // Initial State -> Need Packet OPCode. + s->read.state = NRS_WAITOP; + s->read.head_left = 2; + + + // Detach (if..) all buffers. + if(s->read.buf != NULL){ + _network_free_netbuf_async(s->read.buf); // + s->read.buf = NULL; + } + + if(s->write.buf != NULL){ + b = s->write.buf; + while(1){ + nb = b->next; + + _network_free_netbuf_async(b); + + b = nb; + } + + s->write.buf = NULL; + s->write.buf_last = NULL; + s->write.n_outstanding = 0; + } + + // not changing any limits on outstanding .. + // + +}//end: network_parser_set_ro() + diff --git a/src/common/network.h b/src/common/network.h new file mode 100644 index 000000000..d7b463a2f --- /dev/null +++ b/src/common/network.h @@ -0,0 +1,189 @@ +#ifndef _rA_NETWORK_H_ +#define _rA_NETWORK_H_ + +#include <netinet/in.h> +#include "../common/cbasetypes.h" +#include "../common/netbuffer.h" +#include "../common/evdp.h" + +#ifndef MAXCONN +#define MAXCONN 16384 +#endif + + +typedef struct SESSION{ + EVDP_DATA evdp_data; // Must be always the frist member! (some evdp's may rely on this fact) + + // Connection Type + enum{ NST_FREE=0, NST_LISTENER = 1, NST_CLIENT=2, NST_OUTGOING=3} type; + + // Flags / Settings. + bool v6; // is v6? + bool disconnect_in_progress; // To prevent stack overflows / recursive calls. + + + union{ // union to save memory. + struct sockaddr_in v4; + struct sockaddr_in6 v6; + }addr; + + + // "lowlevel" Handlers + // (Implemented by the protocol specific parser) + // + bool (*onRecv)(int32 fd); // return false = disconnect + bool (*onSend)(int32 fd); // return false = disconnect + + // Event Handlers for LISTENER type sockets + // + // onConnect gets Called when a connection has been + // successfully accepted. + // Session entry is available in this Handler! + // A returncode of false will reejct the connection (disconnect) + // Note: When rejecting a connection in onConnect by returning false + // The onDisconnect handler wont get called! + // Note: the onConnect Handler is also responsible for setting + // the appropriate netparser (which implements onRecv/onSend..) [protocol specific] + // + // onDisconnect gets called when a connection gets disconnected + // (by peer as well as by core) + // + bool (*onConnect)(int32 fd); // return false = disconnect (wont accept) + void (*onDisconnect)(int32 fd); + + + // + // Parser specific data + // + void *netparser_data; // incase of RO Packet Parser, pointer to packet len table (uint16array) + void (*onPacketComplete)(int32 fd, uint16 op, uint16 len, netbuf buf); + + + // + // Buffers + // + struct{ + enum NETREADSTATE { NRS_WAITOP = 0, NRS_WAITLEN = 1, NRS_WAITDATA = 2} state; + + uint32 head_left; + uint16 head[2]; + + netbuf buf; + } read; + + struct{ + uint32 max_outstanding; + uint32 n_outstanding; + + uint32 dataPos; + + netbuf buf, buf_last; + } write; + + // Application Level data Pointer + // (required for backward compatibility with previous athena socket system.) + void *data; + +} SESSION; + + +/** + * Subsystem Initialization / Finalization. + * + */ +void network_init(); +void network_final(); + + +/** + * Will do the net work :) .. + */ +void network_do(); + + +/** + * Adds a new listner. + * + * @param v6 v6 listner? + * @param *addr the address to listen on. + * @param port port to listen on + * + * @return -1 on error otherwise the identifier of the new listener. + */ +int32 network_addlistener(bool v6, const char *addr, uint16 port); + + +/** + * Tries to establish an outgoing connection. + * + * @param v6 operate with IPv6 addresses? + * @param addr the address to connect to + * @param port the port to connect to + * @param from_addr the address to connect from (local source / optional if auto -> NULL) + * @param from_port the port to connect from (local source / optional if auto -> 0) + * @param onConnectionEstablishedHandler callback that gets called when the connection is established. + * @param onConnectionLooseHandler callback that gets called when the connection gets disconnected (or the connection couldnt be established) + * + * @return -1 on error otherwise the identifier of the new connection + */ +int32 network_connect(bool v6, + const char *addr, + uint16 port, + const char *from_addr, + uint16 from_port, + bool (*onConnectionEstablishedHandler)(int32 fd), + void (*onConnectionLooseHandler)(int32 fd) +); + + + +/** + * Disconnects the given connection + * + * @param fd connection identifier. + * + * @Note: + * - onDisconnect callback gets called! + * - cleares (returns) all assigned buffers + * + */ +void network_disconnect(int32 fd); + + +/** + * Attach's a netbuffer at the end of sending queue to the given connection + * + * @param fd connection identifier + * @param buf netbuffer to attach. + */ +void network_send(int32 fd, netbuf buf); + + +/** + * Sets the parser to RO Protocol like Packet Parser. + * + * @param fd connection identifier + * @param *packetlentable pointer to array of uint16 in size of UINT16_MAX, + * @param onComplteProc callback for packet completion. + * + * @note: + * PacketLen Table Fromat: + * each element's offsets represents th ro opcode. + * value is length. + * a length of 0 means the packet is dynamic. + * a length of UINT16_MAX means the packet is unknown. + * + * Static Packets must contain their hader in len so (0x64 == 55 ..) + * + */ +void network_parser_set_ro(int32 fd, + int16 *packetlentable, + void (*onPacketCompleteProc)(int32 fd, uint16 op, uint16 len, netbuf buf) + ); +#define ROPACKET_UNKNOWN UINT16_MAX +#define ROPACKET_DYNLEN 0 + + + + +#endif |