feat async job runner

api_json_module can now fill api_json_module_async_t and return API_JSON_ASYNC to make the function call asynchronized. Useful for long running task like wifi scan.

main/main.c

@@ -10,9 +10,13 @@
 #include "wt_storage.h"
 #include "wifi_manager.h"
 #include "web_server.h"
+#include "static_buffer.h"
+#include "request_runner.h"
 
 void app_main()
 {
+	assert(static_buffer_init() == 0);
+	assert(request_runner_init() == 0);
 	wt_storage_init();
 	ESP_ERROR_CHECK(esp_event_loop_create_default());
 	ESP_ERROR_CHECK(esp_netif_init());

project_components/api_router/CMakeLists.txt

@@ -5,5 +5,5 @@ file(GLOB SOURCES *
 idf_component_register(
         SRCS ${SOURCES}
         INCLUDE_DIRS "."
-        REQUIRES json
+        REQUIRES json request_runner
 )

project_components/api_router/api_json_module.c

@@ -33,6 +33,7 @@ int api_json_module_add(api_json_init_func func)
 	err = func(&api_module);
 	if (err) {
 		printf("module %p init failed\n", func);
+		return 1;
 	}
 
 	if (api_module.module_id >= API_MODULE_MAX) {
@@ -51,10 +52,10 @@ int api_json_module_add(api_json_init_func func)
 	return 0;
 }
 
-int api_json_module_call(uint8_t id, uint16_t cmd, api_json_req_t *in, api_json_resp_t *out)
+int api_json_module_call(uint8_t id, uint16_t cmd, api_json_req_t *in, api_json_module_async_t *out)
 {
 	if (unlikely(id >= API_MODULE_MAX || module_arr[id].on_req == NULL)) {
-		return 1;
+		return API_JSON_BAD_REQUEST;
 	}
 
 	return module_arr[id].on_req(cmd, in, out);

project_components/api_router/api_json_module.h

@@ -1,18 +1,39 @@
 #ifndef API_JSON_MODULE_H_GUARD
 #define API_JSON_MODULE_H_GUARD
 
+#include "request_runner.h"
 #include <cJSON.h>
 #include <stdint.h>
 
+
 typedef struct api_json_req_t {
-	cJSON *json;
+	cJSON *in;
+	cJSON *out;
 } api_json_req_t;
 
-typedef struct api_json_resp_t {
-	cJSON *json;
-} api_json_resp_t;
+typedef struct api_json_module_req_t {
+	int (*func)(api_json_req_t *req);
+	void *arg; /* request context (=api_json_req) */
+} api_json_module_req_t;
 
-typedef int (*api_json_on_req)(uint16_t cmd, api_json_req_t *req, api_json_resp_t *rsp);
+typedef struct api_json_module_async_t {
+	api_json_module_req_t module;
+	req_task_cb_t req_task;
+} api_json_module_async_t;
+
+//typedef struct api_json_send_out_t {
+//	void (*func)(void *arg);
+//	void *arg; /* socket context */
+//} api_json_send_out_t;
+
+
+typedef enum api_json_req_status_e {
+	API_JSON_OK = 0,
+	API_JSON_ASYNC = 1,
+	API_JSON_BAD_REQUEST = 2,
+} api_json_req_status_e;
+
+typedef int (*api_json_on_req)(uint16_t cmd, api_json_req_t *req, api_json_module_async_t *rsp);
 
 typedef struct api_json_module_cfg_t {
 	api_json_on_req on_req;
@@ -32,6 +53,6 @@ int api_json_module_add(api_json_init_func);
   __attribute__((used, constructor(PRI))) void cons_ ## INIT(); \
   void cons_ ## INIT() { api_json_module_add(INIT); }
 
-int api_json_module_call(uint8_t id, uint16_t cmd, api_json_req_t *in, api_json_resp_t *out);
+int api_json_module_call(uint8_t id, uint16_t cmd, api_json_req_t *in, api_json_module_async_t *out);
 
 #endif //API_JSON_MODULE_H_GUARD

project_components/api_router/api_json_router.c

@@ -10,22 +10,22 @@ int api_json_router_init()
 	return 0;
 }
 
-int api_json_route(api_json_req_t *req, api_json_resp_t *rsp)
+int api_json_route(api_json_req_t *req, api_json_module_async_t *rsp)
 {
 	uint16_t cmd;
 	uint8_t module_id;
 	cJSON *cmd_json;
 	cJSON *module_json;
 
-	if (unlikely(req->json == NULL)) {
-		return 1;
+	if (unlikely(req == NULL)) {
+		return API_JSON_BAD_REQUEST;
 	}
 
-	cmd_json = cJSON_GetObjectItem(req->json, "cmd");
-	module_json = cJSON_GetObjectItem(req->json, "module");
+	cmd_json = cJSON_GetObjectItem(req->in, "cmd");
+	module_json = cJSON_GetObjectItem(req->in, "module");
 
 	if (!cJSON_IsNumber(cmd_json) || !cJSON_IsNumber(module_json)) {
-		return 1;
+		return API_JSON_BAD_REQUEST;
 	}
 
 	cmd = cmd_json->valueint;

project_components/api_router/api_json_router.h

@@ -5,6 +5,6 @@
 
 int api_json_router_init();
 
-int api_json_route(api_json_req_t *req, api_json_resp_t *out);
+int api_json_route(api_json_req_t *req, api_json_module_async_t *out);
 
 #endif //API_JSON_ROUTER_H_GUARD

project_components/request_runner/CMakeLists.txt

@@ -0,0 +1,9 @@
+file(GLOB SOURCES
+        *.c
+        )
+
+idf_component_register(
+        SRCS ${SOURCES}
+        INCLUDE_DIRS "."
+        REQUIRES
+)

project_components/request_runner/request_runner.c

@@ -0,0 +1,84 @@
+#include <sys/queue.h>
+#include <sys/cdefs.h>
+#include <sys/queue.h>
+#include <sys/cdefs.h>
+/*
+ * SPDX-FileCopyrightText: 2024 kerms
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+#include "request_runner.h"
+
+#include <freertos/FreeRTOS.h>
+#include <freertos/task.h>
+#include <freertos/queue.h>
+
+static QueueHandle_t long_run_queue = NULL;
+static QueueHandle_t send_out_queue = NULL;
+
+_Noreturn static void req_long_task(void *arg);
+
+_Noreturn static void req_send_out_task(void *arg);
+
+_Noreturn void req_long_task(void *arg)
+{
+	req_task_cb_t *req;
+	while (1) {
+		if (unlikely(xQueueReceive(long_run_queue, &req, portMAX_DELAY) != pdTRUE)) {
+			continue;
+		}
+		int status = req->module.helper_cb(req->module.arg);
+
+		/* if send out queue is busy, set status and let the cb to cancel send out
+		 * */
+		if (req_queue_push_send_out(req, pdMS_TO_TICKS(20)) != 0) {
+			status = -1;
+			req->send_out.cb(req->send_out.arg, status);
+		}
+	}
+}
+
+_Noreturn void req_send_out_task(void *arg)
+{
+	req_task_cb_t *req;
+	while (1) {
+		if (likely(xQueueReceive(send_out_queue, &req, portMAX_DELAY))) {
+			req->send_out.cb(req->send_out.arg, 0);
+		}
+	}
+}
+
+int request_runner_init()
+{
+	BaseType_t res;
+	if (long_run_queue != NULL || send_out_queue != NULL) {
+		return 0;
+	}
+
+	long_run_queue = xQueueCreate(2, sizeof(req_task_cb_t *));
+	send_out_queue = xQueueCreate(4, sizeof(req_task_cb_t *));
+	assert(long_run_queue != NULL && send_out_queue != NULL);
+
+	res = xTaskCreate(req_long_task, "Ltask", 4 * 1024, NULL, 8, NULL);
+	res &= xTaskCreate(req_send_out_task, "send out task", 4 * 1024, NULL, 9, NULL);
+	assert(res == pdPASS);
+	return 0;
+}
+
+int req_queue_push_long_run(req_task_cb_t *req, uint32_t delay)
+{
+	if (unlikely(xQueueSend(long_run_queue, &req, delay) != pdTRUE)) {
+		return 1;
+	}
+
+	return 0;
+}
+
+int req_queue_push_send_out(req_task_cb_t *req, uint32_t delay)
+{
+	if (unlikely(xQueueSend(send_out_queue, &req, delay) != pdTRUE)) {
+		return 1;
+	}
+
+	return 0;
+}

project_components/request_runner/request_runner.h

@@ -0,0 +1,33 @@
+/*
+ * SPDX-FileCopyrightText: 2024 kerms
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef REQUEST_RUNNER_H_GUARD
+#define REQUEST_RUNNER_H_GUARD
+
+#include <stdint.h>
+
+typedef struct req_send_out_cb_t {
+	void (*cb)(void *arg, int status);
+	void *arg; /* socket info */
+} req_send_out_cb_t;
+
+typedef struct req_module_cb_t {
+	int (*helper_cb)(void *arg);
+	void *arg;
+} req_module_cb_t;
+
+typedef struct req_task_cb_t {
+	req_module_cb_t module;
+	req_send_out_cb_t send_out;
+} req_task_cb_t;
+
+int request_runner_init();
+
+int req_queue_push_long_run(req_task_cb_t *req, uint32_t delay);
+int req_queue_push_send_out(req_task_cb_t *req, uint32_t delay);
+
+
+#endif //REQUEST_RUNNER_H_GUARD

project_components/static_buffer/CMakeLists.txt

@@ -0,0 +1,9 @@
+file(GLOB SOURCES
+        *.c
+        )
+
+idf_component_register(
+        SRCS ${SOURCES}
+        INCLUDE_DIRS "."
+        REQUIRES
+)

project_components/static_buffer/static_buffer.c

@@ -0,0 +1,54 @@
+/*
+ * SPDX-FileCopyrightText: 2024 kerms
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "static_buffer.h"
+#include <freertos/FreeRTOS.h>
+#include <freertos/queue.h>
+
+#define BUFFER_NR 4
+#define BUFFER_SZ 2048
+
+static uint8_t buf[BUFFER_NR][BUFFER_SZ];
+static QueueHandle_t buf_queue = NULL;
+
+int static_buffer_init()
+{
+	if (buf_queue != NULL)
+		return 0;
+
+	buf_queue = xQueueCreate(BUFFER_NR, sizeof(void *));
+	if (buf_queue == NULL) {
+		return 1;
+	}
+	for (int i = 0; i < BUFFER_NR; ++i) {
+		uint8_t *buf_ptr = buf[i];
+		if (xQueueSend(buf_queue, &buf_ptr, 0) != pdTRUE) {
+			return 1;
+		}
+	}
+
+	return 0;
+}
+
+void *static_buffer_get(uint32_t tick_wait)
+{
+	void *ptr = NULL;
+	xQueueReceive(buf_queue, &ptr, tick_wait);
+	return ptr;
+}
+
+void static_buffer_put(void *ptr)
+{
+	printf("put buf %d\n", uxQueueMessagesWaiting(buf_queue));
+	if (unlikely(xQueueSend(buf_queue, &ptr, 0) != pdTRUE)) {
+		assert(0);
+	}
+}
+
+uint32_t static_buffer_get_buf_size()
+{
+	return BUFFER_SZ;
+}

project_components/static_buffer/static_buffer.h

@@ -0,0 +1,21 @@
+/*
+ * SPDX-FileCopyrightText: 2024 kerms
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef STATIC_BUFFER_H_GUARD
+#define STATIC_BUFFER_H_GUARD
+
+#include <stdint.h>
+
+int static_buffer_init();
+
+void *static_buffer_get(uint32_t tick_wait);
+
+void static_buffer_put(void *ptr);
+
+uint32_t static_buffer_get_buf_size();
+
+
+#endif //STATIC_BUFFER_H_GUARD

project_components/web_server/CMakeLists.txt

@@ -9,6 +9,7 @@ idf_component_register(
         SRCS ${SOURCES}
         INCLUDE_DIRS "."
         REQUIRES esp_http_server api_router json
+        PRIV_REQUIRES request_runner static_buffer
 )
 
 idf_component_set_property(${COMPONENT_NAME} WHOLE_ARCHIVE ON)

project_components/web_server/uri_modules/uri_api.c

@@ -1,5 +1,7 @@
 #include "web_uri_module.h"
 #include "api_json_router.h"
+#include "request_runner.h"
+#include "static_buffer.h"
 
 #include <esp_http_server.h>
 #include <esp_log.h>
@@ -8,20 +10,46 @@
 
 #define TAG __FILE_NAME__
 
-static char buf[2048];
-static api_json_req_t json_in;
-static api_json_resp_t json_out;
+#define HTTPD_STATUS_503 "503 Busy"
+
+typedef struct post_request_t {
+	api_json_req_t json;
+	api_json_module_async_t async;
+	httpd_req_t *req_out;
+	char buf[0];
+} post_request_t;
+
+static void async_send_out_cb(void *req, int module_status);
+static int uri_api_send_out(httpd_req_t *req, post_request_t *post_req);
+
 
 static esp_err_t api_post_handler(httpd_req_t *req)
 {
+	uint32_t buf_len;
 	int data_len;
 	int err;
+	post_request_t *post_req;
+	char *buf;
 	uint32_t remaining = req->content_len;
 
-	data_len = httpd_req_recv(req, buf, MIN(remaining, sizeof(buf)));
+	buf_len = static_buffer_get_buf_size() - sizeof(post_request_t);
+	if (unlikely(buf_len < remaining)) {
+		ESP_LOGE(TAG, "req size %lu > buf_len %lu", remaining, buf_len);
+		return ESP_FAIL;
+	}
+
+	post_req = static_buffer_get(pdMS_TO_TICKS(20));
+	if (unlikely(post_req == NULL)) {
+		ESP_LOGE(TAG, "static buf busy");
+		return ESP_FAIL;
+	}
+	buf = post_req->buf;
+
+	data_len = httpd_req_recv(req, buf, buf_len);
 	if (unlikely(data_len <= 0)) {
 		ESP_LOGE(TAG, "httpd recv error");
-		return ESP_FAIL;
+		err = ESP_FAIL;
+		goto put_buf;
 	}
 
 	ESP_LOGI(TAG, "=========== RECEIVED DATA ==========");
@@ -30,43 +58,83 @@ static esp_err_t api_post_handler(httpd_req_t *req)
 	ESP_LOGI(TAG, "heap min: %lu, cur: %lu", esp_get_minimum_free_heap_size(), esp_get_free_heap_size());
 
 	/* Decode */
-	json_in.json = cJSON_ParseWithLength(buf, data_len);
-	if (unlikely(json_in.json == NULL)) {
+	post_req->json.in = cJSON_ParseWithLength(buf, data_len);
+	if (unlikely(post_req->json.in == NULL)) {
 		httpd_resp_set_status(req, HTTPD_400);
-		httpd_resp_send(req, NULL, 0);
-		return ESP_OK;
+		goto end;
 	}
 
-	err = api_json_route(&json_in, &json_out);
-	if (err) {
-		httpd_resp_set_status(req, HTTPD_400);
-		httpd_resp_send(req, NULL, 0);
+	post_req->json.out = NULL;
+	err = api_json_route(&post_req->json, &post_req->async);
+	cJSON_Delete(post_req->json.in);
+	if (err == API_JSON_ASYNC) {
+		httpd_req_async_handler_begin(req, &post_req->req_out);
+		post_req->async.req_task.send_out.cb = async_send_out_cb;
+		post_req->async.req_task.send_out.arg = post_req;
+		if (req_queue_push_long_run(&post_req->async.req_task, pdMS_TO_TICKS(20))) {
+			httpd_req_async_handler_complete(post_req->req_out);
+			httpd_resp_set_status(req, HTTPD_STATUS_503);
+			goto end;
+		}
 		return ESP_OK;
+	} else if (unlikely(err != API_JSON_OK)) {
+		httpd_resp_set_status(req, HTTPD_400);
+		goto end;
 	}
-	cJSON_Delete(json_in.json);
 
-	if (json_out.json == NULL) {
-		return httpd_resp_send(req, NULL, 0);
+	if (post_req->json.out == NULL) {
+		goto end;
 	}
 
 	/* api function returns something, send back to http client */
-	httpd_resp_set_type(req, HTTPD_TYPE_JSON);
-	err = !cJSON_PrintPreallocated(json_out.json, buf, sizeof(buf) - 5, 0);
-	cJSON_Delete(json_out.json);
-	json_out.json = NULL;
+	err = uri_api_send_out(req, post_req);
+	goto put_buf;
 
-	if (err) {
+end:
+	err = httpd_resp_send(req, NULL, 0);
+	if (unlikely(err)) {
+		ESP_LOGE(TAG, "resp_send err: %s", esp_err_to_name(err));
+	}
+put_buf:
+	static_buffer_put(post_req);
+	return err;
+}
+
+int uri_api_send_out(httpd_req_t *req, post_request_t *post_req)
+{
+	int err;
+	char *buf;
+	uint32_t buf_len;
+
+	buf = post_req->buf;
+	buf_len = static_buffer_get_buf_size() - sizeof(post_request_t);
+
+	httpd_resp_set_type(req, HTTPD_TYPE_JSON);
+	err = !cJSON_PrintPreallocated(post_req->json.out, buf, buf_len - 5, 0);
+	cJSON_Delete(post_req->json.out);
+	if (unlikely(err)) {
 		httpd_resp_set_status(req, HTTPD_500);
 		return httpd_resp_send(req, NULL, 0);
 	}
 
-	err = httpd_resp_send(req, buf, strlen(buf));
-	if (err) {
-		ESP_LOGE(TAG, "resp_send err: %s", esp_err_to_name(err));
-	}
-	return err;
+	return httpd_resp_send(req, buf, strlen(buf));
 }
 
+void async_send_out_cb(void *arg, int module_status)
+{
+	post_request_t *req = arg;
+	if (module_status != API_JSON_OK) {
+		httpd_sess_trigger_close(req->req_out->handle,
+		                         httpd_req_to_sockfd(req->req_out->handle));
+	}
+
+	uri_api_send_out(req->req_out, req);
+
+	/* clean resources */
+	httpd_req_async_handler_complete(req->req_out);
+	static_buffer_put(req);
+};
+
 /**
  * REGISTER MODULE
  * */
@@ -89,4 +157,6 @@ int URI_API_EXIT(const httpd_uri_t **uri_conf) {
 	return 0;
 }
 
-WEB_URI_MODULE_REGISTER(0x81, URI_API_INIT, URI_API_EXIT);
+WEB_URI_MODULE_REGISTER(0x81, URI_API_INIT, URI_API_EXIT)
+
+

project_components/wifi_manager/wifi_api.c

@@ -20,8 +20,8 @@ void wifi_api_get_ap_info(wifi_api_ap_info_t *ap_info)
 
 static int rssi_comp(const void *a, const void *b)
 {
-	const wifi_ap_record_t *r1 = a;
-	const wifi_ap_record_t *r2 = b;
+	const wifi_api_ap_info_t *r1 = a;
+	const wifi_api_ap_info_t *r2 = b;
 	return r2->rssi - r1->rssi;
 }
 

project_components/wifi_manager/wifi_api_json.c

@@ -1,57 +1,71 @@
 #include "wifi_api_json.h"
 
-#include "api_json_router.h"
+#include "api_json_module.h"
 #include "wifi_api.h"
 #include "wifi_json_utils.h"
 
 #include <stdio.h>
 
-static int wifi_api_json_get_ap_info(api_json_req_t *req, api_json_resp_t *resp);
+static int wifi_api_json_get_ap_info(api_json_req_t *req);
 
-static int wifi_api_json_get_scan(api_json_req_t *req, api_json_resp_t *resp);
+static int wifi_api_json_get_scan(api_json_req_t *req);
 
-static int on_json_req(uint16_t cmd, api_json_req_t *req, api_json_resp_t *rsp)
+static int on_async_call(void *arg)
+{
+	api_json_module_req_t *req = arg;
+	return req->func(req->arg);
+}
+
+static int on_json_req(uint16_t cmd, api_json_req_t *req, api_json_module_async_t *async)
 {
 	wifi_api_json_cmd_t wifi_cmd = cmd;
 	switch (wifi_cmd) {
 	case WIFI_API_JSON_GET_AP_INFO:
-		return wifi_api_json_get_ap_info(req, rsp);
+		return wifi_api_json_get_ap_info(req);
 	case WIFI_API_JSON_CONNECT:
 
 		break;
 	case WIFI_API_JSON_GET_SCAN:
-		return wifi_api_json_get_scan(req, rsp);
+		async->module.func = wifi_api_json_get_scan;
+		async->module.arg = req;
+		async->req_task.module.helper_cb = on_async_call;
+		async->req_task.module.arg = &async->module;
+		return API_JSON_ASYNC;
 	case UNKNOWN:
 	default:
 		break;
 	}
 
-	printf("%d\n", cmd);
+	printf("cmd %d\n", cmd);
 
 	return 0;
 }
 
-static int wifi_api_json_get_ap_info(api_json_req_t *req, api_json_resp_t *resp)
+static int wifi_api_json_get_ap_info(api_json_req_t *req)
 {
 	wifi_api_ap_info_t ap_info;
 	wifi_api_get_ap_info(&ap_info);
-	resp->json = wifi_api_json_serialize_ap_info(&ap_info);
+	req->out = wifi_api_json_serialize_ap_info(&ap_info);
 	return 0;
 }
 
-static int wifi_api_json_get_scan(api_json_req_t *req, api_json_resp_t *resp)
+static int wifi_api_json_get_scan(api_json_req_t *req)
 {
 	wifi_api_ap_info_t ap_info[20];
 	uint16_t max_count = 20;
 	int err;
 
+	printf("get scan\n");
+
 	err = wifi_api_get_scan_list(&max_count, ap_info);
 	if (err == ESP_ERR_NOT_FINISHED) {
-		resp->json = wifi_api_json_create_err_rsp(req->json, "Wi-Fi scan busy");
+		req->out = wifi_api_json_create_err_rsp(req->in, "Wi-Fi scan busy");
 		return 1;
 	}
 
-	resp->json = wifi_api_json_serialize_scan_list(ap_info, max_count);
+	printf("scan ok\n");
+
+	req->out = wifi_api_json_serialize_scan_list(ap_info, max_count);
 	return 0;
 }