v4games
/
v4k-git-backup
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor netsync polling

main
Dominik Madarász 2023-08-21 18:07:45 +02:00
parent 4969ab42d9
commit 27b580dbd2
8 changed files with 457 additions and 521 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
demos/99-syncdemo.c
View File

@ -1,6 +1,6 @@
#include "v4k.h" #include "v4k.h"
enum { MAX_NPCS = 5 }; enum { MAX_NPCS = 5, MAX_CLIENTS = 5 };
struct player_t { struct player_t {
uint64_t seen_until; uint64_t seen_until;
float x,y,z,angle; float x,y,z,angle;
@ -47,7 +47,7 @@ int main() {
// ifdef(win32, FreeConsole()); // tty_detach() // ifdef(win32, FreeConsole()); // tty_detach()
// network setup // network setup
network_create("127.0.0.1", 0, flag("--client") ? NETWORK_CONNECT : 0); network_create(MAX_CLIENTS, "127.0.0.1", 0, flag("--client") ? NETWORK_CONNECT : 0);
int64_t self_id = network_get(NETWORK_RANK); int64_t self_id = network_get(NETWORK_RANK);
bind_netbuffers(self_id); bind_netbuffers(self_id);
@ -71,7 +71,7 @@ int main() {
self_id = network_get(NETWORK_RANK); self_id = network_get(NETWORK_RANK);
if (network_get(NETWORK_LIVE) == 0) { if (network_get(NETWORK_LIVE) == 0) {
network_create("127.0.0.1", 0, flag("--client") ? NETWORK_CONNECT|NETWORK_NOFAIL : 0); network_create(MAX_CLIENTS, "127.0.0.1", 0, flag("--client") ? NETWORK_CONNECT|NETWORK_NOFAIL : 0);
self_id = network_get(NETWORK_RANK); self_id = network_get(NETWORK_RANK);
if (self_id != -1) { if (self_id != -1) {
bind_netbuffers(self_id); bind_netbuffers(self_id);

8
engine/bind/v4k.lua
View File

@ -2100,8 +2100,7 @@ GAMEPAD_GUID, GAMEPAD_NAME,
int tcp_close(int); int tcp_close(int);
int tcp_debug(int); int tcp_debug(int);
enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 }; enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 };
enum { MAX_CLIENTS = 32 }; void network_create(unsigned max_clients, const char *ip, const char *port, unsigned flags);
void network_create(const char *ip, const char *port, unsigned flags);
enum { NETWORK_SEND = 2, NETWORK_RECV = 4 }; enum { NETWORK_SEND = 2, NETWORK_RECV = 4 };
enum { NETWORK_UNRELIABLE = 8, NETWORK_UNORDERED = 16 }; enum { NETWORK_UNRELIABLE = 8, NETWORK_UNORDERED = 16 };
void* network_buffer(void *ptr, unsigned sz, uint64_t flags, int64_t rank); void* network_buffer(void *ptr, unsigned sz, uint64_t flags, int64_t rank);
@ -2119,14 +2118,15 @@ enum { NETWORK_RANK = 0 };
enum { NETWORK_PING = 1 }; enum { NETWORK_PING = 1 };
enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE }; enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
enum { NETWORK_SEND_MS = 4 }; enum { NETWORK_SEND_MS = 4 };
enum { NETWORK_USERID = 5, NETWORK_COUNT }; enum { NETWORK_USERID = 5, NETWORK_COUNT , NETWORK_CAPACITY };
int64_t network_get(uint64_t key); int64_t network_get(uint64_t key);
int64_t network_put(uint64_t key, int64_t value); int64_t network_put(uint64_t key, int64_t value);
void network_rpc(const char *signature, void *function); void network_rpc(const char *signature, void *function);
void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline); void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline);
void network_rpc_send(unsigned id, const char *cmdline); void network_rpc_send(unsigned id, const char *cmdline);
bool server_bind(int max_clients, int port); bool server_bind(int max_clients, int port);
void server_poll(); void server_poll(unsigned timeout_ms);
void client_poll(unsigned timeout_ms);
void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags); void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags);
void server_broadcast_bin(const void *ptr, int len); void server_broadcast_bin(const void *ptr, int len);
void server_broadcast_flags(const char *msg, uint64_t flags); void server_broadcast_flags(const char *msg, uint64_t flags);

280
engine/joint/v4k.h
View File

@ -15916,9 +15916,9 @@ API int tcp_debug(int); // toggle traffic monitoring on/off for given socket
//enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation //enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation
//enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where //enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where
enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 }; enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 };
enum { MAX_CLIENTS = 32 }; API void network_create(unsigned max_clients, const char *ip, const char *port, unsigned flags); // both ip and port can be null
API void network_create(const char *ip, const char *port, unsigned flags); // both ip and port can be null
//enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes //enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes
//enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when //enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when
@ -15948,7 +15948,7 @@ enum { NETWORK_RANK = 0 }; // [0..N] where 0 is server
enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY }; enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY };
enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE }; enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
enum { NETWORK_SEND_MS = 4 }; enum { NETWORK_SEND_MS = 4 };
enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/ }; enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/, NETWORK_CAPACITY };
API int64_t network_get(uint64_t key); API int64_t network_get(uint64_t key);
API int64_t network_put(uint64_t key, int64_t value); API int64_t network_put(uint64_t key, int64_t value);
@ -15960,7 +15960,8 @@ API void network_rpc_send(unsigned id, const char *cmdline);
// low-level api (sockets based) // low-level api (sockets based)
API bool server_bind(int max_clients, int port); API bool server_bind(int max_clients, int port);
API void server_poll(); API void server_poll(unsigned timeout_ms);
API void client_poll(unsigned timeout_ms);
API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags); API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags);
API void server_broadcast_bin(const void *ptr, int len); API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast_flags(const char *msg, uint64_t flags); API void server_broadcast_flags(const char *msg, uint64_t flags);
@ -338615,133 +338616,6 @@ void server_drop_client_peer(ENetPeer *peer) {
map_erase(clients, peer); map_erase(clients, peer);
} }
void server_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
PRINTF( "A new client connected from ::%s:%u.\n", ip, event.peer->address.port );
/* Store any relevant client information here. */
event.peer->data = "Client information";
int64_t client_id = next_client_id++;
map_find_or_add(clients, event.peer, client_id);
map_find_or_add(peers, client_id, event.peer);
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT: {
uint64_t *cid = map_find(clients, event.peer);
if (cid) {
char init_msg[12];
*(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
} else {
PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void client_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT:
/* handled during client_join */
break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) { void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) {
ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0); ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0);
enet_host_broadcast(Server, 0, packet); enet_host_broadcast(Server, 0, packet);
@ -338862,7 +338736,7 @@ static double msg_send_cooldown = 0.0;
static double network_dt = 0.0; static double network_dt = 0.0;
static double last_netsync = 0.0; static double last_netsync = 0.0;
void network_create(const char *ip, const char *port_, unsigned flags) { void network_create(unsigned max_clients, const char *ip, const char *port_, unsigned flags) {
if (buffers) map_clear(buffers); if (buffers) map_clear(buffers);
do_once { do_once {
array_resize(values, 128); array_resize(values, 128);
@ -338878,12 +338752,13 @@ void network_create(const char *ip, const char *port_, unsigned flags) {
network_put(NETWORK_PORT, port); network_put(NETWORK_PORT, port);
network_put(NETWORK_LIVE, -1); network_put(NETWORK_LIVE, -1);
network_put(NETWORK_COUNT, 0); network_put(NETWORK_COUNT, 0);
network_put(NETWORK_CAPACITY, max_clients);
if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) { if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) {
// server, else client // server, else client
PRINTF("Trying to bind server, else we connect as a client...\n"); PRINTF("Trying to bind server, else we connect as a client...\n");
network_put(NETWORK_RANK, 0); network_put(NETWORK_RANK, 0);
if( server_bind(MAX_CLIENTS, port) ) { if( server_bind(max_clients, port) ) {
network_put(NETWORK_LIVE, 1); network_put(NETWORK_LIVE, 1);
PRINTF("Server bound\n"); PRINTF("Server bound\n");
} else { } else {
@ -338994,6 +338869,20 @@ char** network_sync(unsigned timeout_ms) {
msg_send_cooldown -= network_dt; msg_send_cooldown -= network_dt;
} }
if (is_server) {
server_poll(timeout_ms);
} else {
client_poll(timeout_ms);
}
array_push(events, NULL);
return events;
}
void server_poll(unsigned timeout_ms) {
if(timeout_ms < 2) timeout_ms = 2;
// network poll // network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) { for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0; char *msg = 0;
@ -339009,7 +338898,7 @@ char** network_sync(unsigned timeout_ms) {
event.peer->data = STRDUP(ip); /* TEMP */ event.peer->data = STRDUP(ip); /* TEMP */
/* ensure we have free slot for client */ /* ensure we have free slot for client */
if (map_count(clients) >= MAX_CLIENTS) { if (map_count(clients) >= network_get(NETWORK_CAPACITY)) {
msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port)); msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port));
enet_peer_disconnect_now(event.peer, 1); enet_peer_disconnect_now(event.peer, 1);
break; break;
@ -339044,7 +338933,6 @@ char** network_sync(unsigned timeout_ms) {
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
if (is_server) {
uint64_t *cid = map_find(clients, event.peer); uint64_t *cid = map_find(clients, event.peer);
if (cid) { if (cid) {
char init_msg[12]; char init_msg[12];
@ -339056,7 +338944,6 @@ char** network_sync(unsigned timeout_ms) {
} else { } else {
PRINTF("!Ignoring unk MSG_INIT client packet.\n"); PRINTF("!Ignoring unk MSG_INIT client packet.\n");
} }
}
break; break;
case MSG_BUF: { case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0); uint64_t *flags = (uint64_t*)(ptr + 0);
@ -339067,15 +338954,11 @@ char** network_sync(unsigned timeout_ms) {
ptr += 24; ptr += 24;
// validate if peer owns the buffer // validate if peer owns the buffer
uint8_t client_valid = 0;
if (is_server) {
int64_t *cid = map_find(clients, event.peer); int64_t *cid = map_find(clients, event.peer);
client_valid = cid ? *cid == *who : 0; uint8_t client_valid = cid ? *cid == *who : 0;
}
// apply incoming packet. // apply incoming packet.
if( is_client ? *who != whoami : client_valid ) { // clients merge always foreign packets. servers merge foreign packets. if( client_valid ) {
array(netbuffer_t) *list = map_find(buffers, *who); array(netbuffer_t) *list = map_find(buffers, *who);
assert( list ); assert( list );
assert( *idx < array_count(*list) ); assert( *idx < array_count(*list) );
@ -339114,30 +338997,127 @@ char** network_sync(unsigned timeout_ms) {
/* Reset the peer's client information. */ /* Reset the peer's client information. */
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT: case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data)); msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
} }
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg)); if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
} }
}
array_push(events, NULL); void client_poll(unsigned timeout_ms) {
return events; int64_t whoami = network_get(NETWORK_RANK);
if(timeout_ms < 2) timeout_ms = 2;
// network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
switch (event.type) {
default: // case ENET_EVENT_TYPE_NONE:
break;
case ENET_EVENT_TYPE_CONNECT:
/* @TODO: move stuff from client_join here */
break;
case ENET_EVENT_TYPE_RECEIVE:
/*
msg = stringf( "A packet of length %u containing %s was received from %s on channel %u",
(unsigned)event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
*/
char *dbg = (char *)event.peer->data;
char *ptr = (char *)event.packet->data;
unsigned sz = (unsigned)event.packet->dataLength;
unsigned id = (unsigned)event.channelID;
// debug
// puts(dbg);
// hexdump(ptr, sz);
// decapsulate incoming packet.
uint32_t mid = *(uint32_t*)(ptr + 0);
ptr += 4;
switch (mid) {
case MSG_INIT:
/* handled by client_join */
break;
case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0);
uint32_t *idx = (uint32_t*)(ptr + 8);
uint32_t *len = (uint32_t*)(ptr + 12);
uint64_t *who = (uint64_t*)(ptr + 16);
ptr += 24;
// apply incoming packet.
if( *who != whoami ) {
array(netbuffer_t) *list = map_find(buffers, *who);
assert( list );
assert( *idx < array_count(*list) );
netbuffer_t *nb = &(*list)[*idx];
assert( *len == nb->sz );
memcpy(nb->ptr, ptr, *len);
}
} break;
case MSG_RPC: {
event.type = NETWORK_EVENT_RPC;
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 6);
*(uint32_t*)&resp_msg[0] = MSG_RPC_RESP;
memcpy(&resp_msg[4], resp, strlen(resp)+1);
ENetPacket *packet = enet_packet_create(resp_msg, strlen(resp) + 5, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
msg = stringf("%d %s", 0, va("req:%s res:%s", cmdline, resp));
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
event.type = NETWORK_EVENT_RPC_RESP;
msg = stringf("%d %s", 0, va("%s", ptr));
} break;
default:
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port);
break;
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
msg = stringf( "%d %s", 0, va("%s disconnected", (char *)event.peer->data));
/* Reset the peer's client information. */
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
}
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
}
} }
int network_event(const char *msg, int *errcode, char **errstr) { int network_event(const char *msg, int *errcode, char **errstr) {

271
engine/split/v4k_netsync.c
View File

@ -154,133 +154,6 @@ void server_drop_client_peer(ENetPeer *peer) {
map_erase(clients, peer); map_erase(clients, peer);
} }
void server_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
PRINTF( "A new client connected from ::%s:%u.\n", ip, event.peer->address.port );
/* Store any relevant client information here. */
event.peer->data = "Client information";
int64_t client_id = next_client_id++;
map_find_or_add(clients, event.peer, client_id);
map_find_or_add(peers, client_id, event.peer);
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT: {
uint64_t *cid = map_find(clients, event.peer);
if (cid) {
char init_msg[12];
*(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
} else {
PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void client_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT:
/* handled during client_join */
break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) { void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) {
ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0); ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0);
enet_host_broadcast(Server, 0, packet); enet_host_broadcast(Server, 0, packet);
@ -401,7 +274,7 @@ static double msg_send_cooldown = 0.0;
static double network_dt = 0.0; static double network_dt = 0.0;
static double last_netsync = 0.0; static double last_netsync = 0.0;
void network_create(const char *ip, const char *port_, unsigned flags) { void network_create(unsigned max_clients, const char *ip, const char *port_, unsigned flags) {
if (buffers) map_clear(buffers); if (buffers) map_clear(buffers);
do_once { do_once {
array_resize(values, 128); array_resize(values, 128);
@ -417,12 +290,13 @@ void network_create(const char *ip, const char *port_, unsigned flags) {
network_put(NETWORK_PORT, port); network_put(NETWORK_PORT, port);
network_put(NETWORK_LIVE, -1); network_put(NETWORK_LIVE, -1);
network_put(NETWORK_COUNT, 0); network_put(NETWORK_COUNT, 0);
network_put(NETWORK_CAPACITY, max_clients);
if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) { if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) {
// server, else client // server, else client
PRINTF("Trying to bind server, else we connect as a client...\n"); PRINTF("Trying to bind server, else we connect as a client...\n");
network_put(NETWORK_RANK, 0); network_put(NETWORK_RANK, 0);
if( server_bind(MAX_CLIENTS, port) ) { if( server_bind(max_clients, port) ) {
network_put(NETWORK_LIVE, 1); network_put(NETWORK_LIVE, 1);
PRINTF("Server bound\n"); PRINTF("Server bound\n");
} else { } else {
@ -533,6 +407,20 @@ char** network_sync(unsigned timeout_ms) {
msg_send_cooldown -= network_dt; msg_send_cooldown -= network_dt;
} }
if (is_server) {
server_poll(timeout_ms);
} else {
client_poll(timeout_ms);
}
array_push(events, NULL);
return events;
}
void server_poll(unsigned timeout_ms) {
if(timeout_ms < 2) timeout_ms = 2;
// network poll // network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) { for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0; char *msg = 0;
@ -548,7 +436,7 @@ char** network_sync(unsigned timeout_ms) {
event.peer->data = STRDUP(ip); /* TEMP */ event.peer->data = STRDUP(ip); /* TEMP */
/* ensure we have free slot for client */ /* ensure we have free slot for client */
if (map_count(clients) >= MAX_CLIENTS) { if (map_count(clients) >= network_get(NETWORK_CAPACITY)) {
msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port)); msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port));
enet_peer_disconnect_now(event.peer, 1); enet_peer_disconnect_now(event.peer, 1);
break; break;
@ -583,7 +471,6 @@ char** network_sync(unsigned timeout_ms) {
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
if (is_server) {
uint64_t *cid = map_find(clients, event.peer); uint64_t *cid = map_find(clients, event.peer);
if (cid) { if (cid) {
char init_msg[12]; char init_msg[12];
@ -595,7 +482,6 @@ char** network_sync(unsigned timeout_ms) {
} else { } else {
PRINTF("!Ignoring unk MSG_INIT client packet.\n"); PRINTF("!Ignoring unk MSG_INIT client packet.\n");
} }
}
break; break;
case MSG_BUF: { case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0); uint64_t *flags = (uint64_t*)(ptr + 0);
@ -606,15 +492,11 @@ char** network_sync(unsigned timeout_ms) {
ptr += 24; ptr += 24;
// validate if peer owns the buffer // validate if peer owns the buffer
uint8_t client_valid = 0;
if (is_server) {
int64_t *cid = map_find(clients, event.peer); int64_t *cid = map_find(clients, event.peer);
client_valid = cid ? *cid == *who : 0; uint8_t client_valid = cid ? *cid == *who : 0;
}
// apply incoming packet. // apply incoming packet.
if( is_client ? *who != whoami : client_valid ) { // clients merge always foreign packets. servers merge foreign packets. if( client_valid ) {
array(netbuffer_t) *list = map_find(buffers, *who); array(netbuffer_t) *list = map_find(buffers, *who);
assert( list ); assert( list );
assert( *idx < array_count(*list) ); assert( *idx < array_count(*list) );
@ -653,30 +535,127 @@ char** network_sync(unsigned timeout_ms) {
/* Reset the peer's client information. */ /* Reset the peer's client information. */
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT: case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data)); msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
} }
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg)); if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
} }
}
array_push(events, NULL); void client_poll(unsigned timeout_ms) {
return events; int64_t whoami = network_get(NETWORK_RANK);
if(timeout_ms < 2) timeout_ms = 2;
// network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
switch (event.type) {
default: // case ENET_EVENT_TYPE_NONE:
break;
case ENET_EVENT_TYPE_CONNECT:
/* @TODO: move stuff from client_join here */
break;
case ENET_EVENT_TYPE_RECEIVE:
/*
msg = stringf( "A packet of length %u containing %s was received from %s on channel %u",
(unsigned)event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
*/
char *dbg = (char *)event.peer->data;
char *ptr = (char *)event.packet->data;
unsigned sz = (unsigned)event.packet->dataLength;
unsigned id = (unsigned)event.channelID;
// debug
// puts(dbg);
// hexdump(ptr, sz);
// decapsulate incoming packet.
uint32_t mid = *(uint32_t*)(ptr + 0);
ptr += 4;
switch (mid) {
case MSG_INIT:
/* handled by client_join */
break;
case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0);
uint32_t *idx = (uint32_t*)(ptr + 8);
uint32_t *len = (uint32_t*)(ptr + 12);
uint64_t *who = (uint64_t*)(ptr + 16);
ptr += 24;
// apply incoming packet.
if( *who != whoami ) {
array(netbuffer_t) *list = map_find(buffers, *who);
assert( list );
assert( *idx < array_count(*list) );
netbuffer_t *nb = &(*list)[*idx];
assert( *len == nb->sz );
memcpy(nb->ptr, ptr, *len);
}
} break;
case MSG_RPC: {
event.type = NETWORK_EVENT_RPC;
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 6);
*(uint32_t*)&resp_msg[0] = MSG_RPC_RESP;
memcpy(&resp_msg[4], resp, strlen(resp)+1);
ENetPacket *packet = enet_packet_create(resp_msg, strlen(resp) + 5, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
msg = stringf("%d %s", 0, va("req:%s res:%s", cmdline, resp));
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
event.type = NETWORK_EVENT_RPC_RESP;
msg = stringf("%d %s", 0, va("%s", ptr));
} break;
default:
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port);
break;
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
msg = stringf( "%d %s", 0, va("%s disconnected", (char *)event.peer->data));
/* Reset the peer's client information. */
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
}
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
}
} }
int network_event(const char *msg, int *errcode, char **errstr) { int network_event(const char *msg, int *errcode, char **errstr) {

9
engine/split/v4k_netsync.h
View File

@ -13,9 +13,9 @@
//enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation //enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation
//enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where //enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where
enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 }; enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 };
enum { MAX_CLIENTS = 32 }; API void network_create(unsigned max_clients, const char *ip, const char *port, unsigned flags); // both ip and port can be null
API void network_create(const char *ip, const char *port, unsigned flags); // both ip and port can be null
//enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes //enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes
//enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when //enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when
@ -45,7 +45,7 @@ enum { NETWORK_RANK = 0 }; // [0..N] where 0 is server
enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY }; enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY };
enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE }; enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
enum { NETWORK_SEND_MS = 4 }; enum { NETWORK_SEND_MS = 4 };
enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/ }; enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/, NETWORK_CAPACITY };
API int64_t network_get(uint64_t key); API int64_t network_get(uint64_t key);
API int64_t network_put(uint64_t key, int64_t value); API int64_t network_put(uint64_t key, int64_t value);
@ -57,7 +57,8 @@ API void network_rpc_send(unsigned id, const char *cmdline);
// low-level api (sockets based) // low-level api (sockets based)
API bool server_bind(int max_clients, int port); API bool server_bind(int max_clients, int port);
API void server_poll(); API void server_poll(unsigned timeout_ms);
API void client_poll(unsigned timeout_ms);
API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags); API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags);
API void server_broadcast_bin(const void *ptr, int len); API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast_flags(const char *msg, uint64_t flags); API void server_broadcast_flags(const char *msg, uint64_t flags);

271
engine/v4k.c
View File

@ -9631,133 +9631,6 @@ void server_drop_client_peer(ENetPeer *peer) {
map_erase(clients, peer); map_erase(clients, peer);
} }
void server_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
PRINTF( "A new client connected from ::%s:%u.\n", ip, event.peer->address.port );
/* Store any relevant client information here. */
event.peer->data = "Client information";
int64_t client_id = next_client_id++;
map_find_or_add(clients, event.peer, client_id);
map_find_or_add(peers, client_id, event.peer);
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT: {
uint64_t *cid = map_find(clients, event.peer);
if (cid) {
char init_msg[12];
*(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
} else {
PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void client_poll() {
ENetEvent event;
while( enet_host_service(Server, &event, 2 /*timeout,ms*/) > 0 ) {
switch (event.type) {
case ENET_EVENT_TYPE_CONNECT:;
break;
case ENET_EVENT_TYPE_RECEIVE:
PRINTF( "A packet of length %zu containing %s was received from %s on channel %u.\n",
event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
char *dbg = (char *)event.peer->data;
char *ptr = event.packet->data;
unsigned sz = event.packet->dataLength;
uint32_t mid = *(uint32_t*)ptr;
ptr += 4;
// @todo: propagate event to user
switch (mid) {
case MSG_INIT:
/* handled during client_join */
break;
case MSG_RPC:
case MSG_RPC_RESP:
// @todo: process and send a response back
break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
PRINTF( "%s disconnected.\n", (char *)event.peer->data );
/* Reset the peer's client information. */
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
PRINTF( "%s timeout.\n", (char *)event.peer->data );
event.peer->data = NULL;
server_drop_client_peer(event.peer);
break;
case ENET_EVENT_TYPE_NONE: break;
}
}
}
void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) { void server_broadcast_bin_flags(const void *msg, int len, uint64_t flags) {
ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0); ENetPacket *packet = enet_packet_create(msg, len, flags&NETWORK_UNRELIABLE ? ENET_PACKET_FLAG_UNRELIABLE_FRAGMENT : ENET_PACKET_FLAG_RELIABLE | flags&(NETWORK_UNRELIABLE|NETWORK_UNORDERED) ? ENET_PACKET_FLAG_UNSEQUENCED : 0);
enet_host_broadcast(Server, 0, packet); enet_host_broadcast(Server, 0, packet);
@ -9878,7 +9751,7 @@ static double msg_send_cooldown = 0.0;
static double network_dt = 0.0; static double network_dt = 0.0;
static double last_netsync = 0.0; static double last_netsync = 0.0;
void network_create(const char *ip, const char *port_, unsigned flags) { void network_create(unsigned max_clients, const char *ip, const char *port_, unsigned flags) {
if (buffers) map_clear(buffers); if (buffers) map_clear(buffers);
do_once { do_once {
array_resize(values, 128); array_resize(values, 128);
@ -9894,12 +9767,13 @@ void network_create(const char *ip, const char *port_, unsigned flags) {
network_put(NETWORK_PORT, port); network_put(NETWORK_PORT, port);
network_put(NETWORK_LIVE, -1); network_put(NETWORK_LIVE, -1);
network_put(NETWORK_COUNT, 0); network_put(NETWORK_COUNT, 0);
network_put(NETWORK_CAPACITY, max_clients);
if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) { if( !(flags&NETWORK_CONNECT) || flags&NETWORK_BIND ) {
// server, else client // server, else client
PRINTF("Trying to bind server, else we connect as a client...\n"); PRINTF("Trying to bind server, else we connect as a client...\n");
network_put(NETWORK_RANK, 0); network_put(NETWORK_RANK, 0);
if( server_bind(MAX_CLIENTS, port) ) { if( server_bind(max_clients, port) ) {
network_put(NETWORK_LIVE, 1); network_put(NETWORK_LIVE, 1);
PRINTF("Server bound\n"); PRINTF("Server bound\n");
} else { } else {
@ -10010,6 +9884,20 @@ char** network_sync(unsigned timeout_ms) {
msg_send_cooldown -= network_dt; msg_send_cooldown -= network_dt;
} }
if (is_server) {
server_poll(timeout_ms);
} else {
client_poll(timeout_ms);
}
array_push(events, NULL);
return events;
}
void server_poll(unsigned timeout_ms) {
if(timeout_ms < 2) timeout_ms = 2;
// network poll // network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) { for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0; char *msg = 0;
@ -10025,7 +9913,7 @@ char** network_sync(unsigned timeout_ms) {
event.peer->data = STRDUP(ip); /* TEMP */ event.peer->data = STRDUP(ip); /* TEMP */
/* ensure we have free slot for client */ /* ensure we have free slot for client */
if (map_count(clients) >= MAX_CLIENTS) { if (map_count(clients) >= network_get(NETWORK_CAPACITY)) {
msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port)); msg = stringf("%d %s", 1, va("%s", "Server is at maximum capacity, disconnecting the peer (::%s:%u)...", ip, event.peer->address.port));
enet_peer_disconnect_now(event.peer, 1); enet_peer_disconnect_now(event.peer, 1);
break; break;
@ -10060,7 +9948,6 @@ char** network_sync(unsigned timeout_ms) {
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
if (is_server) {
uint64_t *cid = map_find(clients, event.peer); uint64_t *cid = map_find(clients, event.peer);
if (cid) { if (cid) {
char init_msg[12]; char init_msg[12];
@ -10072,7 +9959,6 @@ char** network_sync(unsigned timeout_ms) {
} else { } else {
PRINTF("!Ignoring unk MSG_INIT client packet.\n"); PRINTF("!Ignoring unk MSG_INIT client packet.\n");
} }
}
break; break;
case MSG_BUF: { case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0); uint64_t *flags = (uint64_t*)(ptr + 0);
@ -10083,15 +9969,11 @@ char** network_sync(unsigned timeout_ms) {
ptr += 24; ptr += 24;
// validate if peer owns the buffer // validate if peer owns the buffer
uint8_t client_valid = 0;
if (is_server) {
int64_t *cid = map_find(clients, event.peer); int64_t *cid = map_find(clients, event.peer);
client_valid = cid ? *cid == *who : 0; uint8_t client_valid = cid ? *cid == *who : 0;
}
// apply incoming packet. // apply incoming packet.
if( is_client ? *who != whoami : client_valid ) { // clients merge always foreign packets. servers merge foreign packets. if( client_valid ) {
array(netbuffer_t) *list = map_find(buffers, *who); array(netbuffer_t) *list = map_find(buffers, *who);
assert( list ); assert( list );
assert( *idx < array_count(*list) ); assert( *idx < array_count(*list) );
@ -10130,30 +10012,127 @@ char** network_sync(unsigned timeout_ms) {
/* Reset the peer's client information. */ /* Reset the peer's client information. */
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT: case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data)); msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data); FREE(event.peer->data);
event.peer->data = NULL; event.peer->data = NULL;
if (is_server) {
server_drop_client_peer(event.peer); server_drop_client_peer(event.peer);
network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1); network_put(NETWORK_COUNT, network_get(NETWORK_COUNT)-1);
}
else {network_put(NETWORK_RANK, -1); network_put(NETWORK_LIVE, 0);}
break; break;
} }
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg)); if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
} }
}
array_push(events, NULL); void client_poll(unsigned timeout_ms) {
return events; int64_t whoami = network_get(NETWORK_RANK);
if(timeout_ms < 2) timeout_ms = 2;
// network poll
for( ENetEvent event; Server && enet_host_service(Server, &event, timeout_ms) > 0; ) {
char *msg = 0;
char ip[128]; enet_peer_get_ip(event.peer, ip, 128);
switch (event.type) {
default: // case ENET_EVENT_TYPE_NONE:
break;
case ENET_EVENT_TYPE_CONNECT:
/* @TODO: move stuff from client_join here */
break;
case ENET_EVENT_TYPE_RECEIVE:
/*
msg = stringf( "A packet of length %u containing %s was received from %s on channel %u",
(unsigned)event.packet->dataLength,
event.packet->data,
(char *)event.peer->data,
event.channelID );
*/
char *dbg = (char *)event.peer->data;
char *ptr = (char *)event.packet->data;
unsigned sz = (unsigned)event.packet->dataLength;
unsigned id = (unsigned)event.channelID;
// debug
// puts(dbg);
// hexdump(ptr, sz);
// decapsulate incoming packet.
uint32_t mid = *(uint32_t*)(ptr + 0);
ptr += 4;
switch (mid) {
case MSG_INIT:
/* handled by client_join */
break;
case MSG_BUF: {
uint64_t *flags = (uint64_t*)(ptr + 0);
uint32_t *idx = (uint32_t*)(ptr + 8);
uint32_t *len = (uint32_t*)(ptr + 12);
uint64_t *who = (uint64_t*)(ptr + 16);
ptr += 24;
// apply incoming packet.
if( *who != whoami ) {
array(netbuffer_t) *list = map_find(buffers, *who);
assert( list );
assert( *idx < array_count(*list) );
netbuffer_t *nb = &(*list)[*idx];
assert( *len == nb->sz );
memcpy(nb->ptr, ptr, *len);
}
} break;
case MSG_RPC: {
event.type = NETWORK_EVENT_RPC;
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 6);
*(uint32_t*)&resp_msg[0] = MSG_RPC_RESP;
memcpy(&resp_msg[4], resp, strlen(resp)+1);
ENetPacket *packet = enet_packet_create(resp_msg, strlen(resp) + 5, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
msg = stringf("%d %s", 0, va("req:%s res:%s", cmdline, resp));
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
event.type = NETWORK_EVENT_RPC_RESP;
msg = stringf("%d %s", 0, va("%s", ptr));
} break;
default:
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port);
break;
}
/* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet );
break;
case ENET_EVENT_TYPE_DISCONNECT:
msg = stringf( "%d %s", 0, va("%s disconnected", (char *)event.peer->data));
/* Reset the peer's client information. */
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
case ENET_EVENT_TYPE_DISCONNECT_TIMEOUT:
msg = stringf( "%d %s", 0, va("%s timeout", (char *)event.peer->data));
FREE(event.peer->data);
event.peer->data = NULL;
network_put(NETWORK_RANK, -1);
network_put(NETWORK_LIVE, 0);
break;
}
if(msg) array_push(events, stringf("%d %s", enet_event_to_netsync(event.type), msg));
}
} }
int network_event(const char *msg, int *errcode, char **errstr) { int network_event(const char *msg, int *errcode, char **errstr) {

9
engine/v4k.h
View File

@ -1999,9 +1999,9 @@ API int tcp_debug(int); // toggle traffic monitoring on/off for given socket
//enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation //enum { NETWORK_HANDSHAKE, NETWORK_ENCRYPT, NETWORK_VERSIONED, NETWORK_CHECKSUM }; // negotiation
//enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where //enum { NETWORK_TCP, NETWORK_UDP, NETWORK_KCP, NETWORK_ENET, NETWORK_WEBSOCKET }; // transport, where
enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 }; enum { NETWORK_BIND = 2, NETWORK_CONNECT = 4, NETWORK_NOFAIL = 8 };
enum { MAX_CLIENTS = 32 }; API void network_create(unsigned max_clients, const char *ip, const char *port, unsigned flags); // both ip and port can be null
API void network_create(const char *ip, const char *port, unsigned flags); // both ip and port can be null
//enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes //enum { NETWORK_LOSSY, NETWORK_COMPRESS }; // post-processes
//enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when //enum { NETWORK_PREDICT, NETWORK_RECONCILE, NETWORK_INTERPOLATE, NETWORK_COMPENSATE }; // time authority, when
@ -2031,7 +2031,7 @@ enum { NETWORK_RANK = 0 }; // [0..N] where 0 is server
enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY }; enum { NETWORK_PING = 1 }; // NETWORK_BANDWIDTH, NETWORK_QUALITY };
enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE }; enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
enum { NETWORK_SEND_MS = 4 }; enum { NETWORK_SEND_MS = 4 };
enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/ }; enum { NETWORK_USERID = 5, /*NETWORK_SALT,*/ NETWORK_COUNT/*N users*/ /*...*/, NETWORK_CAPACITY };
API int64_t network_get(uint64_t key); API int64_t network_get(uint64_t key);
API int64_t network_put(uint64_t key, int64_t value); API int64_t network_put(uint64_t key, int64_t value);
@ -2043,7 +2043,8 @@ API void network_rpc_send(unsigned id, const char *cmdline);
// low-level api (sockets based) // low-level api (sockets based)
API bool server_bind(int max_clients, int port); API bool server_bind(int max_clients, int port);
API void server_poll(); API void server_poll(unsigned timeout_ms);
API void client_poll(unsigned timeout_ms);
API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags); API void server_broadcast_bin_flags(const void *ptr, int len, uint64_t flags);
API void server_broadcast_bin(const void *ptr, int len); API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast_flags(const char *msg, uint64_t flags); API void server_broadcast_flags(const char *msg, uint64_t flags);

34
engine/v4k.html
View File

File diff suppressed because one or more lines are too long