LibIPC: Move IPC payloads through post_message

MessageBuffer::transfer_message() handed a MessageDataType to transport
APIs that accepted Vector<u8> const&. That forced the inline-capacity
vector to be materialized as a plain Vector<u8>, and the Mach transport
then copied the same payload again into its pending-send queue.

Make the transport API take MessageDataType by value and pass the
encoded buffer with take_data(). The Mach pending queue now stores the
same type so the payload can move directly to the IO thread. The socket
transport keeps queued messages as owned headers plus moved payloads
instead of copying the payload into an AllocatingMemoryStream, while
preserving the existing chunked send and fd acknowledgement behavior.

Libraries/LibIPC/Message.cpp

@@ -49,7 +49,7 @@ ErrorOr<void> MessageBuffer::transfer_message(Transport& transport)
     VERIFY(m_data.size() <= MAX_MESSAGE_PAYLOAD_SIZE);
     VERIFY(m_attachments.size() <= MAX_MESSAGE_FD_COUNT);
 
-    transport.post_message(m_data, m_attachments);
+    transport.post_message(take_data(), m_attachments);
     return {};
 }
 

Libraries/LibIPC/TransportMachPort.cpp

@@ -372,11 +372,11 @@ void TransportMachPort::wait_until_readable()
         m_incoming_cv.wait();
 }
 
-void TransportMachPort::post_message(Vector<u8> const& bytes, Vector<Attachment>& attachments)
+void TransportMachPort::post_message(MessageDataType bytes, Vector<Attachment>& attachments)
 {
     {
         Sync::MutexLocker locker(m_send_mutex);
-        m_pending_send_messages.append(PendingMessage { bytes, move(attachments) });
+        m_pending_send_messages.append(PendingMessage { move(bytes), move(attachments) });
     }
     wake_io_thread();
 }

Libraries/LibIPC/TransportMachPort.h

@@ -20,6 +20,7 @@
 #include <LibCore/Notifier.h>
 #include <LibIPC/Attachment.h>
 #include <LibIPC/AutoCloseFileDescriptor.h>
+#include <LibIPC/Forward.h>
 #include <LibIPC/TransportHandle.h>
 #include <LibSync/ConditionVariable.h>
 #include <LibSync/Mutex.h>
@@ -49,7 +50,7 @@ public:
 
     void wait_until_readable();
 
-    void post_message(Vector<u8> const&, Vector<Attachment>& attachments);
+    void post_message(MessageDataType, Vector<Attachment>& attachments);
 
     enum class ShouldShutdown {
         No,
@@ -68,7 +69,7 @@ private:
     static constexpr unsigned int IPC_WAKEUP_MESSAGE_ID = 0x4950C003;
 
     struct PendingMessage {
-        Vector<u8> bytes;
+        MessageDataType bytes;
         Vector<Attachment> attachments;
     };
 

Libraries/LibIPC/TransportSocket.cpp

@@ -49,11 +49,11 @@ ErrorOr<TransportSocket::Paired> TransportSocket::create_paired()
     };
 }
 
-void SendQueue::enqueue_message(ReadonlyBytes header, ReadonlyBytes payload, Vector<int>&& fds)
+void SendQueue::enqueue_message(SocketMessageHeader header, MessageDataType payload, Vector<int>&& fds)
 {
     Sync::MutexLocker locker(m_mutex);
-    VERIFY(MUST(m_stream.write_some(header)) == header.size());
-    VERIFY(MUST(m_stream.write_some(payload)) == payload.size());
+    m_queued_byte_count += sizeof(SocketMessageHeader) + payload.size();
+    m_queued_messages.append(QueuedMessage { header, move(payload) });
     m_fds.append(fds.data(), fds.size());
 }
 
@@ -61,9 +61,28 @@ SendQueue::BytesAndFds SendQueue::peek(size_t max_bytes)
 {
     Sync::MutexLocker locker(m_mutex);
     BytesAndFds result;
-    auto bytes_to_send = min(max_bytes, m_stream.used_buffer_size());
+    auto bytes_to_send = min(max_bytes, m_queued_byte_count);
     result.bytes.resize(bytes_to_send);
-    m_stream.peek_some(result.bytes);
+    size_t copied_bytes = 0;
+    for (auto const& queued_message : m_queued_messages) {
+        if (copied_bytes == bytes_to_send)
+            break;
+
+        auto start_offset = queued_message.start_offset;
+        if (start_offset < sizeof(SocketMessageHeader)) {
+            ReadonlyBytes header { reinterpret_cast<u8 const*>(&queued_message.header), sizeof(SocketMessageHeader) };
+            copied_bytes += header.slice(start_offset).copy_trimmed_to(result.bytes.span().slice(copied_bytes));
+            if (copied_bytes == bytes_to_send)
+                break;
+            start_offset = sizeof(SocketMessageHeader);
+        }
+
+        auto payload_offset = start_offset - sizeof(SocketMessageHeader);
+        if (payload_offset < queued_message.payload.size()) {
+            ReadonlyBytes payload { queued_message.payload.data() + payload_offset, queued_message.payload.size() - payload_offset };
+            copied_bytes += payload.copy_trimmed_to(result.bytes.span().slice(copied_bytes));
+        }
+    }
 
     if (m_fds.size() > 0) {
         auto fds_to_send = min(m_fds.size(), Core::LocalSocket::MAX_TRANSFER_FDS);
@@ -76,7 +95,17 @@ SendQueue::BytesAndFds SendQueue::peek(size_t max_bytes)
 void SendQueue::discard(size_t bytes_count, size_t fds_count)
 {
     Sync::MutexLocker locker(m_mutex);
-    MUST(m_stream.discard(bytes_count));
+    VERIFY(bytes_count <= m_queued_byte_count);
+    m_queued_byte_count -= bytes_count;
+    while (bytes_count > 0) {
+        auto& queued_message = m_queued_messages.first();
+        auto available_bytes = queued_message.size() - queued_message.start_offset;
+        auto consumed_bytes = min(available_bytes, bytes_count);
+        queued_message.start_offset += consumed_bytes;
+        bytes_count -= consumed_bytes;
+        if (queued_message.start_offset == queued_message.size())
+            (void)m_queued_messages.remove(m_queued_messages.begin());
+    }
     m_fds.remove(0, fds_count);
 }
 
@@ -262,22 +291,12 @@ static constexpr size_t MAX_UNPROCESSED_BUFFER_SIZE = 128 * MiB;
 // Maximum number of accumulated unprocessed file descriptors before we disconnect the peer
 static constexpr size_t MAX_UNPROCESSED_FDS = 512;
 
-struct MessageHeader {
-    enum class Type : u8 {
-        Payload = 0,
-        FileDescriptorAcknowledgement = 1,
-    };
-    Type type { Type::Payload };
-    u32 payload_size { 0 };
-    u32 fd_count { 0 };
-};
-
-void TransportSocket::post_message(Vector<u8> const& bytes_to_write, Vector<Attachment>& attachments)
+void TransportSocket::post_message(MessageDataType bytes_to_write, Vector<Attachment>& attachments)
 {
     auto num_fds_to_transfer = attachments.size();
 
-    MessageHeader header {
-        .type = MessageHeader::Type::Payload,
+    SocketMessageHeader header {
+        .type = SocketMessageHeader::Type::Payload,
         .payload_size = static_cast<u32>(bytes_to_write.size()),
         .fd_count = static_cast<u32>(num_fds_to_transfer),
     };
@@ -294,7 +313,7 @@ void TransportSocket::post_message(Vector<u8> const& bytes_to_write, Vector<Atta
         }
     }
 
-    m_send_queue->enqueue_message({ reinterpret_cast<u8 const*>(&header), sizeof(header) }, bytes_to_write, move(raw_fds));
+    m_send_queue->enqueue_message(header, move(bytes_to_write), move(raw_fds));
     wake_io_thread();
 }
 
@@ -400,10 +419,10 @@ void TransportSocket::read_incoming_messages()
     Checked<u32> received_fd_count = 0;
     Checked<u32> acknowledged_fd_count = 0;
     size_t index = 0;
-    while (index + sizeof(MessageHeader) <= m_unprocessed_bytes.size()) {
-        MessageHeader header;
-        memcpy(&header, m_unprocessed_bytes.data() + index, sizeof(MessageHeader));
-        if (header.type == MessageHeader::Type::Payload) {
+    while (index + sizeof(SocketMessageHeader) <= m_unprocessed_bytes.size()) {
+        SocketMessageHeader header;
+        memcpy(&header, m_unprocessed_bytes.data() + index, sizeof(SocketMessageHeader));
+        if (header.type == SocketMessageHeader::Type::Payload) {
             if (header.payload_size > MAX_MESSAGE_PAYLOAD_SIZE) {
                 dbgln("TransportSocket: Rejecting message with payload_size {} exceeding limit {}", header.payload_size, MAX_MESSAGE_PAYLOAD_SIZE);
                 m_peer_eof = true;
@@ -415,7 +434,7 @@ void TransportSocket::read_incoming_messages()
                 break;
             }
             Checked<size_t> message_size = header.payload_size;
-            message_size += sizeof(MessageHeader);
+            message_size += sizeof(SocketMessageHeader);
             if (message_size.has_overflow() || message_size.value() > m_unprocessed_bytes.size() - index)
                 break;
             if (header.fd_count > m_unprocessed_attachments.size())
@@ -429,13 +448,13 @@ void TransportSocket::read_incoming_messages()
             }
             for (size_t i = 0; i < header.fd_count; ++i)
                 message->attachments.enqueue(m_unprocessed_attachments.dequeue());
-            if (message->bytes.try_append(m_unprocessed_bytes.data() + index + sizeof(MessageHeader), header.payload_size).is_error()) {
+            if (message->bytes.try_append(m_unprocessed_bytes.data() + index + sizeof(SocketMessageHeader), header.payload_size).is_error()) {
                 dbgln("TransportSocket: Failed to allocate message buffer for payload_size {}", header.payload_size);
                 m_peer_eof = true;
                 break;
             }
             batch.append(move(message));
-        } else if (header.type == MessageHeader::Type::FileDescriptorAcknowledgement) {
+        } else if (header.type == SocketMessageHeader::Type::FileDescriptorAcknowledgement) {
             if (header.payload_size != 0) {
                 dbgln("TransportSocket: FileDescriptorAcknowledgement with non-zero payload_size {}", header.payload_size);
                 m_peer_eof = true;
@@ -454,7 +473,7 @@ void TransportSocket::read_incoming_messages()
         }
         Checked<size_t> new_index = index;
         new_index += header.payload_size;
-        new_index += sizeof(MessageHeader);
+        new_index += sizeof(SocketMessageHeader);
         if (new_index.has_overflow()) {
             dbgln("TransportSocket: index would overflow");
             m_peer_eof = true;
@@ -477,12 +496,12 @@ void TransportSocket::read_incoming_messages()
     }
 
     if (received_fd_count > 0u) {
-        MessageHeader header {
-            .type = MessageHeader::Type::FileDescriptorAcknowledgement,
+        SocketMessageHeader header {
+            .type = SocketMessageHeader::Type::FileDescriptorAcknowledgement,
             .payload_size = 0,
             .fd_count = received_fd_count.value(),
         };
-        m_send_queue->enqueue_message({ reinterpret_cast<u8 const*>(&header), sizeof(header) }, {}, {});
+        m_send_queue->enqueue_message(header, {}, {});
         wake_io_thread();
     }
 

Libraries/LibIPC/TransportSocket.h

@@ -7,11 +7,13 @@
 
 #pragma once
 
-#include <AK/MemoryStream.h>
 #include <AK/Queue.h>
+#include <AK/SinglyLinkedList.h>
+#include <AK/SinglyLinkedListSizePolicy.h>
 #include <LibCore/Socket.h>
 #include <LibIPC/Attachment.h>
 #include <LibIPC/AutoCloseFileDescriptor.h>
+#include <LibIPC/Forward.h>
 #include <LibIPC/TransportHandle.h>
 #include <LibSync/ConditionVariable.h>
 #include <LibSync/Mutex.h>
@@ -19,9 +21,19 @@
 
 namespace IPC {
 
+struct SocketMessageHeader {
+    enum class Type : u8 {
+        Payload = 0,
+        FileDescriptorAcknowledgement = 1,
+    };
+    Type type { Type::Payload };
+    u32 payload_size { 0 };
+    u32 fd_count { 0 };
+};
+
 class SendQueue : public AtomicRefCounted<SendQueue> {
 public:
-    void enqueue_message(ReadonlyBytes header, ReadonlyBytes payload, Vector<int>&& fds);
+    void enqueue_message(SocketMessageHeader, MessageDataType payload, Vector<int>&& fds);
     struct BytesAndFds {
         Vector<u8> bytes;
         Vector<int> fds;
@@ -30,7 +42,16 @@ public:
     void discard(size_t bytes_count, size_t fds_count);
 
 private:
-    AllocatingMemoryStream m_stream;
+    struct QueuedMessage {
+        SocketMessageHeader header;
+        MessageDataType payload;
+        size_t start_offset { 0 };
+
+        size_t size() const { return sizeof(SocketMessageHeader) + payload.size(); }
+    };
+
+    SinglyLinkedList<QueuedMessage, AK::DefaultSizeCalculationPolicy> m_queued_messages;
+    size_t m_queued_byte_count { 0 };
     Vector<int> m_fds;
     Sync::Mutex m_mutex;
 };
@@ -60,7 +81,7 @@ public:
 
     void wait_until_readable();
 
-    void post_message(Vector<u8> const&, Vector<Attachment>& attachments);
+    void post_message(MessageDataType, Vector<Attachment>& attachments);
 
     enum class ShouldShutdown {
         No,

Libraries/LibIPC/TransportSocketWindows.cpp

@@ -175,7 +175,7 @@ Attachment TransportSocketWindows::deserialize_attachment(ReadonlyBytes& seriali
     VERIFY_NOT_REACHED();
 }
 
-void TransportSocketWindows::post_message(Vector<u8> const& bytes, Vector<Attachment>& attachments)
+void TransportSocketWindows::post_message(MessageDataType bytes, Vector<Attachment>& attachments)
 {
     VERIFY(bytes.size() <= MAX_MESSAGE_PAYLOAD_SIZE);
     VERIFY(attachments.size() <= MAX_MESSAGE_FD_COUNT);

Libraries/LibIPC/TransportSocketWindows.h

@@ -10,6 +10,7 @@
 #include <AK/Queue.h>
 #include <LibCore/Socket.h>
 #include <LibIPC/Attachment.h>
+#include <LibIPC/Forward.h>
 #include <LibIPC/TransportHandle.h>
 
 namespace IPC {
@@ -36,7 +37,7 @@ public:
 
     void wait_until_readable();
 
-    void post_message(Vector<u8> const&, Vector<Attachment>& attachments);
+    void post_message(MessageDataType, Vector<Attachment>& attachments);
 
     enum class ShouldShutdown {
         No,