From 02b40ebb11c5d12e8224f7b2d66816a85cf2cc38 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Sch=C3=A4r?= Date: Wed, 13 Apr 2022 19:42:35 +0200 Subject: [PATCH] Fix RPC locking --- include/aos/aos_rpc.h | 4 +- lib/aos/aos_rpc.c | 111 +++++++++++++++++++++++++++--------------- 2 files changed, 74 insertions(+), 41 deletions(-) diff --git a/include/aos/aos_rpc.h b/include/aos/aos_rpc.h index 8d4609f..cdd4d45 100644 --- a/include/aos/aos_rpc.h +++ b/include/aos/aos_rpc.h @@ -44,6 +44,8 @@ struct aos_rpc { // mutually exclusive access for threads accessing rpc 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) { errval_t err; - + if (len > 0) { err = sys_print(buf, len); if (err_is_fail(err)) { diff --git a/lib/aos/aos_rpc.c b/lib/aos/aos_rpc.c index 69e0ac5..90021b4 100644 --- a/lib/aos/aos_rpc.c +++ b/lib/aos/aos_rpc.c @@ -27,18 +27,27 @@ static errval_t do_aos_rpc( ) { errval_t err; - // the calling thread MUST have the rpc lock at this point - // we check to be sure we did not forget to lock in any of the rpc functions! - assert(rpc->lock.locked > 0); - dispatcher_handle_t handle = disp_disable(); - struct dispatcher_generic *disp_gen = get_dispatcher_generic(handle); - assert(rpc->lock.holder == disp_gen->current); - disp_enable(handle); + // Allocate slot if we expect a capability. + // We must do this before taking the lock, because allocating a slot may + // allocate memory and thus do a aos_rpc_get_ram_cap. + struct capref slot; + if (ret_cap != NULL) { + err = slot_alloc(&slot); + 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 - if (ret_cap != NULL && rpc->chan->endpoint->k.recv_cptr == 0){ - lmp_chan_alloc_recv_slot(rpc->chan); - assert(rpc->chan->endpoint->k.recv_cptr != 0); + thread_mutex_lock(&rpc->lock); + + if (ret_cap != NULL) { + 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 @@ -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); if (!err_is_fail(err)) break; if (!lmp_err_is_transient(err)) { + thread_mutex_unlock(&rpc->lock); return err_push(err, AOS_ERR_LMP_SEND_FAILURE); } 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); - if (err_is_fail(err)) return err; + if (err_is_fail(err)) { + thread_mutex_unlock(&rpc->lock); + return err; + } } // Receive return value struct lmp_recv_msg recv = LMP_RECV_MSG_INIT; while (!lmp_endpoint_can_recv(rpc->chan->endpoint)) { 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); - 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); assert(err_is_ok(err)); + thread_mutex_unlock(&rpc->lock); + if (ret_size != NULL) *ret_size = recv.words[1]; if (ret0 != NULL) *ret0 = recv.words[2]; 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. errval_t err; - thread_mutex_lock(&rpc->lock); err = do_aos_rpc( rpc, RPC_MTYPE_SEND_NUMBER, NULL_CAP, 0, num, 0, NULL, NULL, NULL, NULL); - thread_mutex_unlock(&rpc->lock); return err; } @@ -97,14 +119,14 @@ aos_rpc_send_string(struct aos_rpc *rpc, const char *string) { size_t string_size = strlen(string) + 1; 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); err = do_aos_rpc( rpc, RPC_MTYPE_SEND_STRING, NULL_CAP, string_size, 0, 0, NULL, NULL, NULL, NULL ); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); return err; } @@ -117,14 +139,12 @@ aos_rpc_get_ram_cap(struct aos_rpc *rpc, size_t bytes, size_t alignment, errval_t err; - thread_mutex_lock_nested(&rpc->lock); err = do_aos_rpc( rpc, RPC_MTYPE_GET_RAM_CAP, NULL_CAP, 0, bytes, alignment, ret_cap, NULL, ret_bytes, NULL ); - thread_mutex_unlock(&rpc->lock); - + return err; } @@ -136,12 +156,10 @@ aos_rpc_serial_getchar(struct aos_rpc *rpc, char *retc) { errval_t err; uintptr_t retval; - thread_mutex_lock(&rpc->lock); err = do_aos_rpc( rpc, RPC_MTYPE_SERIAL_GETCHAR, NULL_CAP, 0, 0, 0, NULL, NULL, &retval, NULL); - thread_mutex_unlock(&rpc->lock); *retc = retval; return err; @@ -154,13 +172,11 @@ aos_rpc_serial_putchar(struct aos_rpc *rpc, char c) { // serial port. errval_t err; - thread_mutex_lock(&rpc->lock); err = do_aos_rpc( rpc, RPC_MTYPE_SERIAL_PUTCHAR, NULL_CAP, 0, c, 0, NULL, NULL, NULL, NULL ); - thread_mutex_unlock(&rpc->lock); 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 written_bytes; - thread_mutex_lock(&rpc->lock); + thread_mutex_lock(&rpc->shared_mem_lock); memcpy(rpc->shared_mem, buf, chunk_len); err = do_aos_rpc( rpc, RPC_MTYPE_SERIAL_WRITE, NULL_CAP, chunk_len, 0, 0, NULL, NULL, &written_bytes, NULL ); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); if (err_is_fail(err)) { 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 read_bytes; - thread_mutex_lock(&rpc->lock); + thread_mutex_lock(&rpc->shared_mem_lock); err = do_aos_rpc( rpc, RPC_MTYPE_SERIAL_READ, 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)) { DEBUG_ERR(err, "Failed to read the full buffer"); + thread_mutex_unlock(&rpc->shared_mem_lock); return total_bytes; } assert(read_bytes <= chunk_len); memcpy(buf, rpc->shared_mem, read_bytes); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); 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; 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); uintptr_t retval; 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, NULL, &retval, NULL ); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); *newpid = retval; return err; @@ -276,19 +293,26 @@ aos_rpc_process_get_name(struct aos_rpc *rpc, domainid_t pid, char **name) { size_t name_len; errval_t err; - thread_mutex_lock(&rpc->lock); + thread_mutex_lock(&rpc->shared_mem_lock); err = do_aos_rpc( rpc, RPC_MTYPE_PROCESS_GET_NAME, NULL_CAP, 0, pid, 0, 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); - 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); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); assert(name_len != 0 && (*name)[name_len - 1] == '\0'); return err; @@ -302,21 +326,27 @@ aos_rpc_process_get_all_pids(struct aos_rpc *rpc, domainid_t **pids, size_t ret_len; errval_t err; - thread_mutex_lock(&rpc->lock); + thread_mutex_lock(&rpc->shared_mem_lock); err = do_aos_rpc( rpc, RPC_MTYPE_PROCESS_GET_ALL_PIDS, NULL_CAP, 0, 0, 0, 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); *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); - thread_mutex_unlock(&rpc->lock); + thread_mutex_unlock(&rpc->shared_mem_lock); 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; waitset_init(&rpc->ws); thread_mutex_init(&rpc->lock); + thread_mutex_init(&rpc->shared_mem_lock); return SYS_ERR_OK; }