389 lines
15 KiB
Plaintext
389 lines
15 KiB
Plaintext
// Audit Header_Os_Specific.
|
|
// User-level protocol info.
|
|
// Handle wrapping.
|
|
// Assert that we don't read another message without being done with the previous.
|
|
// Assert that we are done reading or writing when destroying channels.
|
|
|
|
// Make the documentation below more accurate:
|
|
|
|
// This is a module for communicating quickly, with other processes on the same machine,
|
|
// using shared memory.
|
|
|
|
// The reader calls reader_create_channel to initialize,
|
|
// reader_release_channel to deinitialize that buffer.
|
|
|
|
// The writer calls writer_connect. If that succeeds,
|
|
// the writer can call writer_get_packet_pointer() to get the position into which
|
|
// we can write, and then writer_send_packet() to actually send.
|
|
|
|
// Document: Circular buffer is big relative to message size, so messages do not wrap.
|
|
// Only one thread at once to write or read.
|
|
|
|
// On Linux, we can't be O_EXCL so we hose ourselves if you run us twice? wtf? How do we get the fd to be tied to the process lifetime?
|
|
|
|
// Keep trying to reconnect if you start the game before the visualizer.
|
|
#if OS == .WINDOWS {
|
|
#load "windows.jai";
|
|
} else #if (OS == .LINUX) || (OS == .MACOS) {
|
|
#load "posix.jai";
|
|
}
|
|
|
|
Header :: struct {
|
|
magic : u32;
|
|
version : u32;
|
|
|
|
reader_status: enum s32 {
|
|
DISCONNECTED :: 0;
|
|
PAUSED :: 1;
|
|
ACTIVELY_READING :: 2;
|
|
};
|
|
|
|
writer_status: enum s32 {
|
|
DISCONNECTED :: 0;
|
|
PAUSED :: 1;
|
|
ACTIVELY_WRITING :: 2;
|
|
};
|
|
|
|
offset_to_os_specific_header: u32; // This is the offset from the start of the header.
|
|
offset_to_message_queue_header: u32; // This is the offset from the start of the header.
|
|
offset_to_message_data: u32; // This is the offset from the start of the header.
|
|
|
|
full_memory_region_size_in_bytes: s64; // The size of the full memory region, from the start of the header to the end of contiguous memory.
|
|
data_size_in_bytes: s64; // The size of the data area (a.k.a. sum of all message sizes that can be queued at once).
|
|
}
|
|
|
|
Read_Channel :: struct {
|
|
header: *Header;
|
|
os_specific: Read_Channel_Os_Specific;
|
|
}
|
|
|
|
Write_Channel :: struct {
|
|
header: *Header;
|
|
os_specific: Write_Channel_Os_Specific;
|
|
}
|
|
|
|
Message_Queue_Header :: struct {
|
|
read_cursor: s64; // Offset of the next data to read from the start of the Header (not this Message_Queue_Header).
|
|
pad: [56] u8 = ---; // Give us at least 1 cache line of padding.
|
|
write_cursor: s64; // Offset of the next data to write from the start of the Header (not this Message_Queue_Header).
|
|
}
|
|
|
|
Reserved_Message :: struct {
|
|
channel: *Write_Channel;
|
|
offset: s64;
|
|
total_bytes: s64; // This includes message overhead.
|
|
|
|
user_write_pointer: *u8; // Write your mesasge here.
|
|
}
|
|
|
|
Read_Message :: struct {
|
|
channel: *Read_Channel;
|
|
offset: s64;
|
|
total_bytes: s64;
|
|
|
|
user_read_pointer: *u8; // Can we not just make this a [] *u8?
|
|
user_read_bytes: s64;
|
|
}
|
|
|
|
OFFSET_TO_HEADER_OS_SPECIFIC :: 64; // Probably 1 or 2 cache lines?
|
|
OFFSET_TO_MESSAGE_QUEUE_HEADER :: 128;
|
|
OFFSET_TO_MESSAGE_DATA :: 256;
|
|
|
|
PER_MESSAGE_OVERHEAD_SIZE :: 16; // 8-byte per-message magic stamp, 8-byte length. The magic stamp helps us with debugging.
|
|
|
|
// All our length stamps are non-negative numbers, so our wrapping sentinel being negative
|
|
// means it will not be mistaken for a length. However, just being boring and writing -1 is not good,
|
|
// since that is a fill pattern others might use. So we make a negative number that is unlikely
|
|
// to be used elsewhere.
|
|
WRAPPING_SENTINEL :: 0xcafebabef00dface; // Reader, if you see this in the length slot, wrap to the start of the buffer.. Must be negative!
|
|
|
|
#assert size_of(Header) < OFFSET_TO_HEADER_OS_SPECIFIC;
|
|
#assert size_of(Header_Os_Specific) + OFFSET_TO_HEADER_OS_SPECIFIC < OFFSET_TO_MESSAGE_QUEUE_HEADER;
|
|
|
|
// already_exists means another reader owns the name; its live header is left untouched.
|
|
reader_create_channel :: (name: string, bytes: s64) -> (success: bool, channel: Read_Channel, already_exists: bool = false) {
|
|
success, channel, already_exists := reader_create_channel_os_specific(name, bytes);
|
|
if !success return false, .{}, already_exists;
|
|
|
|
header := channel.header;
|
|
|
|
Basic.Initialize(header);
|
|
|
|
data := cast(*u8) *channel.header;
|
|
|
|
header.magic = (.*) cast(*u32) CHANNEL_MAGIC.data; // @Endian
|
|
header.version = PROTOCOL_VERSION;
|
|
header.reader_status = .ACTIVELY_READING; // Maybe don't do this until the reader explicitly says so?
|
|
|
|
header.offset_to_os_specific_header = OFFSET_TO_HEADER_OS_SPECIFIC;
|
|
header.offset_to_message_queue_header = OFFSET_TO_MESSAGE_QUEUE_HEADER;
|
|
header.offset_to_message_data = OFFSET_TO_MESSAGE_DATA;
|
|
|
|
header.full_memory_region_size_in_bytes = bytes + OFFSET_TO_MESSAGE_DATA;
|
|
header.data_size_in_bytes = bytes;
|
|
|
|
mq := get_message_queue_header(header);
|
|
mq.read\_cursor = OFFSET_TO_MESSAGE_DATA;
|
|
mq.write_cursor = OFFSET_TO_MESSAGE_DATA;
|
|
|
|
return true, channel;
|
|
}
|
|
|
|
// Discards everything queued, for recovering once from a corrupted queue. A writer that is
|
|
// mid-commit can still land a partial message, which the next poll reports as fatal.
|
|
reader_reset_queue :: (channel: *Read_Channel) {
|
|
mq := get_message_queue_header(channel.header);
|
|
if !mq return;
|
|
|
|
cursor_store(*mq.read_cursor, channel.header.offset_to_message_data);
|
|
cursor_store(*mq.write_cursor, channel.header.offset_to_message_data);
|
|
}
|
|
|
|
get_message_queue_header :: inline (header: *Header) -> *Message_Queue_Header {
|
|
base := cast(*u8) header;
|
|
return cast(*Message_Queue_Header)(base + header.offset_to_message_queue_header);
|
|
}
|
|
|
|
// Each cursor publishes the message data behind it to the other process, which on ARM64 needs
|
|
// acquire and release ordering. compare_and_swap is the atomic documented as acquire-release on
|
|
// every target, so both go through it.
|
|
cursor_load :: inline (cursor: *s64) -> s64 {
|
|
// No cursor is ever -1, so this always fails and only returns the current value.
|
|
_, value := compare_and_swap(cursor, -1, -1);
|
|
return value;
|
|
}
|
|
|
|
cursor_store :: inline (cursor: *s64, value: s64) {
|
|
while true {
|
|
current := cursor.*;
|
|
if compare_and_swap(cursor, current, value) break;
|
|
}
|
|
}
|
|
|
|
/*
|
|
get_data_area :: inline (header: *Header) -> *u8 {
|
|
base := cast(*u8) header;
|
|
return base + header.offset_to_message_data;
|
|
}
|
|
*/
|
|
|
|
//
|
|
// OS-independent writer stuff:
|
|
//
|
|
|
|
writer_reserve_message :: (channel: *Write_Channel, data_bytes: s64) -> (success: bool, reserved: Reserved_Message) {
|
|
// @Incomplete assert if a message is currently reserved!
|
|
|
|
// We assume our buffer is much bigger than the messages.
|
|
// If we cannot get space for a message without wrapping the message in the middle,
|
|
// we skip back to the beginning of the buffer. (This means the user level code
|
|
// on the reading side can just look at the memory without worrying about wrapping around a ring buffer,
|
|
// and without having to copy it out.)
|
|
|
|
total_bytes := data_bytes + PER_MESSAGE_OVERHEAD_SIZE;
|
|
|
|
header := channel.header;
|
|
arena_start := header.offset_to_message_data;
|
|
arena_end := arena_start + header.data_size_in_bytes;
|
|
|
|
mqh := get_message_queue_header(header);
|
|
if !mqh return false, .{};
|
|
|
|
target_position := 0; // 0 means unset. All valid offsets are > 0, so if we see that, we know we found a place for this message.
|
|
write_wrapping_sentinel_at := 0; // If this is not 0, we wrapped, so tell the reader that we wrapped.
|
|
|
|
write_cursor := mqh.write_cursor; // only this side writes it
|
|
read\_cursor := cursor_load(*mqh.read_cursor);
|
|
if write_cursor >= read_cursor {
|
|
// Test against the end of the buffer.
|
|
remaining := arena_end - write_cursor;
|
|
if remaining >= total_bytes {
|
|
target_position = write_cursor;
|
|
} else {
|
|
end := header.offset_to_message_data + header.data_size_in_bytes;
|
|
if end - write_cursor >= PER_MESSAGE_OVERHEAD_SIZE {
|
|
// If there are fewer bytes left than this, the reader will just know to wrap.
|
|
write_wrapping_sentinel_at = write_cursor;
|
|
}
|
|
|
|
write_cursor = header.offset_to_message_data;
|
|
|
|
if total_bytes >= header.data_size_in_bytes - 1 return false, .{}; // No way will this message fit. -1 because we cannot meet the read cursor.
|
|
}
|
|
}
|
|
|
|
if !target_position {
|
|
// Test for space from the beginning of the buffer to the read cursor.
|
|
assert(write_cursor <= read_cursor); // The reader could catch up to us right when we are wrapping.
|
|
remaining := read_cursor - write_cursor - 1; // -1 so we cannot meet the read cursor from the bottom.
|
|
|
|
if remaining >= total_bytes {
|
|
target_position = header.offset_to_message_data;
|
|
} else {
|
|
return false, .{}; // We were unable to find a place for the message!
|
|
}
|
|
}
|
|
|
|
if write_wrapping_sentinel_at {
|
|
assert(write_wrapping_sentinel_at > header.offset_to_message_data); // Writing this to the first buffer position is an error, as it can cause the reader to loop without making progress, and anyway there would be no way to fit a message by wrapping if you were already at the beginning!.
|
|
dest := (cast(*u8) header) + write_wrapping_sentinel_at;
|
|
(.*)cast(*u64)dest = WRAPPING_SENTINEL;
|
|
}
|
|
|
|
assert(target_position > 0);
|
|
|
|
data_start := cast(*u8) header;
|
|
|
|
result: Reserved_Message;
|
|
result.offset = write_cursor;
|
|
result.total_bytes = total_bytes;
|
|
result.channel = channel;
|
|
|
|
result.user_write_pointer = data_start + write_cursor + PER_MESSAGE_OVERHEAD_SIZE;
|
|
|
|
return true, result;
|
|
}
|
|
|
|
writer_commit_message :: (channel: *Write_Channel, message: Reserved_Message) -> success: bool {
|
|
assert(message.channel == channel);
|
|
|
|
if message.channel != channel return false;
|
|
|
|
mqh := get_message_queue_header(channel.header);
|
|
if !mqh return false;
|
|
|
|
data := cast(*u8) channel.header;
|
|
|
|
magic_pointer := cast(*u64) (data + message.offset);
|
|
length_pointer := cast(*s64) (data + message.offset + 8);
|
|
|
|
assert(data + message.offset + 16 == message.user_write_pointer);
|
|
magic_pointer.* = (.*)cast(*u64) MESSAGE_MAGIC.data;
|
|
length_pointer.* = message.total_bytes;
|
|
|
|
cursor_store(*mqh.write_cursor, message.offset + message.total_bytes);
|
|
|
|
return true;
|
|
}
|
|
|
|
|
|
reader_poll_for_message :: (channel: *Read_Channel) -> (found: bool, fatal_error: bool, message: Read_Message) {
|
|
header := channel.header;
|
|
|
|
mqh := get_message_queue_header(header);
|
|
if !mqh return false, true, .{};
|
|
|
|
read_cursor := mqh.read_cursor; // only this side writes it
|
|
if read_cursor == cursor_load(*mqh.write_cursor) {
|
|
return false, false, .{};
|
|
}
|
|
|
|
data := cast(*u8) header;
|
|
|
|
magic_pointer := cast(*u64) (data + read_cursor);
|
|
length_pointer := cast(*s64) (data + read_cursor + 8);
|
|
|
|
desired_magic := (.*) cast(*u64) MESSAGE_MAGIC.data;
|
|
magic := magic_pointer.*;
|
|
if magic != desired_magic {
|
|
if magic == WRAPPING_SENTINEL {
|
|
// It's time to wrap!
|
|
// I hope we are not at the beginning of the buffer though!
|
|
|
|
if read_cursor == header.offset_to_message_data {
|
|
Basic.log_error("In reader_poll_for_message(), we got a WRAPPING_SENTINEL at the start of the buffer, which is invalid. We must stop.\n");
|
|
return false, true, .{};
|
|
}
|
|
|
|
cursor_store(*mqh.read_cursor, header.offset_to_message_data);
|
|
found, fatal_error, message := #this(channel);
|
|
return found, fatal_error, message;
|
|
}
|
|
|
|
Basic.log_error("Invalid magic number 0x% at offset % in reader_poll_for_message(): we got a parse error and must stop.\n", Basic.formatInt(magic, base=16), read_cursor);
|
|
|
|
return false, true, .{};
|
|
}
|
|
|
|
length := length_pointer.*;
|
|
if length < PER_MESSAGE_OVERHEAD_SIZE {
|
|
Basic.log_error("In reader_poll_for_message(), we got a length of %, which is invalid. Failing.\n", length);
|
|
return false, true, .{};
|
|
}
|
|
|
|
end := header.offset_to_message_data + header.data_size_in_bytes;
|
|
remaining := end - read_cursor;
|
|
if remaining < length {
|
|
Basic.log_error("In reader_poll_for_message(), we got a length of %, which is too long (only % bytes are left in the buffer). Failing.\n", length, remaining);
|
|
return false, true, .{};
|
|
}
|
|
|
|
message: Read_Message;
|
|
message.channel = channel;
|
|
message.offset = read_cursor;
|
|
message.total_bytes = length;
|
|
|
|
message.user_read_pointer = data + read_cursor + PER_MESSAGE_OVERHEAD_SIZE;
|
|
message.user_read_bytes = length - PER_MESSAGE_OVERHEAD_SIZE;
|
|
|
|
return true, false, message;
|
|
}
|
|
|
|
reader_done_with_message :: (message: *Read_Message) -> bool {
|
|
channel := message.channel;
|
|
assert(channel != null);
|
|
|
|
header := channel.header;
|
|
|
|
mqh := get_message_queue_header(header);
|
|
if !mqh return false;
|
|
|
|
read_cursor := mqh.read_cursor;
|
|
read_cursor += message.total_bytes;
|
|
|
|
end := header.offset_to_message_data + header.data_size_in_bytes;
|
|
if end - read_cursor < PER_MESSAGE_OVERHEAD_SIZE {
|
|
// Wrap.
|
|
read_cursor = OFFSET_TO_MESSAGE_DATA;
|
|
}
|
|
|
|
cursor_store(*mqh.read_cursor, read_cursor);
|
|
|
|
return true;
|
|
}
|
|
|
|
|
|
PROTOCOL_VERSION :: 1;
|
|
CHANNEL_MAGIC :: "ShmC";
|
|
MESSAGE_MAGIC :: "ShmM2022";
|
|
|
|
#scope_module
|
|
|
|
get_size_of_full_mapping_from_stub_header :: (pointer: *void) -> (success: bool, actual_bytes: s64) {
|
|
// Ininitially, we don't know how big the memory region is, so we just map
|
|
// the header. Then we look in the header to find the size of the full region,
|
|
// then re-map it.
|
|
header_copy := (.*) cast(*Header) pointer;
|
|
desired_magic := (.*) cast(*u32) CHANNEL_MAGIC.data; // @Endian
|
|
|
|
success := true;
|
|
if header_copy.magic != desired_magic {
|
|
Basic.log_error("In writer_connect(), the magic number of the mapped header did not match. (Wanted 0x%, got 0x%).\n", Basic.formatInt(desired_magic, base=16), Basic.formatInt(header_copy.magic, base=16));
|
|
success = false;
|
|
} else if header_copy.version != PROTOCOL_VERSION {
|
|
Basic.log_error("In writer_connect(), the protocol version did not match. (Wanted %, got %).\n", PROTOCOL_VERSION, header_copy.version);
|
|
success = false;
|
|
}
|
|
|
|
actual_bytes := header_copy.full_memory_region_size_in_bytes;
|
|
if actual_bytes < OFFSET_TO_MESSAGE_DATA {
|
|
Basic.log_error("In writer_connect(), the full_memory_region_size_in_bytes we got from the header was %, which is too small.\n", actual_bytes);
|
|
success = false;
|
|
}
|
|
|
|
return success, actual_bytes;
|
|
}
|
|
|
|
Basic :: #import "Basic";
|
|
assert :: Basic.assert;
|