Fix some netplay issues

4 files changed, 59 insertions, 33 deletions

diff --git a/event_log.c b/event_log.c
index 63122cf..877d910 100644
--- a/event_log.c
+++ b/event_log.c
@@ -124,7 +124,7 @@ static void event_header(uint8_t type, uint32_t cycle)
 			}
 			return;
 		}
-	} else if (type == last_event_type && delta == last_delta) {
+	} else if (type == last_event_type && delta == last_delta && type != EVENT_FLUSH) {
 		//make some room
 		save_int8(&buffer, 0);
 		//shift existing command
@@ -246,6 +246,11 @@ void event_log(uint8_t type, uint32_t cycle, uint8_t size, uint8_t *payload)
 	last = cycle;
 	save_buffer8(&buffer, payload, size);
 	if (listen_sock && buffer.size > 1280) {
+		if (multi_count) {
+			buffer.data[multi_start] |= multi_count - 2;
+			multi_count = 0;
+			last_event_type = 0xFF;
+		}
 		flush_socket();
 	}
 }
@@ -394,6 +399,40 @@ void init_event_reader_tcp(event_reader *reader, char *address, char *port)
 	setsockopt(reader->socket, IPPROTO_TCP, TCP_NODELAY, (const char *)&flag, sizeof(flag));
 }
 
+static void read_from_socket(event_reader *reader)
+{
+	if (reader->storage - (reader->buffer.size - reader->buffer.cur_pos) < 128 * 1024) {
+		reader->storage *= 2;
+		uint8_t *new_buf = malloc(reader->storage);
+		memcpy(new_buf, reader->buffer.data + reader->buffer.cur_pos, reader->buffer.size - reader->buffer.cur_pos);
+		free(reader->buffer.data);
+		reader->buffer.data = new_buf;
+		reader->buffer.size -= reader->buffer.cur_pos;
+		reader->buffer.cur_pos = 0;
+	} else if (reader->buffer.cur_pos >= reader->buffer.size/2 && reader->buffer.size >= reader->storage/2) {
+		memmove(reader->buffer.data, reader->buffer.data + reader->buffer.cur_pos, reader->buffer.size - reader->buffer.cur_pos);
+		reader->buffer.size -= reader->buffer.cur_pos;
+		reader->buffer.cur_pos = 0;
+	}
+	int bytes = recv(reader->socket, reader->buffer.data + reader->buffer.size, reader->storage - reader->buffer.size, 0);
+	if (bytes >= 0) {
+		reader->buffer.size += bytes;
+	} else if (!socket_error_is_wouldblock()) {
+		fatal_error("Connection closed, error = %X\n", socket_last_error());
+	}
+}
+
+void reader_ensure_data(event_reader *reader, size_t bytes)
+{
+	if (reader->socket && reader->buffer.size - reader->buffer.cur_pos < bytes) {
+		socket_blocking(reader->socket, 1);
+		while (reader->buffer.size - reader->buffer.cur_pos < bytes) {
+			read_from_socket(reader);
+		}
+		socket_blocking(reader->socket, 0);
+	}
+}
+
 uint8_t reader_next_event(event_reader *reader, uint32_t *cycle_out)
 {
 	if (reader->repeat_remaining) {
@@ -403,34 +442,8 @@ uint8_t reader_next_event(event_reader *reader, uint32_t *cycle_out)
 		return reader->repeat_event;
 	}
 	if (reader->socket) {
-		uint8_t blocking = 0;
-		if (reader->buffer.size - reader->buffer.cur_pos < 9) {
-			//set back to block mode
-			socket_blocking(reader->socket, 1);
-			blocking = 1;
-		}
-		if (reader->storage - (reader->buffer.size - reader->buffer.cur_pos) < 128 * 1024) {
-			reader->storage *= 2;
-			uint8_t *new_buf = malloc(reader->storage);
-			memcpy(new_buf, reader->buffer.data + reader->buffer.cur_pos, reader->buffer.size - reader->buffer.cur_pos);
-			free(reader->buffer.data);
-			reader->buffer.data = new_buf;
-			reader->buffer.size -= reader->buffer.cur_pos;
-			reader->buffer.cur_pos = 0;
-		} else if (reader->buffer.cur_pos >= reader->buffer.size/2 && reader->buffer.size >= reader->storage/2) {
-			memmove(reader->buffer.data, reader->buffer.data + reader->buffer.cur_pos, reader->buffer.size - reader->buffer.cur_pos);
-			reader->buffer.size -= reader->buffer.cur_pos;
-			reader->buffer.cur_pos = 0;
-		}
-		int bytes = recv(reader->socket, reader->buffer.data + reader->buffer.size, reader->storage - reader->buffer.size, 0);
-		if (bytes >= 0) {
-			reader->buffer.size += bytes;
-			if (blocking && reader->buffer.size - reader->buffer.cur_pos >= 9) {
-				socket_blocking(reader->socket, 0);
-			}
-		} else if (!socket_error_is_wouldblock()) {
-			printf("Connection closed, error = %X\n", socket_last_error());
-		}
+		read_from_socket(reader);
+		reader_ensure_data(reader, 1);
 	}
 	uint8_t header = load_int8(&reader->buffer);
 	uint8_t ret;
@@ -439,15 +452,18 @@ uint8_t reader_next_event(event_reader *reader, uint32_t *cycle_out)
 	if ((header & 0xF0) == (EVENT_MULTI << 4)) {
 		reader->repeat_remaining = (header & 0xF) + 1;
 		multi_start = 1;
+		reader_ensure_data(reader, 1);
 		header = load_int8(&reader->buffer);
 	}
 	if ((header & 0xF0) < FORMAT_3BYTE) {
 		delta = (header & 0xF) + 16;
 		ret = header >> 4;
 	} else if ((header & 0xF0) == FORMAT_3BYTE) {
+		reader_ensure_data(reader, 2);
 		delta = load_int16(&reader->buffer);
 		ret = header & 0xF;
 	} else {
+		reader_ensure_data(reader, 3);
 		delta = load_int8(&reader->buffer) << 16;
 		//sign extend 24-bit delta to 32-bit
 		if (delta & 0x800000) {
@@ -463,11 +479,13 @@ uint8_t reader_next_event(event_reader *reader, uint32_t *cycle_out)
 	*cycle_out = reader->last_cycle + delta;
 	reader->last_cycle = *cycle_out;
 	if (ret == EVENT_ADJUST) {
+		reader_ensure_data(reader, 4);
 		size_t old_pos = reader->buffer.cur_pos;
 		uint32_t adjust = load_int32(&reader->buffer);
 		reader->buffer.cur_pos = old_pos;
 		reader->last_cycle -= adjust;
 	} else if (ret == EVENT_STATE) {
+		reader_ensure_data(reader, 8);
 		reader->last_cycle = load_int32(&reader->buffer);
 		reader->last_word_address = load_int8(&reader->buffer) << 16;
 		reader->last_word_address |= load_int16(&reader->buffer);
diff --git a/event_log.h b/event_log.h
index 5080f80..db6ddf8 100644
--- a/event_log.h
+++ b/event_log.h
@@ -48,6 +48,7 @@ void event_flush(uint32_t cycle);
 void init_event_reader(event_reader *reader, uint8_t *data, size_t size);
 void init_event_reader_tcp(event_reader *reader, char *address, char *port);
 uint8_t reader_next_event(event_reader *reader, uint32_t *cycle_out);
+void reader_ensure_data(event_reader *reader, size_t bytes);
 uint8_t reader_system_type(event_reader *reader);
 void reader_send_gamepad_event(event_reader *reader, uint8_t pad, uint8_t button, uint8_t down);
 
diff --git a/gen_player.c b/gen_player.c
index c282517..91045e2 100644
--- a/gen_player.c
+++ b/gen_player.c
@@ -47,10 +47,12 @@ void start_context(system_header *sys, char *statefile)
 			break;
 		case EVENT_PSG_REG:
 			sync_sound(player, cycle);
+			reader_ensure_data(&player->reader, 1);
 			psg_write(player->psg, load_int8(&player->reader.buffer));
 			break;
 		case EVENT_YM_REG: {
 			sync_sound(player, cycle);
+			reader_ensure_data(&player->reader, 3);
 			uint8_t part = load_int8(&player->reader.buffer);
 			uint8_t reg = load_int8(&player->reader.buffer);
 			uint8_t value = load_int8(&player->reader.buffer);
@@ -62,12 +64,10 @@ void start_context(system_header *sys, char *statefile)
 			ym_data_write(player->ym, value);
 			break;
 		case EVENT_STATE: {
+			reader_ensure_data(&player->reader, 3);
 			uint32_t size = load_int8(&player->reader.buffer) << 16;
 			size |= load_int16(&player->reader.buffer);
-			if (player->reader.buffer.size - player->reader.buffer.cur_pos < size) {
-				puts("State has not been fully loaded!");
-				exit(1);
-			}
+			reader_ensure_data(&player->reader, size);
 			deserialize_buffer buffer;
 			init_deserialize(&buffer, player->reader.buffer.data + player->reader.buffer.cur_pos, size);
 			register_section_handler(&buffer, (section_handler){.fun = vdp_deserialize, .data = player->vdp}, SECTION_VDP);
diff --git a/vdp.c b/vdp.c
index 3bf05cd..8138ec6 100644
--- a/vdp.c
+++ b/vdp.c
@@ -4571,26 +4571,33 @@ void vdp_replay_event(vdp_context *context, uint8_t event, event_reader *reader)
 	switch (event)
 	{
 	case EVENT_VRAM_BYTE:
+		reader_ensure_data(reader, 3);
 		address = load_int16(buffer);
 		break;
 	case EVENT_VRAM_BYTE_DELTA:
+		reader_ensure_data(reader, 2);
 		address = reader->last_byte_address + load_int8(buffer);
 		break;
 	case EVENT_VRAM_BYTE_ONE:
+		reader_ensure_data(reader, 1);
 		address = reader->last_byte_address + 1;
 		break;
 	case EVENT_VRAM_BYTE_AUTO:
+		reader_ensure_data(reader, 1);
 		address = reader->last_byte_address + context->regs[REG_AUTOINC];
 		break;
 	case EVENT_VRAM_WORD:
+		reader_ensure_data(reader, 4);
 		address = load_int8(buffer) << 16;
 		address |= load_int16(buffer);
 		break;
 	case EVENT_VRAM_WORD_DELTA:
+		reader_ensure_data(reader, 3);
 		address = reader->last_word_address + load_int8(buffer);
 		break;
 	case EVENT_VDP_REG:
 	case EVENT_VDP_INTRAM:
+		reader_ensure_data(reader, event == EVENT_VDP_REG ? 2 : 3);
 		address = load_int8(buffer);
 		break;
 	}