djsundog/circuitpython
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Web Workflow sockets and threads handling improvements.

Browse Source 
Fixes polling thread looping forever hangs preventing new connections.
Don't lose listening sockets on mp resets and re-init.
Keep better separation of "system" and "user" sockets.
Track socket states to prevent re-use of sockets before closed.
Close REST socket when transaction completes. No post-init.
Remove unnecessary state flags.
This commit is contained in:
Ted Hess 2023-04-04 15:07:04 -04:00
parent 4e7bbf70a6
commit 9825b7fbb7
7 changed files with 199 additions and 182 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

217
ports/espressif/common-hal/socketpool/Socket.c
View File

@ -45,70 +45,79 @@
StackType_t socket_select_stack[2 * configMINIMAL_STACK_SIZE];
STATIC int open_socket_fds[CONFIG_LWIP_MAX_SOCKETS];
/* Socket state table:
* 0 := Closed (unused)
* 1 := Open
* 2 := Closing (remove from rfds)
* Index into socket_fd_state is calculated from actual lwip fd. idx := fd - LWIP_SOCKET_OFFSET
*/
#define FDSTATE_CLOSED 0
#define FDSTATE_OPEN 1
#define FDSTATE_CLOSING 2
STATIC uint8_t socket_fd_state[CONFIG_LWIP_MAX_SOCKETS];
STATIC socketpool_socket_obj_t *user_socket[CONFIG_LWIP_MAX_SOCKETS];
StaticTask_t socket_select_task_handle;
StaticTask_t socket_select_task_buffer;
TaskHandle_t socket_select_task_handle;
STATIC int socket_change_fd = -1;
STATIC void socket_select_task(void *arg) {
uint64_t signal;
fd_set readfds;
fd_set excptfds;
while (true) {
fd_set readfds;
fd_set errfds;
FD_ZERO(&readfds);
FD_ZERO(&errfds);
FD_ZERO(&excptfds);
FD_SET(socket_change_fd, &readfds);
FD_SET(socket_change_fd, &errfds);
int max_fd = socket_change_fd;
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
int sockfd = open_socket_fds[i];
if (sockfd < 0) {
continue;
}
for (size_t i = 0; i < MP_ARRAY_SIZE(socket_fd_state); i++) {
if ((socket_fd_state[i] == FDSTATE_OPEN) && (user_socket[i] == NULL)) {
int sockfd = i + LWIP_SOCKET_OFFSET;
max_fd = MAX(max_fd, sockfd);
FD_SET(sockfd, &readfds);
FD_SET(sockfd, &errfds);
FD_SET(sockfd, &excptfds);
}
}
int num_triggered = select(max_fd + 1, &readfds, NULL, &errfds, NULL);
// Check for bad file descriptor and queue up the background task before
// circling around.
if (num_triggered == -1 && errno == EBADF) {
// One for the change fd and one for the closed socket.
num_triggered = 2;
}
// Try and find the bad file and remove it from monitoring.
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
int sockfd = open_socket_fds[i];
if (sockfd < 0) {
int num_triggered = select(max_fd + 1, &readfds, NULL, &excptfds, NULL);
// Hard error (or someone closed a socket on another thread)
if (num_triggered == -1) {
assert(errno == EBADF);
continue;
}
int err;
int optlen = sizeof(int);
int ret = getsockopt(sockfd, SOL_SOCKET, SO_ERROR, &err, (socklen_t *)&optlen);
if (ret < 0) {
open_socket_fds[i] = -1;
// Raise num_triggered so that we skip the assert and queue the background task.
num_triggered = 2;
}
}
assert(num_triggered >= 0);
assert(num_triggered > 0);
assert(!FD_ISSET(socket_change_fd, &excptfds));
// Notice event trigger
if (FD_ISSET(socket_change_fd, &readfds)) {
read(socket_change_fd, &signal, sizeof(signal));
num_triggered -= 1;
num_triggered--;
}
// Handle active FDs, close the dead ones
for (size_t i = 0; i < MP_ARRAY_SIZE(socket_fd_state); i++) {
int sockfd = i + LWIP_SOCKET_OFFSET;
if (socket_fd_state[i] != FDSTATE_CLOSED) {
if (FD_ISSET(sockfd, &readfds) || FD_ISSET(sockfd, &excptfds)) {
if (socket_fd_state[i] == FDSTATE_CLOSING) {
socket_fd_state[i] = FDSTATE_CLOSED;
num_triggered--;
}
}
}
}
if (num_triggered > 0) {
// Wake up CircuitPython by queuing request
supervisor_workflow_request_background();
// Wake up CircuitPython. We know it is asleep because we are lower
// priority.
port_wake_main_task();
ulTaskNotifyTake(pdTRUE, portMAX_DELAY);
}
}
}
close(socket_change_fd);
socket_change_fd = -1;
vTaskDelete(NULL);
}
@ -117,75 +126,62 @@ void socket_user_reset(void) {
esp_vfs_eventfd_config_t config = ESP_VFS_EVENTD_CONFIG_DEFAULT();
ESP_ERROR_CHECK(esp_vfs_eventfd_register(&config));
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
open_socket_fds[i] = -1;
// Clear initial socket states
for (size_t i = 0; i < MP_ARRAY_SIZE(socket_fd_state); i++) {
socket_fd_state[i] = FDSTATE_CLOSED;
user_socket[i] = NULL;
}
socket_change_fd = eventfd(0, 0);
// Run this at the same priority as CP so that the web workflow background task can be
// queued while CP is running. Both tasks can still sleep and, therefore, sleep overall.
(void)xTaskCreateStaticPinnedToCore(socket_select_task,
socket_select_task_handle = xTaskCreateStaticPinnedToCore(socket_select_task,
"socket_select",
2 * configMINIMAL_STACK_SIZE,
NULL,
uxTaskPriorityGet(NULL),
socket_select_stack,
&socket_select_task_handle,
&socket_select_task_buffer,
xPortGetCoreID());
}
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
if (open_socket_fds[i] >= 0 && user_socket[i]) {
} else {
// Not init - close open user sockets
for (size_t i = 0; i < MP_ARRAY_SIZE(socket_fd_state); i++) {
if ((socket_fd_state[i] == FDSTATE_OPEN) && user_socket[i]) {
common_hal_socketpool_socket_close(user_socket[i]);
int num = open_socket_fds[i];
// Close automatically clears socket handle
lwip_shutdown(num, SHUT_RDWR);
lwip_close(num);
open_socket_fds[i] = -1;
user_socket[i] = NULL;
}
}
}
}
// Unblock select task (ok if not blocked yet)
void socketpool_socket_poll_resume(void) {
if (socket_select_task_handle) {
xTaskNotifyGive(socket_select_task_handle);
}
}
// The writes below send an event to the socket select task so that it redoes the
// select with the new open socket set.
STATIC bool register_open_socket(int fd) {
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
if (open_socket_fds[i] == -1) {
open_socket_fds[i] = fd;
user_socket[i] = false;
if (fd < FD_SETSIZE) {
socket_fd_state[fd - LWIP_SOCKET_OFFSET] = FDSTATE_OPEN;
user_socket[fd - LWIP_SOCKET_OFFSET] = NULL;
uint64_t signal = 1;
write(socket_change_fd, &signal, sizeof(signal));
socketpool_socket_poll_resume();
return true;
}
}
return false;
}
STATIC void unregister_open_socket(int fd) {
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
if (open_socket_fds[i] == fd) {
open_socket_fds[i] = -1;
user_socket[i] = false;
// Write must be 8 bytes for an eventfd.
uint64_t signal = 1;
write(socket_change_fd, &signal, sizeof(signal));
return;
}
}
}
STATIC void mark_user_socket(int fd, socketpool_socket_obj_t *obj) {
for (size_t i = 0; i < MP_ARRAY_SIZE(open_socket_fds); i++) {
if (open_socket_fds[i] == fd) {
user_socket[i] = obj;
return;
}
}
socket_fd_state[fd - LWIP_SOCKET_OFFSET] = FDSTATE_OPEN;
user_socket[fd - LWIP_SOCKET_OFFSET] = obj;
// No need to wakeup select task
}
bool socketpool_socket(socketpool_socketpool_obj_t *self,
STATIC bool _socketpool_socket(socketpool_socketpool_obj_t *self,
socketpool_socketpool_addressfamily_t family, socketpool_socketpool_sock_t type,
socketpool_socket_obj_t *sock) {
int addr_family;
@ -193,9 +189,11 @@ bool socketpool_socket(socketpool_socketpool_obj_t *self,
if (family == SOCKETPOOL_AF_INET) {
addr_family = AF_INET;
ipproto = IPPROTO_IP;
#if LWIP_IPV6
} else { // INET6
addr_family = AF_INET6;
ipproto = IPPROTO_IPV6;
#endif
}
int socket_type;
@ -218,14 +216,28 @@ bool socketpool_socket(socketpool_socketpool_obj_t *self,
if (socknum < 0) {
return false;
}
// This shouldn't happen since we have room for the same number of sockets as LWIP.
if (!register_open_socket(socknum)) {
lwip_close(socknum);
return false;
}
sock->num = socknum;
// Sockets should be nonblocking in most cases
lwip_fcntl(socknum, F_SETFL, O_NONBLOCK);
return true;
}
// special entry for workflow listener (register system socket)
bool socketpool_socket(socketpool_socketpool_obj_t *self,
socketpool_socketpool_addressfamily_t family, socketpool_socketpool_sock_t type,
socketpool_socket_obj_t *sock) {
if (!_socketpool_socket(self, family, type, sock)) {
return false;
}
// This shouldn't happen since we have room for the same number of sockets as LWIP.
if (!register_open_socket(sock->num)) {
lwip_close(sock->num);
return false;
}
return true;
}
@ -238,7 +250,7 @@ socketpool_socket_obj_t *common_hal_socketpool_socket(socketpool_socketpool_obj_
socketpool_socket_obj_t *sock = m_new_obj_with_finaliser(socketpool_socket_obj_t);
sock->base.type = &socketpool_socket_type;
if (!socketpool_socket(self, family, type, sock)) {
if (!_socketpool_socket(self, family, type, sock)) {
mp_raise_RuntimeError(translate("Out of sockets"));
}
mark_user_socket(sock->num, sock);
@ -279,17 +291,16 @@ int socketpool_socket_accept(socketpool_socket_obj_t *self, uint8_t *ip, uint32_
// We got a socket. New client socket will not be non-blocking by default, so make it non-blocking.
lwip_fcntl(newsoc, F_SETFL, O_NONBLOCK);
if (accepted != NULL) {
// Error if called with open socket object.
assert(common_hal_socketpool_socket_get_closed(accepted));
// Register if system socket
if (!register_open_socket(newsoc)) {
lwip_close(newsoc);
return -MP_EBADF;
}
if (accepted != NULL) {
// Close the active socket because we have another we accepted.
if (!common_hal_socketpool_socket_get_closed(accepted)) {
common_hal_socketpool_socket_close(accepted);
}
// Replace the old accepted socket with the new one.
accepted->num = newsoc;
accepted->pool = self->pool;
@ -353,12 +364,21 @@ void socketpool_socket_close(socketpool_socket_obj_t *self) {
return;
}
self->connected = false;
if (self->num >= 0) {
lwip_shutdown(self->num, SHUT_RDWR);
lwip_close(self->num);
unregister_open_socket(self->num);
self->num = -1;
int fd = self->num;
// Ignore bogus/closed sockets
if (fd >= LWIP_SOCKET_OFFSET) {
if (user_socket[fd - LWIP_SOCKET_OFFSET] == NULL) {
socket_fd_state[fd - LWIP_SOCKET_OFFSET] = FDSTATE_CLOSING;
lwip_shutdown(fd, SHUT_RDWR);
lwip_close(fd);
} else {
lwip_shutdown(fd, SHUT_RDWR);
lwip_close(fd);
socket_fd_state[fd - LWIP_SOCKET_OFFSET] = FDSTATE_CLOSED;
user_socket[fd - LWIP_SOCKET_OFFSET] = NULL;
}
}
self->num = -1;
}
void common_hal_socketpool_socket_close(socketpool_socket_obj_t *self) {
@ -420,7 +440,7 @@ bool common_hal_socketpool_socket_get_connected(socketpool_socket_obj_t *self) {
}
bool common_hal_socketpool_socket_listen(socketpool_socket_obj_t *self, int backlog) {
return lwip_listen(self->num, backlog);
return lwip_listen(self->num, backlog) == 0;
}
mp_uint_t common_hal_socketpool_socket_recvfrom_into(socketpool_socket_obj_t *self,
@ -479,10 +499,9 @@ int socketpool_socket_recv_into(socketpool_socket_obj_t *self,
}
RUN_BACKGROUND_TASKS;
received = lwip_recv(self->num, (void *)buf, len, 0);
// In non-blocking mode, fail instead of looping
if (received == -1 && self->timeout_ms == 0) {
if (errno == ENOTCONN) {
if (received < 1 && self->timeout_ms == 0) {
if ((received == 0) || (errno == ENOTCONN)) {
self->connected = false;
return -MP_ENOTCONN;
}

2
ports/espressif/common-hal/socketpool/Socket.h
View File

@ -48,3 +48,5 @@ typedef struct {
} socketpool_socket_obj_t;
void socket_user_reset(void);
// Unblock workflow socket select thread (platform specific)
void socketpool_socket_poll_resume(void);

3
ports/raspberrypi/common-hal/socketpool/Socket.h
View File

@ -75,4 +75,7 @@ typedef struct _lwip_socket_obj_t {
socketpool_socketpool_obj_t *pool;
} socketpool_socket_obj_t;
// Not required for RPi socket positive callbacks
#define socketpool_socket_poll_resume(x)
void socket_user_reset(void);

77
supervisor/shared/web_workflow/web_workflow.c
View File

@ -303,15 +303,12 @@ void supervisor_start_web_workflow(void) {
return;
}
mp_int_t new_port = web_api_port;
// (leaves new_port unchanged on any failure)
(void)common_hal_os_getenv_int("CIRCUITPY_WEB_API_PORT", &new_port);
(void)common_hal_os_getenv_int("CIRCUITPY_WEB_API_PORT", &web_api_port);
bool first_start = pool.base.type != &socketpool_socketpool_type;
bool port_changed = new_port != web_api_port;
if (first_start) {
port_changed = false;
pool.base.type = &socketpool_socketpool_type;
common_hal_socketpool_socketpool_construct(&pool, &common_hal_wifi_radio_obj);
@ -320,6 +317,11 @@ void supervisor_start_web_workflow(void) {
websocket_init();
}
if (!common_hal_socketpool_socket_get_closed(&active)) {
common_hal_socketpool_socket_close(&active);
}
#if CIRCUITPY_MDNS
// Try to start MDNS if the user deinited it.
if (mdns.base.type != &mdns_server_type ||
@ -330,24 +332,10 @@ void supervisor_start_web_workflow(void) {
common_hal_mdns_server_set_instance_name(&mdns, MICROPY_HW_BOARD_NAME);
}
}
#endif
if (port_changed) {
common_hal_socketpool_socket_close(&listening);
}
if (first_start || port_changed) {
web_api_port = new_port;
#if CIRCUITPY_MDNS
if (!common_hal_mdns_server_deinited(&mdns)) {
common_hal_mdns_server_advertise_service(&mdns, "_circuitpython", "_tcp", web_api_port);
}
#endif
socketpool_socket(&pool, SOCKETPOOL_AF_INET, SOCKETPOOL_SOCK_STREAM, &listening);
common_hal_socketpool_socket_settimeout(&listening, 0);
// Bind to any ip.
common_hal_socketpool_socket_bind(&listening, "", 0, web_api_port);
common_hal_socketpool_socket_listen(&listening, 1);
}
const size_t api_password_len = sizeof(_api_password) - 1;
result = common_hal_os_getenv_str("CIRCUITPY_WEB_API_PASSWORD", _api_password + 1, api_password_len);
@ -355,6 +343,16 @@ void supervisor_start_web_workflow(void) {
_api_password[0] = ':';
_base64_in_place(_api_password, strlen(_api_password), sizeof(_api_password) - 1);
}
if (common_hal_socketpool_socket_get_closed(&listening)) {
socketpool_socket(&pool, SOCKETPOOL_AF_INET, SOCKETPOOL_SOCK_STREAM, &listening);
common_hal_socketpool_socket_settimeout(&listening, 0);
// Bind to any ip. (Not checking for failures)
common_hal_socketpool_socket_bind(&listening, "", 0, web_api_port);
common_hal_socketpool_socket_listen(&listening, 1);
}
// Wake polling thread (maybe)
socketpool_socket_poll_resume();
#endif
}
@ -513,7 +511,7 @@ static void _cors_header(socketpool_socket_obj_t *socket, _request *request) {
_send_strs(socket,
"Access-Control-Allow-Credentials: true\r\n",
"Vary: Origin, Accept, Upgrade\r\n",
"Access-Control-Allow-Origin: ", request->origin, "\r\n",
"Access-Control-Allow-Origin: *\r\n",
NULL);
}
@ -1080,6 +1078,10 @@ static bool _reply(socketpool_socket_obj_t *socket, _request *request) {
#else
_reply_missing(socket, request);
#endif
// For now until CORS is sorted, allow always the origin requester.
// Note: caller knows who we are better than us. CORS is not security
// unless browser cooperates. Do not rely on mDNS or IP.
} else if (strlen(request->origin) > 0 && !_origin_ok(request->origin)) {
_reply_forbidden(socket, request);
} else if (strncmp(request->path, "/fs/", 4) == 0) {
@ -1323,9 +1325,14 @@ static void _process_request(socketpool_socket_obj_t *socket, _request *request)
uint8_t c;
// This code assumes header lines are terminated with \r\n
while (more && !error) {
int len = socketpool_socket_recv_into(socket, &c, 1);
if (len != 1) {
more = false;
if (len == 0 || len == -MP_ENOTCONN) {
// Disconnect - clear 'in-progress'
_reset_request(request);
}
break;
}
if (!request->in_progress) {
@ -1452,6 +1459,7 @@ static void _process_request(socketpool_socket_obj_t *socket, _request *request)
}
bool reload = _reply(socket, request);
_reset_request(request);
common_hal_socketpool_socket_close(socket);
autoreload_resume(AUTORELOAD_SUSPEND_WEB);
if (reload) {
autoreload_trigger();
@ -1459,45 +1467,52 @@ static void _process_request(socketpool_socket_obj_t *socket, _request *request)
}
void supervisor_web_workflow_background(void) {
// Track if we have more to do. For example, we should start processing a
// request immediately after we accept the socket.
bool more_to_do = true;
while (more_to_do) {
more_to_do = false;
void supervisor_web_workflow_background(void *data) {
while (true) {
// If we have a request in progress, continue working on it. Do this first
// so that we can accept another socket after finishing this request.
if (common_hal_socketpool_socket_get_connected(&active)) {
_process_request(&active, &active_request);
if (active_request.in_progress) {
break;
}
} else {
// Close the active socket if it is no longer connected.
// Close the active socket if necessary
if (!common_hal_socketpool_socket_get_closed(&active)) {
common_hal_socketpool_socket_close(&active);
}
}
// Otherwise, see if we have another socket to accept.
if ((!common_hal_socketpool_socket_get_connected(&active) ||
(!active_request.in_progress && !active_request.new_socket)) &&
!common_hal_socketpool_socket_get_closed(&listening)) {
uint32_t ip;
uint32_t port;
if (!common_hal_socketpool_socket_get_closed(&active)) {
common_hal_socketpool_socket_close(&active);
}
int newsoc = socketpool_socket_accept(&listening, (uint8_t *)&ip, &port, &active);
if (newsoc == -EBADF) {
common_hal_socketpool_socket_close(&listening);
return;
break;
}
if (newsoc > 0) {
common_hal_socketpool_socket_settimeout(&active, 0);
_reset_request(&active_request);
// Mark new sockets, otherwise we may accept another before the first
// could start its request.
active_request.new_socket = true;
more_to_do = true;
continue;
}
break;
}
websocket_background();
break;
}
// Resume polling
socketpool_socket_poll_resume();
return;
}
void supervisor_stop_web_workflow(void) {

2
supervisor/shared/web_workflow/web_workflow.h
View File

@ -33,7 +33,7 @@
// This background function should be called repeatedly. It cannot be done based
// on events.
void supervisor_web_workflow_background(void);
void supervisor_web_workflow_background(void *data);
bool supervisor_web_workflow_status_dirty(void);
void supervisor_web_workflow_status(void);
void supervisor_start_web_workflow(void);

14
supervisor/shared/web_workflow/websocket.c
View File

@ -42,7 +42,6 @@ typedef struct {
uint8_t frame_len;
uint8_t payload_len_size;
bool masked;
bool closed;
uint8_t mask[4];
int frame_index;
size_t payload_remaining;
@ -59,17 +58,16 @@ static _websocket cp_serial;
void websocket_init(void) {
socketpool_socket_reset(&cp_serial.socket);
cp_serial.closed = true;
ringbuf_init(&_incoming_ringbuf, _buf, sizeof(_buf) - 1);
}
void websocket_handoff(socketpool_socket_obj_t *socket) {
if (!cp_serial.closed) {
if (!common_hal_socketpool_socket_get_closed(&cp_serial.socket)) {
common_hal_socketpool_socket_close(&cp_serial.socket);
}
socketpool_socket_move(socket, &cp_serial.socket);
cp_serial.closed = false;
cp_serial.opcode = 0;
cp_serial.frame_index = 0;
cp_serial.frame_len = 2;
@ -81,12 +79,14 @@ void websocket_handoff(socketpool_socket_obj_t *socket) {
}
bool websocket_connected(void) {
return _incoming_ringbuf.size > 0 && !cp_serial.closed && common_hal_socketpool_socket_get_connected(&cp_serial.socket);
return _incoming_ringbuf.size > 0 &&
!common_hal_socketpool_socket_get_closed(&cp_serial.socket) &&
common_hal_socketpool_socket_get_connected(&cp_serial.socket);
}
static bool _read_byte(uint8_t *c) {
int len = socketpool_socket_recv_into(&cp_serial.socket, c, 1);
if (len != 1) {
if (len < 1) {
return false;
}
return true;
@ -160,8 +160,6 @@ static void _read_next_frame_header(void) {
if (cp_serial.payload_remaining == 0) {
cp_serial.frame_index = 0;
if (cp_serial.opcode == 0x8) {
cp_serial.closed = true;
common_hal_socketpool_socket_close(&cp_serial.socket);
}
}

38
supervisor/shared/workflow.c
View File

@ -46,15 +46,7 @@
#if CIRCUITPY_WEB_WORKFLOW
#include "supervisor/shared/web_workflow/web_workflow.h"
#endif
static background_callback_t workflow_background_cb;
static bool workflow_started = false;
static void workflow_background(void *data) {
#if CIRCUITPY_WEB_WORKFLOW
supervisor_web_workflow_background();
#endif
}
static background_callback_t workflow_background_cb = {NULL, NULL};
// Called during a VM reset. Doesn't actually reset things.
void supervisor_workflow_reset(void) {
@ -63,31 +55,18 @@ void supervisor_workflow_reset(void) {
#endif
#if CIRCUITPY_WEB_WORKFLOW
if (workflow_background_cb.fun) {
supervisor_start_web_workflow();
#endif
workflow_background_cb.fun = workflow_background;
workflow_background_cb.data = NULL;
supervisor_workflow_request_background();
}
#endif
}
void supervisor_workflow_request_background(void) {
if (!workflow_started) {
return;
}
if (workflow_background_cb.fun) {
workflow_background_cb.data = NULL;
background_callback_add_core(&workflow_background_cb);
}
// Return true as soon as USB communication with host has started,
// even before enumeration is done.
// Not that some chips don't notice when USB is unplugged after first being plugged in,
// so this is not perfect, but tud_suspended() check helps.
bool supervisor_workflow_connecting(void) {
#if CIRCUITPY_USB
return tud_connected() && !tud_suspended();
#else
return false;
#endif
}
}
// Return true if host has completed connection to us (such as USB enumeration).
@ -120,9 +99,10 @@ void supervisor_workflow_start(void) {
#if CIRCUITPY_WEB_WORKFLOW
supervisor_start_web_workflow();
memset(&workflow_background_cb, 0, sizeof(workflow_background_cb));
workflow_background_cb.fun = supervisor_web_workflow_background;
#endif
workflow_started = true;
}
FRESULT supervisor_workflow_mkdir_parents(FATFS *fs, char *path) {