Thread Safety

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

// NOT THREAD-SAFE: This counter can produce incorrect results under concurrent access
class UnsafeCounter {
    private count: number = 0;
    
    // Problem: read-modify-write is three separate operations
    increment(): void {
        this.count = this.count + 1;  // Thread A reads 5, Thread B reads 5
                                       // Both write 6, losing an increment!
    }
    
    getCount(): number {
        return this.count;
    }
}
 
// THREAD-SAFE (conceptually): Proper synchronization ensures correctness
class ThreadSafeCounter {
    private count: number = 0;
    private lock: Mutex = new Mutex();  // Conceptual mutex
    
    async increment(): Promise<void> {
        await this.lock.acquire();  // Only one thread can hold the lock
        try {
            this.count = this.count + 1;  // Now atomic from callers' perspective
        } finally {
            this.lock.release();
        }
    }
    
    async getCount(): Promise<number> {
        await this.lock.acquire();
        try {
            return this.count;
        } finally {
            this.lock.release();
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

import java.util.Collections;
import java.util.List;
import java.util.ArrayList;
 
public class ConditionalSafetyExample {
    
    // This list is "conditionally thread-safe"
    private List<String> syncList = Collections.synchronizedList(new ArrayList<>());
    
    // SAFE: Individual operations are synchronized
    public void addItem(String item) {
        syncList.add(item);  // Thread-safe - uses internal lock
    }
    
    public int getSize() {
        return syncList.size();  // Thread-safe - uses internal lock
    }
    
    // UNSAFE: Compound operation without synchronization
    public void addIfAbsentBroken(String item) {
        // BUG: Between contains() and add(), another thread could add the same item
        if (!syncList.contains(item)) {
            syncList.add(item);  // Not atomic!
        }
    }
    
    // UNSAFE: Iteration without synchronization
    public void printAllBroken() {
        // BUG: Another thread could modify list during iteration
        for (String item : syncList) {  // ConcurrentModificationException possible!
            System.out.println(item);
        }
    }
    
    // SAFE: Proper synchronization during iteration
    public void printAllCorrect() {
        synchronized (syncList) {  // Must hold lock for entire iteration
            for (String item : syncList) {
                System.out.println(item);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

public class VisibilityProblem {
    
    // Without 'volatile', this change may NEVER be seen by the reader thread!
    private boolean running = true;   // BUG: Missing volatile
    private int value = 0;
    
    // Writer thread
    public void stop() {
        value = 42;      // This write might not be visible either
        running = false; // This might be cached in register, never written to main memory
    }
    
    // Reader thread - may loop FOREVER even after stop() is called
    public void run() {
        while (running) {
            // JVM is free to optimize this to: if (running) { while(true) {} }
            // because it sees 'running' is never modified in this thread's view
        }
        System.out.println("Value: " + value);  // Might print 0!
    }
}
 
// CORRECT: Using volatile for visibility
public class VisibilityFixed {
    
    private volatile boolean running = true;  // volatile ensures visibility
    private volatile int value = 0;           // visibility guaranteed
    
    public void stop() {
        value = 42;       // Happens-before the write to running
        running = false;  // This write is guaranteed to be visible
    }
    
    public void run() {
        while (running) {
            // Loop will terminate when stop() is called
        }
        System.out.println("Value: " + value);  // Guaranteed to print 42
    }
}

Thread A: A1, A2
Thread B: B1, B2
 
Possible interleavings (6 total):
1. A1 → A2 → B1 → B2
2. A1 → B1 → A2 → B2
3. A1 → B1 → B2 → A2
4. B1 → A1 → A2 → B2
5. B1 → A1 → B2 → A2
6. B1 → B2 → A1 → A2
 
For n threads with m operations each: (n * m)! / (m!)^n interleavings
 
Example: 3 threads × 10 operations each = 30!/(10!)³ ≈ 5.5 × 10¹⁸ interleavings
 
This is why testing CANNOT prove thread safety—you can only test a tiny fraction 
of possible executions.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

public class InvariantViolation {
    
    // Invariant: items.length == capacity, 0 <= count <= capacity
    private final Object[] items;
    private final int capacity;
    private int count = 0;
    private int putIndex = 0;
    private int takeIndex = 0;
    
    public InvariantViolation(int capacity) {
        this.capacity = capacity;
        this.items = new Object[capacity];
    }
    
    // BUG: Invariant can be violated between check and modification
    public void put(Object item) throws InterruptedException {
        // WINDOW OF VULNERABILITY BEGINS
        while (count == capacity) {  // <- Thread A checks: count=10, capacity=10
            wait();                   // <- Thread A waits
        }
        // Thread B could decrement count here, but Thread C might also be 
        // waking up and about to put—both think there's space!
        
        items[putIndex] = item;       // <- Both threads might write to same slot!
        putIndex = (putIndex + 1) % capacity;
        count++;                      // <- count could exceed capacity!
        // WINDOW OF VULNERABILITY ENDS
        
        notifyAll();
    }
    
    // CORRECT: Synchronized to protect invariant
    public synchronized void putSafe(Object item) throws InterruptedException {
        while (count == capacity) {
            wait();  // Releases lock while waiting
        }
        // No other thread can execute putSafe/takeSafe while we hold the lock
        items[putIndex] = item;
        putIndex = (putIndex + 1) % capacity;
        count++;
        notifyAll();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

/**
 * A thread-safe counter with atomic increment and decrement operations.
 * 
 * <h2>Thread Safety</h2>
 * This class is <b>unconditionally thread-safe</b>. All individual operations
 * are atomic with respect to concurrent access.
 * 
 * <h3>Atomicity</h3>
 * Each method (increment, decrement, getValue) is atomic. However, compound operations
 * combining multiple methods are NOT atomic without external synchronization.
 * 
 * For example, the following is NOT safe:
 * <pre>
 * // BUG: Not atomic - another thread could modify between calls
 * if (counter.getValue() > 0) {
 *     counter.decrement();
 * }
 * </pre>
 * 
 * <h3>Visibility</h3>
 * All writes are immediately visible to subsequent reads from any thread.
 * 
 * <h3>Ordering</h3>
 * Operations by a single thread appear in program order. Operations across
 * threads are linearizable - they appear to occur in some total order consistent
 * with the real-time ordering of operations.
 * 
 * @ThreadSafe
 */
public class AtomicCounter {
    private final AtomicLong value = new AtomicLong(0);
    
    /** Atomically increments and returns the new value. */
    public long increment() {
        return value.incrementAndGet();
    }
    
    /** Atomically decrements and returns the new value. */
    public long decrement() {
        return value.decrementAndGet();
    }
    
    /** Returns the current value. */
    public long getValue() {
        return value.get();
    }
    
    /**
     * Atomically decrements if value is positive.
     * <p>
     * Use this instead of check-then-decrement for safe conditional decrement.
     * 
     * @return true if decrement was performed, false if value was already 0
     */
    public boolean decrementIfPositive() {
        while (true) {
            long current = value.get();
            if (current <= 0) {
                return false;
            }
            if (value.compareAndSet(current, current - 1)) {
                return true;
            }
            // CAS failed due to concurrent modification - retry
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import java.util.concurrent.ConcurrentHashMap;
 
public class CompositionFallacy {
    
    // ConcurrentHashMap is a thread-safe data structure
    private ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();
    
    // BUG: Not thread-safe despite using ConcurrentHashMap!
    public void incrementBroken(String key) {
        Integer current = counts.get(key);      // Thread A gets 5
                                                 // Thread B gets 5
        if (current == null) {
            counts.put(key, 1);                  // Both might put 1
        } else {
            counts.put(key, current + 1);        // Both put 6, should be 7!
        }
    }
    
    // CORRECT: Use atomic operation
    public void incrementCorrect(String key) {
        counts.merge(key, 1, Integer::sum);  // Atomic compute operation
        
        // Or use compute:
        // counts.compute(key, (k, v) -> v == null ? 1 : v + 1);
    }
    
    // ALSO CORRECT: Compare-and-swap loop
    public void incrementCAS(String key) {
        while (true) {
            Integer current = counts.get(key);
            Integer next = (current == null) ? 1 : current + 1;
            
            if (current == null) {
                if (counts.putIfAbsent(key, next) == null) return;
            } else {
                if (counts.replace(key, current, next)) return;
            }
            // CAS failed - retry
        }
    }
}