diff --git a/src/base/all.h b/src/base/all.h index d673d99..b2f8500 100644 --- a/src/base/all.h +++ b/src/base/all.h @@ -442,7 +442,25 @@ union Range1f32 DWORD input_mode; DWORD output_mode; } TermIOs; + // networking shim for windows +# define POLLIN 0x0001 +# define POLLPRI 0x0002 +# define POLLOUT 0x0004 +# define POLLERR 0x0008 +# define POLLHUP 0x0010 +# define POLLNVAL 0x0020 + + typedef struct pollfd { + SOCKET fd; + short events; + short revents; + } pollfd_t; + + typedef int nfds_t; + + int poll(struct pollfd *fds, nfds_t nfds, int timeout); #else +# include # include # include # include diff --git a/src/base/win32_os.c b/src/base/win32_os.c index 35b09ea..09398cc 100644 --- a/src/base/win32_os.c +++ b/src/base/win32_os.c @@ -326,5 +326,68 @@ bool osThreadJoin(Thread handle, u64 endt_us) { return true; } - fn void osBarrierWait(Barrier barrier) { - } +fn void osBarrierWait(Barrier barrier) { +} + +// poll() shim using select() on windows +int poll(struct pollfd *fds, nfds_t nfds, int timeout) { + assert(nfds <= FD_SETSIZE); + + fd_set read_fds, write_fds, except_fds; + struct timeval tv; + int ret, i; + SOCKET max_fd = 0; + + if (!fds && nfds > 0) { + WSASetLastError(WSAEFAULT); + return -1; + } + + FD_ZERO(&read_fds); + FD_ZERO(&write_fds); + FD_ZERO(&except_fds); + + for (i = 0; i < nfds; i++) { + fds[i].revents = 0; + + if (fds[i].fd == INVALID_SOCKET) { + fds[i].revents |= POLLNVAL; + continue; + } + + if (fds[i].fd > max_fd) + max_fd = fds[i].fd; + + if (fds[i].events & POLLIN) + FD_SET(fds[i].fd, &read_fds); + if (fds[i].events & POLLOUT) + FD_SET(fds[i].fd, &write_fds); + /* always watch for errors */ + FD_SET(fds[i].fd, &except_fds); + } + + if (timeout >= 0) { + tv.tv_sec = timeout / 1000; + tv.tv_usec = (timeout % 1000) * 1000; + } + + ret = select((int)(max_fd + 1), &read_fds, &write_fds, &except_fds, &tv); + if (ret == SOCKET_ERROR) + return -1; + if (ret == 0) + return 0; + + ret = 0; + for (i = 0; i < nfds; i++) { + if (fds[i].fd == INVALID_SOCKET) + continue; + + if (FD_ISSET(fds[i].fd, &read_fds)) fds[i].revents |= POLLIN; + if (FD_ISSET(fds[i].fd, &write_fds)) fds[i].revents |= POLLOUT; + if (FD_ISSET(fds[i].fd, &except_fds)) fds[i].revents |= POLLERR; + + if (fds[i].revents) ret++; + } + + return ret; +} diff --git a/src/client.c b/src/client.c index 73c48fd..db272fb 100644 --- a/src/client.c +++ b/src/client.c @@ -451,6 +451,8 @@ fn void handleIncomingMessage(u8* message, i32 len) { fn void* receiveNetworkUpdates(void* net_client) { TCPClient client = *(TCPClient*)net_client; dbg("receiveNetworkUpdates() sock=%d\n", client.socket); + // TODO: handle the server DCing with out a segfault here. + // should try to reconnect with some kind of backoff infiniteReadTCPClient(client.socket, &should_quit, handleIncomingMessage, addSystemMessage); return NULL; } diff --git a/src/lib/network.c b/src/lib/network.c index 52830be..9a7e63a 100644 --- a/src/lib/network.c +++ b/src/lib/network.c @@ -1,5 +1,4 @@ #include "../base/all.h" -#include // https://stackoverflow.com/questions/1098897/what-is-the-largest-safe-udp-packet-size-on-the-internet #define UDP_MAX_MESSAGE_LEN 508 @@ -284,6 +283,7 @@ i32 recvMessage(i32 socket, u8* message_buffer, void (*addSystemMessage)(u8* msg void infiniteReadTCPServer( TCPServer* server, + bool* should_quit, void (*handleMessage)(u8* udp_message, u32 udp_len, SocketAddress sending_address, i32 socket), void (*closeConnection)(i32 socket_fd), void (*addSystemMessage)(u8* msg) @@ -300,12 +300,11 @@ void infiniteReadTCPServer( u32 pollable_fd_count = 1; i32 poll_event_count; - bool should_keep_serving = true; - while (should_keep_serving) { - poll_event_count = poll(pollable_fds, pollable_fd_count, -1); + while (*should_quit == false) { + poll_event_count = poll(pollable_fds, pollable_fd_count, 3000); // times out after 3 seconds so that quitting the server will actually quit the server process in relatively short order. if (poll_event_count == -1) { // TODO handle error better - should_keep_serving = false; + *should_quit = true; continue; } for(u32 i = 0; i < pollable_fd_count; i++) { diff --git a/src/server.c b/src/server.c index 81f96c1..e781789 100644 --- a/src/server.c +++ b/src/server.c @@ -80,6 +80,7 @@ typedef struct Account { typedef struct Client { bool active; + bool needs_systems_init; i32 socket_fd; u64 account_id; SocketAddress address; @@ -424,7 +425,7 @@ fn void sendMessageAuctionBidResult(i32 socket_fd, SocketAddress addr, AuctionBi fn void handleIncomingMessage(u8* message, u32 len, SocketAddress sender, i32 socket) { char sbuf[SBUFLEN] = {0}; - sprintf(sbuf, "%d: %s from %s:%d\n", len, command_type_strings[message[0]], inet_ntoa(sender.sin_addr), sender.sin_port); + sprintf(sbuf, "%d: %s from %s:%d\n", len, COMMAND_TYPE_STRINGS[message[0]], inet_ntoa(sender.sin_addr), sender.sin_port); addSystemMessage((u8*)sbuf); u32 msg_idx = 0; ParsedClientCommand parsed = { @@ -474,7 +475,7 @@ fn void handleIncomingMessage(u8* message, u32 len, SocketAddress sender, i32 so msg_idx += temp_str.length; MemoryZero(sbuf, SBUFLEN); - sprintf(sbuf, "Logging in player: %s %d %s\n", command_type_strings[parsed.type], name_len, message + 7); + sprintf(sbuf, "Logging in player: %s %d %s\n", COMMAND_TYPE_STRINGS[parsed.type], name_len, message + 7); addSystemMessage((u8*)sbuf); } break; case CommandPayMortgage: { @@ -530,7 +531,7 @@ fn void* receiveNetworkUpdates(void* tcp_server) { char sbuf[SBUFLEN] = {0}; sprintf(sbuf, "receiveNetworkUpdates() sock=%d\n", server.socket_fd); addSystemMessage((u8*)sbuf); - infiniteReadTCPServer(&server, handleIncomingMessage, removeClientBySocketFd, addSystemMessage); + infiniteReadTCPServer(&server, &should_quit, handleIncomingMessage, removeClientBySocketFd, addSystemMessage); return NULL; } @@ -566,9 +567,6 @@ fn void* sendNetworkUpdates(void* tcp_server) { } } - NetworkMessage base_sys_msg = makeMessageSystemCommodities(&state.map[send_loop/2 % STAR_SYSTEM_COUNT]); - NetworkMessage base_pass_msg = makeMessageSystemPassengers(&state.map[send_loop/2 % STAR_SYSTEM_COUNT]); - lockMutex(&state.client_mutex); { // WARNING the `i` starts at 1 here because state.clients.items[0] is a "null" Client for (u32 i = 1; i < state.clients.length; i++) { @@ -578,14 +576,6 @@ fn void* sendNetworkUpdates(void* tcp_server) { } if (send_loop % 2 == 0) { - // every other "send-frame" we send each connnected client the current prices for a different system - // so that the prices mostly stay up to date pretty quickly without having to track changes - base_sys_msg.socket_fd = client.socket_fd; - sendTCPMessage(base_sys_msg); - // and the passenger offers - base_pass_msg.socket_fd = client.socket_fd; - sendTCPMessage(base_pass_msg); - // send the client the current auction details for their current system Account* account = &state.accounts[client.account_id]; if (!accountIsEmpty(account)) { @@ -610,7 +600,7 @@ fn void* sendNetworkUpdates(void* tcp_server) { // update all the changed systems NetworkMessage msg_data; for (u32 ii = 0; ii < STAR_SYSTEM_COUNT; ii++) { - if (state.map[ii].changed == true) { + if (state.map[ii].changed == true || client.needs_systems_init) { // send commodities msg_data = makeMessageSystemCommodities(&state.map[ii]); msg_data.address = client.address; @@ -639,6 +629,8 @@ fn void* sendNetworkUpdates(void* tcp_server) { sendTCPMessage(msg_data); } } + + state.clients.items[i].needs_systems_init = false; } } unlockMutex(&state.client_mutex); @@ -783,12 +775,13 @@ fn void* gameLoop(void* params) { if (client_handle == 0) break; Account* account = &state.accounts[client->account_id]; account->destination_sys_idx = msg.byte; + account->changed = true; } break; case CommandReadyStatus: { if (client_handle == 0) break; Account* account = &state.accounts[client->account_id]; account->ship.ready_to_depart = msg.byte; - sendMessagePlayerDetails(client->socket_fd, account->ship, sender); + account->changed = true; state.all_accounts_ready = true; for (u32 i = 0; i < ACCOUNT_LEN; i++) { if (!accountIsEmpty(&state.accounts[i])) { @@ -852,6 +845,7 @@ fn void* gameLoop(void* params) { } } sys->changed = true; + account->changed = true; sendMessagePlayerDetails(client->socket_fd, account->ship, sender); sendMessageTransactionResult(client->socket_fd, sender, qty_traded, is_buying_from_system, credit_value); } break; @@ -933,6 +927,7 @@ fn void* gameLoop(void* params) { outgoingMessageQueuePush(state.network_send_queue, &outgoing_message); addSystemMessage((u8*)"MessageNewAccountCreated sent\n"); } + client->needs_systems_init = true; MemoryZero(sbuf, SBUFLEN); sprintf(sbuf, "client_handle=%d, acct_id=%d\n", client_handle, existing_account->id); addSystemMessage((u8*)sbuf); diff --git a/src/shared.h b/src/shared.h index 34b35f7..92e19f0 100644 --- a/src/shared.h +++ b/src/shared.h @@ -378,10 +378,11 @@ typedef enum CommandType { CommandType_Count, } CommandType; -static const char* command_type_strings[CommandType_Count] = { +static const char* COMMAND_TYPE_STRINGS[CommandType_Count] = { "Invalid", "Login", "CreateCharacter", + "Transact", "ReadyStatus", "SetDestination", "PayMortgage",