nrf_rpc: implement timeout for response

Introduce necessary changes to support timing out waiting
for a command response:
1. Handle nrf_rpc_os_msg_get() returning null data pointer,
   indicating that the response has not arrived within
   the required maximum time.
2. Introduce context mutex to protect all context members.
   This is necessary because now the response may arrive
   after the initiating thread has already released the
   context.

Note that 2 may not sufficient solution to address another
scenario where the context is freed by the initiator but
is then reused before a delayed response arrives - to solve
this, we likely need to introduce the concept of session
ID or conversation ID to detect outdated packets.

Signed-off-by: Damian Krolik <[email protected]>

Author: Damian-Nordic

nrf_rpc/CHANGELOG.rst

@@ -9,6 +9,18 @@ Changelog
 
 All the notable changes to this project are documented on this page.
 
+nRF Connect SDK v3.0.99
+***********************
+
+Added
+=====
+
+* Added a timeout feature for awaiting command responses.
+  This change required making the following changes to the nRF RPC OS abstraction layer:
+
+  * Introduced the :c:struct:`nrf_rpc_os_mutex` structure to protect the nRF RPC thread-local context.
+  * Updated the :c:func:`nrf_rpc_os_msg_get` function to unlock the thread-local context mutex and to allow returning no data from the function if a timeout occurs.
+
 nRF Connect SDK v2.8.0
 **********************
 

nrf_rpc/nrf_rpc.c

@@ -52,6 +52,8 @@ struct nrf_rpc_cmd_ctx {
 				    */
 	nrf_rpc_handler_t handler; /* Response handler provided be the user. */
 	void *handler_data;	   /* Pointer for the response handler. */
+	struct nrf_rpc_os_mutex mutex; /* Mutex for protecting all context
+					* members. */
 	struct nrf_rpc_os_msg recv_msg;
 				   /* Message passing between transport
 				    * receive callback and a thread that waits
@@ -187,6 +189,7 @@ static struct nrf_rpc_cmd_ctx *cmd_ctx_alloc(void)
 	NRF_RPC_ASSERT(index < CONFIG_NRF_RPC_CMD_CTX_POOL_SIZE);
 
 	ctx = &cmd_ctx_pool[index];
+	nrf_rpc_os_mutex_lock(&ctx->mutex);
 	ctx->handler = NULL;
 	ctx->remote_id = NRF_RPC_ID_UNKNOWN;
 	ctx->use_count = 1;
@@ -200,6 +203,7 @@ static struct nrf_rpc_cmd_ctx *cmd_ctx_alloc(void)
 
 static void cmd_ctx_free(struct nrf_rpc_cmd_ctx *ctx)
 {
+	nrf_rpc_os_mutex_unlock(&ctx->mutex);
 	nrf_rpc_os_tls_set(NULL);
 	nrf_rpc_os_ctx_pool_release(ctx->id);
 }
@@ -741,7 +745,7 @@ static void receive_handler(const struct nrf_rpc_tr *transport, const uint8_t *p
 	int err;
 	int remote_err;
 	struct header hdr;
-	struct nrf_rpc_cmd_ctx *cmd_ctx;
+	struct nrf_rpc_cmd_ctx *cmd_ctx = NULL;
 	const struct nrf_rpc_group *group = NULL;
 
 	err = header_decode(packet, len, &hdr);
@@ -797,6 +801,17 @@ static void receive_handler(const struct nrf_rpc_tr *transport, const uint8_t *p
 			err = -NRF_EBADMSG;
 			goto cleanup_and_exit;
 		}
+
+		nrf_rpc_os_mutex_lock(&cmd_ctx->mutex);
+
+		if (cmd_ctx->use_count == 0) {
+			/* Context initiator has likely timed out and is no longer interested
+			 * in receiving messages for this conversation.
+			 */
+			nrf_rpc_os_mutex_unlock(&cmd_ctx->mutex);
+			goto cleanup_and_exit;
+		}
+
 		if (cmd_ctx->handler != NULL &&
 		    hdr.type == NRF_RPC_PACKET_TYPE_RSP &&
 		    auto_free_rx_buf(transport)) {
@@ -805,16 +820,17 @@ static void receive_handler(const struct nrf_rpc_tr *transport, const uint8_t *p
 					 cmd_ctx->handler_data);
 			nrf_rpc_os_msg_set(&cmd_ctx->recv_msg,
 					   RESPONSE_HANDLED_PTR, 0);
+			nrf_rpc_os_mutex_unlock(&cmd_ctx->mutex);
 			goto cleanup_and_exit;
 
 		} else {
 			nrf_rpc_os_msg_set(&cmd_ctx->recv_msg, packet, len);
+			nrf_rpc_os_mutex_unlock(&cmd_ctx->mutex);
 
 			if (auto_free_rx_buf(transport)) {
 				nrf_rpc_os_event_wait(&group->data->decode_done_event,
 						      NRF_RPC_OS_WAIT_FOREVER);
 			}
-
 		}
 		return;
 
@@ -895,7 +911,7 @@ void nrf_rpc_decoding_done(const struct nrf_rpc_group *group, const uint8_t *pac
  * @param[out] rsp_len    If not NULL contains response packet length.
  * @return 0 on success or negative error code.
  */
-static void wait_for_response(const struct nrf_rpc_group *group, struct nrf_rpc_cmd_ctx *cmd_ctx,
+static int wait_for_response(const struct nrf_rpc_group *group, struct nrf_rpc_cmd_ctx *cmd_ctx,
 			      const uint8_t **rsp_packet, size_t *rsp_len)
 {
 	size_t len;
@@ -907,12 +923,14 @@ static void wait_for_response(const struct nrf_rpc_group *group, struct nrf_rpc_
 	NRF_RPC_DBG("Waiting for a response");
 
 	do {
-		nrf_rpc_os_msg_get(&cmd_ctx->recv_msg, &packet, &len);
+		nrf_rpc_os_msg_get(&cmd_ctx->recv_msg, &cmd_ctx->mutex, &packet, &len);
 
-		NRF_RPC_ASSERT(packet != NULL);
+		if (packet == NULL) {
+			return -NRF_ETIMEDOUT;
+		}
 
 		if (packet == RESPONSE_HANDLED_PTR) {
-			return;
+			return 0;
 		}
 
 		type = parse_incoming_packet(cmd_ctx, packet, len);
@@ -934,6 +952,8 @@ static void wait_for_response(const struct nrf_rpc_group *group, struct nrf_rpc_
 
 		nrf_rpc_decoding_done(group, &packet[NRF_RPC_HEADER_SIZE]);
 	}
+
+	return 0;
 }
 
 int nrf_rpc_cmd_common(const struct nrf_rpc_group *group, uint32_t cmd,
@@ -986,7 +1006,7 @@ int nrf_rpc_cmd_common(const struct nrf_rpc_group *group, uint32_t cmd,
 	err = send(group, full_packet, len + NRF_RPC_HEADER_SIZE);
 
 	if (err >= 0) {
-		wait_for_response(group, cmd_ctx, rsp_packet, rsp_len);
+		err = wait_for_response(group, cmd_ctx, rsp_packet, rsp_len);
 	}
 
 	cmd_ctx->handler = old_handler;
@@ -1153,6 +1173,10 @@ int nrf_rpc_init(nrf_rpc_err_handler_t err_handler)
 
 	for (i = 0; i < CONFIG_NRF_RPC_CMD_CTX_POOL_SIZE; i++) {
 		cmd_ctx_pool[i].id = i;
+		err = nrf_rpc_os_mutex_init(&cmd_ctx_pool[i].mutex);
+		if (err < 0) {
+			return err;
+		}
 		err = nrf_rpc_os_msg_init(&cmd_ctx_pool[i].recv_msg);
 		if (err < 0) {
 			return err;

nrf_rpc/template/nrf_rpc_os_tmpl.h

@@ -32,6 +32,9 @@ extern "C" {
 /** @brief Structure to pass events between threads. */
 struct nrf_rpc_os_event;
 
+/** @brief Stucture to assure exclusive access to thread contexts. */
+struct nrf_rpc_os_mutex;
+
 /** @brief Structure to pass messages between threads. */
 struct nrf_rpc_os_msg;
 
@@ -94,6 +97,26 @@ void nrf_rpc_os_event_set(struct nrf_rpc_os_event *event);
  */
 int nrf_rpc_os_event_wait(struct nrf_rpc_os_event *event, int32_t timeout);
 
+/** @brief Initialize mutex structure.
+ *
+ * @param mutex Pointer to mutex structure.
+ *
+ * @return    0 on success or negative error code.
+ */
+int nrf_rpc_os_mutex_init(struct nrf_rpc_os_mutex *mutex);
+
+/** @brief Lock mutex.
+ *
+ * @param mutex Pointer to mutex structure.
+ */
+void nrf_rpc_os_mutex_lock(struct nrf_rpc_os_mutex *mutex);
+
+/** @brief Unlock mutex.
+ *
+ * @param mutex Pointer to mutex structure.
+ */
+void nrf_rpc_os_mutex_unlock(struct nrf_rpc_os_mutex *mutex);
+
 /** @brief Initialize message passing structure.
  *
  * @param msg Structure to initialize.
@@ -118,14 +141,19 @@ void nrf_rpc_os_msg_set(struct nrf_rpc_os_msg *msg, const uint8_t *data,
 /** @brief Get a message.
  *
  * If message was not set yet then this function waits.
+ * When this function starts waiting, it atomically unlocks the passed mutex.
+ *
+ * This function MAY time out, which is indicated by returning a NULL data
+ * pointer.
  *
- * @param[in]  msg  Message passing structure.
- * @param[out] data Received data pointer. Data is passed as a pointer, so no
- *                  copying is done.
- * @param[out] len  Length of the `data`.
+ * @param[in]  msg   Message passing structure.
+ * @param[in]  mutex Pointer to mutex to unlock before entering the wait state.
+ * @param[out] data  Received data pointer. Data is passed as a pointer, so no
+ *                   copying is done.
+ * @param[out] len   Length of the `data`.
  */
-void nrf_rpc_os_msg_get(struct nrf_rpc_os_msg *msg, const uint8_t **data,
-			size_t *len);
+void nrf_rpc_os_msg_get(struct nrf_rpc_os_msg *msg, struct nrf_rpc_os_mutex *mutex,
+			const uint8_t **data, size_t *len);
 
 /** @brief Get TLS (Thread Local Storage) for nRF RPC.
  *