diff --git a/src/algo_naive.c b/src/algo_naive.c index 04593f0..0f9d8f8 100644 --- a/src/algo_naive.c +++ b/src/algo_naive.c @@ -2,14 +2,27 @@ struct naive_ctx { int ref_count; - GHashTable* packet_buffer; + struct buffer_packet bps[10]; + GQueue* free_buffer; // Available buffers + GHashTable* used_buffer; // Buffers used for reading or writing + GQueue* read_waiting; // Who wait to be notified for a read + GHashTable* write_waiting; // Structure to track packets waiting to be written }; +void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); +void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo); + void free_nothing(void* app_ctx) {} void free_naive(void* app_ctx) { struct naive_ctx* ctx = (struct naive_ctx*) app_ctx; ctx->ref_count--; - if (ctx->ref_count <= 0) free(ctx); + if (ctx->ref_count > 0) return; + g_queue_free(ctx->free_buffer); + g_queue_free(ctx->read_waiting); + g_hash_table_destroy (ctx->used_buffer); + free(ctx); } char* get_port(char* out, char* in) { @@ -17,39 +30,35 @@ char* get_port(char* out, char* in) { return out; } -void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { - int conn_sock1, conn_sock2, port; +void on_tcp_co(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + int conn_sock1, conn_sock2; struct sockaddr_in addr; socklen_t in_len; - char url[1024]; + char url[1024], port[6]; struct evt_core_cat local_cat = {0}; - struct evt_core_fdinfo fdinfo = {0}; - fdinfo.cat = &local_cat; - fdinfo.url = url; + struct evt_core_fdinfo to_fdinfo = {0}; + to_fdinfo.cat = &local_cat; + to_fdinfo.url = url; in_len = sizeof(addr); - conn_sock1 = accept(fd, (struct sockaddr*)&addr, &in_len); + conn_sock1 = accept(fdinfo->fd, (struct sockaddr*)&addr, &in_len); if (conn_sock1 == -1) goto co_error; conn_sock2 = dup(conn_sock1); if (conn_sock2 == -1) goto co_error; //printf("fd=%d accepts, creating fds=%d,%d\n", fd, conn_sock1, conn_sock2); - struct evt_core_fdinfo* listen_info = g_hash_table_lookup(ctx->socklist, &fd); - if (listen_info == NULL) goto co_error; - sscanf(listen_info->url, "tcp:listen:127.0.0.1:%d", &port); + get_port(port, fdinfo->url); - fdinfo.fd = conn_sock1; - fdinfo.cat->name = "tcp-read"; - sprintf(fdinfo.url, "tcp:read:127.0.0.1:%d", port); - evt_core_add_fd (ctx, &fdinfo); + to_fdinfo.fd = conn_sock1; + to_fdinfo.cat->name = "tcp-read"; + sprintf(to_fdinfo.url, "tcp:read:127.0.0.1:%s", port); + evt_core_add_fd (ctx, &to_fdinfo); - fdinfo.fd = conn_sock2; - fdinfo.cat->name = "tcp-write"; - sprintf(fdinfo.url, "tcp:write:127.0.0.1:%d", port); - evt_core_add_fd (ctx, &fdinfo); - - printf("Selected port: %d\n", port); + to_fdinfo.fd = conn_sock2; + to_fdinfo.cat->name = "tcp-write"; + sprintf(to_fdinfo.url, "tcp:write:127.0.0.1:%s", port); + evt_core_add_fd (ctx, &to_fdinfo); return; @@ -58,59 +67,115 @@ co_error: exit(EXIT_FAILURE); } -struct buffer_packet* get_bp(struct evt_core_cat* cat, int fd, int is_read_buffer) { +/** + * Returns a buffer if available, NULL otherwise + */ +struct buffer_packet* get_read_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) { struct buffer_packet* bp; - struct naive_ctx* ctx = cat->app_ctx; - bp = g_hash_table_lookup (ctx->packet_buffer, &fd); + // 1. Check if we don't have a buffer + bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd); if (bp != NULL) return bp; - if (!is_read_buffer) goto alloc_error; - bp = malloc(sizeof(struct buffer_packet)); - if (bp == NULL) goto alloc_error; - memset(bp, 0, sizeof(struct buffer_packet)); - bp->fdread = fd; - bp->fdwrite = -1; - bp->ref_count++; - g_hash_table_insert(ctx->packet_buffer, &(bp->fdread), bp); + // 2. Get a new buffer otherwise + bp = g_queue_pop_head(app_ctx->free_buffer); + if (bp == NULL) { + // 2.1 If no buffer is available, we subscribe to be notified later + g_queue_push_tail (app_ctx->read_waiting, &(fdinfo->fd)); + return NULL; + } + + // 3. Update state + g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp); + return bp; - -alloc_error: - perror("alloc error"); - exit(EXIT_FAILURE); } -void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); -void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); -void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); -void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd); +/** + * Returns a buffer if available, NULL otherwise + */ +struct buffer_packet* get_write_buffer(struct naive_ctx *app_ctx, struct evt_core_fdinfo *fdinfo) { + struct buffer_packet* bp; + GQueue* q; -void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { + // 1. Check if we don't have a buffer + bp = g_hash_table_lookup (app_ctx->used_buffer, &fdinfo->fd); + if (bp != NULL) return bp; + + // 2. Check our waiting queue otherwise + if ((q = g_hash_table_lookup(app_ctx->write_waiting, &(fdinfo->fd))) == NULL) return NULL; + bp = g_queue_pop_head(q); + if (bp == NULL) return NULL; // No packet to process + + // 3. Update state + g_hash_table_insert(app_ctx->used_buffer, &(fdinfo->fd), bp); + + return bp; +} + +void mv_buffer_rtow(struct naive_ctx* app_ctx, + struct evt_core_fdinfo* from, + struct evt_core_fdinfo* to, + struct buffer_packet* bp) { + + // 1. We get the target writing queue + GQueue* q; + q = g_hash_table_lookup(app_ctx->write_waiting, &(to->fd)); + if (q == NULL) { + q = g_queue_new (); + g_hash_table_insert(app_ctx->write_waiting, &(to->fd), q); + } + + // 2. We move the buffer to the target queue + g_hash_table_remove(app_ctx->used_buffer, &from->fd); + g_queue_push_tail(q, bp); +} + +void mv_buffer_wtor(struct naive_ctx* app_ctx, struct evt_core_fdinfo* fdinfo, struct buffer_packet* bp) { + g_queue_push_tail (app_ctx->free_buffer, bp); + g_hash_table_remove(app_ctx->used_buffer, &(fdinfo->fd)); +} + +void notify_read(struct evt_core_ctx* ctx, struct naive_ctx* app_ctx) { + struct evt_core_fdinfo* next_fdinfo = NULL; + while (next_fdinfo == NULL) { + int fd = GPOINTER_TO_INT(g_queue_pop_head(app_ctx->read_waiting)); + if (fd == 0) break; + next_fdinfo = evt_core_get_from_fd (ctx, fd); + if (strcmp(next_fdinfo->cat->name, "tcp-read") == 0) on_tcp_read(ctx, next_fdinfo); + else if (strcmp(next_fdinfo->cat->name, "udp-read") == 0) on_udp_read(ctx, next_fdinfo); + else { + fprintf(stderr, "A fd from category %s can't be stored in read_waiting\n", next_fdinfo->cat->name); + exit(EXIT_FAILURE); + } + } +} + +void on_tcp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { struct buffer_packet* bp; struct evt_core_fdinfo *to_fdinfo = NULL; - struct naive_ctx* app_ctx = cat->app_ctx; + struct naive_ctx* app_ctx = fdinfo->cat->app_ctx; int read_res = FDS_READY; char url[255]; - // 1. Read in our buffer packet - bp = get_bp(cat, fd, TRUE); + // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later + if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return; + + // 2. Try to read a whole packet in the buffer while (read_res != FDS_AGAIN && bp->mode == BP_READING) { - read_res = read_packet_from_tcp (fd, bp); + read_res = read_packet_from_tcp (fdinfo->fd, bp); if (read_res == FDS_ERR) goto co_error; } - if (bp->mode != BP_SWITCH_WRITE) return; + if (bp->mode != BP_WRITING) return; - // 2. Packet has just been read, choose a destination + // 3. A whole packet has been read, we will find someone to write it sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); to_fdinfo = evt_core_get_from_url (ctx, url); if (to_fdinfo == NULL) goto co_error; - // 3. Configure packet buffer for the destination - bp->mode = BP_WRITING; - bp->fdwrite = to_fdinfo->fd; - bp->ref_count++; - g_hash_table_insert(app_ctx->packet_buffer, &(bp->fdwrite), bp); - on_udp_write(ctx, to_fdinfo->cat, to_fdinfo->fd); + // 4. We move the buffer and notify the target + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); + on_udp_write(ctx, to_fdinfo); return; co_error: @@ -118,140 +183,136 @@ co_error: exit(EXIT_FAILURE); } -void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { +void on_tcp_write(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { struct buffer_packet* bp; - struct evt_core_fdinfo *to_fdinfo; + struct naive_ctx* app_ctx = fdinfo->cat->app_ctx; int write_res = FDS_READY; - char url[255]; - bp = get_bp(cat, fd, FALSE); - if (bp == NULL) return; + // 1. Get current write buffer OR a buffer from the waiting queue OR leave + if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return; + // 2. Write data from the buffer to the socket while (write_res != FDS_AGAIN && bp->mode == BP_WRITING) { - write_res = write_packet_to_tcp(fd, bp); + write_res = write_packet_to_tcp(fdinfo->fd, bp); if (write_res == FDS_ERR) goto co_error; } + if (bp->mode != BP_READING) return; + // 3. A whole packet has been written + // Release the buffer and notify + mv_buffer_wtor(app_ctx, fdinfo, bp); + notify_read(ctx, app_ctx); + + return; co_error: perror("Failed to TCP write"); exit(EXIT_FAILURE); } -void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { - -} - -void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { - -} - -void tcp_to_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { - // Init data structures for the transfer +void on_udp_read(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + struct buffer_packet* bp; + struct evt_core_fdinfo *to_fdinfo; + struct naive_ctx* app_ctx = fdinfo->cat->app_ctx; + int read_res = FDS_READY; char url[255]; - struct naive_ctx* app_ctx = cat->app_ctx; - struct buffer_packet* bp = &(app_ctx->tcp_to_udp); - struct evt_core_fdinfo* fdinfo; - int read_res = FDS_READY; - int write_res = FDS_READY; - // Either we can read something, either we can write something - while ((read_res != FDS_AGAIN && bp->mode == BP_READING) - || (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) { + // 1. Get current read buffer OR a new read buffer OR subscribe to be notified later + if ((bp = get_read_buffer(app_ctx, fdinfo)) == NULL) return; - // 1. Read packet from TCP socket - if (bp->mode == BP_READING) { - read_res = read_packet_from_tcp (fd, bp); - if (read_res == FDS_ERR) goto co_error; - } + // 2. Read packet from socket + read_res = read_packet_from_udp (fdinfo->fd, bp); + if (read_res == FDS_ERR) goto co_error; + if (bp->mode != BP_WRITING) return; - // 2. Write packet to UDP socket - if (bp->mode == BP_WRITING) { - sprintf(url, "udp:write:127.0.0.1:%d", bp->ip.ap.str.port); - fdinfo = evt_core_get_from_url(ctx, url); - if (fdinfo == NULL) goto co_error; - write_res = write_packet_to_udp(fdinfo->fd, bp); - if (read_res == FDS_ERR) goto co_error; - } - } - return; + // 3. A whole packet has been read, we will find someone to write it + sprintf(url, "tcp:write:127.0.0.1:7500"); + to_fdinfo = evt_core_get_from_url (ctx, url); + if (to_fdinfo == NULL) goto co_error; + + // 4. We move the buffer and notify the target + mv_buffer_rtow (app_ctx, fdinfo, to_fdinfo, bp); + on_tcp_write(ctx, to_fdinfo); co_error: - perror("Failed to handle read write for tcp_to_udp"); + perror("Failed to UDP read"); exit(EXIT_FAILURE); } -void udp_to_tcp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { - // Get target file descriptor - struct evt_core_cat* tcp = g_hash_table_lookup (ctx->catlist, "tcp-write"); - if (tcp == NULL || tcp->socklist->len < 1) goto co_error; - int tcp_fd = g_array_index(tcp->socklist, int, 0); - - // Init data structures for the transfer - struct naive_ctx* app_ctx = cat->app_ctx; - struct buffer_packet* bp = &(app_ctx->udp_to_tcp); - int read_res = FDS_READY; +void on_udp_write (struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + struct buffer_packet* bp; + struct naive_ctx* app_ctx = fdinfo->cat->app_ctx; int write_res = FDS_READY; - // Either we can read something, either we can write something - while ((read_res != FDS_AGAIN && bp->mode == BP_READING) - || (write_res != FDS_AGAIN && bp->mode == BP_WRITING)) { + // 1. Get current write buffer OR a buffer from the waiting queue OR leave + if ((bp = get_write_buffer(app_ctx, fdinfo)) == NULL) return; - // 1. Read packet from UDP socket - if (bp->mode == BP_READING) { - read_res = read_packet_from_udp (fd, bp); - if (read_res == FDS_ERR) goto co_error; - } + // 2. Write buffer + write_res = write_packet_to_udp(fdinfo->fd, bp); + if (write_res == FDS_ERR) goto co_error; + if (bp->mode != BP_READING) return; - // 2. Write packet to TCP socket - if (bp->mode == BP_WRITING) { - write_res = write_packet_to_tcp(tcp_fd, bp); - if (write_res == FDS_ERR) goto co_error; - } - } - - return; + // 3. A whole packet has been written + // Release the buffer and notify + mv_buffer_wtor(app_ctx, fdinfo, bp); + notify_read(ctx, app_ctx); co_error: - perror("Failed to handle read write for udp_to_tcp"); + perror("Failed to UDP write"); exit(EXIT_FAILURE); } -void free_buffer_packet(void *v) { - struct buffer_packet* bp = (struct buffer_packet*) v; - bp->ref_count--; - if (bp->ref_count == 0) free(v); -} -void free_simple(void* v) { +void naive_free_simple(void* v) { free(v); } +void on_err(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { + struct naive_ctx* app_ctx = fdinfo->cat->app_ctx; + struct buffer_packet* bp; + + // 1. If has a "used" buffer, remove it + bp = g_hash_table_lookup (app_ctx->used_buffer, &(fdinfo->fd)); + if (bp != NULL) { + g_hash_table_remove (app_ctx->used_buffer, &(fdinfo->fd)); + memset(bp, 0, sizeof(struct buffer_packet)); + g_queue_push_tail(app_ctx->free_buffer, bp); + } + + // 2. If appears in the write waiting queue, remove it + GQueue* writew = g_hash_table_lookup (app_ctx->write_waiting, &(fdinfo->fd)); + while (writew != NULL && (bp = g_queue_pop_head (writew)) != NULL) { + memset(bp, 0, sizeof(struct buffer_packet)); + g_queue_push_tail(app_ctx->free_buffer, bp); + } + g_hash_table_remove (app_ctx->write_waiting, &(fdinfo->fd)); + + // 3. If appears in the read waiting queue, remove it + g_queue_remove_all (app_ctx->read_waiting, &(fdinfo->fd)); +} + void algo_naive(struct algo_skel* as) { struct naive_ctx* ctx = malloc(sizeof(struct naive_ctx)); if (ctx == NULL) goto init_err; memset(ctx, 0, sizeof(struct naive_ctx)); - ctx->packet_buffer = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, free_buffer_packet); - - as->on_init.name = "init"; - as->on_init.flags = 0; - as->on_init.app_ctx = NULL; - as->on_init.free_app_ctx = NULL; - as->on_init.cb = NULL; - as->on_init.socklist = NULL; + ctx->free_buffer = g_queue_new (); + ctx->read_waiting = g_queue_new (); + ctx->used_buffer = g_hash_table_new(g_int_hash, g_int_equal); + ctx->write_waiting = g_hash_table_new_full (g_int_hash, g_int_equal, NULL, naive_free_simple); + for (int i = 0; i < sizeof(ctx->bps) / sizeof(ctx->bps[0]); i++) { + g_queue_push_tail(ctx->free_buffer, &(ctx->bps[i])); + } as->on_tcp_co.name = "tcp-listen"; as->on_tcp_co.flags = EPOLLIN; - as->on_tcp_co.app_ctx = NULL; as->on_tcp_co.free_app_ctx = free_nothing; as->on_tcp_co.cb = on_tcp_co; - as->on_tcp_co.socklist = NULL; as->on_tcp_read.name = "tcp-read"; as->on_tcp_read.flags = EPOLLIN | EPOLLET | EPOLLRDHUP; as->on_tcp_read.app_ctx = ctx; as->on_tcp_read.free_app_ctx = free_naive; as->on_tcp_read.cb = on_tcp_read; - as->on_tcp_read.socklist = NULL; + as->on_tcp_read.err_cb = on_err; ctx->ref_count++; as->on_udp_read.name = "udp-read"; @@ -259,7 +320,7 @@ void algo_naive(struct algo_skel* as) { as->on_udp_read.app_ctx = ctx; as->on_udp_read.free_app_ctx = free_naive; as->on_udp_read.cb = on_udp_read; - as->on_udp_read.socklist = NULL; + as->on_udp_read.err_cb = on_err; ctx->ref_count++; as->on_tcp_write.name = "tcp-write"; @@ -267,7 +328,7 @@ void algo_naive(struct algo_skel* as) { as->on_tcp_write.app_ctx = ctx; as->on_tcp_write.free_app_ctx = free_naive; as->on_tcp_write.cb = on_tcp_write; - as->on_tcp_write.socklist = NULL; + as->on_tcp_write.err_cb = on_err; ctx->ref_count++; as->on_udp_write.name = "udp-write"; @@ -275,7 +336,7 @@ void algo_naive(struct algo_skel* as) { as->on_udp_write.app_ctx = ctx; as->on_udp_write.free_app_ctx = free_naive; as->on_udp_write.cb = on_udp_write; - as->on_udp_write.socklist = NULL; + as->on_udp_write.err_cb = on_err; ctx->ref_count++; return; diff --git a/src/algo_skel.h b/src/algo_skel.h index bdfe972..e4a21ef 100644 --- a/src/algo_skel.h +++ b/src/algo_skel.h @@ -8,7 +8,6 @@ #include "utils.h" struct algo_skel { - struct evt_core_cat on_init; struct evt_core_cat on_udp_read; struct evt_core_cat on_tcp_read; struct evt_core_cat on_udp_write; diff --git a/src/donar.c b/src/donar.c index 8e40703..7dfe2fe 100644 --- a/src/donar.c +++ b/src/donar.c @@ -74,8 +74,8 @@ in_error: terminate: if (onion_file != NULL) free(onion_file); - if (port != NULL) free(port); if (algo != NULL) free(algo); + g_ptr_array_free(ports, TRUE); return errored; } diff --git a/src/donar_client.c b/src/donar_client.c index 08f9ea9..f21f1eb 100644 --- a/src/donar_client.c +++ b/src/donar_client.c @@ -150,7 +150,6 @@ void donar_client(struct donar_client_ctx* ctx, struct algo_skel* algo, char* on .socklist = NULL }; evt_core_add_cat (&(ctx->evts), &init_socks5); - evt_core_add_cat (&(ctx->evts), &(algo->on_init)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co)); evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read)); diff --git a/src/donar_server.c b/src/donar_server.c index 8437cb8..bcebeab 100644 --- a/src/donar_server.c +++ b/src/donar_server.c @@ -82,7 +82,6 @@ socket_failed: void donar_server(struct donar_server_ctx* ctx, struct algo_skel* algo, GPtrArray* ports) { evt_core_init (&(ctx->evts)); - evt_core_add_cat (&(ctx->evts), &(algo->on_init)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_co)); evt_core_add_cat (&(ctx->evts), &(algo->on_udp_read)); evt_core_add_cat (&(ctx->evts), &(algo->on_tcp_read)); diff --git a/src/evt_core.c b/src/evt_core.c index 5a6e457..2354b9b 100644 --- a/src/evt_core.c +++ b/src/evt_core.c @@ -52,6 +52,7 @@ void evt_core_add_cat(struct evt_core_ctx* ctx, struct evt_core_cat* cat) { dyn->cb = cat->cb; dyn->name = strdup(cat->name); dyn->flags = cat->flags; + dyn->err_cb = cat->err_cb; dyn->socklist = g_array_new (FALSE, FALSE, sizeof(struct evt_core_fdinfo*)); if (dyn->name == NULL) { @@ -133,11 +134,6 @@ void evt_core_loop(struct evt_core_ctx* ctx) { struct evt_core_fdinfo* fdinfo; struct evt_core_cat* cat; - cat = g_hash_table_lookup(ctx->catlist, "init"); - if (cat != NULL) { - cat->cb(ctx, cat, -1); - } - printf("--- Start main loop\n"); int num_fd, n = 0; while(1) { @@ -156,9 +152,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) { if (evt & EPOLLHUP) fprintf(stderr, "Epoll Hup Event. "); if (evt & EPOLLERR) fprintf(stderr, "Epoll Err Event. "); - cat = evt_core_rm_fd (ctx, err_fd); - if (cat != NULL) { - fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, cat->name); + fdinfo = evt_core_get_from_fd(ctx, err_fd); + if (fdinfo != NULL) { + fprintf(stderr, "Clearing fd=%d on cat=%s\n", err_fd, fdinfo->cat->name); + if (fdinfo->cat->err_cb != NULL) fdinfo->cat->err_cb(ctx, fdinfo); + evt_core_rm_fd (ctx, err_fd); } else { fprintf(stderr, "The file descriptor is not registered in a category, this is probably a logic error\n"); close (err_fd); @@ -172,16 +170,11 @@ void evt_core_loop(struct evt_core_ctx* ctx) { fprintf(stderr, "Ignoring file descriptor %d as it is not registered. This is a bug.\n", events[n].data.fd); continue; } - fdinfo->cat->cb(ctx, fdinfo->cat, events[n].data.fd); + fdinfo->cat->cb(ctx, fdinfo); } } - cat = g_hash_table_lookup(ctx->catlist, "destroy"); - if (cat != NULL) { - cat->cb(ctx, cat, -1); - } - evt_core_free(ctx); } diff --git a/src/evt_core.h b/src/evt_core.h index 63b1922..523617c 100644 --- a/src/evt_core.h +++ b/src/evt_core.h @@ -12,14 +12,16 @@ struct evt_core_ctx; struct evt_core_cat; +struct evt_core_fdinfo; typedef void (*evt_core_free_app_ctx)(void*); -typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_cat*, int fd); +typedef void (*evt_core_cb)(struct evt_core_ctx*, struct evt_core_fdinfo*); struct evt_core_cat { void* app_ctx; evt_core_free_app_ctx free_app_ctx; evt_core_cb cb; + evt_core_cb err_cb; char* name; int flags; GArray* socklist; diff --git a/src/meas_lat.c b/src/meas_lat.c index 0d069ba..5f93b68 100644 --- a/src/meas_lat.c +++ b/src/meas_lat.c @@ -7,10 +7,10 @@ struct timer_ctx { uint64_t counter; }; -void on_udp(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { +void on_udp(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { } -void on_timer(struct evt_core_ctx* ctx, struct evt_core_cat* cat, int fd) { +void on_timer(struct evt_core_ctx* ctx, struct evt_core_fdinfo* fdinfo) { } int main(int argc, char** argv) { @@ -23,6 +23,7 @@ int main(int argc, char** argv) { .app_ctx = NULL, .free_app_ctx = NULL, .cb = on_udp, + .err_cb = NULL, .name = "udp-read", .flags = EPOLLIN | EPOLLET, .socklist = NULL @@ -31,6 +32,7 @@ int main(int argc, char** argv) { .app_ctx = &apptime, .free_app_ctx = NULL, .cb = on_timer, + .err_cb = NULL, .name = "timer", .flags = EPOLLIN | EPOLLET, .socklist = NULL diff --git a/src/packet.c b/src/packet.c index 568d926..e89ad8f 100644 --- a/src/packet.c +++ b/src/packet.c @@ -21,7 +21,7 @@ enum FD_STATE read_packet_from_tcp(int fd, struct buffer_packet* bp) { bp->aread += nread; } - bp->mode = BP_SWITCH_WRITE; + bp->mode = BP_WRITING; bp->awrite = 0; return FDS_READY; @@ -38,7 +38,7 @@ enum FD_STATE write_packet_to_tcp(int fd, struct buffer_packet* bp) { bp->awrite += nwrite; } - bp->mode = BP_SWITCH_READ; + bp->mode = BP_READING; bp->aread = 0; return FDS_READY; @@ -60,7 +60,7 @@ enum FD_STATE write_packet_to_udp(int fd, struct buffer_packet* bp) { if (nwrite == -1 && errno == EAGAIN) return FDS_AGAIN; if (nwrite != bytes_to_send) return FDS_ERR; - bp->mode = BP_SWITCH_READ; + bp->mode = BP_READING; bp->aread = 0; return FDS_READY; @@ -81,7 +81,7 @@ enum FD_STATE read_packet_from_udp (int fd, struct buffer_packet* bp) { bp->ip.ap.str.size = nread + pkt_header_size; - bp->mode = BP_SWITCH_WRITE; + bp->mode = BP_WRITING; bp->awrite = 0; return FDS_READY; diff --git a/src/packet.h b/src/packet.h index 357e956..f8d1f8f 100644 --- a/src/packet.h +++ b/src/packet.h @@ -24,9 +24,7 @@ enum FD_STATE { enum BP_MODE { BP_READING, - BP_WRITING, - BP_SWITCH_WRITE, - BP_SWITCH_READ + BP_WRITING }; union abstract_packet { @@ -45,10 +43,7 @@ struct internet_packet { }; struct buffer_packet { - uint64_t ref_count; uint8_t mode; - int fdread; - int fdwrite; uint16_t aread; uint16_t awrite; struct internet_packet ip;