[Zion] Pass data to message queue as IpcMessage obj.

lib/glacier/container/array.h

@@ -21,7 +21,7 @@ class Array {
 
   Array(const Array&) = delete;
 
-  Array(Array&& other) : data_(other.data), size_(other.size_) {
+  Array(Array&& other) : data_(other.data_), size_(other.size_) {
     other.data_ = nullptr;
     other.size_ = 0;
   }

lib/glacier/container/linked_list.h

@@ -2,6 +2,8 @@
 
 #include <stdint.h>
 
+#include "glacier/memory/move.h"
+
 namespace glcr {
 
 template <typename T>
@@ -15,20 +17,19 @@ class LinkedList {
   uint64_t size() const { return size_; }
 
   void PushBack(const T& item) {
-    size_++;
     ListItem* new_item = new ListItem{
         .item = item,
         .next = nullptr,
     };
-    if (front_ == nullptr) {
-      front_ = new_item;
-      return;
-    }
-    ListItem* litem = front_;
-    while (litem->next != nullptr) {
-      litem = litem->next;
-    }
-    litem->next = new_item;
+    PushBackInternal(new_item);
+  }
+
+  void PushBack(T&& item) {
+    ListItem* new_item = new ListItem{
+        .item = glcr::Move(item),
+        .next = nullptr,
+    };
+    PushBackInternal(new_item);
   }
 
   T PopFront() {
@@ -36,12 +37,12 @@ class LinkedList {
 
     ListItem* old_front = front_;
     front_ = front_->next;
-    T ret = old_front->item;
+    T ret = glcr::Move(old_front->item);
     delete old_front;
     return ret;
   }
 
-  T PeekFront() const { return front_->item; }
+  T& PeekFront() const { return front_->item; }
 
   struct ListItem {
     T item;
@@ -73,6 +74,19 @@ class LinkedList {
   uint64_t size_ = 0;
 
   ListItem* front_ = nullptr;
+
+  void PushBackInternal(ListItem* new_item) {
+    size_++;
+    if (front_ == nullptr) {
+      front_ = new_item;
+      return;
+    }
+    ListItem* litem = front_;
+    while (litem->next != nullptr) {
+      litem = litem->next;
+    }
+    litem->next = new_item;
+  }
 };
 
 }  // namespace glcr

zion/interrupt/interrupt.cpp

@@ -153,7 +153,7 @@ glcr::RefPtr<Port> pci1_port;
 extern "C" void isr_pci1();
 extern "C" void interrupt_pci1(InterruptFrame*) {
   dbgln("Interrupt PCI line 1");
-  pci1_port->Send({}, {});
+  pci1_port->Send({});
   gApic->SignalEOI();
 }
 

zion/lib/message_queue.cpp

@@ -3,41 +3,9 @@
 #include "debug/debug.h"
 #include "scheduler/scheduler.h"
 
-glcr::ErrorCode UnboundedMessageQueue::PushBack(
-    const glcr::ArrayView<uint8_t>& message,
-    const glcr::ArrayView<z_cap_t>& caps, z_cap_t reply_cap) {
-  if (message.size() > 0x1000) {
-    dbgln("Large message size unimplemented: %x", message.size());
-    return glcr::UNIMPLEMENTED;
-  }
-
-  auto msg_struct = glcr::MakeShared<Message>();
-  msg_struct->message = glcr::Array<uint8_t>(message);
-
-  if (reply_cap != kZionInvalidCapability) {
-    // FIXME: We're just trusting that capability has the correct permissions.
-    msg_struct->reply_cap =
-        gScheduler->CurrentProcess().ReleaseCapability(reply_cap);
-  }
-
-  for (uint64_t i = 0; i < caps.size(); i++) {
-    // FIXME: This would feel safer closer to the relevant syscall.
-    // FIXME: Race conditions on get->check->release here. Would be better to
-    // have that as a single call on the process. (This pattern repeats other
-    // places too).
-    auto cap = gScheduler->CurrentProcess().GetCapability(caps[i]);
-    if (!cap) {
-      return glcr::CAP_NOT_FOUND;
-    }
-    if (!cap->HasPermissions(kZionPerm_Transmit)) {
-      return glcr::CAP_PERMISSION_DENIED;
-    }
-    cap = gScheduler->CurrentProcess().ReleaseCapability(caps[i]);
-    msg_struct->caps.PushBack(cap);
-  }
-
+glcr::ErrorCode UnboundedMessageQueue::PushBack(IpcMessage&& message) {
   MutexHolder h(mutex_);
-  pending_messages_.PushBack(msg_struct);
+  pending_messages_.PushBack(glcr::Move(message));
 
   if (blocked_threads_.size() > 0) {
     auto thread = blocked_threads_.PopFront();
@@ -47,10 +15,8 @@ glcr::ErrorCode UnboundedMessageQueue::PushBack(
   return glcr::OK;
 }
 
-glcr::ErrorCode UnboundedMessageQueue::PopFront(uint64_t* num_bytes,
-                                                void* bytes, uint64_t* num_caps,
-                                                z_cap_t* caps,
-                                                z_cap_t* reply_cap) {
+glcr::ErrorOr<IpcMessage> UnboundedMessageQueue::PopFront(
+    uint64_t data_buf_size, uint64_t cap_buf_size) {
   mutex_->Lock();
   while (pending_messages_.empty()) {
     auto thread = gScheduler->CurrentThread();
@@ -63,80 +29,45 @@ glcr::ErrorCode UnboundedMessageQueue::PopFront(uint64_t* num_bytes,
   mutex_->Release();
 
   MutexHolder lock(mutex_);
-  auto next_msg = pending_messages_.PeekFront();
-  if (next_msg->message.size() > *num_bytes) {
+  auto& next_msg = pending_messages_.PeekFront();
+  if (next_msg.data.size() > data_buf_size) {
     return glcr::BUFFER_SIZE;
   }
-  if (next_msg->caps.size() > *num_caps) {
+  if (next_msg.caps.size() > cap_buf_size) {
     return glcr::BUFFER_SIZE;
   }
 
-  next_msg = pending_messages_.PopFront();
-
-  *num_bytes = next_msg->message.size();
-
-  for (uint64_t i = 0; i < *num_bytes; i++) {
-    static_cast<uint8_t*>(bytes)[i] = next_msg->message[i];
-  }
-
-  auto& proc = gScheduler->CurrentProcess();
-  if (reply_cap != nullptr) {
-    if (!next_msg->reply_cap) {
-      dbgln("Tried to read reply capability off of a message without one");
-      return glcr::INTERNAL;
-    }
-    *reply_cap = proc.AddExistingCapability(next_msg->reply_cap);
-  }
-
-  *num_caps = next_msg->caps.size();
-  for (uint64_t i = 0; i < *num_caps; i++) {
-    caps[i] = proc.AddExistingCapability(next_msg->caps.PopFront());
-  }
-  return glcr::OK;
+  return pending_messages_.PopFront();
 }
 
 void UnboundedMessageQueue::WriteKernel(uint64_t init,
                                         glcr::RefPtr<Capability> cap) {
   // FIXME: Add synchronization here in case it is ever used outside of init.
-  auto msg = glcr::MakeShared<Message>();
-  msg->message = glcr::Array<uint8_t>(sizeof(init));
+  IpcMessage msg;
+  msg.data = glcr::Array<uint8_t>(sizeof(init));
 
   uint8_t* data = reinterpret_cast<uint8_t*>(&init);
   for (uint8_t i = 0; i < sizeof(init); i++) {
-    msg->message[i] = data[i];
+    msg.data[i] = data[i];
   }
-  msg->caps.PushBack(cap);
+  msg.caps.PushBack(cap);
 
-  pending_messages_.PushBack(msg);
+  pending_messages_.PushBack(glcr::Move(msg));
 }
 
-glcr::ErrorCode SingleMessageQueue::PushBack(
-    const glcr::ArrayView<uint8_t>& message,
-    const glcr::ArrayView<z_cap_t>& caps, z_cap_t reply_port) {
-  MutexHolder h(mutex_);
-  if (has_written_) {
-    return glcr::FAILED_PRECONDITION;
-  }
-  message_ = message;
-
-  if (reply_port != kZionInvalidCapability) {
+glcr::ErrorCode SingleMessageQueue::PushBack(IpcMessage&& message) {
+  if (message.reply_cap) {
     dbgln("Sent a reply port to a single message queue");
     return glcr::INTERNAL;
   }
 
-  for (uint64_t i = 0; i < caps.size(); i++) {
-    // FIXME: This would feel safer closer to the relevant syscall.
-    auto cap = gScheduler->CurrentProcess().GetCapability(caps[i]);
-    if (!cap) {
-      return glcr::CAP_NOT_FOUND;
-    }
-    if (!cap->HasPermissions(kZionPerm_Transmit)) {
-      return glcr::CAP_PERMISSION_DENIED;
-    }
-    cap = gScheduler->CurrentProcess().ReleaseCapability(caps[i]);
-    caps_.PushBack(cap);
+  MutexHolder h(mutex_);
+  if (has_written_) {
+    dbgln("Double write to reply port.");
+    return glcr::FAILED_PRECONDITION;
   }
 
+  message_ = glcr::Move(message);
   has_written_ = true;
 
   if (blocked_threads_.size() > 0) {
@@ -148,9 +79,8 @@ glcr::ErrorCode SingleMessageQueue::PushBack(
   return glcr::OK;
 }
 
-glcr::ErrorCode SingleMessageQueue::PopFront(uint64_t* num_bytes, void* bytes,
-                                             uint64_t* num_caps, z_cap_t* caps,
-                                             z_cap_t* reply_port) {
+glcr::ErrorOr<IpcMessage> SingleMessageQueue::PopFront(uint64_t data_buf_size,
+                                                       uint64_t cap_buf_size) {
   mutex_->Lock();
   while (!has_written_) {
     auto thread = gScheduler->CurrentThread();
@@ -164,32 +94,18 @@ glcr::ErrorCode SingleMessageQueue::PopFront(uint64_t* num_bytes, void* bytes,
 
   MutexHolder lock(mutex_);
   if (has_read_) {
+    dbgln("Double read from reply port.");
     return glcr::FAILED_PRECONDITION;
   }
 
-  if (message_.size() > *num_bytes) {
+  if (message_.data.size() > data_buf_size) {
     return glcr::BUFFER_SIZE;
   }
-  if (caps_.size() > *num_caps) {
+  if (message_.caps.size() > cap_buf_size) {
     return glcr::BUFFER_SIZE;
   }
 
-  *num_bytes = message_.size();
-  for (uint64_t i = 0; i < message_.size(); i++) {
-    reinterpret_cast<uint8_t*>(bytes)[i] = message_[i];
-  }
-
-  if (reply_port != nullptr) {
-    dbgln("Tried to read a reply port a single message queue");
-    return glcr::INTERNAL;
-  }
-
-  *num_caps = caps_.size();
-  auto& proc = gScheduler->CurrentProcess();
-  for (uint64_t i = 0; i < *num_caps; i++) {
-    caps[i] = proc.AddExistingCapability(caps_.PopFront());
-  }
   has_read_ = true;
 
-  return glcr::OK;
+  return glcr::Move(message_);
 }

zion/lib/message_queue.h

@@ -4,24 +4,31 @@
 #include <glacier/container/array_view.h>
 #include <glacier/container/intrusive_list.h>
 #include <glacier/container/linked_list.h>
+#include <glacier/container/vector.h>
 #include <glacier/memory/ref_ptr.h>
 #include <glacier/memory/shared_ptr.h>
 #include <glacier/status/error.h>
+#include <glacier/status/error_or.h>
 
 #include "capability/capability.h"
 #include "include/ztypes.h"
 #include "object/mutex.h"
 
+struct IpcMessage {
+  glcr::Array<uint8_t> data;
+
+  glcr::Vector<glcr::RefPtr<Capability>> caps;
+
+  glcr::RefPtr<Capability> reply_cap;
+};
+
 class MessageQueue {
  public:
   virtual ~MessageQueue() {}
 
-  virtual glcr::ErrorCode PushBack(const glcr::ArrayView<uint8_t>& message,
-                                   const glcr::ArrayView<z_cap_t>& caps,
-                                   z_cap_t reply_cap) = 0;
-  virtual glcr::ErrorCode PopFront(uint64_t* num_bytes, void* bytes,
-                                   uint64_t* num_caps, z_cap_t* caps,
-                                   z_cap_t* reply_cap) = 0;
+  virtual glcr::ErrorCode PushBack(IpcMessage&&) = 0;
+  virtual glcr::ErrorOr<IpcMessage> PopFront(uint64_t data_buf_size,
+                                             uint64_t cap_buf_size) = 0;
   virtual bool empty() = 0;
 
  protected:
@@ -38,11 +45,9 @@ class UnboundedMessageQueue : public MessageQueue {
   UnboundedMessageQueue& operator=(const UnboundedMessageQueue&) = delete;
   virtual ~UnboundedMessageQueue() override {}
 
-  glcr::ErrorCode PushBack(const glcr::ArrayView<uint8_t>& message,
-                           const glcr::ArrayView<z_cap_t>& caps,
-                           z_cap_t reply_cap) override;
-  glcr::ErrorCode PopFront(uint64_t* num_bytes, void* bytes, uint64_t* num_caps,
-                           z_cap_t* caps, z_cap_t* reply_cap) override;
+  glcr::ErrorCode PushBack(IpcMessage&& message) override;
+  glcr::ErrorOr<IpcMessage> PopFront(uint64_t data_buf_size,
+                                     uint64_t cap_buf_size) override;
 
   void WriteKernel(uint64_t init, glcr::RefPtr<Capability> cap);
 
@@ -52,14 +57,7 @@ class UnboundedMessageQueue : public MessageQueue {
   }
 
  private:
-  struct Message {
-    glcr::Array<uint8_t> message;
-
-    glcr::LinkedList<glcr::RefPtr<Capability>> caps;
-    glcr::RefPtr<Capability> reply_cap;
-  };
-
-  glcr::LinkedList<glcr::SharedPtr<Message>> pending_messages_;
+  glcr::LinkedList<IpcMessage> pending_messages_;
 };
 
 class SingleMessageQueue : public MessageQueue {
@@ -69,11 +67,9 @@ class SingleMessageQueue : public MessageQueue {
   SingleMessageQueue(SingleMessageQueue&&) = delete;
   virtual ~SingleMessageQueue() override {}
 
-  glcr::ErrorCode PushBack(const glcr::ArrayView<uint8_t>& message,
-                           const glcr::ArrayView<z_cap_t>& caps,
-                           z_cap_t reply_cap) override;
-  glcr::ErrorCode PopFront(uint64_t* num_bytes, void* bytes, uint64_t* num_caps,
-                           z_cap_t* caps, z_cap_t* reply_cap) override;
+  glcr::ErrorCode PushBack(IpcMessage&&) override;
+  glcr::ErrorOr<IpcMessage> PopFront(uint64_t data_buf_size,
+                                     uint64_t cap_buf_size) override;
 
   bool empty() override {
     MutexHolder h(mutex_);
@@ -83,6 +79,5 @@ class SingleMessageQueue : public MessageQueue {
  private:
   bool has_written_ = false;
   bool has_read_ = false;
-  glcr::Array<uint8_t> message_;
-  glcr::LinkedList<glcr::RefPtr<Capability>> caps_;
+  IpcMessage message_;
 };

zion/object/ipc_object.cpp

@@ -2,26 +2,11 @@
 
 #include "scheduler/scheduler.h"
 
-glcr::ErrorCode IpcObject::Send(const glcr::ArrayView<uint8_t>& message,
-                                const glcr::ArrayView<z_cap_t>& caps) {
-  return Send(message, caps, kZionInvalidCapability);
+glcr::ErrorCode IpcObject::Send(IpcMessage&& message) {
+  return GetSendMessageQueue().PushBack(glcr::Move(message));
 }
 
-glcr::ErrorCode IpcObject::Send(const glcr::ArrayView<uint8_t>& message,
-                                const glcr::ArrayView<z_cap_t>& caps,
-                                const z_cap_t reply_port) {
-  auto& message_queue = GetSendMessageQueue();
-  return message_queue.PushBack(message, caps, reply_port);
-}
-
-glcr::ErrorCode IpcObject::Recv(uint64_t* num_bytes, void* bytes,
-                                uint64_t* num_caps, z_cap_t* caps) {
-  return Recv(num_bytes, bytes, num_caps, caps, nullptr);
-}
-
-glcr::ErrorCode IpcObject::Recv(uint64_t* num_bytes, void* bytes,
-                                uint64_t* num_caps, z_cap_t* caps,
-                                z_cap_t* reply_port) {
-  auto& message_queue = GetRecvMessageQueue();
-  return message_queue.PopFront(num_bytes, bytes, num_caps, caps, reply_port);
+glcr::ErrorOr<IpcMessage> IpcObject::Recv(uint64_t data_buf_size,
+                                          uint64_t cap_buf_size) {
+  return GetRecvMessageQueue().PopFront(data_buf_size, cap_buf_size);
 }

zion/object/ipc_object.h

@@ -11,17 +11,10 @@ class IpcObject : public KernelObject {
   IpcObject(){};
   virtual ~IpcObject() {}
 
-  virtual glcr::ErrorCode Send(const glcr::ArrayView<uint8_t>& message,
-                               const glcr::ArrayView<z_cap_t>& caps) final;
-  virtual glcr::ErrorCode Send(const glcr::ArrayView<uint8_t>& message,
-                               const glcr::ArrayView<z_cap_t>& caps,
-                               const z_cap_t reply_port) final;
+  virtual glcr::ErrorCode Send(IpcMessage&& message) final;
 
-  virtual glcr::ErrorCode Recv(uint64_t* num_bytes, void* bytes,
-                               uint64_t* num_caps, z_cap_t* caps) final;
-  virtual glcr::ErrorCode Recv(uint64_t* num_bytes, void* bytes,
-                               uint64_t* num_caps, z_cap_t* caps,
-                               z_cap_t* reply_port) final;
+  virtual glcr::ErrorOr<IpcMessage> Recv(uint64_t data_buf_size,
+                                         uint64_t cap_buf_size) final;
 
   bool HasMessages() { return !GetRecvMessageQueue().empty(); }
 

zion/syscall/ipc.cpp

@@ -1,6 +1,7 @@
 #include "syscall/ipc.h"
 
 #include "capability/capability.h"
+#include "debug/debug.h"
 #include "interrupt/interrupt.h"
 #include "object/endpoint.h"
 #include "object/reply_port.h"
@@ -12,6 +13,76 @@ glcr::ArrayView<uint8_t> Buffer(const void* bytes, uint64_t num_bytes) {
   return glcr::ArrayView(reinterpret_cast<uint8_t*>(const_cast<void*>(bytes)),
                          num_bytes);
 }
+
+template <typename T>
+glcr::ErrorOr<IpcMessage> TranslateRequestToIpcMessage(const T& req) {
+  if (req.num_bytes > 0x1000) {
+    dbgln("Large message size unimplemented: %x", req.num_bytes);
+    return glcr::UNIMPLEMENTED;
+  }
+
+  IpcMessage message;
+  message.data = Buffer(req.data, req.num_bytes);
+
+  glcr::ArrayView<const z_cap_t> caps(req.caps, req.num_caps);
+
+  message.caps.Resize(caps.size());
+  for (uint64_t i = 0; i < caps.size(); i++) {
+    // FIXME: This would feel safer closer to the relevant syscall.
+    // FIXME: Race conditions on get->check->release here. Would be better to
+    // have that as a single call on the process. (This pattern repeats other
+    // places too).
+    auto cap = gScheduler->CurrentProcess().GetCapability(caps[i]);
+    if (!cap) {
+      return glcr::CAP_NOT_FOUND;
+    }
+    if (!cap->HasPermissions(kZionPerm_Transmit)) {
+      return glcr::CAP_PERMISSION_DENIED;
+    }
+    message.caps.PushBack(
+        gScheduler->CurrentProcess().ReleaseCapability(caps[i]));
+  }
+
+  return message;
+}
+
+template <typename T>
+glcr::ErrorCode TranslateIpcMessageToResponse(const IpcMessage& message,
+                                              T* resp) {
+  if (message.data.size() > *resp->num_bytes) {
+    return glcr::BUFFER_SIZE;
+  }
+  if (message.caps.size() > *resp->num_caps) {
+    return glcr::BUFFER_SIZE;
+  }
+
+  *resp->num_bytes = message.data.size();
+  for (uint64_t i = 0; i < message.data.size(); i++) {
+    reinterpret_cast<uint8_t*>(resp->data)[i] = message.data[i];
+  }
+
+  *resp->num_caps = message.caps.size();
+  auto& proc = gScheduler->CurrentProcess();
+  for (uint64_t i = 0; i < *resp->num_caps; i++) {
+    resp->caps[i] = proc.AddExistingCapability(message.caps[i]);
+  }
+  return glcr::OK;
+}
+
+template <typename T>
+glcr::ErrorCode TranslateIpcMessageToResponseWithReplyPort(
+    const IpcMessage& message, T* resp) {
+  TranslateIpcMessageToResponse(message, resp);
+
+  if (!message.reply_cap) {
+    dbgln("Tried to read reply capability off of a message without one");
+    return glcr::INTERNAL;
+  }
+  auto& proc = gScheduler->CurrentProcess();
+  *resp->reply_port_cap = proc.AddExistingCapability(message.reply_cap);
+  return glcr::OK;
+}
+
 }  // namespace
 
 glcr::ErrorCode ChannelCreate(ZChannelCreateReq* req) {
@@ -28,8 +99,8 @@ glcr::ErrorCode ChannelSend(ZChannelSendReq* req) {
   RET_ERR(ValidateCapability<Channel>(chan_cap, kZionPerm_Write));
 
   auto chan = chan_cap->obj<Channel>();
-  return chan->Send(Buffer(req->data, req->num_bytes),
-                    glcr::ArrayView<z_cap_t>(req->caps, req->num_caps));
+  ASSIGN_OR_RETURN(IpcMessage message, TranslateRequestToIpcMessage(*req));
+  return chan->Send(glcr::Move(message));
 }
 
 glcr::ErrorCode ChannelRecv(ZChannelRecvReq* req) {
@@ -38,7 +109,8 @@ glcr::ErrorCode ChannelRecv(ZChannelRecvReq* req) {
   RET_ERR(ValidateCapability<Channel>(chan_cap, kZionPerm_Read));
 
   auto chan = chan_cap->obj<Channel>();
-  return chan->Recv(req->num_bytes, req->data, req->num_caps, req->caps);
+  ASSIGN_OR_RETURN(IpcMessage msg, chan->Recv(*req->num_bytes, *req->num_caps));
+  return TranslateIpcMessageToResponse(msg, req);
 }
 
 glcr::ErrorCode PortCreate(ZPortCreateReq* req) {
@@ -54,8 +126,8 @@ glcr::ErrorCode PortSend(ZPortSendReq* req) {
   RET_ERR(ValidateCapability<Port>(port_cap, kZionPerm_Write));
 
   auto port = port_cap->obj<Port>();
-  return port->Send(Buffer(req->data, req->num_bytes),
-                    glcr::ArrayView<z_cap_t>(req->caps, req->num_caps));
+  ASSIGN_OR_RETURN(IpcMessage message, TranslateRequestToIpcMessage(*req));
+  return port->Send(glcr::Move(message));
 }
 
 glcr::ErrorCode PortRecv(ZPortRecvReq* req) {
@@ -64,7 +136,8 @@ glcr::ErrorCode PortRecv(ZPortRecvReq* req) {
   RET_ERR(ValidateCapability<Port>(port_cap, kZionPerm_Read));
 
   auto port = port_cap->obj<Port>();
-  return port->Recv(req->num_bytes, req->data, req->num_caps, req->caps);
+  ASSIGN_OR_RETURN(IpcMessage msg, port->Recv(*req->num_bytes, *req->num_caps));
+  return TranslateIpcMessageToResponse(msg, req);
 }
 
 glcr::ErrorCode PortPoll(ZPortPollReq* req) {
@@ -78,7 +151,8 @@ glcr::ErrorCode PortPoll(ZPortPollReq* req) {
   if (!port->HasMessages()) {
     return glcr::EMPTY;
   }
-  return port->Recv(req->num_bytes, req->data, req->num_caps, req->caps);
+  ASSIGN_OR_RETURN(IpcMessage msg, port->Recv(*req->num_bytes, *req->num_caps));
+  return TranslateIpcMessageToResponse(msg, req);
 }
 
 glcr::ErrorCode IrqRegister(ZIrqRegisterReq* req) {
@@ -108,12 +182,11 @@ glcr::ErrorCode EndpointSend(ZEndpointSendReq* req) {
 
   auto reply_port = ReplyPort::Create();
   *req->reply_port_cap = proc.AddNewCapability(reply_port, kZionPerm_Read);
-  uint64_t reply_port_cap_to_send =
-      proc.AddNewCapability(reply_port, kZionPerm_Write | kZionPerm_Transmit);
-  return endpoint->Send(
-      Buffer(req->data, req->num_bytes),
-      glcr::ArrayView<z_cap_t>(const_cast<z_cap_t*>(req->caps), req->num_caps),
-      reply_port_cap_to_send);
+
+  ASSIGN_OR_RETURN(IpcMessage message, TranslateRequestToIpcMessage(*req));
+  message.reply_cap = glcr::MakeRefCounted<Capability>(
+      reply_port, kZionPerm_Write | kZionPerm_Transmit);
+  return endpoint->Send(glcr::Move(message));
 }
 
 glcr::ErrorCode EndpointRecv(ZEndpointRecvReq* req) {
@@ -123,13 +196,9 @@ glcr::ErrorCode EndpointRecv(ZEndpointRecvReq* req) {
   ValidateCapability<Endpoint>(endpoint_cap, kZionPerm_Read);
   auto endpoint = endpoint_cap->obj<Endpoint>();
 
-  uint64_t num_caps = 1;
-  RET_ERR(endpoint->Recv(req->num_bytes, req->data, req->num_caps, req->caps,
-                         req->reply_port_cap));
-  if (num_caps != 1) {
-    return glcr::INTERNAL;
-  }
-  return glcr::OK;
+  ASSIGN_OR_RETURN(IpcMessage msg,
+                   endpoint->Recv(*req->num_bytes, *req->num_caps));
+  return TranslateIpcMessageToResponseWithReplyPort(msg, req);
 }
 
 glcr::ErrorCode ReplyPortSend(ZReplyPortSendReq* req) {
@@ -138,8 +207,8 @@ glcr::ErrorCode ReplyPortSend(ZReplyPortSendReq* req) {
   ValidateCapability<ReplyPort>(reply_port_cap, kZionPerm_Read);
   auto reply_port = reply_port_cap->obj<ReplyPort>();
 
-  return reply_port->Send(Buffer(req->data, req->num_bytes),
-                          glcr::ArrayView<z_cap_t>(req->caps, req->num_caps));
+  ASSIGN_OR_RETURN(IpcMessage message, TranslateRequestToIpcMessage(*req));
+  return reply_port->Send(glcr::Move(message));
 }
 glcr::ErrorCode ReplyPortRecv(ZReplyPortRecvReq* req) {
   auto& proc = gScheduler->CurrentProcess();
@@ -148,5 +217,7 @@ glcr::ErrorCode ReplyPortRecv(ZReplyPortRecvReq* req) {
   ValidateCapability<ReplyPort>(reply_port_cap, kZionPerm_Read);
   auto reply_port = reply_port_cap->obj<ReplyPort>();
 
-  return reply_port->Recv(req->num_bytes, req->data, req->num_caps, req->caps);
+  ASSIGN_OR_RETURN(IpcMessage msg,
+                   reply_port->Recv(*req->num_bytes, *req->num_caps));
+  return TranslateIpcMessageToResponse(msg, req);
 }