Commit 608eee47 authored by Marek Vavrusa's avatar Marek Vavrusa

Code reindented.

parent 1b2f9a9f
This diff is collapsed.
......@@ -37,12 +37,12 @@ struct dt_unit_t;
* The reason is, you can compare like: "state > Dead" etc.
*/
enum {
ThreadJoined = 1 << 0, /*!< Thread is finished and joined. */
ThreadJoinable = 1 << 1, /*!< Thread is dead, waiting to be reclaimed. */
ThreadCancelled = 1 << 2, /*!< Thread is cancelled, finishing task. */
ThreadDead = 1 << 3, /*!< Thread is finished, exiting. */
ThreadIdle = 1 << 4, /*!< Thread is idle, waiting for purpose. */
ThreadActive = 1 << 5 /*!< Thread is active, working on a task. */
ThreadJoined = 1 << 0, /*!< Thread is finished and joined. */
ThreadJoinable = 1 << 1, /*!< Thread is dead, waiting to be reclaimed. */
ThreadCancelled = 1 << 2, /*!< Thread is cancelled, finishing task. */
ThreadDead = 1 << 3, /*!< Thread is finished, exiting. */
ThreadIdle = 1 << 4, /*!< Thread is idle, waiting for purpose. */
ThreadActive = 1 << 5 /*!< Thread is active, working on a task. */
} dt_state_t;
......@@ -67,14 +67,14 @@ typedef int (*runnable_t)(struct dthread_t*);
* while not breaking an array of dthread_t.
*/
typedef struct dthread_t {
volatile unsigned state; /*!< Bitfield of dt_flag flags. */
runnable_t run; /*!< Runnable function or 0. */
void *data; /*!< Currently active data */
struct dt_unit_t *unit; /*!< Reference to assigned unit. */
void *_adata; /* Thread-specific data. */
pthread_t _thr; /* Implementation specific thread */
pthread_attr_t _attr; /* Implementation specific thread attributes */
pthread_mutex_t _mx; /* Thread state change lock. */
volatile unsigned state; /*!< Bitfield of dt_flag flags. */
runnable_t run; /*!< Runnable function or 0. */
void *data; /*!< Currently active data */
struct dt_unit_t *unit; /*!< Reference to assigned unit. */
void *_adata; /* Thread-specific data. */
pthread_t _thr; /* Implementation specific thread */
pthread_attr_t _attr; /* Implementation specific thread attributes */
pthread_mutex_t _mx; /* Thread state change lock. */
} dthread_t;
/*!
......@@ -85,13 +85,13 @@ typedef struct dthread_t {
* the same runnable.
*/
typedef struct dt_unit_t {
int size; /*!< Unit width (number of allocated threads) */
struct dthread_t **threads; /*!< Array of threads */
pthread_cond_t _notify; /* Notify thread */
pthread_mutex_t _notify_mx; /* Condition mutex */
pthread_cond_t _report; /* Report thread state */
pthread_mutex_t _report_mx; /* Condition mutex */
pthread_mutex_t _mx; /* Unit lock */
int size; /*!< Unit width (number of allocated threads) */
struct dthread_t **threads; /*!< Array of threads */
pthread_cond_t _notify; /* Notify thread */
pthread_mutex_t _notify_mx; /* Condition mutex */
pthread_cond_t _report; /* Report thread state */
pthread_mutex_t _report_mx; /* Condition mutex */
pthread_mutex_t _mx; /* Unit lock */
} dt_unit_t;
/*!
......
This diff is collapsed.
......@@ -31,7 +31,7 @@ typedef struct ns_nameserver {
* @brief Pointer to the zone database structure used for answering
* queries.
*/
zdb_database *zone_db;
zdb_database *zone_db;
uint8_t *err_response;
size_t err_resp_size;
} ns_nameserver;
......@@ -64,7 +64,7 @@ ns_nameserver *ns_create( zdb_database *database );
* @retval -1 if an error occured and the response is not valid.
*/
int ns_answer_request( ns_nameserver *nameserver, const uint8_t *query_wire,
size_t qsize, uint8_t *response_wire, size_t *rsize );
size_t qsize, uint8_t *response_wire, size_t *rsize );
/*!
* @brief Properly destroys the name server structure.
......
......@@ -10,201 +10,201 @@
cute_server *cute_create()
{
debug_server("Creating Server structure..\n");
cute_server *server = malloc(sizeof(cute_server));
server->handlers = NULL;
server->state = Idle;
if (server == NULL) {
ERR_ALLOC_FAILED;
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating Zone Database structure..\n");
server->zone_db = zdb_create();
if (server->zone_db == NULL) {
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating Name Server structure..\n");
server->nameserver = ns_create(server->zone_db);
if (server->nameserver == NULL) {
zdb_destroy(&server->zone_db);
free(server);
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating workers..\n");
// Estimate number of threads/manager
int thr_count = dt_optimal_size();
debug_server("Estimated number of threads per handler: %d\n", thr_count);
// Create socket handlers
int sock = socket_create(PF_INET, SOCK_STREAM);
socket_bind(sock, "0.0.0.0", DEFAULT_PORT);
socket_listen(sock, TCP_BACKLOG_SIZE);
// Create threading unit
dt_unit_t *unit = dt_create(thr_count);
dt_repurpose(unit->threads[0], &tcp_master, 0);
cute_create_handler(server, sock, unit);
// Create UDP socket
sock = socket_create(PF_INET, SOCK_DGRAM);
socket_bind(sock, "0.0.0.0", DEFAULT_PORT);
// Create threading unit
unit = dt_create_coherent(thr_count, &udp_master, 0);
cute_create_handler(server, sock, unit);
debug_server("Done\n\n");
return server;
debug_server("Creating Server structure..\n");
cute_server *server = malloc(sizeof(cute_server));
server->handlers = NULL;
server->state = Idle;
if (server == NULL) {
ERR_ALLOC_FAILED;
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating Zone Database structure..\n");
server->zone_db = zdb_create();
if (server->zone_db == NULL) {
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating Name Server structure..\n");
server->nameserver = ns_create(server->zone_db);
if (server->nameserver == NULL) {
zdb_destroy(&server->zone_db);
free(server);
return NULL;
}
debug_server("Done\n\n");
debug_server("Creating workers..\n");
// Estimate number of threads/manager
int thr_count = dt_optimal_size();
debug_server("Estimated number of threads per handler: %d\n", thr_count);
// Create socket handlers
int sock = socket_create(PF_INET, SOCK_STREAM);
socket_bind(sock, "0.0.0.0", DEFAULT_PORT);
socket_listen(sock, TCP_BACKLOG_SIZE);
// Create threading unit
dt_unit_t *unit = dt_create(thr_count);
dt_repurpose(unit->threads[0], &tcp_master, 0);
cute_create_handler(server, sock, unit);
// Create UDP socket
sock = socket_create(PF_INET, SOCK_DGRAM);
socket_bind(sock, "0.0.0.0", DEFAULT_PORT);
// Create threading unit
unit = dt_create_coherent(thr_count, &udp_master, 0);
cute_create_handler(server, sock, unit);
debug_server("Done\n\n");
return server;
}
iohandler_t* cute_create_handler(cute_server *server, int fd, dt_unit_t* unit)
{
// Create new worker
iohandler_t* handler = malloc(sizeof(iohandler_t));
if(handler == 0)
return 0;
// Initialize
handler->fd = fd;
handler->state = Idle;
handler->next = server->handlers;
handler->server = server;
handler->unit = unit;
// Update unit data object
for(int i = 0; i < unit->size; ++i) {
dthread_t *thread = unit->threads[i];
dt_repurpose(thread, thread->run, handler);
}
// Update list
server->handlers = handler;
// Run if server is online
if(server->state & Running) {
dt_start(handler->unit);
}
return handler;
// Create new worker
iohandler_t* handler = malloc(sizeof(iohandler_t));
if(handler == 0)
return 0;
// Initialize
handler->fd = fd;
handler->state = Idle;
handler->next = server->handlers;
handler->server = server;
handler->unit = unit;
// Update unit data object
for(int i = 0; i < unit->size; ++i) {
dthread_t *thread = unit->threads[i];
dt_repurpose(thread, thread->run, handler);
}
// Update list
server->handlers = handler;
// Run if server is online
if(server->state & Running) {
dt_start(handler->unit);
}
return handler;
}
int cute_remove_handler(cute_server *server, iohandler_t *ref)
{
// Find worker
iohandler_t *w = 0, *p = 0;
for(w = server->handlers; w != NULL; p = w,w = w->next) {
// Compare fd
if(w == ref) {
// Disconnect
if(p == 0) {
server->handlers = w->next;
}
else {
p->next = w->next;
}
break;
}
}
// Check
if(w == 0) {
return -1;
}
// Wait for dispatcher to finish
if(w->state & Running) {
w->state = Idle;
dt_stop(w->unit);
dt_join(w->unit);
}
// Close socket
socket_close(w->fd);
// Destroy dispatcher and worker
dt_delete(&w->unit);
free(w);
return 0;
// Find worker
iohandler_t *w = 0, *p = 0;
for(w = server->handlers; w != NULL; p = w,w = w->next) {
// Compare fd
if(w == ref) {
// Disconnect
if(p == 0) {
server->handlers = w->next;
}
else {
p->next = w->next;
}
break;
}
}
// Check
if(w == 0) {
return -1;
}
// Wait for dispatcher to finish
if(w->state & Running) {
w->state = Idle;
dt_stop(w->unit);
dt_join(w->unit);
}
// Close socket
socket_close(w->fd);
// Destroy dispatcher and worker
dt_delete(&w->unit);
free(w);
return 0;
}
int cute_start( cute_server *server, char **filenames, uint zones )
{
debug_server("Starting server with %u zone files.\n", zones);
//stat
stat_static_gath_start();
//!stat
for (uint i = 0; i < zones; ++i) {
debug_server("Parsing zone file %s..\n", filenames[i]);
if (zp_parse_zone(filenames[i], server->zone_db) != 0) {
return -1;
}
}
debug_server("\nDone\n\n");
debug_server("Starting servers..\n");
// Start dispatchers
int ret = 0;
server->state |= Running;
for(iohandler_t* w = server->handlers; w != NULL; w = w->next) {
w->state = Running;
ret += dt_start(w->unit);
}
return ret;
debug_server("Starting server with %u zone files.\n", zones);
//stat
stat_static_gath_start();
//!stat
for (uint i = 0; i < zones; ++i) {
debug_server("Parsing zone file %s..\n", filenames[i]);
if (zp_parse_zone(filenames[i], server->zone_db) != 0) {
return -1;
}
}
debug_server("\nDone\n\n");
debug_server("Starting servers..\n");
// Start dispatchers
int ret = 0;
server->state |= Running;
for(iohandler_t* w = server->handlers; w != NULL; w = w->next) {
w->state = Running;
ret += dt_start(w->unit);
}
return ret;
}
int cute_wait(cute_server *server)
{
// Wait for dispatchers to finish
int ret = 0;
while(server->handlers != NULL) {
debug_server("server: [%p] joining threading unit\n", server->handlers);
ret += dt_join(server->handlers->unit);
cute_remove_handler(server, server->handlers);
debug_server("server: joined threading unit\n", p);
}
return ret;
// Wait for dispatchers to finish
int ret = 0;
while(server->handlers != NULL) {
debug_server("server: [%p] joining threading unit\n", server->handlers);
ret += dt_join(server->handlers->unit);
cute_remove_handler(server, server->handlers);
debug_server("server: joined threading unit\n", p);
}
return ret;
}
void cute_stop( cute_server *server )
{
// Notify servers to stop
server->state &= ~Running;
for(iohandler_t* w = server->handlers; w != NULL; w = w->next) {
w->state = Idle;
dt_stop(w->unit);
}
// Notify servers to stop
server->state &= ~Running;
for(iohandler_t* w = server->handlers; w != NULL; w = w->next) {
w->state = Idle;
dt_stop(w->unit);
}
}
void cute_destroy( cute_server **server )
{
// Free workers
iohandler_t* w = (*server)->handlers;
while(w != NULL) {
iohandler_t* n = w->next;
cute_remove_handler(*server, w);
w = n;
}
stat_static_gath_free();
ns_destroy(&(*server)->nameserver);
zdb_destroy(&(*server)->zone_db);
free(*server);
*server = NULL;
// Free workers
iohandler_t* w = (*server)->handlers;
while(w != NULL) {
iohandler_t* n = w->next;
cute_remove_handler(*server, w);
w = n;
}
stat_static_gath_free();
ns_destroy(&(*server)->nameserver);
zdb_destroy(&(*server)->zone_db);
free(*server);
*server = NULL;
}
......@@ -23,11 +23,11 @@
*/
typedef struct iohandler_t {
int fd; /* I/O filedescripto r */
unsigned state; /* Handler state */
struct iohandler_t* next; /* Next handler */
dt_unit_t* unit; /* Threading unit */
struct cute_server* server; /* Reference to server */
int fd; /* I/O filedescripto r */
unsigned state; /* Handler state */
struct iohandler_t* next; /* Next handler */
dt_unit_t* unit; /* Threading unit */
struct cute_server* server; /* Reference to server */
} iohandler_t;
......@@ -39,8 +39,8 @@ typedef struct iohandler_t {
/*! Server state flags.
*/
typedef enum {
Idle = 0 << 0,
Running = 1 << 0
Idle = 0 << 0,
Running = 1 << 0
} server_state;
struct cute_server;
......@@ -54,17 +54,17 @@ struct iohandler_t;
*/
typedef struct cute_server {
/*! @brief Server state tracking. */
unsigned state;
/*! @brief Server state tracking. */
unsigned state;
/*! @brief Reference to the name server structure. */
ns_nameserver *nameserver;
/*! @brief Reference to the name server structure. */
ns_nameserver *nameserver;
/*! @brief Reference to the zone database structure. */
zdb_database *zone_db;
/*! @brief Reference to the zone database structure. */
zdb_database *zone_db;
/*! @brief I/O handlers list. */
struct iohandler_t *handlers;
/*! @brief I/O handlers list. */
struct iohandler_t *handlers;
} cute_server;
......
......@@ -12,100 +12,100 @@
int socket_create( int family, int type )
{
// Create socket
int sock = socket(family, type, 0);
// Create socket
int sock = socket(family, type, 0);
// Reuse open socket
int on = 1;
if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
(const char*)&on, sizeof(on)) < 0) {
return socket_close(sock);
}
// Reuse open socket
int on = 1;
if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
(const char*)&on, sizeof(on)) < 0) {
return socket_close(sock);
}
return sock;
return sock;
}
int socket_connect( int socket, const char* addr, unsigned short port )
{
// Create socket
struct hostent* hent = gethostbyname(addr);
if(hent == 0)
return -1;
// Prepare host address
struct sockaddr_in saddr;
socklen_t addrlen = sizeof(struct sockaddr_in);
saddr.sin_family = AF_INET;
saddr.sin_port = htons(port);
memcpy(&saddr.sin_addr, hent->h_addr, hent->h_length);
// Connect to host
return connect(socket, (struct sockaddr *)&saddr, addrlen);
// Create socket
struct hostent* hent = gethostbyname(addr);
if(hent == 0)
return -1;
// Prepare host address
struct sockaddr_in saddr;
socklen_t addrlen = sizeof(struct sockaddr_in);
saddr.sin_family = AF_INET;
saddr.sin_port = htons(port);
memcpy(&saddr.sin_addr, hent->h_addr, hent->h_length);
// Connect to host
return connect(socket, (struct sockaddr *)&saddr, addrlen);
}
int socket_bind( int socket, const char* addr, unsigned short port )
{
// Initialize socket address
struct sockaddr_in saddr;
socklen_t addrlen = sizeof(struct sockaddr_in);
if(getsockname(socket, &saddr, &addrlen) < 0) {
return -1;
}
// Set address and port
saddr.sin_port = htons(port);
saddr.sin_addr.s_addr = inet_addr(addr);
if(saddr.sin_addr.s_addr == INADDR_NONE) {
log_error("socket_listen: address %s is invalid, using 0.0.0.0 instead",
addr);
saddr.sin_addr.s_addr = htonl(INADDR_ANY);
}
// Reuse old address if taken
int flag = 1;
if(setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag)) < 0) {
return -2;
}
// Bind to specified address
int res = bind(socket, (struct sockaddr *)& saddr, sizeof(saddr));
if (res == -1) {
log_error("cannot bind socket (errno %d): %s.\n",
errno, strerror(errno));
return -3;
}
return 0;
// Initialize socket address
struct sockaddr_in saddr;
socklen_t addrlen = sizeof(struct sockaddr_in);
if(getsockname(socket, &saddr, &addrlen) < 0) {
return -1;
}
// Set address and port
saddr.sin_port = htons(port);
saddr.sin_addr.s_addr = inet_addr(addr);
if(saddr.sin_addr.s_addr == INADDR_NONE) {
log_error("socket_listen: address %s is invalid, using 0.0.0.0 instead",
addr);
saddr.sin_addr.s_addr = htonl(INADDR_ANY);
}
// Reuse old address if taken
int flag = 1;
if(setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(flag)) < 0) {
return -2;
}
// Bind to specified address
int res = bind(socket, (struct sockaddr *)& saddr, sizeof(saddr));
if (res == -1) {
log_error("cannot bind socket (errno %d): %s.\n",
errno, strerror(errno));
return -3;
}
return 0;
}
int socket_listen( int socket, int backlog_size )
{
return listen(socket, backlog_size);
return listen(socket, backlog_size);
}
ssize_t socket_recv( int socket, void *buf, size_t len, int flags )
{
return recv(socket, buf, len, flags);
return recv(socket, buf, len, flags);
}
ssize_t socket_recvfrom( int socket, void *buf, size_t len, int flags,
struct sockaddr *from, socklen_t *fromlen )
{
return recvfrom(socket, buf, len, flags, from, fromlen);
return recvfrom(socket, buf, len, flags, from, fromlen);
}
ssize_t socket_send( int socket, const void *buf, size_t len, int flags )
{
return send(socket, buf, len, flags);
return send(socket, buf, len, flags);
}
ssize_t socket_sendto( int socket, const void *buf, size_t len, int flags,
const struct sockaddr *to, socklen_t tolen )
const struct sockaddr *to, socklen_t tolen )
{
return sendto(socket, buf, len, flags, to, tolen);
return sendto(socket, buf, len, flags, to, tolen);
}
int socket_close( int socket )
{
return close(socket);
return close(socket);
}
......@@ -20,7 +20,7 @@
/*! \brief Socket-related constants. */
enum {
SOCKET_MTU_SZ = 8192, //!< \todo <= Determine UDP MTU size.
SOCKET_MTU_SZ = 8192, //!< \todo <= Determine UDP MTU size.
} socket_const_t;
/*!
......
This diff is collapsed.
......@@ -9,164 +9,164 @@
/** Event descriptor.
*/
typedef struct sm_event {
struct sm_manager* manager;
int fd;
uint32_t events;
void* inbuf;
void* outbuf;
size_t size_in;
size_t size_out;
struct sm_manager* manager;
int fd;
uint32_t events;
void* inbuf;
void* outbuf;
size_t size_in;
size_t size_out;
} sm_event;
static inline void udp_epoll_handler(sm_event *ev)
{
struct sockaddr_in faddr;
int addrsize = sizeof(faddr);
struct sockaddr_in faddr;
int addrsize = sizeof(faddr);
int n = 0;
int n = 0;
// Loop until all data is read
while(n >= 0) {
// Loop until all data is read
while(n >= 0) {
// Receive data
// \todo Global I/O lock means ~ 8% overhead; recvfrom() should be thread-safe
n = socket_recvfrom(ev->fd, ev->inbuf, ev->size_in, MSG_DONTWAIT, (struct sockaddr *)&faddr, (socklen_t *)&addrsize);
//char _str[INET_ADDRSTRLEN];
//inet_ntop(AF_INET, &(faddr.sin_addr), _str, INET_ADDRSTRLEN);
//fprintf(stderr, "recvfrom() in %p: received %d bytes from %s:%d.\n", (void*)pthread_self(), n, _str, faddr.sin_port);
// Receive data
// \todo Global I/O lock means ~ 8% overhead; recvfrom() should be thread-safe
n = socket_recvfrom(ev->fd, ev->inbuf, ev->size_in, MSG_DONTWAIT, (struct sockaddr *)&faddr, (socklen_t *)&addrsize);