1import threading
2
3counter = 0
4lock = threading.Lock()
5
6# WITHOUT with statement — error-prone
7def increment_unsafe():
8    global counter
9    lock.acquire()
10    try:
11        counter += 1
12    finally:
13        lock.release()  # Must remember to release!
14
15# WITH statement — clean and safe
16def increment_safe():
17    global counter
18    with lock:  # Acquires lock, releases automatically
19        counter += 1

1import threading
2
3class SafeCounter:
4    def __init__(self):
5        self._count = 0
6        self._lock = threading.Lock()
7
8    def increment(self):
9        with self._lock:
10            self._count += 1
11
12    def decrement(self):
13        with self._lock:
14            self._count -= 1
15
16    @property
17    def value(self):
18        with self._lock:
19            return self._count
20
21counter = SafeCounter()
22
23def worker(n):
24    for _ in range(100000):
25        counter.increment()
26
27threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
28for t in threads:
29    t.start()
30for t in threads:
31    t.join()
32
33print(counter.value)  # 400000 — always correct with locking

1import threading
2
3rlock = threading.RLock()
4
5def outer():
6    with rlock:
7        print("Outer acquired lock")
8        inner()  # Same thread can acquire again
9
10def inner():
11    with rlock:  # Would deadlock with a regular Lock
12        print("Inner acquired lock")
13
14outer()

1import threading
2import time
3
4# Allow at most 3 concurrent connections
5pool = threading.Semaphore(3)
6
7def access_database(thread_id):
8    with pool:  # Blocks if 3 threads already hold the semaphore
9        print(f"Thread {thread_id}: connected")
10        time.sleep(1)  # Simulate database work
11        print(f"Thread {thread_id}: disconnected")
12
13threads = [threading.Thread(target=access_database, args=(i,)) for i in range(10)]
14for t in threads:
15    t.start()
16for t in threads:
17    t.join()
18# Only 3 threads run concurrently at a time

1import threading
2import time
3
4queue = []
5condition = threading.Condition()
6
7def producer():
8    for i in range(5):
9        time.sleep(0.5)
10        with condition:
11            queue.append(i)
12            print(f"Produced: {i}")
13            condition.notify()  # Wake up one waiting consumer
14
15def consumer():
16    while True:
17        with condition:
18            while not queue:
19                condition.wait()  # Release lock and wait for notify
20            item = queue.pop(0)
21            print(f"Consumed: {item}")
22        if item == 4:
23            break
24
25t1 = threading.Thread(target=producer)
26t2 = threading.Thread(target=consumer)
27t2.start()
28t1.start()
29t1.join()
30t2.join()

1import threading
2
3stop_event = threading.Event()
4
5def worker():
6    while not stop_event.is_set():
7        print("Working...")
8        stop_event.wait(timeout=1)  # Sleep 1 second or until set
9    print("Stopped")
10
11t = threading.Thread(target=worker)
12t.start()
13
14import time
15time.sleep(3)
16stop_event.set()  # Signal the thread to stop
17t.join()

1import threading
2import time
3
4class TimedLock:
5    """A lock that logs how long it was held."""
6    def __init__(self, name="lock"):
7        self._lock = threading.Lock()
8        self._name = name
9
10    def __enter__(self):
11        self._lock.acquire()
12        self._start = time.monotonic()
13        return self
14
15    def __exit__(self, exc_type, exc_val, exc_tb):
16        elapsed = time.monotonic() - self._start
17        self._lock.release()
18        if elapsed > 0.1:
19            print(f"Warning: {self._name} held for {elapsed:.3f}s")
20        return False  # Don't suppress exceptions
21
22lock = TimedLock("db_lock")
23with lock:
24    time.sleep(0.2)  # Warning: db_lock held for 0.200s