dinosauriel/aos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix RPC locking

Browse Source
This commit is contained in:
Jan Schär 2022-04-13 19:42:35 +02:00
parent 3c85b4d971
commit 02b40ebb11
2 changed files with 74 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/aos/aos_rpc.h
View File

@ -44,6 +44,8 @@ struct aos_rpc {
// mutually exclusive access for threads accessing rpc // mutually exclusive access for threads accessing rpc
struct thread_mutex lock; struct thread_mutex lock;
// take this lock if you use shared_mem, to ensure no-one else writes to it
struct thread_mutex shared_mem_lock;
}; };
/** /**
@ -152,7 +154,7 @@ struct aos_rpc *aos_rpc_get_serial_channel(void);
static inline size_t dummy_terminal_write(const char *buf, size_t len) static inline size_t dummy_terminal_write(const char *buf, size_t len)
{ {
errval_t err; errval_t err;
if (len > 0) { if (len > 0) {
err = sys_print(buf, len); err = sys_print(buf, len);
if (err_is_fail(err)) { if (err_is_fail(err)) {

111
lib/aos/aos_rpc.c
View File

@ -27,18 +27,27 @@ static errval_t do_aos_rpc(
) { ) {
errval_t err; errval_t err;
// the calling thread MUST have the rpc lock at this point // Allocate slot if we expect a capability.
// we check to be sure we did not forget to lock in any of the rpc functions! // We must do this before taking the lock, because allocating a slot may
assert(rpc->lock.locked > 0); // allocate memory and thus do a aos_rpc_get_ram_cap.
dispatcher_handle_t handle = disp_disable(); struct capref slot;
struct dispatcher_generic *disp_gen = get_dispatcher_generic(handle); if (ret_cap != NULL) {
assert(rpc->lock.holder == disp_gen->current); err = slot_alloc(&slot);
disp_enable(handle); if (err_is_fail(err)) {
return err_push(err, LIB_ERR_SLOT_ALLOC);
}
}
// allocate capability if we expect one and the recv slot is empty thread_mutex_lock(&rpc->lock);
if (ret_cap != NULL && rpc->chan->endpoint->k.recv_cptr == 0){
lmp_chan_alloc_recv_slot(rpc->chan); if (ret_cap != NULL) {
assert(rpc->chan->endpoint->k.recv_cptr != 0); if (rpc->chan->endpoint->k.recv_cptr == 0) {
lmp_chan_set_recv_slot(rpc->chan, slot);
} else {
// Freeing slots never allocates memory, so we can do this
// while we have the lock.
slot_free(slot);
}
} }
// Send call // Send call
@ -46,25 +55,40 @@ static errval_t do_aos_rpc(
err = lmp_chan_send4(rpc->chan, LMP_FLAG_YIELD, arg_cap, msg_type, arg_size, arg0, arg1); err = lmp_chan_send4(rpc->chan, LMP_FLAG_YIELD, arg_cap, msg_type, arg_size, arg0, arg1);
if (!err_is_fail(err)) break; if (!err_is_fail(err)) break;
if (!lmp_err_is_transient(err)) { if (!lmp_err_is_transient(err)) {
thread_mutex_unlock(&rpc->lock);
return err_push(err, AOS_ERR_LMP_SEND_FAILURE); return err_push(err, AOS_ERR_LMP_SEND_FAILURE);
} }
err = lmp_chan_register_send(rpc->chan, &rpc->ws, MKCLOSURE(noop_callback, NULL)); err = lmp_chan_register_send(rpc->chan, &rpc->ws, MKCLOSURE(noop_callback, NULL));
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->lock);
return err;
}
err = event_dispatch(&rpc->ws); err = event_dispatch(&rpc->ws);
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->lock);
return err;
}
} }
// Receive return value // Receive return value
struct lmp_recv_msg recv = LMP_RECV_MSG_INIT; struct lmp_recv_msg recv = LMP_RECV_MSG_INIT;
while (!lmp_endpoint_can_recv(rpc->chan->endpoint)) { while (!lmp_endpoint_can_recv(rpc->chan->endpoint)) {
err = lmp_chan_register_recv(rpc->chan, &rpc->ws, MKCLOSURE(noop_callback, NULL)); err = lmp_chan_register_recv(rpc->chan, &rpc->ws, MKCLOSURE(noop_callback, NULL));
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->lock);
return err;
}
err = event_dispatch(&rpc->ws); err = event_dispatch(&rpc->ws);
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->lock);
return err;
}
} }
err = lmp_chan_recv(rpc->chan, &recv, ret_cap); err = lmp_chan_recv(rpc->chan, &recv, ret_cap);
assert(err_is_ok(err)); assert(err_is_ok(err));
thread_mutex_unlock(&rpc->lock);
if (ret_size != NULL) *ret_size = recv.words[1]; if (ret_size != NULL) *ret_size = recv.words[1];
if (ret0 != NULL) *ret0 = recv.words[2]; if (ret0 != NULL) *ret0 = recv.words[2];
if (ret1 != NULL) *ret1 = recv.words[3]; if (ret1 != NULL) *ret1 = recv.words[3];
@ -78,12 +102,10 @@ aos_rpc_send_number(struct aos_rpc *rpc, uintptr_t num) {
// given channel and wait until the ack gets returned. // given channel and wait until the ack gets returned.
errval_t err; errval_t err;
thread_mutex_lock(&rpc->lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SEND_NUMBER, rpc, RPC_MTYPE_SEND_NUMBER,
NULL_CAP, 0, num, 0, NULL_CAP, 0, num, 0,
NULL, NULL, NULL, NULL); NULL, NULL, NULL, NULL);
thread_mutex_unlock(&rpc->lock);
return err; return err;
} }
@ -97,14 +119,14 @@ aos_rpc_send_string(struct aos_rpc *rpc, const char *string) {
size_t string_size = strlen(string) + 1; size_t string_size = strlen(string) + 1;
if (string_size > RPC_SHARED_SIZE) return AOS_ERR_RPC_ARG_TOO_BIG; if (string_size > RPC_SHARED_SIZE) return AOS_ERR_RPC_ARG_TOO_BIG;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
memcpy(rpc->shared_mem, string, string_size); memcpy(rpc->shared_mem, string, string_size);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SEND_STRING, rpc, RPC_MTYPE_SEND_STRING,
NULL_CAP, string_size, 0, 0, NULL_CAP, string_size, 0, 0,
NULL, NULL, NULL, NULL NULL, NULL, NULL, NULL
); );
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
return err; return err;
} }
@ -117,14 +139,12 @@ aos_rpc_get_ram_cap(struct aos_rpc *rpc, size_t bytes, size_t alignment,
errval_t err; errval_t err;
thread_mutex_lock_nested(&rpc->lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_GET_RAM_CAP, rpc, RPC_MTYPE_GET_RAM_CAP,
NULL_CAP, 0, bytes, alignment, NULL_CAP, 0, bytes, alignment,
ret_cap, NULL, ret_bytes, NULL ret_cap, NULL, ret_bytes, NULL
); );
thread_mutex_unlock(&rpc->lock);
return err; return err;
} }
@ -136,12 +156,10 @@ aos_rpc_serial_getchar(struct aos_rpc *rpc, char *retc) {
errval_t err; errval_t err;
uintptr_t retval; uintptr_t retval;
thread_mutex_lock(&rpc->lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SERIAL_GETCHAR, rpc, RPC_MTYPE_SERIAL_GETCHAR,
NULL_CAP, 0, 0, 0, NULL_CAP, 0, 0, 0,
NULL, NULL, &retval, NULL); NULL, NULL, &retval, NULL);
thread_mutex_unlock(&rpc->lock);
*retc = retval; *retc = retval;
return err; return err;
@ -154,13 +172,11 @@ aos_rpc_serial_putchar(struct aos_rpc *rpc, char c) {
// serial port. // serial port.
errval_t err; errval_t err;
thread_mutex_lock(&rpc->lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SERIAL_PUTCHAR, rpc, RPC_MTYPE_SERIAL_PUTCHAR,
NULL_CAP, 0, c, 0, NULL_CAP, 0, c, 0,
NULL, NULL, NULL, NULL NULL, NULL, NULL, NULL
); );
thread_mutex_unlock(&rpc->lock);
return err; return err;
} }
@ -177,14 +193,14 @@ aos_rpc_serial_write(struct aos_rpc *rpc, const char *buf, size_t buf_len) {
size_t chunk_len = MIN(buf_len, RPC_SHARED_SIZE); size_t chunk_len = MIN(buf_len, RPC_SHARED_SIZE);
size_t written_bytes; size_t written_bytes;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
memcpy(rpc->shared_mem, buf, chunk_len); memcpy(rpc->shared_mem, buf, chunk_len);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SERIAL_WRITE, rpc, RPC_MTYPE_SERIAL_WRITE,
NULL_CAP, chunk_len, 0, 0, NULL_CAP, chunk_len, 0, 0,
NULL, NULL, &written_bytes, NULL NULL, NULL, &written_bytes, NULL
); );
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
if (err_is_fail(err)) { if (err_is_fail(err)) {
DEBUG_ERR(err, "Failed to write the full buffer"); DEBUG_ERR(err, "Failed to write the full buffer");
@ -218,7 +234,7 @@ aos_rpc_serial_read(struct aos_rpc *rpc, char *buf, size_t buf_len) {
size_t chunk_len = MIN(buf_len, RPC_SHARED_SIZE); size_t chunk_len = MIN(buf_len, RPC_SHARED_SIZE);
size_t read_bytes; size_t read_bytes;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_SERIAL_READ, rpc, RPC_MTYPE_SERIAL_READ,
NULL_CAP, 0, chunk_len, 0, NULL_CAP, 0, chunk_len, 0,
@ -226,12 +242,13 @@ aos_rpc_serial_read(struct aos_rpc *rpc, char *buf, size_t buf_len) {
); );
if (err_is_fail(err)) { if (err_is_fail(err)) {
DEBUG_ERR(err, "Failed to read the full buffer"); DEBUG_ERR(err, "Failed to read the full buffer");
thread_mutex_unlock(&rpc->shared_mem_lock);
return total_bytes; return total_bytes;
} }
assert(read_bytes <= chunk_len); assert(read_bytes <= chunk_len);
memcpy(buf, rpc->shared_mem, read_bytes); memcpy(buf, rpc->shared_mem, read_bytes);
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
total_bytes += read_bytes; total_bytes += read_bytes;
@ -254,7 +271,7 @@ aos_rpc_process_spawn(struct aos_rpc *rpc, char *cmdline,
size_t cmdline_size = strlen(cmdline) + 1; size_t cmdline_size = strlen(cmdline) + 1;
if (cmdline_size > RPC_SHARED_SIZE) return AOS_ERR_RPC_ARG_TOO_BIG; if (cmdline_size > RPC_SHARED_SIZE) return AOS_ERR_RPC_ARG_TOO_BIG;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
memcpy(rpc->shared_mem, cmdline, cmdline_size); memcpy(rpc->shared_mem, cmdline, cmdline_size);
uintptr_t retval; uintptr_t retval;
errval_t err = do_aos_rpc( errval_t err = do_aos_rpc(
@ -262,7 +279,7 @@ aos_rpc_process_spawn(struct aos_rpc *rpc, char *cmdline,
NULL_CAP, cmdline_size, core, 0, NULL_CAP, cmdline_size, core, 0,
NULL, NULL, &retval, NULL NULL, NULL, &retval, NULL
); );
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
*newpid = retval; *newpid = retval;
return err; return err;
@ -276,19 +293,26 @@ aos_rpc_process_get_name(struct aos_rpc *rpc, domainid_t pid, char **name) {
size_t name_len; size_t name_len;
errval_t err; errval_t err;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_PROCESS_GET_NAME, rpc, RPC_MTYPE_PROCESS_GET_NAME,
NULL_CAP, 0, pid, 0, NULL_CAP, 0, pid, 0,
NULL, &name_len, NULL, NULL NULL, &name_len, NULL, NULL
); );
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->shared_mem_lock);
return err;
}
// malloc may use RPC but not the shared memory, so this is fine.
*name = malloc(name_len); *name = malloc(name_len);
if (*name == NULL) return LIB_ERR_MALLOC_FAIL; if (*name == NULL) {
thread_mutex_unlock(&rpc->shared_mem_lock);
return LIB_ERR_MALLOC_FAIL;
}
memcpy(*name, rpc->shared_mem, name_len); memcpy(*name, rpc->shared_mem, name_len);
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
assert(name_len != 0 && (*name)[name_len - 1] == '\0'); assert(name_len != 0 && (*name)[name_len - 1] == '\0');
return err; return err;
@ -302,21 +326,27 @@ aos_rpc_process_get_all_pids(struct aos_rpc *rpc, domainid_t **pids,
size_t ret_len; size_t ret_len;
errval_t err; errval_t err;
thread_mutex_lock(&rpc->lock); thread_mutex_lock(&rpc->shared_mem_lock);
err = do_aos_rpc( err = do_aos_rpc(
rpc, RPC_MTYPE_PROCESS_GET_ALL_PIDS, rpc, RPC_MTYPE_PROCESS_GET_ALL_PIDS,
NULL_CAP, 0, 0, 0, NULL_CAP, 0, 0, 0,
NULL, &ret_len, NULL, NULL NULL, &ret_len, NULL, NULL
); );
if (err_is_fail(err)) return err; if (err_is_fail(err)) {
thread_mutex_unlock(&rpc->shared_mem_lock);
return err;
}
*pid_count = ret_len / sizeof(**pids); *pid_count = ret_len / sizeof(**pids);
*pids = malloc(ret_len); *pids = malloc(ret_len);
if (*pids == NULL) return LIB_ERR_MALLOC_FAIL; if (*pids == NULL) {
thread_mutex_unlock(&rpc->shared_mem_lock);
return LIB_ERR_MALLOC_FAIL;
}
memcpy(*pids, rpc->shared_mem, ret_len); memcpy(*pids, rpc->shared_mem, ret_len);
thread_mutex_unlock(&rpc->lock); thread_mutex_unlock(&rpc->shared_mem_lock);
return err; return err;
} }
@ -328,6 +358,7 @@ errval_t aos_rpc_init(struct aos_rpc *rpc, struct lmp_chan *chan, void *shared_m
rpc->shared_mem = shared_mem; rpc->shared_mem = shared_mem;
waitset_init(&rpc->ws); waitset_init(&rpc->ws);
thread_mutex_init(&rpc->lock); thread_mutex_init(&rpc->lock);
thread_mutex_init(&rpc->shared_mem_lock);
return SYS_ERR_OK; return SYS_ERR_OK;
} }