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

rpc support in v4k

main
Dominik Madarász 2023-08-10 19:55:53 +02:00
parent ee76c55b0e
commit 5cb2c38b25
10 changed files with 411 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
demos/99-net_rpc.c
View File

@ -1,128 +0,0 @@
// core for remote procedure calls
// - rlyeh, public domain.
//
// format:
// - query: id method [args.....] ('subject' is alias for arg[0]; and 'object' can be any arg[1..])
// - answer: id error [values...]
//
// todo:
// - [ ] promote rpc_function to (int argc, void **args) ?
#define V4K_IMPLEMENTATION
#include "engine/joint/v4k.h"
#ifndef RPC_H
#define RPC_H
void rpc_insert(const char *signature, void * function );
char *rpc(const char *cmdline); // for debugging purposes
char* rpc_quick(unsigned query_number, const char* cmdline);
char *rpc_full(unsigned id, const char* method, unsigned num_args, char *args[]);
#endif
// -----------------------------------------------------------------------------
#define RPC_C
#ifdef RPC_C
#pragma once
typedef void* (*rpc_function)();
typedef struct rpc_call {
char *method;
rpc_function function;
uint64_t function_hash;
} rpc_call;
#define RPC_SIGNATURE_i_iii UINT64_C(0x9830e90d3327e74b) // HASH_STR("int(int,int,int)")
#define RPC_SIGNATURE_i_ii UINT64_C(0xa7fcab437d38c750) // HASH_STR("int(int,int)")
#define RPC_SIGNATURE_s_s UINT64_C(0xc4db3d7818162463) // HASH_STR("char*(char*)")
#define RPC_SIGNATURE_s_v UINT64_C(0x8857a7c1cd20bd7b) // HASH_STR("char*(void)")
rpc_call rpc_new_call(const char *signature, rpc_function function) {
if( signature && function ) {
array(char*)tokens = strsplit(signature, "(,)"); array_pop(tokens);
if( array_count(tokens) >= 1 ) {
char *method = strrchr(tokens[0], ' ')+1;
char *rettype = va("%.*s", (int)(method - tokens[0] - 1), tokens[0]);
int num_args = array_count(tokens) - 1;
uint64_t hash = hash_str(va("%s(%s)", rettype, num_args ? strjoin(num_args, &tokens[1], ",") : "void" ));
method = va("%s%d", method, num_args );
#ifdef _DEBUG
printf("%p %p %s %s(", function, (void*)hash, rettype, method); for(int i = 1, end = array_count(tokens); i < end; ++i) printf("%s%s", tokens[i], i == (end-1)? "":", "); puts(");");
#endif
return (rpc_call) { strdup(method), function, hash }; // LEAK
}
}
return (rpc_call) {0};
}
static map(char*, rpc_call) rpc_calls = 0;
void rpc_insert(const char *signature, void *function ) {
rpc_call call = rpc_new_call(signature, function);
if( call.method ) {
if( !rpc_calls ) map_init(rpc_calls, less_str, hash_str);
if( map_find(rpc_calls, call.method)) {
map_erase(rpc_calls, call.method);
}
map_insert(rpc_calls, call.method, call);
}
}
char *rpc_full(unsigned id, const char* method, unsigned num_args, char *args[]) {
#ifdef _DEBUG
printf("id:%x method:%s args:", id, method );
for( int i = 0; i < num_args; ++i ) printf("%s", args[i]); puts("");
#endif
method = va("%s%d", method, num_args);
rpc_call *found = map_find(rpc_calls, (char*)method);
if( found ) {
switch(found->function_hash) {
default:
case RPC_SIGNATURE_i_iii: return va("%d \"%d %s\" %d", id, errno, strerror(errno), (int)(uintptr_t)found->function(atoi(args[0]), atoi(args[1]), atoi(args[2])) );
case RPC_SIGNATURE_i_ii: return va("%d \"%d %s\" %d", id, errno, strerror(errno), (int)(uintptr_t)found->function(atoi(args[0]), atoi(args[1])) );
case RPC_SIGNATURE_s_s: return va("%d \"%d %s\" %s", id, errno, strerror(errno), (char*)found->function(args[0]) );
case RPC_SIGNATURE_s_v: return va("%d \"%d %s\" %s", id, errno, strerror(errno), (char*)found->function() );
}
}
return va("%d \"0\" ?", id);
}
char* rpc_quick(unsigned query_number, const char* cmdline) {
array(char*) args = os_argparse(cmdline, false);
int num_args = array_count(args);
char *ret = num_args ? rpc_full(query_number, args[0], num_args - 1, &args[1]) : rpc_full(query_number, "", 0, NULL);
array_free(args);
return ret;
}
char *rpc(const char *cmdline) { // for debugging purposes
char *rc = rpc_quick(0, cmdline);
puts(rc);
return rc;
}
// -----------------------------------------------------------------------------
int rpc_add2(int num1, int num2) {
return num1+num2;
}
int rpc_add3(int num1, int num2, int num3) {
return num1+num2+num3;
}
char *rpc_echo(char *text) {
return text;
}
int main() {
rpc_insert("int add(int,int)", rpc_add2);
rpc_insert("int add(int,int,int)", rpc_add3);
rpc_insert("char* echo(char*)", rpc_echo);
rpc("add 1 2"); // -> 3
rpc("add 100 3 -3"); // -> 100
rpc("echo \"hello world\""); // -> hello world
}
#endif // RPC_C

146
demos/99-rpc_demo.c 100644
View File

@ -0,0 +1,146 @@
#include "v4k.h"
typedef void* (*rpc_function)();
typedef struct rpc_call {
char *method;
rpc_function function;
uint64_t function_hash;
} rpc_call;
#define RPC_SIGNATURE_i_iii UINT64_C(0x78409099752fa48a) // printf("%llx\n, HASH_STR("int(int,int,int)"));
#define RPC_SIGNATURE_i_ii UINT64_C(0x258290edf43985a5) // printf("%llx\n, HASH_STR("int(int,int)"));
#define RPC_SIGNATURE_s_s UINT64_C(0x97deedd17d9afb12) // printf("%llx\n, HASH_STR("char*(char*)"));
#define RPC_SIGNATURE_s_v UINT64_C(0x09c16a1242049b80) // printf("%llx\n, HASH_STR("char*(void)"));
static
rpc_call rpc_new_call(const char *signature, rpc_function function) {
if( signature && function ) {
array(char*)tokens = strsplit(signature, "(,)");
if( array_count(tokens) >= 1 ) {
char *method = strrchr(tokens[0], ' ')+1;
char *rettype = va("%.*s", (int)(method - tokens[0] - 1), tokens[0]);
int num_args = array_count(tokens) - 1;
char* hash_sig = va("%s(%s)", rettype, num_args ? (array_pop_front(tokens), strjoin(tokens, ",")) : "void");
uint64_t hash = hash_str(hash_sig);
method = va("%s%d", method, num_args );
#if RPC_DEBUG
printf("%p %p %s `%s` %s(", function, (void*)hash, rettype, hash_sig, method); for(int i = 0, end = array_count(tokens); i < end; ++i) printf("%s%s", tokens[i], i == (end-1)? "":", "); puts(");");
#endif
return (rpc_call) { strdup(method), function, hash }; // LEAK
}
}
return (rpc_call) {0};
}
static map(char*, rpc_call) rpc_calls = 0;
static
void rpc_insert(const char *signature, void *function ) {
rpc_call call = rpc_new_call(signature, function);
if( call.method ) {
if( !rpc_calls ) map_init(rpc_calls, less_str, hash_str);
if( map_find(rpc_calls, call.method)) {
map_erase(rpc_calls, call.method);
}
map_insert(rpc_calls, call.method, call);
}
}
static
char *rpc_full(unsigned id, const char* method, unsigned num_args, char *args[]) {
#if RPC_DEBUG
printf("id:%x method:%s args:", id, method );
for( int i = 0; i < num_args; ++i ) printf("%s,", args[i]); puts("");
#endif
method = va("%s%d", method, num_args);
rpc_call *found = map_find(rpc_calls, (char*)method);
if( found ) {
switch(found->function_hash) {
case RPC_SIGNATURE_i_iii: return va("%d %d", id, (int)(uintptr_t)found->function(atoi(args[0]), atoi(args[1]), atoi(args[2])) );
case RPC_SIGNATURE_i_ii: return va("%d %d", id, (int)(uintptr_t)found->function(atoi(args[0]), atoi(args[1])) );
case RPC_SIGNATURE_s_s: return va("%d %s", id, (char*)found->function(args[0]) );
case RPC_SIGNATURE_s_v: return va("%d %s", id, (char*)found->function() );
default: break;
}
}
return va("%d -1", id);
}
static
array(char*) rpc_parse_args( const char *cmdline, bool quote_whitespaces ) { // parse cmdline arguments. must array_free() after use
// - supports quotes: "abc" "abc def" "abc \"def\"" "abc \"def\"""ghi" etc.
// - #comments removed
array(char*) args = 0; // LEAK
for( int i = 0; cmdline[i]; ) {
char buf[256] = {0}, *ptr = buf;
while(cmdline[i] && isspace(cmdline[i])) ++i;
bool quoted = cmdline[i] == '\"';
if( quoted ) {
while(cmdline[++i]) {
char ch = cmdline[i];
/**/ if (ch == '\\' && cmdline[i + 1] == '\"') *ptr++ = '\"', ++i;
else if (ch == '\"' && cmdline[i + 1] == '\"') ++i;
else if (ch == '\"' && (!cmdline[i + 1] || isspace(cmdline[i + 1]))) {
++i; break;
}
else *ptr++ = ch;
}
} else {
while(cmdline[i] && !isspace(cmdline[i])) *ptr++ = cmdline[i++];
}
if (buf[0] && buf[0] != '#') { // exclude empty args + comments
if( quote_whitespaces && quoted )
array_push(args, va("\"%s\"",buf));
else
array_push(args, va("%s",buf));
}
}
return args;
}
static
char* rpc(unsigned id, const char* cmdline) {
array(char*) args = rpc_parse_args(cmdline, false);
int num_args = array_count(args);
char *ret = num_args ? rpc_full(id, args[0], num_args - 1, &args[1]) : rpc_full(id, "", 0, NULL);
array_free(args);
return ret;
}
static void enet_quit(void) {
do_once {
// enet_deinitialize();
}
}
static void enet_init() {
do_once {
if( enet_initialize() != 0 ) {
PANIC("cannot initialize enet");
}
atexit( enet_quit );
}
}
// -----------------------------------------------------------------------------
// demo
int rpc_add2(int num1, int num2) {
return num1+num2;
}
int rpc_add3(int num1, int num2, int num3) {
return num1+num2+num3;
}
char *rpc_echo(char *text) {
return text;
}
int main() {
rpc_insert("int add(int,int)", rpc_add2);
rpc_insert("int add(int,int,int)", rpc_add3);
rpc_insert("char* echo(char*)", rpc_echo);
puts(rpc(0,"add 1 2")); // -> 3
puts(rpc(1,"add 100 3 -3")); // -> 100
puts(rpc(2,"echo \"hello world\"")); // -> hello world
}

17
demos/99-syncdemo.c
View File

@ -15,6 +15,12 @@ struct world_t {
struct npc_t npc[MAX_NPCS]; struct npc_t npc[MAX_NPCS];
} world = {0}; } world = {0};
char *show_notification(char *msg) {
printf("notif %s\n", msg);
ui_notify("server", msg);
return NULL;
}
void bind_netbuffers(int64_t self_id) { void bind_netbuffers(int64_t self_id) {
uint32_t colors[] = { ORANGE,GREEN,RED,CYAN,PURPLE,YELLOW,GRAY,PINK,AQUA }; uint32_t colors[] = { ORANGE,GREEN,RED,CYAN,PURPLE,YELLOW,GRAY,PINK,AQUA };
for (int64_t i=0; i<MAX_NPCS; ++i) { for (int64_t i=0; i<MAX_NPCS; ++i) {
@ -30,6 +36,11 @@ void bind_netbuffers(int64_t self_id) {
world.player[i].color = colors[i%(sizeof colors / sizeof colors[0])]; world.player[i].color = colors[i%(sizeof colors / sizeof colors[0])];
network_buffer(&world.player[i], sizeof(struct player_t), i!=self_id ? NETWORK_RECV : NETWORK_SEND, i /* each client owns exactly 1 buffer */); network_buffer(&world.player[i], sizeof(struct player_t), i!=self_id ? NETWORK_RECV : NETWORK_SEND, i /* each client owns exactly 1 buffer */);
}; };
// register server->client rpc
if (self_id > 0) {
network_rpc("char* show_notification(char*)", show_notification);
}
} }
int main() { int main() {
@ -61,6 +72,12 @@ int main() {
continue; continue;
} }
/* quick hack to broadcast notif from host */
if (self_id == 0 && input_down(KEY_F3)) {
printf("rpc %s\n", "show_notification \"hi, sailor!\"");
network_rpc_send(rand()%4, "show_notification \"hi, sailor!\"");
}
// camera tracking // camera tracking
cam.position = vec3(self->x,100,self->z); cam.position = vec3(self->x,100,self->z);
camera_lookat(&cam, vec3(self->x,0,self->z)); camera_lookat(&cam, vec3(self->x,0,self->z));

105
engine/joint/v4k.h
View File

@ -15932,8 +15932,9 @@ enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
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);
API void network_rpc_insert(const char *signature, void *function); API void network_rpc(const char *signature, void *function);
API char *network_rpc(unsigned id, const char *cmdline); API void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline);
API void network_rpc_send(unsigned id, const char *cmdline);
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
// low-level api (sockets based) // low-level api (sockets based)
@ -15944,6 +15945,7 @@ API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast(const char *msg); API void server_broadcast(const char *msg);
API void server_terminate(); API void server_terminate();
API void server_send(int64_t handle, const char *msg); API void server_send(int64_t handle, const char *msg);
API void server_send_bin(int64_t handle, const void *ptr, int len);
API void server_drop(int64_t handle); API void server_drop(int64_t handle);
API int64_t client_join(const char *ip, int port); API int64_t client_join(const char *ip, int port);
@ -338521,12 +338523,11 @@ static void enet_init() {
} }
} }
static ENetHost *Server; static ENetHost *Server;
static map(ENetPeer *, int64_t) clients; static map(ENetPeer *, int64_t) clients;
static map(int64_t, ENetPeer *) peers; static map(int64_t, ENetPeer *) peers;
static int64_t next_client_id = 1; // assumes ID 0 is server static int64_t next_client_id = 1; // assumes ID 0 is server
enum { MSG_INIT, MSG_BUF, MSG_RPC }; enum { MSG_INIT, MSG_BUF, MSG_RPC, MSG_RPC_RESP };
bool server_bind(int max_clients, int port) { bool server_bind(int max_clients, int port) {
map_init(clients, less_64, hash_64); map_init(clients, less_64, hash_64);
@ -338581,21 +338582,25 @@ void server_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: { case MSG_INIT: {
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];
*(uint32_t*)&init_msg[0] = MSG_INIT; *(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid; *(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet); enet_peer_send(event.peer, 0, packet);
} else { } else {
PRINTF("ignoring unk MSG_INIT client packet.\n"); PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
} }
} 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -338641,11 +338646,15 @@ void client_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
/* handled during client_join */ /* handled during client_join */
break; break;
default: case MSG_RPC:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg); 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. */ /* Clean up the packet now that we're done using it. */
@ -338757,11 +338766,15 @@ void server_drop(int64_t handle) {
server_drop_client(handle); server_drop_client(handle);
} }
void server_send(int64_t handle, const char *msg) { void server_send_bin(int64_t handle, const void *ptr, int len) {
ENetPacket *packet = enet_packet_create(msg, strlen(msg) + 1, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(ptr, len, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet); enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet);
} }
void server_send(int64_t handle, const char *msg) {
server_send_bin(handle, msg, strlen(msg)+1);
}
// --- // ---
typedef struct netbuffer_t { typedef struct netbuffer_t {
@ -338975,9 +338988,24 @@ char** network_sync(unsigned timeout_ms) {
memcpy(nb->ptr, ptr, *len); memcpy(nb->ptr, ptr, *len);
} }
} break; } break;
case MSG_RPC: {
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 5);
*(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) + 4, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
// @todo: react on response?
msg = ptr;
} break;
default: default:
break;
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port); // 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -339006,6 +339034,31 @@ char** network_sync(unsigned timeout_ms) {
array_push(events, NULL); array_push(events, NULL);
return events; return events;
} }
void network_rpc(const char *signature, void *function) {
rpc_insert(signature, function);
}
void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline) {
assert(network_get(NETWORK_RANK) == 0); /* must be a host */
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_send_bin(rank, msg, sz);
FREE(msg);
}
void network_rpc_send(unsigned id, const char *cmdline) {
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_broadcast_bin(msg, sz);
FREE(msg);
}
#line 0 #line 0
#line 1 "v4k_render.c" #line 1 "v4k_render.c"

99
engine/split/v4k_netsync.c
View File

@ -121,12 +121,11 @@ static void enet_init() {
} }
} }
static ENetHost *Server; static ENetHost *Server;
static map(ENetPeer *, int64_t) clients; static map(ENetPeer *, int64_t) clients;
static map(int64_t, ENetPeer *) peers; static map(int64_t, ENetPeer *) peers;
static int64_t next_client_id = 1; // assumes ID 0 is server static int64_t next_client_id = 1; // assumes ID 0 is server
enum { MSG_INIT, MSG_BUF, MSG_RPC }; enum { MSG_INIT, MSG_BUF, MSG_RPC, MSG_RPC_RESP };
bool server_bind(int max_clients, int port) { bool server_bind(int max_clients, int port) {
map_init(clients, less_64, hash_64); map_init(clients, less_64, hash_64);
@ -181,21 +180,25 @@ void server_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: { case MSG_INIT: {
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];
*(uint32_t*)&init_msg[0] = MSG_INIT; *(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid; *(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet); enet_peer_send(event.peer, 0, packet);
} else { } else {
PRINTF("ignoring unk MSG_INIT client packet.\n"); PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
} }
} 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -241,11 +244,15 @@ void client_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
/* handled during client_join */ /* handled during client_join */
break; break;
default: case MSG_RPC:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg); 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. */ /* Clean up the packet now that we're done using it. */
@ -357,11 +364,15 @@ void server_drop(int64_t handle) {
server_drop_client(handle); server_drop_client(handle);
} }
void server_send(int64_t handle, const char *msg) { void server_send_bin(int64_t handle, const void *ptr, int len) {
ENetPacket *packet = enet_packet_create(msg, strlen(msg) + 1, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(ptr, len, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet); enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet);
} }
void server_send(int64_t handle, const char *msg) {
server_send_bin(handle, msg, strlen(msg)+1);
}
// --- // ---
typedef struct netbuffer_t { typedef struct netbuffer_t {
@ -575,9 +586,24 @@ char** network_sync(unsigned timeout_ms) {
memcpy(nb->ptr, ptr, *len); memcpy(nb->ptr, ptr, *len);
} }
} break; } break;
case MSG_RPC: {
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 5);
*(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) + 4, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
// @todo: react on response?
msg = ptr;
} break;
default: default:
break;
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port); // 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -606,3 +632,28 @@ char** network_sync(unsigned timeout_ms) {
array_push(events, NULL); array_push(events, NULL);
return events; return events;
} }
void network_rpc(const char *signature, void *function) {
rpc_insert(signature, function);
}
void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline) {
assert(network_get(NETWORK_RANK) == 0); /* must be a host */
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_send_bin(rank, msg, sz);
FREE(msg);
}
void network_rpc_send(unsigned id, const char *cmdline) {
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_broadcast_bin(msg, sz);
FREE(msg);
}

6
engine/split/v4k_netsync.h
View File

@ -33,8 +33,9 @@ enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
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);
API void network_rpc_insert(const char *signature, void *function); API void network_rpc(const char *signature, void *function);
API char *network_rpc(unsigned id, const char *cmdline); API void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline);
API void network_rpc_send(unsigned id, const char *cmdline);
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
// low-level api (sockets based) // low-level api (sockets based)
@ -45,6 +46,7 @@ API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast(const char *msg); API void server_broadcast(const char *msg);
API void server_terminate(); API void server_terminate();
API void server_send(int64_t handle, const char *msg); API void server_send(int64_t handle, const char *msg);
API void server_send_bin(int64_t handle, const void *ptr, int len);
API void server_drop(int64_t handle); API void server_drop(int64_t handle);
API int64_t client_join(const char *ip, int port); API int64_t client_join(const char *ip, int port);

99
engine/v4k.c
View File

@ -9598,12 +9598,11 @@ static void enet_init() {
} }
} }
static ENetHost *Server; static ENetHost *Server;
static map(ENetPeer *, int64_t) clients; static map(ENetPeer *, int64_t) clients;
static map(int64_t, ENetPeer *) peers; static map(int64_t, ENetPeer *) peers;
static int64_t next_client_id = 1; // assumes ID 0 is server static int64_t next_client_id = 1; // assumes ID 0 is server
enum { MSG_INIT, MSG_BUF, MSG_RPC }; enum { MSG_INIT, MSG_BUF, MSG_RPC, MSG_RPC_RESP };
bool server_bind(int max_clients, int port) { bool server_bind(int max_clients, int port) {
map_init(clients, less_64, hash_64); map_init(clients, less_64, hash_64);
@ -9658,21 +9657,25 @@ void server_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: { case MSG_INIT: {
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];
*(uint32_t*)&init_msg[0] = MSG_INIT; *(uint32_t*)&init_msg[0] = MSG_INIT;
*(uint64_t*)&init_msg[4] = *cid; *(uint64_t*)&init_msg[4] = *cid;
ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(init_msg, 12, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet); enet_peer_send(event.peer, 0, packet);
} else { } else {
PRINTF("ignoring unk MSG_INIT client packet.\n"); PRINTF("ignoring unk MSG_INIT client packet.\n");
}
} break;
default:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg);
} }
} 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -9718,11 +9721,15 @@ void client_poll() {
// @todo: propagate event to user // @todo: propagate event to user
switch (mid) { switch (mid) {
case MSG_INIT: case MSG_INIT:
/* handled during client_join */ /* handled during client_join */
break; break;
default: case MSG_RPC:
PRINTF("recving unk %d sz %d from peer %s\n", mid, sz, dbg); 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. */ /* Clean up the packet now that we're done using it. */
@ -9834,11 +9841,15 @@ void server_drop(int64_t handle) {
server_drop_client(handle); server_drop_client(handle);
} }
void server_send(int64_t handle, const char *msg) { void server_send_bin(int64_t handle, const void *ptr, int len) {
ENetPacket *packet = enet_packet_create(msg, strlen(msg) + 1, ENET_PACKET_FLAG_RELIABLE); ENetPacket *packet = enet_packet_create(ptr, len, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet); enet_peer_send(*(ENetPeer **)map_find(peers, handle), 0, packet);
} }
void server_send(int64_t handle, const char *msg) {
server_send_bin(handle, msg, strlen(msg)+1);
}
// --- // ---
typedef struct netbuffer_t { typedef struct netbuffer_t {
@ -10052,9 +10063,24 @@ char** network_sync(unsigned timeout_ms) {
memcpy(nb->ptr, ptr, *len); memcpy(nb->ptr, ptr, *len);
} }
} break; } break;
case MSG_RPC: {
unsigned id = *(uint32_t*)ptr; ptr += 4;
char *cmdline = ptr;
char *resp = rpc(id, cmdline);
char *resp_msg = MALLOC(strlen(resp) + 5);
*(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) + 4, ENET_PACKET_FLAG_RELIABLE);
enet_peer_send(event.peer, 0, packet);
FREE(resp_msg);
} break;
case MSG_RPC_RESP: {
// @todo: react on response?
msg = ptr;
} break;
default: default:
break;
// PRINTF("!Receiving unk %d sz %d from peer ::%s:%u\n", mid, sz, ip, event.peer->address.port); // 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. */ /* Clean up the packet now that we're done using it. */
enet_packet_destroy( event.packet ); enet_packet_destroy( event.packet );
@ -10083,6 +10109,31 @@ char** network_sync(unsigned timeout_ms) {
array_push(events, NULL); array_push(events, NULL);
return events; return events;
} }
void network_rpc(const char *signature, void *function) {
rpc_insert(signature, function);
}
void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline) {
assert(network_get(NETWORK_RANK) == 0); /* must be a host */
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_send_bin(rank, msg, sz);
FREE(msg);
}
void network_rpc_send(unsigned id, const char *cmdline) {
unsigned sz = strlen(cmdline) + 8;
char *msg = MALLOC(sz);
*(uint32_t*)&msg[0] = MSG_RPC;
*(uint32_t*)&msg[4] = id;
memcpy(&msg[8], cmdline, sz-8);
server_broadcast_bin(msg, sz);
FREE(msg);
}
#line 0 #line 0
#line 1 "v4k_render.c" #line 1 "v4k_render.c"

6
engine/v4k.h
View File

@ -2015,8 +2015,9 @@ enum { NETWORK_PORT = 2, NETWORK_IP, NETWORK_LIVE };
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);
API void network_rpc_insert(const char *signature, void *function); API void network_rpc(const char *signature, void *function);
API char *network_rpc(unsigned id, const char *cmdline); API void network_rpc_send_to(int64_t rank, unsigned id, const char *cmdline);
API void network_rpc_send(unsigned id, const char *cmdline);
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
// low-level api (sockets based) // low-level api (sockets based)
@ -2027,6 +2028,7 @@ API void server_broadcast_bin(const void *ptr, int len);
API void server_broadcast(const char *msg); API void server_broadcast(const char *msg);
API void server_terminate(); API void server_terminate();
API void server_send(int64_t handle, const char *msg); API void server_send(int64_t handle, const char *msg);
API void server_send_bin(int64_t handle, const void *ptr, int len);
API void server_drop(int64_t handle); API void server_drop(int64_t handle);
API int64_t client_join(const char *ip, int port); API int64_t client_join(const char *ip, int port);

11
tools/uuid.c 100644
View File

@ -0,0 +1,11 @@
#include <stdio.h>
#include "v4k.h"
int main(int argc, char **argv) {
if (argc != 2) {
return -1;
}
printf("%llx", hash_str(argv[1]));
return 0;
}

BIN
tools/uuid.exe 100644
View File

Binary file not shown.