// 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;