Independent worlds stepped at once: worker slots of their own per world

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

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)

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 }

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 {

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