diff --git a/aephysics/dynamics/module.ae b/aephysics/dynamics/module.ae index ffd7fa1..7df1328 100644 --- a/aephysics/dynamics/module.ae +++ b/aephysics/dynamics/module.ae @@ -142,7 +142,7 @@ exports ( world_body_move_event_count, world_body_move_event, world_contact_hit_event_count, world_contact_hit_event, world_joint_event_count, world_joint_event, world_set_pre_solve, world_clear_pre_solve, shape_user_material_id, make_relative_sweep, body_move_event_at, - TaskContext, task_context, task_context_at, task_sensor_hit_at + TaskContext, task_context, task_context_at, task_sensor_hit_at, enter_world, leave_world ) extern memset(block: ptr, value: int, size: int) -> ptr @@ -1133,6 +1133,8 @@ struct World { constraint_graph: ConstraintGraph task_contexts: ptr // TaskContext[worker_count] worker_count: int + worker_base: int // the first of the worker slots the world's steps run as + owns_worker_slots: bool // claimed for this world alone (else base 0, shared) scheduler: ptr // *Scheduler (aephysics.parallel), the step's threads owns_scheduler: bool // made with the world, or the host's contact_state: BitSet // the contacts whose touching changed this pass, every worker's merged @@ -1332,6 +1334,12 @@ create_world(def: *WorldDef) -> WorldId { w.owns_scheduler = true } reserve_scratch() + // The world's steps run as worker slots of its own, so worlds stepped + // at once on different threads use different scratch; with none free, + // slot 0 on, shared, as before (such worlds step one at a time). + w.worker_base = claim_worker_slots(w.worker_count) + w.owns_worker_slots = w.worker_base >= 0 + if w.worker_base < 0 { w.worker_base = 0 } w.task_contexts = basics.alloc(w.worker_count * sizeof(TaskContext)) i = 0 while i < w.worker_count { @@ -1513,6 +1521,7 @@ destroy_world(id: WorldId) { i = i + 1 } basics.free_bytes(w.task_contexts, w.worker_count * sizeof(TaskContext)) + if w.owns_worker_slots { release_worker_slots(w.worker_base, w.worker_count) } if w.owns_scheduler { parallel.destroy_scheduler(w.scheduler as *Scheduler) } basics.bitset_destroy(&w.contact_state) basics.buffer_destroy(&w.contact_begin_events) @@ -4510,10 +4519,74 @@ collide_range(start: int, end: int, worker: int, context: ptr) { } // The calling worker's context, and any worker's. -task_context(w: *World) -> *TaskContext { return (w.task_contexts + native.worker_index() * sizeof(TaskContext)) as *TaskContext } +// The calling worker's context: its slot less the world's base. A serial +// caller outside a step (a test driving the collide pass) is worker 0. +task_context(w: *World) -> *TaskContext { + local = native.worker_index() - w.worker_base + if local < 0 || local >= w.worker_count { local = 0 } + return (w.task_contexts + local * sizeof(TaskContext)) as *TaskContext +} + +// --- worker slots ------------------------------------------------------ + +// The worker slots the worlds hold, a bit each (MAX_WORKERS is 64). A world +// claims a run of its worker count when made and gives it back when +// destroyed; worlds are made and destroyed on one thread at a time. +var g_worker_slots: long = 0 + +worker_slot_bit(slot: int) -> long { return (1 as long) << (slot as long) } + +// The first slot of a free run of count, claimed; -1 when there is none. +claim_worker_slots(count: int) -> int { + start = 0 + while start + count <= native.MAX_WORKERS { + free = true + k = 0 + while k < count && free { + if (g_worker_slots & worker_slot_bit(start + k)) != (0 as long) { free = false } + k = k + 1 + } + if free { + k = 0 + while k < count { + g_worker_slots = g_worker_slots | worker_slot_bit(start + k) + k = k + 1 + } + return start + } + start = start + 1 + } + return 0 - 1 +} + +release_worker_slots(start: int, count: int) { + k = 0 + while k < count { + g_worker_slots = g_worker_slots & (worker_slot_bit(start + k) ^ ((0 as long) - (1 as long))) + k = k + 1 + } +} + +// The calling thread made the world's first worker for a pass over it (a +// step, the collide pass, the solve); the index it had before, to give back. +enter_world(w: *World) -> int { + caller = native.worker_index() + native.set_worker_index(w.worker_base) + return caller +} + +leave_world(caller: int) { native.set_worker_index(caller) } task_context_at(w: *World, index: int) -> *TaskContext { return (w.task_contexts + index * sizeof(TaskContext)) as *TaskContext } +// The collide pass as the world's workers, whatever the calling thread's index. collide(w: *World) -> int { + caller = enter_world(w) + touching = collide_pass(w) + leave_world(caller) + return touching +} + +collide_pass(w: *World) -> int { basics.buffer_clear(&w.contact_begin_events) touching_count = 0 i = 0 diff --git a/aephysics/parallel/module.ae b/aephysics/parallel/module.ae index 69d62f8..8e01aa6 100644 --- a/aephysics/parallel/module.ae +++ b/aephysics/parallel/module.ae @@ -67,6 +67,8 @@ struct Scheduler { semaphore: ptr shutdown: int // atomic live: int // atomic + for_tasks: ptr // ForTask[MAX_WORKERS], parallel_for's, the scheduler's own + for_handles: ptr // *Task[MAX_WORKERS] } task_at(s: *Scheduler, index: int) -> *Task { return (s.tasks + index * sizeof(Task)) as *Task } @@ -112,6 +114,8 @@ create_scheduler(worker_count: int) -> *Scheduler { s = calloc(1, sizeof(Scheduler)) as *Scheduler s.worker_count = count s.tasks = calloc(MAX_TASKS, sizeof(Task)) + s.for_tasks = calloc(MAX_WORKERS, sizeof(ForTask)) + s.for_handles = calloc(MAX_WORKERS, 8) s.semaphore = native.semaphore_new(0) i = 1 while i < count { @@ -133,6 +137,8 @@ destroy_scheduler(s: *Scheduler) { } native.semaphore_free(s.semaphore) free(s.tasks) + free(s.for_tasks) + free(s.for_handles) free(s as ptr) } @@ -171,9 +177,10 @@ enqueue(s: *Scheduler, body: fn(ptr) -> void, context: ptr, worker_index: int) - t.body = body t.context = context t.worker_index = worker_index + caller = native.worker_index() native.set_worker_index(worker_index) body(context) - native.set_worker_index(0) + native.set_worker_index(caller) native.atomic_store(status_of(t), TASK_COMPLETE) return t } @@ -196,10 +203,11 @@ enqueue(s: *Scheduler, body: fn(ptr) -> void, context: ptr, worker_index: int) - // the thread when there are none; the caller is worker 0 again after. finish(s: *Scheduler, t: *Task) { if t == null { return } + caller = native.worker_index() while native.atomic_load(status_of(t)) != TASK_COMPLETE { if execute_one(s) == false { native.yield_thread() } } - native.set_worker_index(0) + native.set_worker_index(caller) } // --- parallel_for -------------------------------------------------------------------------------------------- @@ -233,9 +241,6 @@ for_trampoline(context: ptr) { } } -var g_for_tasks: ptr = null // ForTask[MAX_WORKERS] -var g_for_handles: ptr = null // *Task[MAX_WORKERS] - // [0, item_count) in blocks of at least min_range, over the scheduler's // workers; the caller works as worker 0 and returns when every block is // done, with how many blocks there were. Blocks stay at min_range until @@ -244,21 +249,15 @@ var g_for_handles: ptr = null // *Task[MAX_WORKERS] parallel_for(s: *Scheduler, body: fn(int, int, int, ptr) -> void, item_count: int, min_range: int, context: ptr) -> int { if item_count <= 0 { return 0 } worker_count = s.worker_count - // The caller works as worker 0 -- the task contexts and every module's - // scratch are chosen by the thread's worker index -- and gets its own - // index back after: a host stepping a one-worker world from one of its - // pool's threads (index k) indexed the world's single task context at k. + // The caller is the range's worker 0 and keeps its own slot (a world's + // first, set by whoever entered the world); task i runs as the slot i on + // from it, so worlds in their own slots share no scratch. The body is + // given the worker's index within the range. caller_index = native.worker_index() if worker_count <= 1 { - native.set_worker_index(0) body(0, item_count, 0, context) - native.set_worker_index(caller_index) return 1 } - if g_for_tasks == null { - g_for_tasks = calloc(MAX_WORKERS, sizeof(ForTask)) - g_for_handles = calloc(MAX_WORKERS, 8) - } max_block_count = 32 * worker_count block_size = min_range if item_count > min_range * max_block_count { block_size = (item_count + max_block_count - 1) / max_block_count } @@ -268,18 +267,17 @@ parallel_for(s: *Scheduler, body: fn(int, int, int, ptr) -> void, item_count: in shared = ForShared { next_block: 0, block_count: block_count, block_size: block_size, item_count: item_count, body: body, context: context } begin_round(s) - tasks = g_for_tasks as ForTask[] - handles = g_for_handles as ptr[] + tasks = s.for_tasks as ForTask[] + handles = s.for_handles as ptr[] i = 1 while i < task_count { tasks[i] = ForTask { shared: &shared, worker_index: i } - handles[i] = enqueue(s, for_trampoline, (g_for_tasks + i * sizeof(ForTask)), i) as ptr + handles[i] = enqueue(s, for_trampoline, (s.for_tasks + i * sizeof(ForTask)), caller_index + i) as ptr i = i + 1 } // The calling thread takes blocks as worker 0. tasks[0] = ForTask { shared: &shared, worker_index: 0 } - native.set_worker_index(0) - for_trampoline(g_for_tasks) + for_trampoline(s.for_tasks) i = 1 while i < task_count { finish(s, handles[i] as *Task) diff --git a/aephysics/physics_world/module.ae b/aephysics/physics_world/module.ae index 266ae8a..c0867d6 100644 --- a/aephysics/physics_world/module.ae +++ b/aephysics/physics_world/module.ae @@ -133,11 +133,10 @@ is_valid(id: WorldId) -> bool { return dynamics.world_is_valid(id) } step(id: WorldId, time_step: float, sub_step_count: int) { w = world_of(id) if w == null { return } - // The step runs on the calling thread as worker 0 (its task context and - // scratch), whatever the thread's index in a host's pool; the index is - // given back at the end. - caller_index = native.worker_index() - native.set_worker_index(0) + // The step runs on the calling thread as the world's first worker slot + // (its task context and scratch), whatever the thread's index in a host's + // pool; the index is given back at the end. + caller_index = dynamics.enter_world(w) solver.stage_step() w.locked = true basics.buffer_clear(&w.body_move_events) @@ -187,7 +186,7 @@ step(id: WorldId, time_step: float, sub_step_count: int) { dynamics.swap_end_events(w) w.profile.step = ms(os.now_monotonic_ns() - t0) w.locked = false - native.set_worker_index(caller_index) + dynamics.leave_world(caller_index) } ms(ns: long) -> float { return (ns as float) / 1000000.0 } diff --git a/aephysics/solver/module.ae b/aephysics/solver/module.ae index b46c67f..8b849e7 100644 --- a/aephysics/solver/module.ae +++ b/aephysics/solver/module.ae @@ -274,7 +274,9 @@ continuous_query_visitor(proxy_id: int, user_data: long, context: ptr) -> bool { if shape.should_shapes_collide(fast_shape.filter, s.filter) == false { return true } body = dynamics.body_at(w, s.body_id) sim = dynamics.get_body_sim(w, body) - if (sim.flags & dynamics.IS_BULLET) != 0 { return true } + // Another bullet's flags, which its own sweep may be setting bits in on + // another worker: read atomically (the bullet bit itself never changes). + if (native.atomic_load((&sim.flags) as ptr) & dynamics.IS_BULLET) != 0 { return true } fast_body = dynamics.body_at(w, ctx.fast_sim.body_id) if dynamics.should_bodies_collide(w, fast_body, body) == false { return true } if w.has_custom_filter && ((s.flags & dynamics.SHAPE_ENABLE_CUSTOM_FILTERING) != 0 || (fast_shape.flags & dynamics.SHAPE_ENABLE_CUSTOM_FILTERING) != 0) { @@ -298,7 +300,7 @@ continuous_query_visitor(proxy_id: int, user_data: long, context: ptr) -> bool { did_hit = pre_solve(dynamics.make_shape_id(w, s), dynamics.make_shape_id(w, fast_shape), math.add(ctx.base, output.point), output.normal, w.pre_solve_context) } if did_hit { - ctx.fast_sim.flags = ctx.fast_sim.flags | dynamics.HAD_TIME_OF_IMPACT + native.atomic_or((&ctx.fast_sim.flags) as ptr, dynamics.HAD_TIME_OF_IMPACT) ctx.fraction = output.fraction } } @@ -310,7 +312,7 @@ grow_bounds(w: *World, s: *WorldShape, sim: *BodySim, is_bullet: bool) { if math.aabb_contains(s.fat_aabb, s.aabb) { return } margin = math.vec3(s.aabb_margin, s.aabb_margin, s.aabb_margin) s.fat_aabb = AABB { lower: math.sub(s.aabb.lower, margin), upper: math.add(s.aabb.upper, margin) } - sim.flags = sim.flags | dynamics.ENLARGE_BOUNDS + native.atomic_or((&sim.flags) as ptr, dynamics.ENLARGE_BOUNDS) // A bullet's proxies are enlarged later, serially. if is_bullet == false { broad_phase.broad_phase_mark_proxy_moved(&w.broad_phase, s.proxy_key, s.fat_aabb) } } @@ -1038,7 +1040,14 @@ orchestrate(sc: *SolverContext) { // The step's solve (b3Solve). The world's step counter advances; with // no awake body nothing else happens. +// The solve as the world's workers, whatever the calling thread's index. solve(w: *World, ctx: *StepContext) { + caller = dynamics.enter_world(w) + solve_step(w, ctx) + dynamics.leave_world(caller) +} + +solve_step(w: *World, ctx: *StepContext) { w.step_index = w.step_index + (1 as long) awake = dynamics.get_solver_set(w, dynamics.AWAKE_SET) awake_body_count = awake.body_sims.count @@ -1207,11 +1216,11 @@ solve(w: *World, ctx: *StepContext) { i = 1 while i < worker_count { workers[i] = WorkerContext { sc: &sc, worker_index: i } - handles[i] = parallel.enqueue(scheduler, solver_task, workers_block + i * sizeof(WorkerContext), i) as ptr + handles[i] = parallel.enqueue(scheduler, solver_task, workers_block + i * sizeof(WorkerContext), w.worker_base + i) as ptr i = i + 1 } workers[0] = WorkerContext { sc: &sc, worker_index: 0 } - native.set_worker_index(0) + native.set_worker_index(w.worker_base) solver_task(workers_block) i = 1 while i < worker_count { diff --git a/tests/test_determinism.ae b/tests/test_determinism.ae index 0fa9abd..b9984da 100644 --- a/tests/test_determinism.ae +++ b/tests/test_determinism.ae @@ -480,7 +480,86 @@ mesh_drop(workers: int) -> Outcome { return outcome } +// Independent worlds stepped at the same time on different threads: each +// claims worker slots of its own, so they share no scratch, and each ends +// where the same world stepped alone does, bit for bit. Worlds are made +// one at a time (as the reference's are); only the stepping is concurrent. +struct ConcurrentWorld { + world_id: WorldId + bodies: ptr // BodyId[PYRAMID_BOXES] + steps: int +} + +const PYRAMID_BOXES = 55 // a base of ten + +pyramid_world(box: *HullData, slab: *HullData) -> ConcurrentWorld { + wdef = dynamics.default_world_def() + wdef.worker_count = 2 + world_id = dynamics.create_world(&wdef) + gdef = dynamics.default_body_def() + gdef.position = math.vec3(0.0, 0.0 - 0.5, 0.0) + ground = dynamics.create_body(world_id, &gdef) + sdef = dynamics.default_shape_def() + // The hulls are the world's by reference: they outlive it. + dynamics.create_hull_shape(ground, &sdef, slab) + block = calloc(PYRAMID_BOXES, sizeof(BodyId)) + bodies = block as BodyId[] + bdef = dynamics.default_body_def() + bdef.kind = dynamics.BODY_DYNAMIC + n = 0 + row = 0 + while row < 10 { + k = 0 + while k < 10 - row { + bdef.position = math.vec3(1.05 * (k as float) - 0.525 * ((9 - row) as float), 0.5 + 1.0 * (row as float), 0.0) + bodies[n] = dynamics.create_body(world_id, &bdef) + dynamics.create_hull_shape(bodies[n], &sdef, box) + n = n + 1 + k = k + 1 + } + row = row + 1 + } + return ConcurrentWorld { world_id: world_id, bodies: block, steps: 120 } +} + +step_concurrent_world(context: ptr) { + cw = context as *ConcurrentWorld + i = 0 + while i < cw.steps { + physics_world.step(cw.world_id, 1.0 / 60.0, 4) + i = i + 1 + } +} + +test_concurrent_worlds() { + box = hull.make_box_hull(0.5, 0.5, 0.5) + slab = hull.make_box_hull(20.0, 0.5, 20.0) + alone = pyramid_world(box, slab) + step_concurrent_world((&alone) as ptr) + expected = checksum(alone.bodies as BodyId[], PYRAMID_BOXES) + a = pyramid_world(box, slab) + b = pyramid_world(box, slab) + host = parallel.create_scheduler(2) + parallel.begin_round(host) + task = parallel.enqueue(host, step_concurrent_world, (&a) as ptr, 1) + step_concurrent_world((&b) as ptr) + parallel.finish(host, task) + parallel.destroy_scheduler(host) + hash_a = checksum(a.bodies as BodyId[], PYRAMID_BOXES) + hash_b = checksum(b.bodies as BodyId[], PYRAMID_BOXES) + ensure("concurrent worlds: the world on the host thread as alone, bit for bit", bits_of(hash_a) == bits_of(expected)) + ensure("concurrent worlds: the world on the main thread as alone, bit for bit", bits_of(hash_b) == bits_of(expected)) + dynamics.destroy_world(alone.world_id) + dynamics.destroy_world(a.world_id) + dynamics.destroy_world(b.world_id) + free(alone.bodies) + free(a.bodies) + free(b.bodies) + hull.destroy_hull(box) + hull.destroy_hull(slab) +} main() { + test_concurrent_worlds() value = os.getenv("AEPHYSICS_TRACE") g_trace = value != null && string.length(value) > 0 g_trace_run = true