app: Implement sm_at_host_lock() for multi part responses

Implement sm_at_host_lock() that can be used to block URC handling
and RX processing while outputting multi-part responses using
rsp_send(), rsp_send_to() and data_send().

Signed-off-by: Seppo Takalo <seppo.takalo@nordicsemi.no>

Author: SeppoTakalo

app/src/sm_at_host.c

@@ -152,7 +152,7 @@ struct sm_at_host_ctx {
 
 	/* AT command reception state (for cmd_rx_handler) */
 	bool inside_quotes;
-	bool executing;
+	atomic_t executing_lock;
 	size_t at_cmd_len;
 	size_t echo_len;
 	uint8_t prev_character;
@@ -442,7 +442,7 @@ static void at_pipe_rx_work_fn(struct k_work *work)
 	}
 
 	/* Do not parse more commands, if we are still executing */
-	if (ctx->executing) {
+	if (atomic_get(&ctx->executing_lock) > 0) {
 		return;
 	}
 
@@ -1168,7 +1168,8 @@ static void cmd_send(struct sm_at_host_ctx *ctx, uint8_t *buf, size_t cmd_length
 	}
 
 	/* Block CMDs & URCs while command is executing, even if idle timer triggers */
-	ctx->executing = true;
+	atomic_inc(&ctx->executing_lock);
+
 	/* Send to modem.
 	 * Reserve space for CRLF in the response buffer.
 	 */
@@ -1178,12 +1179,12 @@ static void cmd_send(struct sm_at_host_ctx *ctx, uint8_t *buf, size_t cmd_length
 	if (err == -AT_COMMAND_CONTINUE_RET) {
 		return;
 	} else if (err == -SILENT_AT_COMMAND_RET) {
-		ctx->executing = false;
+		atomic_dec(&ctx->executing_lock);
 		return;
 	} else if (err < 0) {
 		LOG_ERR("AT command failed: %d", err);
 		rsp_send_error();
-		ctx->executing = false;
+		atomic_dec(&ctx->executing_lock);
 		return;
 	} else if (err > 0) {
 		LOG_ERR("AT command error (%d), type: %d: value: %d", err,
@@ -1204,22 +1205,36 @@ static void cmd_send(struct sm_at_host_ctx *ctx, uint8_t *buf, size_t cmd_length
 			LOG_ERR("AT command response failed: %d", err);
 		}
 	}
-	ctx->executing = false;
+	atomic_dec(&ctx->executing_lock);
 }
 
 void sm_at_host_cmd_done(struct sm_at_host_ctx *ctx)
 {
 	if (!sm_at_ctx_check(ctx)) {
 		return;
 	}
-	ctx->executing = false;
 
 	/* Continue processing received data, if there is any */
-	if (is_open(ctx)) {
+	if (atomic_dec(&ctx->executing_lock) == 1 && is_open(ctx)) {
+		sm_at_host_event_notify(ctx, SM_EVENT_URC);
 		k_work_submit_to_queue(&sm_work_q, &ctx->rx_work);
 	}
 }
 
+void sm_at_host_lock_ctx(struct sm_at_host_ctx *ctx)
+{
+	if (!sm_at_ctx_check(ctx)) {
+		LOG_ERR("invalid context");
+		return;
+	}
+
+	if (is_idle(ctx)) {
+		flush_pipe_urcs(ctx);
+	}
+
+	atomic_inc(&ctx->executing_lock);
+}
+
 static size_t cmd_rx_handler(struct sm_at_host_ctx *ctx, uint8_t c)
 {
 	bool send = false;
@@ -1618,7 +1633,8 @@ bool in_at_mode_pipe(struct modem_pipe *pipe)
 bool is_idle_ctx(struct sm_at_host_ctx *ctx)
 {
 	return (sm_at_ctx_check(ctx) && in_at_mode_ctx(ctx) &&
-		ctx->executing == false && k_timer_remaining_ticks(&ctx->idle_timer) == 0);
+		atomic_get(&ctx->executing_lock) == 0 &&
+		k_timer_remaining_ticks(&ctx->idle_timer) == 0);
 }
 
 bool is_idle_pipe(struct modem_pipe *pipe)
@@ -1885,7 +1901,7 @@ static struct sm_at_host_ctx *sm_at_host_create(struct modem_pipe *pipe)
 	ctx = sm_at_host_get_urc_ctx();
 	if (ctx && atomic_ptr_cas(&ctx->pipe, NULL, pipe)) {
 		LOG_DBG("Reusing first AT host instance %p for pipe %p", (void *)ctx, (void *)pipe);
-		ctx->executing = false;
+		atomic_set(&ctx->executing_lock, 0);
 		modem_pipe_attach(pipe, at_pipe_event_handler, ctx);
 		if (urcs_queued) {
 			sm_at_host_event_notify(ctx, SM_EVENT_URC);

app/src/sm_at_host.h

@@ -342,6 +342,40 @@ static inline void sm_at_host_cmd_done_pipe(struct modem_pipe *pipe)
 			struct modem_pipe * : sm_at_host_cmd_done_pipe \
 			)(X)
 
+/** Lock the AT host context from processing or URC handling.
+ *
+ * This blocks RX handling and URC sending until
+ * sm_at_host_cmd_done() is called.
+ * This is used for sending multiple messages in a row without
+ * interleaving URCs or other command responses.
+ *
+ * AT context have internal refcount to allow nested locks.
+ *
+ * Only safe to call from sm_work_q.
+ */
+void sm_at_host_lock_ctx(struct sm_at_host_ctx *ctx);
+
+/** See sm_at_host_lock_ctx()
+ */
+static inline void sm_at_host_lock_pipe(struct modem_pipe *pipe)
+{
+	return sm_at_host_lock_ctx(sm_at_host_get_ctx_from(pipe));
+}
+
+/** See sm_at_host_lock_ctx()
+ */
+#define sm_at_host_lock(X) \
+		_Generic((X), \
+			struct sm_at_host_ctx * : sm_at_host_lock_ctx, \
+			struct modem_pipe * : sm_at_host_lock_pipe \
+			)(X)
+
+/** Unlock the AT host context from processing or URC handling.
+ *
+ * Alias for sm_at_host_cmd_done() to improve readability when the intention is to block/unblock.
+ */
+#define sm_at_host_unlock(x) cmd_done(x)
+
 /**
  * @brief Submit a work to be executed when the current AT command processing is done.
  *

app/src/sm_at_mqtt.c

@@ -96,6 +96,7 @@ static int handle_mqtt_publish_evt(struct mqtt_client *const c, const struct mqt
 	/* MQTT client does not track the packet identifiers, so MQTT_QOS_2_EXACTLY_ONCE
 	 * promise is not kept. This deviates from MQTT v3.1.1.
 	 */
+	sm_at_host_lock(ctx.pipe);
 	urc_send_to(ctx.pipe, "\r\n#XMQTTMSG: %d,%d\r\n",
 		evt->param.publish.message.topic.topic.size,
 		evt->param.publish.message.payload.len);
@@ -111,7 +112,7 @@ static int handle_mqtt_publish_evt(struct mqtt_client *const c, const struct mqt
 	} while (ret >= 0 && size_read < evt->param.publish.message.payload.len);
 	data_send(ctx.pipe, "\r\n", 2);
 
-
+	sm_at_host_unlock(ctx.pipe);
 	return 0;
 }
 

app/src/sm_at_socket.c

@@ -242,6 +242,8 @@ static void auto_reception(struct sm_socket *sock)
 		return;
 	}
 
+	sm_at_host_lock(sock->pipe);
+
 	if (sock->connected || sock->type == SOCK_RAW) {
 		err = do_recv(sock, 0, MSG_DONTWAIT,
 			      sock->async_poll.adr_hex ? AT_SOCKET_MODE_HEX
@@ -255,12 +257,14 @@ static void auto_reception(struct sm_socket *sock)
 	}
 	if (err) {
 		LOG_ERR("auto_reception() error: %d", err);
+		sm_at_host_unlock(sock->pipe);
 		return;
 	}
 	if (!in_datamode(sock->pipe)) {
 		/* <CR><LF> after the data. */
 		rsp_send_to(sock->pipe, "\r\n");
 	}
+	sm_at_host_unlock(sock->pipe);
 }
 
 static int update_poll_events(struct sm_socket *sock, uint8_t events, bool update_xapoll)
@@ -1126,19 +1130,22 @@ static int do_recv(struct sm_socket *sock, int timeout, int flags,
 	if (ret == 0) {
 		LOG_WRN("zsock_recv() return 0");
 	} else {
+		sm_at_host_lock(sock->pipe);
 		if (!in_datamode(sock->pipe)) {
 			rsp_send_to(sock->pipe, "\r\n#XRECV: %d,%d,%d\r\n", sock->fd, mode, ret);
 		}
 
 		if (mode == AT_SOCKET_MODE_HEX) {
 			ret = data_send_hex(sock, sm_data_buf, ret);
 			if (ret) {
+				sm_at_host_unlock(sock->pipe);
 				return ret;
 			}
 		} else {
 			data_send(sock->pipe, sm_data_buf, ret);
 		}
 		ret = 0;
+		sm_at_host_unlock(sock->pipe);
 
 		update_poll_events(sock, ZSOCK_POLLIN, true);
 	}
@@ -1237,6 +1244,7 @@ static int do_recvfrom(struct sm_socket *sock, int timeout, int flags,
 	if (ret == 0) {
 		LOG_WRN("zsock_recvfrom() return 0");
 	} else {
+		sm_at_host_lock(sock->pipe);
 		if (!in_datamode(sock->pipe)) {
 			char peer_addr[INET6_ADDRSTRLEN] = {0};
 			uint16_t peer_port = 0;
@@ -1249,11 +1257,13 @@ static int do_recvfrom(struct sm_socket *sock, int timeout, int flags,
 		if (mode == AT_SOCKET_MODE_HEX) {
 			ret = data_send_hex(sock, sm_data_buf, ret);
 			if (ret) {
+				sm_at_host_unlock(sock->pipe);
 				return ret;
 			}
 		} else {
 			data_send(sock->pipe, sm_data_buf, ret);
 		}
+		sm_at_host_unlock(sock->pipe);
 
 		update_poll_events(sock, ZSOCK_POLLIN, true);
 	}