diff --git a/design/mvp/canonical-abi/definitions.py b/design/mvp/canonical-abi/definitions.py index 8f04f093..97214606 100644 --- a/design/mvp/canonical-abi/definitions.py +++ b/design/mvp/canonical-abi/definitions.py @@ -516,6 +516,16 @@ def unregister_thread(self, thread): assert(thread.index is not None) self.inst.threads.remove(thread.index) + def move_thread_registration_to(self, thread, new_task): + assert(thread in self.threads and thread.task is self) + trap_if(thread is self.implicit_thread) + self.threads.remove(thread) + new_task.threads.append(thread) + if len(self.threads) == 0: + trap_if(self.state != Task.State.RESOLVED) + assert(self.num_borrows == 0) + thread.task = new_task + def request_cancellation(self, caller: Optional[ComponentInstance]): if self.state == Task.State.INITIAL: self.state = Task.State.CANCEL_DELIVERED @@ -2684,6 +2694,15 @@ def canon_thread_index(): assert(thread.index is not None) return [thread.index] +### 🧵 `canon thread.join-task-of` + +def canon_thread_join_task_of(i): + thread = current_thread() + trap_if(not thread.task.inst.may_leave) + other_thread = thread.task.inst.threads.get(i) + thread.task.move_thread_registration_to(thread, other_thread.task) + return [] + ### 🧵 `canon thread.new-indirect` @dataclass @@ -2699,7 +2718,7 @@ def canon_thread_new_indirect(ft, ftbl: Table[CoreFuncRef], fi, c): trap_if(f.t != ft) def thread_func(): [] = call_and_trap_on_throw(f.callee, [c]) - task.unregister_thread(new_thread) + new_thread.task.unregister_thread(new_thread) new_thread = Thread(task, thread_func) assert(new_thread.suspended()) task.register_thread(new_thread) diff --git a/design/mvp/canonical-abi/run_tests.py b/design/mvp/canonical-abi/run_tests.py index 619e5f42..63bcf59c 100644 --- a/design/mvp/canonical-abi/run_tests.py +++ b/design/mvp/canonical-abi/run_tests.py @@ -515,6 +515,11 @@ def core_consumer_realloc(args): fail("thread.index must trap during realloc") except Trap: pass + try: + canon_thread_join_task_of(consumer_thread.index) + fail("thread.join-task-of must trap during realloc") + except Trap: + pass return consumer_heap.realloc(args) consumer_opts = mk_opts(MemInst(consumer_heap.memory, 'i32'), realloc = core_consumer_realloc) @@ -3020,6 +3025,122 @@ def core_consumer(args): lift_and_run(mk_opts(), consumer_inst, consumer_ft, core_consumer, lambda:[], lambda _:()) +def test_thread_join_task_of(): + store = Store() + inst = ComponentInstance(store) + opts = mk_opts(async_ = True) + + ftbl = Table() + ft = CoreFuncType(['i32'],[]) + + t1i = None + t2i = None + bi = None + + def thread_func1(args): + assert(args == [201]) + task_a = current_task() + assert(task_a.state == Task.State.RESOLVED) + assert(len(task_a.threads) == 2) + + assert(canon_thread_index() == [t1i]) + [] = canon_thread_join_task_of(t1i) + assert(current_task() is task_a) + [] = canon_thread_join_task_of(t2i) + assert(current_task() is task_a) + assert(canon_thread_index() == [t1i]) + + [] = canon_thread_join_task_of(bi) + assert(current_task() is not task_a) + assert(current_task() is inst.threads.get(bi).task) + assert(canon_thread_index() == [t1i]) + assert(len(task_a.threads) == 1) + + [] = canon_task_return([U8Type()], opts, [55]) + [] = canon_thread_resume_later(t2i) + return [] + fi1 = ftbl.add(CoreFuncRef(ft, thread_func1)) + + def thread_func2(args): + assert(args == [202]) + task_a = current_task() + assert(task_a.state == Task.State.RESOLVED) + assert(len(task_a.threads) == 1) + + try: + canon_thread_join_task_of(t1i) + fail("thread.join-task-of must trap on an exited thread's index") + except Trap: + pass + + [] = canon_thread_join_task_of(bi) + assert(current_task() is not task_a) + assert(len(task_a.threads) == 0) + assert(canon_thread_index() == [t2i]) + + [] = canon_thread_resume_later(bi) + return [] + fi2 = ftbl.add(CoreFuncRef(ft, thread_func2)) + + def core_func_a(args): + assert(not args) + nonlocal t1i, t2i + + [ai] = canon_thread_index() + try: + canon_thread_join_task_of(ai) + fail("thread.join-task-of must trap on an implicit thread") + except Trap: + pass + + [t1i] = canon_thread_new_indirect(ft, ftbl, fi1, 201) + [t2i] = canon_thread_new_indirect(ft, ftbl, fi2, 202) + [] = canon_thread_resume_later(t1i) + [] = canon_task_return([U8Type()], opts, [11]) + return [] + + def core_func_b(args): + assert(not args) + nonlocal bi + [bi] = canon_thread_index() + [cancelled] = canon_thread_suspend(False) + assert(cancelled == Cancelled.FALSE) + return [] + + a_result = None + def on_resolve_a(v): + nonlocal a_result + [a_result] = v + + b_result = None + def on_resolve_b(v): + nonlocal b_result + [b_result] = v + + caller_ft = FuncType([], [U8Type()], async_ = True) + _ = store.invoke(store.lift(core_func_a, caller_ft, opts, inst), lambda:[], on_resolve_a) + _ = store.invoke(store.lift(core_func_b, caller_ft, opts, inst), lambda:[], on_resolve_b) + while store.waiting: + store.tick() + assert(a_result == 11) + assert(b_result == 55) + + # Moving the last thread of an unresolved task traps. Since this trap fires + # in the middle of mutating spec-internal state, exercise it directly on + # scratch objects rather than catching the (normally store-fatal) trap + # inside a live component instance and continuing. + scratch_inst = ComponentInstance(Store()) + scratch_task_a = Task(caller_ft, opts, scratch_inst, lambda:[], lambda _:()) + scratch_task_b = Task(caller_ft, opts, scratch_inst, lambda:[], lambda _:()) + scratch_thread = Thread(scratch_task_a, lambda:()) + scratch_task_a.register_thread(scratch_thread) + try: + scratch_task_a.move_thread_registration_to(scratch_thread, scratch_task_b) + fail("moving the last thread of an unresolved task must trap") + except Trap: + pass + scratch_thread.resume() + test_roundtrips() test_cross_component_realloc() test_handles() @@ -3047,5 +3168,6 @@ def core_consumer(args): test_threads() test_sync_threads() test_thread_cancel_callback() +test_thread_join_task_of() print("All tests passed")