Add protocol for signalling consumption from consumer to producer

IPC message can optionally transfer buffers from the producer to
the consumer. These buffers are not copied while the message is in
transit. Consequently, the consumer must inform the producer when
it has finished processing the buffer.

This patch adds a monitor variable that the producer uses to wait
for the competion of the consumers message handling. Once done, the
consumer informs the producer to continue.

The reply can also return a buffer back to the consumer. The lifetime
of this buffer is managed by a higher-level protocol.

Author: tdz

src/IPCQueue.c

@@ -4,46 +4,150 @@
 
 #include "IPCQueue.h"
 
+typedef struct
+{
+  uint32_t mDWord0;
+  uint32_t mDWord1;
+  uint32_t mStatus;
+  void*    mBuffer;
+} IPCMessageReply;
+
 /*
  * IPCMessage
  */
 
 int
 IPCMessageInit(IPCMessage* aMsg)
 {
+  aMsg->mMonitor = xQueueCreate(1, sizeof(IPCMessageReply));
+  if (!aMsg->mMonitor) {
+    return -1;
+  }
+
   aMsg->mDWord0 = 0;
   aMsg->mDWord1 = 0;
-  aMsg->mStatus = 0;
+  aMsg->mStatus = IPC_MESSAGE_STATE_CLEAR;
   aMsg->mBuffer = NULL;
 
   return 0;
 }
 
+void
+IPCMessageUninit(IPCMessage* aMsg)
+{
+  vQueueDelete(aMsg->mMonitor);
+}
+
 uint32_t
 IPCMessageGetBufferLength(const IPCMessage* aMsg)
 {
   return aMsg->mStatus & 0x00ffffff;
 }
 
 int
-IPCMessageProduce(IPCMessage* aMsg)
+IPCMessageProduce(IPCMessage* aMsg, uint32_t aLength, void* aBuffer)
+{
+  switch (aMsg->mStatus & 0xf0000000) {
+  case IPC_MESSAGE_STATE_CLEAR:      /* fall through */
+  case IPC_MESSAGE_STATE_PRODUCED:   /* fall through */
+  case IPC_MESSAGE_STATE_ERROR:
+    /* We're good if the message is currently not in transit. */
+    break;
+  case IPC_MESSAGE_STATE_PENDING:    /* fall through */
+  default:
+    /* If the message is currently in transit or the status is
+     * unknown, we don't produce a new one. Better abort here. */
+    return -1;
+  }
+
+  aMsg->mDWord0 = 0;
+  aMsg->mDWord1 = 0;
+  aMsg->mStatus = 0;
+  aMsg->mStatus |= IPC_MESSAGE_STATE_PRODUCED;
+  aMsg->mStatus |= aLength;
+  aMsg->mBuffer = aBuffer;
+
+  return 0;
+}
+
+static int
+WaitForConsumption(IPCMessage* aMsg, IPCMessageReply* aReply)
+{
+  uint32_t state = aMsg->mStatus & 0xf0000000;
+  if (state != IPC_MESSAGE_STATE_PENDING) {
+    return -1;
+  }
+  BaseType_t ret = xQueueReceive(aMsg->mMonitor, aReply, portMAX_DELAY);
+  if (ret != pdPASS) {
+    return -1;
+  };
+  return 0;
+}
+
+int
+IPCMessageWaitForReply(IPCMessage* aMsg)
 {
-  /* TODO: At some point we have to implement efficient IPC with
-   * large buffers. IPCMessageProduce() will signal the end of the
-   * message constrcution **on the producer tast.** A produced
-   * message can be send over over an IPC queue to a consumer task.
-   * The consumer calls IPCMessageConsume() after it processed the
-   * buffer. The producer can then release the buffer. */
+  IPCMessageReply reply;
+  int res = WaitForConsumption(aMsg, &reply);
+  if (res < 0) {
+    return -1;
+  };
+  aMsg->mDWord0 = reply.mDWord0;
+  aMsg->mDWord1 = reply.mDWord1;
+  aMsg->mStatus = reply.mStatus;
+  aMsg->mBuffer = reply.mBuffer;
   return 0;
 }
 
 int
-IPCMessageConsume(IPCMessage* aMsg)
+IPCMessageWaitForConsumption(IPCMessage* aMsg)
 {
-  /* TODO: See IPCMessageProduce() */
+  IPCMessageReply reply;
+  int res = WaitForConsumption(aMsg, &reply);
+  if (res < 0) {
+    return -1;
+  };
+  aMsg->mStatus &= 0x0fffffff; /* clear pending status */
+  return 0;
+}
+
+static int
+ConsumeAndReply(IPCMessage* aMsg, const IPCMessageReply* aReply)
+{
+  BaseType_t res = xQueueSend(aMsg->mMonitor, &aReply, 0);
+  if (res != pdPASS) {
+    return -1;
+  }
   return 0;
 }
 
+int
+IPCMessageConsumeAndReply(IPCMessage* aMsg,
+                          uint32_t aDWord0, uint32_t aDWord1,
+                          uint32_t aFlags, uint32_t aLength,
+                          void* aBuffer)
+{
+  IPCMessageReply reply = {
+    .mDWord0 = aDWord0,
+    .mDWord1 = aDWord1,
+    .mStatus = aFlags | aLength,
+    .mBuffer = aBuffer
+  };
+  return ConsumeAndReply(aMsg, &reply);
+}
+
+int
+IPCMessageConsume(IPCMessage* aMsg)
+{
+  static const IPCMessageReply sReply = {
+    .mDWord0 = 0,
+    .mDWord1 = 0,
+    .mStatus = IPC_MESSAGE_STATE_CLEAR,
+    .mBuffer = NULL,
+  };
+  return ConsumeAndReply(aMsg, &sReply);
+}
+
 /*
  * IPCMessageQueue
  */
@@ -58,14 +162,42 @@ IPCMessageQueueInit(IPCMessageQueue* aMsgQueue)
   return 0;
 }
 
+void
+IPCMessageQueueUninit(IPCMessageQueue* aMsgQueue)
+{
+  vQueueDelete(aMsgQueue->mWaitQueue);
+}
+
 int
 IPCMessageQueueConsume(IPCMessageQueue* aMsgQueue, IPCMessage* aMsg)
 {
-	  BaseType_t res = xQueueSend(aMsgQueue->mWaitQueue, aMsg, 0);
-    if (res != pdPASS){
-      return -1;
-    }
-    return 0;
+  uint32_t status = aMsg->mStatus;
+
+  switch (status & 0xf0000000) {
+  case IPC_MESSAGE_STATE_PRODUCED:    /* fall through */
+    /* We're good if the message has been produced correctly. */
+    break;
+  case IPC_MESSAGE_STATE_CLEAR:       /* fall through */
+  case IPC_MESSAGE_STATE_PENDING:     /* fall through */
+  case IPC_MESSAGE_STATE_ERROR:       /* fall through */
+  default:
+    /* In any other case, the message is probably not ready for
+     * consumption. Better abort here. */
+    return -1;
+  }
+
+  aMsg->mStatus &= 0x0fffffff;
+  aMsg->mStatus |= IPC_MESSAGE_STATE_PENDING;
+
+	BaseType_t res = xQueueSend(aMsgQueue->mWaitQueue, aMsg, 0);
+  if (res != pdPASS){
+    goto err_xQueueSend;
+  }
+  return 0;
+
+err_xQueueSend:
+  aMsg->mStatus = status;
+  return -1;
 }
 
 int

src/IPCQueue.h

@@ -13,25 +13,50 @@
  * IPCMessage
  */
 
+enum IPCMessageStatus {
+  IPC_MESSAGE_STATE_CLEAR    = 0x00000000,
+  IPC_MESSAGE_STATE_PRODUCED = 0x10000000,
+  IPC_MESSAGE_STATE_PENDING  = 0x20000000,
+  IPC_MESSAGE_STATE_ERROR    = 0x30000000
+};
+
 typedef struct
 {
+  /* Internal monitor for signalling */
+  QueueHandle_t mMonitor;
+
   /* Two dword for data transfers. */
   uint32_t mDWord0;
   uint32_t mDWord1;
-  /* Message flags [31:24] and buffer length [23:0] */
+  /* Message state [31:28], flags [27:24], and buffer length [23:0] */
   uint32_t mStatus;
   /* Message buffer */
-  const void* mBuffer;
+  void* mBuffer;
 } IPCMessage;
 
 int
 IPCMessageInit(IPCMessage* aMsg);
 
+void
+IPCMessageUninit(IPCMessage* aMsg);
+
 uint32_t
 IPCMessageGetBufferLength(const IPCMessage* aMsg);
 
 int
-IPCMessageProduce(IPCMessage* aMsg);
+IPCMessageProduce(IPCMessage* aMsg, uint32_t aLength, void* aBuffer);
+
+int
+IPCMessageWaitForReply(IPCMessage* aMsg);
+
+int
+IPCMessageWaitForConsumption(IPCMessage* aMsg);
+
+int
+IPCMessageConsumeAndReply(IPCMessage* aMsg,
+                          uint32_t aDWord0, uint32_t aDWord1,
+                          uint32_t aFlags, uint32_t aLength,
+                          void* aBuffer);
 
 int
 IPCMessageConsume(IPCMessage* aMsg);

src/Producer.c

@@ -19,22 +19,34 @@ Run(ProducerTask* aProducer)
     "This"," is"," a"," SensorWeb"," device"," by"," Mozilla.\n\r"
   };
 
+  IPCMessage msg;
+  int res = IPCMessageInit(&msg);
+  if (res < 0) {
+    return;
+  }
+
   for (unsigned long i = 0;; i = (i + 1) % ArrayLength(sMessage)) {
-    IPCMessage msg;
-    int res = IPCMessageInit(&msg);
+
+    res = IPCMessageProduce(&msg, strlen(sMessage[i]) + 1, (void*)sMessage[i]);
     if (res < 0) {
       return;
     }
-    msg.mBuffer = sMessage[i];
-    msg.mStatus = strlen(msg.mBuffer) + 1;
-
-    IPCMessageProduce(&msg);
 
     res = IPCMessageQueueConsume(aProducer->mSendQueue, &msg);
     if (res < 0) {
       return;
     }
+
     vTaskDelay(TicksOfMSecs(200));
+
+    /* While we have been waiting in vTaskDelay(), the consumer
+     * probably processed our message. Waiting for consumption should
+     * have the reply ready.
+     */
+    res = IPCMessageWaitForConsumption(&msg);
+    if (res < 0) {
+      return;
+    }
   }
 }