yystopf/xiuos
1
0
Fork 1
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Support multithread server

Browse Source
This commit is contained in:
TXuian
2024-05-31 01:17:57 +08:00
parent d05754a98e
commit 3a99cc550c
8 changed files with 128 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Ubiquitous/XiZi_AIoT/services/lib/ipc/libipc.c
+71 -57
View File
@@ -168,25 +168,6 @@ int cur_session_id(void)
return cur_sess_id;
}
static bool session_delayed = false;
void delay_session(void)
{
session_delayed = true;
}
bool is_cur_session_delayed(void)
{
return session_delayed;
}
bool is_cur_handler_been_delayed()
{
if (ipc_server_loop_cur_msg == NULL) {
return false;
}
return ipc_server_loop_cur_msg->header.delayed == 1;
}
void ipc_server_loop(struct IpcNode* ipc_node)
{
struct Session session_list[NR_MAX_SESSION];
@@ -199,49 +180,82 @@ void ipc_server_loop(struct IpcNode* ipc_node)
*/
poll_session(session_list, NR_MAX_SESSION);
/* handle each session */
bool has_delayed = true;
for (int repeat = 0; repeat <= 1 && has_delayed; repeat++) {
has_delayed = false;
for (int i = 0; i < NR_MAX_SESSION; i++) {
session_delayed = false;
if (session_list[i].buf == NULL) {
yield(SYS_TASK_YIELD_NO_REASON);
for (int i = 0; i < NR_MAX_SESSION; i++) {
if (session_list[i].buf == NULL) {
yield(SYS_TASK_YIELD_NO_REASON);
break;
}
cur_sess_id = session_list[i].id;
ipc_server_loop_cur_msg = IPCSESSION_MSG(&session_list[i]);
if (ipc_server_loop_cur_msg->header.magic == IPC_MSG_MAGIC && ipc_server_loop_cur_msg->header.valid == 1 && //
ipc_server_loop_cur_msg->header.done == 1) {
session_forward_head(&session_list[i], ipc_server_loop_cur_msg->header.len);
ipc_server_loop_cur_msg = IPCSESSION_MSG(&session_list[i]);
}
/* handle every message in current session
a session could be delay in case one of its message(current message) needs to wait for an interrupt message's arrival
interfaces[opcode] should explicitly call delay_session() and return to delay this session
*/
while (ipc_server_loop_cur_msg->header.magic == IPC_MSG_MAGIC && ipc_server_loop_cur_msg->header.valid == 1 && //
ipc_server_loop_cur_msg->header.handling == 0 && ipc_server_loop_cur_msg->header.done == 0) {
if (session_used_size(&session_list[i]) == 0 && session_forward_tail(&session_list[i], ipc_server_loop_cur_msg->header.len) < 0) {
break;
}
cur_sess_id = session_list[i].id;
ipc_server_loop_cur_msg = IPCSESSION_MSG(&session_list[i]);
/* handle every message in current session
a session could be delay in case one of its message(current message) needs to wait for an interrupt message's arrival
interfaces[opcode] should explicitly call delay_session() and return to delay this session
*/
while (ipc_server_loop_cur_msg->header.magic == IPC_MSG_MAGIC && ipc_server_loop_cur_msg->header.valid == 1 && ipc_server_loop_cur_msg->header.done == 0) {
if (session_used_size(&session_list[i]) == 0 && session_forward_tail(&session_list[i], ipc_server_loop_cur_msg->header.len) < 0) {
break;
}
// this is a message needs to handle
if (ipc_node->interfaces[ipc_server_loop_cur_msg->header.opcode]) {
ipc_node->interfaces[ipc_server_loop_cur_msg->header.opcode](ipc_server_loop_cur_msg);
// check if this session is delayed by op handler, all messages after the delayed message in current session is blocked.
if (ipc_server_loop_cur_msg->header.done == 0) {
ipc_server_loop_cur_msg->header.delayed = 1;
has_delayed = true;
break;
}
} else {
printf("Unsupport opcode(%u) for server: %s\n", ipc_server_loop_cur_msg->header.opcode, ipc_node->name);
}
// current msg is a message that needs to ignore
// finish this message in server's perspective
if (session_forward_head(&session_list[i], ipc_server_loop_cur_msg->header.len) < 0) {
break;
}
ipc_server_loop_cur_msg = IPCSESSION_MSG(&session_list[i]);
// this is a message needs to handle
if (ipc_node->interfaces[ipc_server_loop_cur_msg->header.opcode]) {
ipc_node->interfaces[ipc_server_loop_cur_msg->header.opcode](ipc_server_loop_cur_msg);
} else {
printf("Unsupport opcode(%u) for server: %s\n", ipc_server_loop_cur_msg->header.opcode, ipc_node->name);
}
// stop handle this session
cur_sess_id = -1;
ipc_server_loop_cur_msg = NULL;
// current msg is a message that needs to ignore
// finish this message in server's perspective
if (ipc_server_loop_cur_msg->header.done == 0) {
break;
}
session_forward_head(&session_list[i], ipc_server_loop_cur_msg->header.len);
ipc_server_loop_cur_msg = IPCSESSION_MSG(&session_list[i]);
}
// stop handle this session
cur_sess_id = -1;
ipc_server_loop_cur_msg = NULL;
}
}
}
// utils
void _ipc_addr_to_buf(uintptr_t addr, char buf[17])
{
int buf_idx = 0;
while (addr != 0) {
int x = addr % 16;
if (x < 10) {
buf[buf_idx] = x + '0';
} else {
buf[buf_idx] = x - 10 + 'A';
}
buf_idx++;
addr /= 16;
}
buf[buf_idx] = '\0';
}
uintptr_t _ipc_buf_to_addr(char* buf)
{
uintptr_t addr = 0;
int buf_idx = 0;
uintptr_t multiplier = 1;
while (buf_idx < 17 && buf[buf_idx] != '\0') {
uint8_t x = (uint8_t)buf[buf_idx];
if (x >= '0' && x <= '9') {
x -= '0';
} else if (x >= 'A' && x <= 'F') {
x -= 'A';
x += 10;
}
addr += (uintptr_t)x * multiplier;
multiplier *= 16;
buf_idx++;
}
return addr;
}
Ubiquitous/XiZi_AIoT/services/lib/ipc/libipc.h
+40 -12
View File
@@ -33,6 +33,7 @@ Modification:
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>
#include <string.h>
#include "ipcargs.h"
#include "session.h"
@@ -47,7 +48,7 @@ typedef struct {
uint64_t valid : 1; // for server to peek new msg
uint64_t done : 1; // for client to check request done
uint64_t init : 1; // for client to check request done
uint64_t delayed : 1;
uint64_t handling : 1;
uint64_t nr_args : 4;
uint64_t opcode : 8;
uint64_t len : 16;
@@ -191,6 +192,7 @@ void ipc_server_loop(struct IpcNode* ipc_node);
#define IPC_CALL(ipc_name) ipc_call_copy_args_##ipc_name
#define IPC_SERVE(ipc_name) ipc_serve_##ipc_name
#define IPC_THREAD_SERVE(ipc_name) ipc_thread_serve_##ipc_name
#define IPC_DO_SERVE_FUNC(ipc_name) ipc_do_serve_##ipc_name
/// when defining a ipc server:
@@ -236,7 +238,6 @@ void ipc_server_loop(struct IpcNode* ipc_node);
return res; \
}
bool is_cur_session_delayed(void);
#define IPC_SERVER_INTERFACE(ipc_name, argc) \
static int IPC_SERVE(ipc_name)(struct IpcMsg * msg) \
{ \
@@ -245,16 +246,43 @@ bool is_cur_session_delayed(void);
argv[i] = ipc_msg_get_nth_arg_buf(msg, i); \
} \
int32_t _ret = IPC_DO_SERVE##argc(ipc_name); \
if (!is_cur_session_delayed()) { \
ipc_msg_set_return(msg, &_ret); \
msg->header.done = 1; \
} \
ipc_msg_set_return(msg, &_ret); \
msg->header.done = 1; \
return 0; \
}
int cur_session_id(void);
/// @brief delay the session(message, or a inter-process-call)
/// the delayed call will be handled again later from begining, not from the position where delay_session() is called.
/// @param
void delay_session(void);
bool is_cur_handler_been_delayed();
void _ipc_addr_to_buf(uintptr_t addr, char buf[17]);
uintptr_t _ipc_buf_to_addr(char* buf);
#define IPC_SERVER_THREAD_INTERFACE(ipc_name, argc) \
static int IPC_THREAD_SERVE(ipc_name)(int ipc_argc, char** ipc_argv) \
{ \
if (ipc_argc != 2) { \
printf("[%s] Error server thread creation.\n", __func__); \
exit(1); \
return -1; \
} \
struct IpcMsg* msg = (struct IpcMsg*)_ipc_buf_to_addr(ipc_argv[1]); \
void* argv[argc]; \
for (int i = 0; i < argc; i++) { \
argv[i] = ipc_msg_get_nth_arg_buf(msg, i); \
} \
int32_t _ret = IPC_DO_SERVE##argc(ipc_name); \
ipc_msg_set_return(msg, &_ret); \
msg->header.done = 1; \
exit(0); \
return 0; \
} \
static int IPC_SERVE(ipc_name)(struct IpcMsg * msg) \
{ \
char addr_buf[17]; \
_ipc_addr_to_buf((uintptr_t)msg, addr_buf); \
char* param[] = { #ipc_name, addr_buf, NULL }; \
int tid = thread(IPC_THREAD_SERVE(ipc_name), #ipc_name, param); \
if (tid > 0) { \
msg->header.handling = 1; \
} \
return 0; \
}
int cur_session_id(void);