Mercurial > repos > blastem

comparison event_log.c @ 1974:04b79a725b7f

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Better handling of pad assignment to remotes
author Michael Pavone <pavone@retrodev.com>
date Fri, 08 May 2020 16:38:58 -0700
parents cd163b230cf9
children a042e046f7f2
comparison
equal deleted inserted replaced
1973:cd163b230cf9 1974:04b79a725b7f
77 event_log_common_init(); 77 event_log_common_init();
78 fully_active = 1; 78 fully_active = 1;
79 atexit(file_finish); 79 atexit(file_finish);
80 } 80 }
81 81
82 static int listen_sock, remotes[7]; 82 typedef struct {
83 uint8_t *send_progress;
84 int sock;
85 uint8_t players[1]; //TODO: Expand when support for multiple players per remote is added
86 uint8_t num_players;
87 } remote;
88
89 static int listen_sock;
90 static remote remotes[7];
83 static int num_remotes; 91 static int num_remotes;
92 static uint8_t available_players[7] = {2,3,4,5,6,7,8};
93 static int num_available_players = 7;
84 void event_log_tcp(char *address, char *port) 94 void event_log_tcp(char *address, char *port)
85 { 95 {
86 struct addrinfo request, *result; 96 struct addrinfo request, *result;
87 socket_init(); 97 socket_init();
88 memset(&request, 0, sizeof(request)); 98 memset(&request, 0, sizeof(request));
195 event_header(EVENT_ADJUST, cycle); 205 event_header(EVENT_ADJUST, cycle);
196 last = cycle - deduction; 206 last = cycle - deduction;
197 save_int32(&buffer, deduction); 207 save_int32(&buffer, deduction);
198 } 208 }
199 209
200 static uint8_t *remote_send_progress[7]; 210 static uint8_t next_available_player(void)
201 static uint8_t remote_needs_state[7]; 211 {
212 uint8_t lowest = 0xFF;
213 int lowest_index = -1;
214 for (int i = 0; i < num_available_players; i++)
215 {
216 if (available_players[i] < lowest) {
217 lowest = available_players[i];
218 lowest_index = i;
219 }
220 }
221 if (lowest_index >= 0) {
222 available_players[lowest_index] = available_players[num_available_players - 1];
223 --num_available_players;
224 }
225 return lowest;
226 }
227
202 static void flush_socket(void) 228 static void flush_socket(void)
203 { 229 {
204 int remote = accept(listen_sock, NULL, NULL); 230 int remote_sock = accept(listen_sock, NULL, NULL);
205 if (remote != -1) { 231 if (remote_sock != -1) {
206 if (num_remotes == 7) { 232 if (num_remotes == 7) {
207 socket_close(remote); 233 socket_close(remote_sock);
208 } else { 234 } else {
209 printf("remote %d connected\n", num_remotes); 235 printf("remote %d connected\n", num_remotes);
210 remotes[num_remotes] = remote; 236 uint8_t player = next_available_player();
211 remote_needs_state[num_remotes++] = 1; 237 remotes[num_remotes++] = (remote){
238 .sock = remote_sock,
239 .send_progress = NULL,
240 .players = {player},
241 .num_players = player == 0xFF ? 0 : 1
242 };
212 current_system->save_state = EVENTLOG_SLOT + 1; 243 current_system->save_state = EVENTLOG_SLOT + 1;
213 } 244 }
214 } 245 }
215 uint8_t *min_progress = compressed; 246 uint8_t *min_progress = compressed;
216 for (int i = 0; i < num_remotes; i++) { 247 for (int i = 0; i < num_remotes; i++) {
217 int sent = 1; 248 if (remotes[i].send_progress) {
218 if (remote_needs_state[i]) { 249 uint8_t recv_buffer[1500];
219 remote_send_progress[i] = output_stream.next_out; 250 int bytes = recv(remotes[i].sock, recv_buffer, sizeof(recv_buffer), 0);
220 } else {
221 uint8_t buffer[1500];
222 int bytes = recv(remotes[i], buffer, sizeof(buffer), 0);
223 for (int j = 0; j < bytes; j++) 251 for (int j = 0; j < bytes; j++)
224 { 252 {
225 uint8_t cmd = buffer[j]; 253 uint8_t cmd = recv_buffer[j];
226 switch(cmd) 254 switch(cmd)
227 { 255 {
228 case CMD_GAMEPAD_DOWN: 256 case CMD_GAMEPAD_DOWN:
229 case CMD_GAMEPAD_UP: { 257 case CMD_GAMEPAD_UP: {
230 ++j; 258 ++j;
231 if (j < bytes) { 259 if (j < bytes) {
232 uint8_t button = buffer[j]; 260 uint8_t button = recv_buffer[j];
233 uint8_t pad = (button >> 5) + i + 1; 261 uint8_t pad = (button >> 5) - 1;
234 button &= 0x1F; 262 button &= 0x1F;
235 if (cmd == CMD_GAMEPAD_DOWN) { 263 if (pad < remotes[i].num_players) {
236 current_system->gamepad_down(current_system, pad, button); 264 pad = remotes[i].players[pad];
237 } else { 265 if (cmd == CMD_GAMEPAD_DOWN) {
238 current_system->gamepad_up(current_system, pad, button); 266 current_system->gamepad_down(current_system, pad, button);
267 } else {
268 current_system->gamepad_up(current_system, pad, button);
269 }
239 } 270 }
240 } else { 271 } else {
241 warning("Received incomplete command %X\n", cmd); 272 warning("Received incomplete command %X\n", cmd);
242 } 273 }
243 break; 274 break;
245 default: 276 default:
246 warning("Unrecognized remote command %X\n", cmd); 277 warning("Unrecognized remote command %X\n", cmd);
247 j = bytes; 278 j = bytes;
248 } 279 }
249 } 280 }
250 } 281 int sent = 1;
251 while (sent && output_stream.next_out > remote_send_progress[i]) 282 while (sent && output_stream.next_out > remotes[i].send_progress)
252 { 283 {
253 sent = send(remotes[i], remote_send_progress[i], output_stream.next_out - remote_send_progress[i], 0); 284 sent = send(remotes[i].sock, remotes[i].send_progress, output_stream.next_out - remotes[i].send_progress, 0);
254 if (sent >= 0) { 285 if (sent >= 0) {
255 remote_send_progress[i] += sent; 286 remotes[i].send_progress += sent;
256 } else if (!socket_error_is_wouldblock()) { 287 } else if (!socket_error_is_wouldblock()) {
257 socket_close(remotes[i]); 288 socket_close(remotes[i].sock);
258 remotes[i] = remotes[num_remotes-1]; 289 for (int j = 0; j < remotes[i].num_players; j++) {
259 remote_send_progress[i] = remote_send_progress[num_remotes-1]; 290 available_players[num_available_players++] = remotes[i].players[j];
260 remote_needs_state[i] = remote_needs_state[num_remotes-1]; 291 }
261 num_remotes--; 292 remotes[i] = remotes[num_remotes-1];
262 if (!num_remotes) { 293 num_remotes--;
263 //last remote disconnected, reset buffers/deflate 294 if (!num_remotes) {
264 fully_active = 0; 295 //last remote disconnected, reset buffers/deflate
265 deflateReset(&output_stream); 296 fully_active = 0;
266 output_stream.next_out = compressed; 297 deflateReset(&output_stream);
267 output_stream.avail_out = compressed_storage; 298 output_stream.next_out = compressed;
268 buffer.size = 0; 299 output_stream.avail_out = compressed_storage;
300 buffer.size = 0;
301 }
302 i--;
303 break;
269 } 304 }
270 i--; 305 if (remotes[i].send_progress > min_progress) {
271 break; 306 min_progress = remotes[i].send_progress;
272 } 307 }
273 if (remote_send_progress[i] > min_progress) {
274 min_progress = remote_send_progress[i];
275 } 308 }
276 } 309 }
277 } 310 }
278 if (min_progress == output_stream.next_out) { 311 if (min_progress == output_stream.next_out) {
279 output_stream.next_out = compressed; 312 output_stream.next_out = compressed;
280 output_stream.avail_out = compressed_storage; 313 output_stream.avail_out = compressed_storage;
281 for (int i = 0; i < num_remotes; i++) { 314 for (int i = 0; i < num_remotes; i++) {
282 remote_send_progress[i] = compressed; 315 if (remotes[i].send_progress) {
316 remotes[i].send_progress = compressed;
317 }
283 } 318 }
284 } 319 }
285 } 320 }
286 321
287 uint8_t wrote_since_last_flush; 322 uint8_t wrote_since_last_flush;
374 compressed_storage *= 2; 409 compressed_storage *= 2;
375 compressed = realloc(compressed, compressed_storage); 410 compressed = realloc(compressed, compressed_storage);
376 output_stream.next_out = compressed + old_storage; 411 output_stream.next_out = compressed + old_storage;
377 output_stream.avail_out = old_storage; 412 output_stream.avail_out = old_storage;
378 for (int i = 0; i < num_remotes; i++) { 413 for (int i = 0; i < num_remotes; i++) {
379 if (!remote_needs_state[i]) { 414 if (remotes[i].send_progress) {
380 remote_send_progress[i] = compressed + (remote_send_progress[i] - old_compressed); 415 remotes[i].send_progress = compressed + (remotes[i].send_progress - old_compressed);
381 } 416 }
382 } 417 }
383 } 418 }
384 int result = deflate(&output_stream, full ? Z_FINISH : Z_SYNC_FLUSH); 419 int result = deflate(&output_stream, full ? Z_FINISH : Z_SYNC_FLUSH);
385 if (result != (full ? Z_STREAM_END : Z_OK)) { 420 if (result != (full ? Z_STREAM_END : Z_OK)) {
409 state->size >> 16, state->size >> 8, state->size 444 state->size >> 16, state->size >> 8, state->size
410 }; 445 };
411 uint8_t sent_system_start = 0; 446 uint8_t sent_system_start = 0;
412 for (int i = 0; i < num_remotes; i++) 447 for (int i = 0; i < num_remotes; i++)
413 { 448 {
414 if (remote_needs_state[i]) { 449 if (!remotes[i].send_progress) {
415 if (send_all(remotes[i], system_start, system_start_size, 0) == system_start_size) { 450 if (send_all(remotes[i].sock, system_start, system_start_size, 0) == system_start_size) {
416 sent_system_start = 1; 451 sent_system_start = 1;
417 } else { 452 } else {
418 socket_close(remotes[i]); 453 socket_close(remotes[i].sock);
419 remotes[i] = remotes[num_remotes-1]; 454 remotes[i] = remotes[num_remotes-1];
420 remote_send_progress[i] = remote_send_progress[num_remotes-1];
421 remote_needs_state[i] = remote_needs_state[num_remotes-1];
422 num_remotes--; 455 num_remotes--;
423 i--; 456 i--;
424 } 457 }
425 } 458 }
426 } 459 }
436 save_buffer8(&buffer, state->data, state->size); 469 save_buffer8(&buffer, state->data, state->size);
437 size_t old_compressed_size = output_stream.next_out - compressed; 470 size_t old_compressed_size = output_stream.next_out - compressed;
438 deflate_flush(1); 471 deflate_flush(1);
439 size_t state_size = output_stream.next_out - compressed - old_compressed_size; 472 size_t state_size = output_stream.next_out - compressed - old_compressed_size;
440 for (int i = 0; i < num_remotes; i++) { 473 for (int i = 0; i < num_remotes; i++) {
441 if (remote_needs_state[i]) { 474 if (!remotes[i].send_progress) {
442 if (send_all(remotes[i], compressed + old_compressed_size, state_size, 0) == state_size) { 475 if (send_all(remotes[i].sock, compressed + old_compressed_size, state_size, 0) == state_size) {
443 remote_send_progress[i] = compressed + old_compressed_size; 476 remotes[i].send_progress = compressed + old_compressed_size;
444 remote_needs_state[i] = 0; 477 socket_blocking(remotes[i].sock, 0);
445 socket_blocking(remotes[i], 0);
446 int flag = 1; 478 int flag = 1;
447 setsockopt(remotes[i], IPPROTO_TCP, TCP_NODELAY, (const char *)&flag, sizeof(flag)); 479 setsockopt(remotes[i].sock, IPPROTO_TCP, TCP_NODELAY, (const char *)&flag, sizeof(flag));
448 fully_active = 1; 480 fully_active = 1;
449 } else { 481 } else {
450 socket_close(remotes[i]); 482 socket_close(remotes[i].sock);
451 remotes[i] = remotes[num_remotes-1]; 483 remotes[i] = remotes[num_remotes-1];
452 remote_send_progress[i] = remote_send_progress[num_remotes-1];
453 remote_needs_state[i] = remote_needs_state[num_remotes-1];
454 num_remotes--; 484 num_remotes--;
455 i--; 485 i--;
456 } 486 }
457 } 487 }
458 } 488 }