How to Mock Mutual Exclusion for A/B Experiment Groups in Python

Simulate mutual exclusion for experiment groups using a thread-safe lock, ensuring only one member updates the shared counter at a time.

Medium Python 3.9+ Aug 9, 2026 A/B testing & experimentation 13 views 0 copies

threading mutual-exclusion ab-testing mocking experiment-groups

Python code

34 lines

Python 3.9+

import threading
import time
import random


class CountingGate:
    """A mock mutual exclusion gate using a lock."""
    def __init__(self):
        self.counter = 0
        self.lock = threading.Lock()

    def enter(self, group_id, member_id):
        with self.lock:
            current = self.counter
            time.sleep(random.uniform(0, 0.01))
            self.counter = current + 1
            print(f"Group {group_id}, member {member_id}: count={self.counter}")


def experiment_group(gate, group_id, members=5):
    for member in range(members):
        gate.enter(group_id, member)


if __name__ == "__main__":
    gate = CountingGate()
    threads = []
    for group in range(5):
        t = threading.Thread(target=experiment_group, args=(gate, group))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print("Final count (expected 25):", gate.counter)

Output

stdout

Group 0, member 0: count=1
Group 1, member 0: count=2
Group 2, member 0: count=3
Group 3, member 0: count=4
Group 4, member 0: count=5
Group 0, member 1: count=6
Group 1, member 1: count=7
Group 2, member 1: count=8
Group 3, member 1: count=9
Group 4, member 1: count=10
Group 0, member 2: count=11
Group 1, member 2: count=12
Group 2, member 2: count=13
Group 3, member 2: count=14
Group 4, member 2: count=15
Group 0, member 3: count=16
Group 1, member 3: count=17
Group 2, member 3: count=18
Group 3, member 3: count=19
Group 4, member 3: count=20
Group 0, member 4: count=21
Group 1, member 4: count=22
Group 2, member 4: count=23
Group 3, member 4: count=24
Group 4, member 4: count=25
Final count (expected 25): 25

How it works

The CountingGate uses a threading.Lock to make the increment operation atomic. Inside enter, the with self.lock: block ensures that only one thread can read, modify, and write the counter at a time, preventing race conditions. The time.sleep simulates variable processing time to highlight the need for mutual exclusion. Each group spawns multiple threads for its members, and the main thread joins them to guarantee completion before printing the final count. This pattern mimics how experiment groups are isolated in A/B testing to avoid interference.

Common mistakes

Forgetting to acquire the lock before modifying the shared counter, leading to race conditions.
Not joining threads, so the final count prints before all increments complete.
Using a global variable without a lock, causing inconsistent counts under concurrency.

Variations

Use `threading.RLock` for reentrant locks if the same thread may acquire the lock multiple times.
Replace the lock with `queue.Queue` to serialize access through a single worker thread.

Real-world use cases

Simulating hash-based bucketing to ensure users are consistently assigned to one experiment group.
Testing that simultaneous user requests to an A/B assignment service do not cause conflicting group assignments.
Validating that invariant checks (like user count per group) hold under concurrent load in a mock environment.

How to Mock Mutual Exclusion for A/B Experiment Groups in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from A/B testing & experimentation

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from A/B testing & experimentation

Keep learning

Tutorials

Quizzes