diff --git a/examples/queue.c b/examples/queue.c new file mode 100644 index 0000000..2bcfd32 --- /dev/null +++ b/examples/queue.c @@ -0,0 +1,302 @@ +#include "queue.h" + +typedef struct queue_entry_t queue_entry_t; +typedef struct queue_event_t queue_event_t; + +struct queue_entry_t { + queue_entry_t *prev; + queue_entry_t *next; + void *data; + int events; + microtcp_queue_t *queue; + microtcp_socket_t *socket; + queue_event_t *event_entry; +}; + +struct queue_event_t { + + queue_event_t *prev; + queue_event_t *next; + + int events; + queue_entry_t *socket_entry; +}; + +struct microtcp_queue_t { + + microtcp_t *mtcp; + + pthread_cond_t something_happened; + + queue_event_t *queue_head; + queue_event_t *queue_tail; + + queue_entry_t *entry_free_list; + queue_entry_t entry_pool[MICROTCP_QUEUE_ENTRIES_MAX]; + + queue_event_t *event_free_list; + queue_event_t event_pool[MICROTCP_QUEUE_ENTRIES_MAX]; +}; + +static event_entry_t* +pop_event_entry_from_queue(microtcp_queue_t *queue) +{ + event_entry_t *event; + + if (queue->queue_tail) { + + // An event is present in the queue. Pop it. + + event = queue->queue_tail; + + if (event->prev) + event->prev->next = NULL; + else + queue->queue_head = NULL; + queue->queue_tail = event->prev; + + } else + event = NULL; + + return event; +} + +microtcp_event_t microtcp_queue_next(microtcp_queue_t *queue, bool no_block) +{ + microtcp_t *mtcp; + + event_entry_t *event = pop_event_entry_from_queue(queue); + + while (!event && !no_block) { + pthread_cond_wait(&queue->something_happened, &mtcp->lock); + event = pop_event_entry_from_queue(queue); + } + + microtcp_event_t result; + + if (event) { + queue_entry_t *socket_entry = event->socket_entry; + + result.type = event->events; + result.data = socket_entry->data; + result.socket = socket_entry->socket; + + } else { + result.type = 0; + result.data = NULL; + result.socket = NULL; + } + + return result; +} + +static void signal_socket_events_to_queues(microtcp_socket_t *socket, int events) +{ + queue_entry_t *entry = socket->queue_entry; + while (entry) { + + int interesting_events = events & entry->events; + + if (interesting_events) { + // Some events that are of interest to + // this queue are being signaled. + // Push them onto the queue's event queue + + microtcp_queue_t *queue = entry->queue; + + if (entry->event_entry) { + // At least one event was already signaled, so + // an event structure was already created. + // We just need to add to it the newly signaled + // events. + event_entry_t *event = entry->event_entry; + event->events |= interesting_events; + } else { + + // No events related to this socket are in the + // queue, so we need to create a new event structure. + // + // Unused event structure are organized in a free + // list, so we just need to pop one from there. + // + // We know for sure that a free event structure is + // available at this point because there is one for + // each socket registered into the queue. + assert(queue->event_free_list); + + event_entry_t *event = queue->event_free_list; + queue->event_free_list = event->next; + + // Set-up the event structure + event->prev = NULL; + event->next = NULL; + event->events = interesting_events; + event->socket_entry = entry; + + // Now push the event structure into the queue's queue + event->next = queue->queue_head; + if (queue->queue_head) + queue->queue_head->prev = event; + else + queue->queue_tail = event; + queue->queue_head = event; + } + + pthread_cond_signal(&queue->something_happened); + } + + entry = entry->next; + } +} + +microtcp_queue_t *microtcp_queue_create(microtcp_t *mtcp) +{ + microtcp_queue_t *queue = malloc(sizeof(microtcp_queue_t)); + if (!queue) + return NULL; + + queue->mtcp = mtcp; + queue->queue_head = NULL; + queue->queue_tail = NULL; + queue->entry_free_list = queue->entry_pool; + queue->event_free_list = queue->event_pool; + + for (size_t i = 0; i < MICROTCP_QUEUE_ENTRIES_MAX-1; i++) { + mtcp->entry_pool[i].mtcp = NULL; + mtcp->entry_pool[i].prev = NULL; + mtcp->entry_pool[i].next = mtcp->entry_pool + i+1; + + mtcp->event_pool[i].mtcp = NULL; + mtcp->event_pool[i].prev = NULL; + mtcp->event_pool[i].next = mtcp->event_pool + i+1; + } + mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].mtcp = NULL; + mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].prev = NULL; + mtcp->entry_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].next = NULL; + + mtcp->event_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].prev = NULL; + mtcp->event_pool[MICROTCP_QUEUE_ENTRIES_MAX-1].next = NULL; + + if (pthread_cond_init(&queue->something_happened, NULL)) { + free(queue); + return NULL; + } + + return queue; +} + +void microtcp_queue_destroy(microtcp_queue_t *queue) +{ + for (size_t i = 0; i < MICROTCP_QUEUE_ENTRIES_MAX; i++) { + if (queue->entries[i].mtcp == NULL) + continue; + microtcp_queue_unregister(queue, queue->entries[i].socket, MICROTCP_EVENT_ALL); + } + free(queue); +} + +bool microtcp_queue_register(microtcp_queue_t *queue, microtcp_socket_t *socket, void *data, int events) +{ + if (!events) + return true; // No events registered + + if (!queue->free_list) + return false; // Registration limit reached + + queue_entry_t *entry = queue->free_list; + queue->free_list = entry->next; + + entry->prev = NULL; + entry->next = NULL; + entry->data = data; + entry->queue = queue; + entry->events = events; + entry->socket = socket; + + entry->next = socket->queue_entry; + if (socket->queue_entry) + socket->queue_entry->prev = entry; + socket->queue_entry = entry; + + return true; +} + +static queue_entry_t *find_entry(microtcp_queue_t *queue, microtcp_socket_t *socket) +{ + queue_entry_t *entry = socket->queue_entry; + while (entry) { + if (entry->queue == queue) + return entry; + entry = entry->next; + } + return NULL; +} + +static void +unregister_socket_from_all_queues(microtcp_socket_t *socket) +{ + while (socket->queue_entry) + microtcp_queue_unregister(socket->queue_entry->queue, socket, MICROTCP_EVENT_ALL); +} + +bool microtcp_queue_unregister(microtcp_queue_t *queue, microtcp_socket_t *socket, int unregister_events) +{ + if (!queue || !socket || queue->mtcp != socket->mtcp) + return false; + + queue_entry_t *entry = find_entry(queue, socket); + if (!entry) + return false; // Socket wasn't registered in this queue + + int remaining_events = entry->events & ~unregister_events; + + if (!remaining_events) { + + // All events were unregistered, so the + // socket must be removed from the queue. + + // Unlink any events related to this socket + // from the queue + queue_event_t *event = entry->event_entry; + if (event) { + + // Remove the event structure from the queue + if (event->prev) + event->prev->next = event->next; + else + queue->queue_head = event->next; + + if (event->next) + event->next->prev = event->prev; + else + queue->queue_tail = event->prev; + + // Push the now unlinked event structure onto + // the free list + event->prev = NULL; + event->next = queue->event_free_list; + queue->event_free_list = event; + } + + // Unlink the entry from the socket's list + if (entry->prev) + entry->prev->next = entry->next; + else + socket->queue_entry = entry->next; + + if (entry->next) + entry->next->prev = entry->prev; + + // Put the entry structure back into the + // free list of the queue + entry->mtcp = NULL; // It's important to set this to NULL because + // this way the queue knows it's not used anymore. + entry->prev = NULL; + entry->next = queue->free_list; + queue->free_list = entry; + + } else + entry->events = remaining_events; + + return true; +} \ No newline at end of file diff --git a/examples/queue.h b/examples/queue.h new file mode 100644 index 0000000..bf2c36c --- /dev/null +++ b/examples/queue.h @@ -0,0 +1,25 @@ + +enum { + MICROTCP_EVENT_NONE = 0, + MICROTCP_EVENT_SEND = 1, + MICROTCP_EVENT_RECV = 2, + MICROTCP_EVENT_ACCEPT = 4, + + MICROTCP_EVENT_ALL = MICROTCP_EVENT_RECV + | MICROTCP_EVENT_SEND + | MICROTCP_EVENT_ACCEPT, +}; + +typedef struct microtcp_queue_t microtcp_queue_t; + +typedef struct { + int type; + void *data; + microtcp_socket_t *socket; +} microtcp_event_t; + +microtcp_queue_t *microtcp_queue_create(microtcp_t *mtcp); +void microtcp_queue_destroy(microtcp_queue_t *queue); +microtcp_event_t microtcp_queue_next(microtcp_queue_t *queue, bool no_block); +bool microtcp_queue_register(microtcp_queue_t *queue, microtcp_socket_t *socket, void *data, int events); +bool microtcp_queue_unregister(microtcp_queue_t *queue, microtcp_socket_t *socket, int unregister_events); diff --git a/examples/queue2.c b/examples/queue2.c new file mode 100644 index 0000000..cb5e4a7 --- /dev/null +++ b/examples/queue2.c @@ -0,0 +1,54 @@ + +typedef struct { + size_t size; + size_t used; + uint8_t *data; +} buffer_t; + +typedef struct client_t client_t; +struct client_t { + client_t *prev; + client_t *next; + microtcp_socket_t *socket; + buffer_t input; + buffer_t output; +}; + +int main(void) +{ + microtcp_t *mtcp = microtcp_create(); + + microtcp_socket_t *server = microtcp_open(mtcp, 80); + + microtcp_queue_t *queue = microtcp_queue_create(mtcp); + microtcp_queue_register(queue, server, NULL, MICROTCP_EVENT_ACCEPT); + while (1) { + microtcp_event_t event = microtcp_queue_next(queue, no_block); + switch (event.type) { + + case MICROTCP_EVENT_NONE: + break; + + case MICROTCP_EVENT_READ: + break; + + case MICROTCP_EVENT_RECV: + { + microtcp_socket_t *client_socket = event.socket; + microtcp_recv(client_socket, ); + } + break; + + case MICROTCP_EVENT_ACCEPT: + { + microtcp_socket_t *client_socket = microtcp_accept(event.socket, true); + microtcp_queue_register(queue, client_socket, NULL, MICROTCP_EVENT_SEND | MICROTCP_EVENT_RECV); + } + break; + } + } + microtcp_queue_destroy(queue); + + microtcp_destroy(mtcp); + return 0; +} \ No newline at end of file diff --git a/src/defs.h b/src/defs.h index f77cde6..a68b828 100644 --- a/src/defs.h +++ b/src/defs.h @@ -15,6 +15,11 @@ static_assert(sizeof(ip_address_t) == 4); #define MAC_ZERO (mac_address_t) {.data = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00}} #define MAC_BROADCAST (mac_address_t) {.data = {0xff, 0xff, 0xff, 0xff, 0xff, 0xff}} +typedef struct { + const void *src; + size_t len; +} slice_list_t; + #define MIN(X, Y) ((X) < (Y) ? (X) : (Y)) #define MAX(X, Y) ((X) > (Y) ? (X) : (Y)) diff --git a/src/ip.c b/src/ip.c index bc24a9d..aa53ce0 100644 --- a/src/ip.c +++ b/src/ip.c @@ -120,11 +120,26 @@ void ip_seconds_passed(ip_state_t *state, size_t seconds) (void) seconds; } -int ip_send(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, bool no_fragm, const void *src, size_t len) +int ip_send(ip_state_t *state, ip_protocol_t protocol, + ip_address_t dst, bool no_fragm, + const void *src, size_t len) { + const slice_list_t slices[] = { + {src, len} + }; + return ip_send_2(state, protocol, dst, no_fragm, slices, 1); +} + +int ip_send_2(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, + bool no_fragm, const slice_list_t *slices, size_t num_slices) +{ + size_t total_len = 0; + for (size_t i = 0; i < num_slices; i++) + total_len += slices[i].len; + size_t managed_payload = 0; - while (managed_payload < len && (managed_payload == 0 || !no_fragm)) { + while (managed_payload < total_len && (managed_payload == 0 || !no_fragm)) { if (state->output_ptr == NULL) { // Lower layers of the network stack didn't specify an output @@ -144,7 +159,7 @@ int ip_send(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, bool no return -1; size_t current_payload_limit = state->output_max - sizeof(ip_packet_t); - size_t remaining_payload = len - managed_payload; + size_t remaining_payload = total_len - managed_payload; size_t considered_payload = MIN(current_payload_limit, remaining_payload); ip_packet_t *packet = state->output_ptr; // This changes every iteration @@ -159,9 +174,15 @@ int ip_send(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, bool no packet->checksum = 0; // Temporary value packet->src_ip = state->ip; packet->dst_ip = dst; - memcpy(packet->payload, - src + managed_payload, - considered_payload); + + size_t copied_bytes = 0; + size_t copied_slices = 0; + while (copied_bytes < considered_payload) { + size_t copying = MIN(slices[copied_slices].len, considered_payload - copied_bytes); + memcpy(packet->payload + copied_bytes, slices[copied_slices].src, copying); + copied_bytes += copying; + copied_slices++; + } packet->checksum = calculate_checksum_ip((uint16_t*) packet, 4 * packet->header_length); diff --git a/src/ip.h b/src/ip.h index d3cfa5e..2b639d2 100644 --- a/src/ip.h +++ b/src/ip.h @@ -61,5 +61,6 @@ void ip_change_output_buffer(ip_state_t *state, void *ptr, size_t max); void ip_init(ip_state_t *state, ip_address_t ip, void *send_data, void (*send)(void*, ip_address_t, size_t)); void ip_free(ip_state_t *state); int ip_send(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, bool no_fragm, const void *src, size_t len); +int ip_send_2(ip_state_t *state, ip_protocol_t protocol, ip_address_t dst, bool no_fragm, const slice_list_t *slices, size_t num_slices); void ip_process_packet(ip_state_t *state, const void *packet, size_t len); bool ip_plug_protocol(ip_state_t *ip_state, uint8_t protocol, void *data, void (*process_packet)(void *data, ip_address_t sender, const void *packet, size_t len)); diff --git a/src/microtcp.c b/src/microtcp.c index 7a2f6cd..ae0d1cf 100644 --- a/src/microtcp.c +++ b/src/microtcp.c @@ -160,11 +160,12 @@ static void send_arp_packet(void *data, mac_address_t dst) mtcp->used_buffer->used = 0; } -static int send_tcp_segment(void *data, ip_address_t dst, - const void *str, size_t len) +static int send_tcp_segment(void *data, ip_address_t ip, + const slice_list_t *slices, + size_t num_slices) { microtcp_t *mtcp = data; - return ip_send(&mtcp->ip_state, IP_PROTOCOL_TCP, dst, true, str, len); + return ip_send_2(&mtcp->ip_state, IP_PROTOCOL_TCP, ip, true, slices, num_slices); } static void move_wait_buffer_to_free_list(buffer_t *buffer) diff --git a/src/tcp.c b/src/tcp.c index 998af39..466f194 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -10,12 +10,6 @@ #define TCP_DEBUG_LOG(...) #endif -static int tcp_send(tcp_state_t *tcp_state, ip_address_t ip, - const void *src, size_t len) -{ - return tcp_state->callbacks.send(tcp_state->callbacks.data, ip, src, len); -} - void tcp_init(tcp_state_t *tcp_state, ip_address_t ip, tcp_callbacks_t callbacks) { tcp_state->ip = ip; @@ -47,6 +41,10 @@ void tcp_seconds_passed(tcp_state_t *state, size_t seconds) (void) seconds; } +typedef enum { + TCP_CONNECTION_ +} tcp_connection_state; + static tcp_connection_t* connection_create_waiting_for_ack(tcp_listener_t *listener, uint32_t seq_no, uint32_t ack_no, @@ -133,7 +131,9 @@ static tcp_connection_t *find_connection(tcp_connection_t *list, ip_address_t pe return NULL; } -static connection_state_t find_connection_associated_to(tcp_listener_t *listener, ip_address_t peer_ip, uint16_t peer_port, tcp_connection_t **connection) +static connection_state_t +find_connection_associated_to(tcp_listener_t *listener, ip_address_t peer_ip, + uint16_t peer_port, tcp_connection_t **connection) { tcp_connection_t *connection2 = find_connection(listener->connections, peer_ip, peer_port); if (connection2) { @@ -175,25 +175,21 @@ typedef struct { } tcp_pseudoheader_t; static uint16_t -calculate_checksum_tcp(const void *a, const void *b, - size_t a_len, size_t b_len) +calculate_checksum(const slice_list_t *slices, size_t num_slices) { - assert((a_len & 1) == 0 - && (b_len & 1) == 0); - - const uint16_t *a2 = a; - const uint16_t *b2 = b; - uint32_t sum = 0xffff; - for (size_t i = 0; i < a_len/2; i++) { - sum += ntohs(a2[i]); - if (sum > 0xffff) - sum -= 0xffff; - } - for (size_t i = 0; i < b_len/2; i++) { - sum += ntohs(b2[i]); - if (sum > 0xffff) - sum -= 0xffff; + + for (size_t slice_idx = 0; slice_idx < num_slices; slice_idx++) { + + assert((slices[slice_idx].len & 1) == 0); + const uint16_t *src = slices[slice_idx].src; + const size_t len = slices[slice_idx].len; + + for (size_t i = 0; i < len/2; i++) { + sum += ntohs(src[i]); + if (sum > 0xffff) + sum -= 0xffff; + } } return htons(~sum); @@ -252,7 +248,7 @@ static void emit_segment(tcp_connection_t *connection, bool ack, bool syn, size_ //if (payload_being_sent > 0) // seq_no++; - connection->out_header = (tcp_segment_t) { + tcp_segment_t header = { .src_port = htons(listener->port), .dst_port = htons(connection->peer_port), .flags = flags, @@ -273,10 +269,17 @@ static void emit_segment(tcp_connection_t *connection, bool ack, bool syn, size_ .tcp_length = htons(total_segment_size), }; - connection->out_header.checksum = calculate_checksum_tcp(&pseudo_header, &connection->out_header, sizeof(pseudo_header), total_segment_size); + header.checksum = calculate_checksum((slice_list_t[]) { + {&pseudo_header, sizeof(tcp_pseudoheader_t)}, + {&header, sizeof(tcp_segment_t)}, + {connection->out_buffer, connection->snd_wnd}, + }, 3); + + int result = state->callbacks.send(state->callbacks.data, connection->peer_ip, (slice_list_t[]) { + {&header, sizeof(tcp_segment_t)}, + {connection->out_buffer, payload_being_sent}, + }, 2); - int result = tcp_send(state, connection->peer_ip, &connection->out_header, total_segment_size); - if (result < 0) { // It wasn't possible to send out bytes. We'll try again later! } else { @@ -449,6 +452,7 @@ tcp_listener_create(tcp_state_t *state, uint16_t port, void *callback_data, listener->port = port; listener->connections = NULL; listener->connections_waiting_for_ack = NULL; + listener->connections_waiting_for_fin = NULL; listener->connections_waiting_for_accept_head = NULL; listener->connections_waiting_for_accept_tail = NULL; listener->callback_data = callback_data; @@ -533,13 +537,18 @@ void tcp_connection_destroy(tcp_connection_t *connection) // NOTE: This can only be called when the // connection was accepted. + // Make sure the connection was first finished + // by being moved from the idle list to the + // waiting-for-fin list. + tcp_connection_finish(connection); + tcp_listener_t *listener = connection->listener; - // Pop connection from the idle connection list + // Pop connection from the waiting-for-fin list if (connection->prev) connection->prev->next = connection->next; else - listener->connections = connection->next; + listener->connections_waiting_for_fin = connection->next; // Push it into the free connection list tcp_state_t *state = listener->state; @@ -583,7 +592,49 @@ append_to_output_buffer(tcp_connection_t *connection, size_t tcp_connection_send(tcp_connection_t *connection, const void *src, size_t len) { - size_t num = append_to_output_buffer(connection, src, len); - emit_segment(connection, false, false, SIZE_MAX); + size_t num; + if (connection->read_only) + num = 0; + else { + num = append_to_output_buffer(connection, src, len); + emit_segment(connection, false, false, SIZE_MAX); + } return num; } + +void tcp_connection_finish(tcp_connection_t *connection) +{ + if (!connection->read_only) { + + // Move connection from idle list to + // waiting-for-fin list + + tcp_listener_t *listener = connection->listener; + + // Pop it from the idle list + { + if (connection->prev) + connection->prev->next = connection->next; + else + listener->connections = connection->next; + + if (connection->next) + connection->next->prev = connection->prev; + + connection->prev = NULL; + connection->next = NULL; + } + + #warning "The FIN segment should be sent here" + + // Push it to the waiting-for-fin list + { + connection->prev = NULL; + connection->next = listener->connections_waiting_for_fin; + listener->connections_waiting_for_fin = connection; + } + + // Now mark the connection as read-only + connection->read_only = true; + } +} \ No newline at end of file diff --git a/src/tcp.h b/src/tcp.h index 24f1e2e..59920c9 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -47,6 +47,8 @@ struct tcp_listener_t { tcp_connection_t *connections_waiting_for_ack; tcp_connection_t *connections_waiting_for_accept_head; tcp_connection_t *connections_waiting_for_accept_tail; + tcp_connection_t *connections_waiting_for_fin; + void (*callback_ready_to_accept)(void*); void *callback_data; }; @@ -60,6 +62,8 @@ struct tcp_connection_t { void (*callback_ready_to_recv)(void*); void (*callback_ready_to_send)(void*); + bool read_only; + ip_address_t peer_ip; // Network byte order uint16_t peer_port; // CPU byte order @@ -92,16 +96,14 @@ struct tcp_connection_t { uint32_t snd_una; // SND.UNA from RFC 793 // It's the sequence number of the last // byte sent and acknowledged by the peer. - - tcp_segment_t out_header; // There must be no padding between - char out_buffer[TCP_OUTPUT_BUFFER_SIZE]; // these two - char in_buffer[TCP_INPUT_BUFFER_SIZE]; + + char out_buffer[TCP_OUTPUT_BUFFER_SIZE]; + char in_buffer[TCP_INPUT_BUFFER_SIZE]; }; -static_assert(offsetof(tcp_connection_t, out_buffer) == offsetof(tcp_connection_t, out_header) + sizeof(tcp_segment_t)); typedef struct { void *data; - int (*send)(void *data, ip_address_t ip, const void *src, size_t len); + int (*send)(void *data, ip_address_t ip, const slice_list_t *slices, size_t num_slices); } tcp_callbacks_t; struct tcp_state_t { @@ -123,5 +125,6 @@ tcp_listener_t *tcp_listener_create(tcp_state_t *state, uint16_t port, void *d void tcp_listener_destroy(tcp_listener_t *listener); tcp_connection_t *tcp_listener_accept(tcp_listener_t *listener, void *callback_data, void (*callback_ready_to_recv)(void*), void (*callback_ready_to_send)(void*)); void tcp_connection_destroy(tcp_connection_t *connection); +void tcp_connection_finish(tcp_connection_t *connection); size_t tcp_connection_recv(tcp_connection_t *connection, void *dst, size_t len); size_t tcp_connection_send(tcp_connection_t *connection, const void *src, size_t len);