Last active
July 8, 2026 22:13
-
-
Save birddevelper/21793b07679680123942130d46bc8bd3 to your computer and use it in GitHub Desktop.
double-checked-lock-benchmark.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| """ | |
| Part 1: Basic Double-Checked Locking Benchmark | |
| Comparing: No Lock (Unsafe) vs Always Lock vs Double-Checked Locking | |
| """ | |
| import threading | |
| import time | |
| import matplotlib.pyplot as plt | |
| import numpy as np | |
| # ============================================================================ | |
| # Three Basic Approaches | |
| # ============================================================================ | |
| class NoLockUnsafe: | |
| """β UNSAFE: No synchronization at all""" | |
| _instance = None | |
| def __new__(cls): | |
| if cls._instance is None: | |
| time.sleep(0.001) # Simulate expensive creation | |
| cls._instance = super().__new__(cls) | |
| return cls._instance | |
| class AlwaysLock: | |
| """π SAFE but SLOW: Locks every single time""" | |
| _instance = None | |
| _lock = threading.Lock() | |
| def __new__(cls): | |
| with cls._lock: | |
| if cls._instance is None: | |
| time.sleep(0.001) | |
| cls._instance = super().__new__(cls) | |
| return cls._instance | |
| class DoubleCheckedLocking: | |
| """β‘ SAFE and FAST: Double-checked locking""" | |
| _instance = None | |
| _lock = threading.Lock() | |
| def __new__(cls): | |
| if cls._instance is None: # Quick check (no lock) | |
| with cls._lock: # Acquire lock | |
| if cls._instance is None: # Second check (with lock) | |
| time.sleep(0.001) | |
| cls._instance = super().__new__(cls) | |
| return cls._instance | |
| # ============================================================================ | |
| # Part 1: Benchmark Runner | |
| # ============================================================================ | |
| class Benchmark: | |
| def __init__(self, name, target_class, num_threads=50, num_ops=10000): | |
| self.name = name | |
| self.target_class = target_class | |
| self.num_threads = num_threads | |
| self.num_ops = num_ops | |
| def _reset(self): | |
| if hasattr(self.target_class, '_instance'): | |
| self.target_class._instance = None | |
| def _worker(self): | |
| for _ in range(self.num_ops // self.num_threads): | |
| instance = self.target_class() | |
| assert instance is not None | |
| def run(self): | |
| self._reset() | |
| # Warm-up | |
| for _ in range(100): | |
| self.target_class() | |
| self._reset() | |
| start = time.perf_counter() | |
| threads = [] | |
| for _ in range(self.num_threads): | |
| t = threading.Thread(target=self._worker) | |
| threads.append(t) | |
| t.start() | |
| for t in threads: | |
| t.join() | |
| elapsed = time.perf_counter() - start | |
| return { | |
| 'name': self.name, | |
| 'ops': self.num_ops, | |
| 'time': elapsed, | |
| 'ops_per_sec': self.num_ops / elapsed | |
| } | |
| def run_part1(): | |
| """Run Part 1 benchmarks""" | |
| print("=" * 70) | |
| print("π¬ PART 1: Basic Double-Checked Locking") | |
| print("=" * 70) | |
| print("\nComparing three approaches:") | |
| print(" β No Lock (Unsafe) - Fast but broken") | |
| print(" π Always Lock - Safe but slow") | |
| print(" β‘ Double-Checked - Safe and fast") | |
| print() | |
| patterns = [ | |
| ("β No Lock (Unsafe)", NoLockUnsafe), | |
| ("π Always Lock", AlwaysLock), | |
| ("β‘ Double-Checked", DoubleCheckedLocking), | |
| ] | |
| results = [] | |
| ops_counts = [1000, 5000, 10000, 50000] | |
| for ops in ops_counts: | |
| print(f"\nπ Testing with {ops:,} operations...") | |
| for name, cls in patterns: | |
| bench = Benchmark(name, cls, num_threads=50, num_ops=ops) | |
| result = bench.run() | |
| results.append(result) | |
| print(f" {name}: {result['ops_per_sec']:>10,.0f} ops/sec") | |
| return results | |
| def plot_part1(results): | |
| """Visualize Part 1 results""" | |
| fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5)) | |
| # Color coding | |
| colors = { | |
| 'β No Lock (Unsafe)': 'red', | |
| 'π Always Lock': 'orange', | |
| 'β‘ Double-Checked': 'green' | |
| } | |
| # Throughput plot | |
| for name in set(r['name'] for r in results): | |
| data = [r for r in results if r['name'] == name] | |
| x = [r['ops'] for r in data] | |
| y = [r['ops_per_sec'] for r in data] | |
| ax1.plot(x, y, marker='o', label=name, color=colors.get(name), linewidth=2, markersize=8) | |
| ax1.set_xlabel('Number of Operations', fontsize=12) | |
| ax1.set_ylabel('Operations per Second', fontsize=12) | |
| ax1.set_title('Throughput (Higher is Better)', fontsize=14, fontweight='bold') | |
| ax1.legend(loc='best') | |
| ax1.grid(True, alpha=0.3) | |
| ax1.set_xscale('log') | |
| # Speedup compared to Always Lock | |
| for name in set(r['name'] for r in results): | |
| if name == 'π Always Lock': | |
| continue | |
| data = [r for r in results if r['name'] == name] | |
| baseline = [r for r in results if r['name'] == 'π Always Lock'] | |
| x = [r['ops'] for r in data] | |
| y = [data[i]['ops_per_sec'] / baseline[i]['ops_per_sec'] for i in range(len(data))] | |
| ax2.plot(x, y, marker='s', label=name, color=colors.get(name), linewidth=2, markersize=8) | |
| ax2.axhline(y=1, color='gray', linestyle='--', alpha=0.5) | |
| ax2.set_xlabel('Number of Operations', fontsize=12) | |
| ax2.set_ylabel('Speedup (x times faster)', fontsize=12) | |
| ax2.set_title('Speedup vs Always Lock', fontsize=14, fontweight='bold') | |
| ax2.legend(loc='best') | |
| ax2.grid(True, alpha=0.3) | |
| ax2.set_xscale('log') | |
| plt.suptitle('Part 1: Basic Double-Checked Locking Performance', fontsize=16, fontweight='bold') | |
| plt.tight_layout() | |
| plt.savefig('part1_basic_dcl.png', dpi=300, bbox_inches='tight') | |
| print("\nπ Part 1 results saved as 'part1_basic_dcl.png'") | |
| plt.show() | |
| # ============================================================================ | |
| # Part 2: Metaclass Double-Checked Locking | |
| # ============================================================================ | |
| class SingletonMeta(type): | |
| """Metaclass-based double-checked locking""" | |
| _instances = {} | |
| _lock = threading.Lock() | |
| def __call__(cls, *args, **kwargs): | |
| if cls not in cls._instances: | |
| with cls._lock: | |
| if cls not in cls._instances: | |
| time.sleep(0.001) # Simulate expensive creation | |
| instance = super().__call__(*args, **kwargs) | |
| cls._instances[cls] = instance | |
| return cls._instances[cls] | |
| class MetaSingleton(metaclass=SingletonMeta): | |
| """Singleton using metaclass approach""" | |
| def __init__(self): | |
| self.config = {"db": "postgresql://localhost"} | |
| # Also test the __new__ approach again for fair comparison | |
| class NewDoubleChecked: | |
| """__new__ based double-checked locking (same as above)""" | |
| _instance = None | |
| _lock = threading.Lock() | |
| def __new__(cls): | |
| if cls._instance is None: | |
| with cls._lock: | |
| if cls._instance is None: | |
| time.sleep(0.001) | |
| cls._instance = super().__new__(cls) | |
| return cls._instance | |
| def run_part2(): | |
| """Run Part 2 benchmarks comparing __new__ vs metaclass""" | |
| print("\n" + "=" * 70) | |
| print("π¬ PART 2: __new__ vs Metaclass Double-Checked Locking") | |
| print("=" * 70) | |
| print("\nComparing two double-checked locking implementations:") | |
| print(" π¦ __new__ override - Traditional approach") | |
| print(" 𧬠Metaclass - Alternative approach") | |
| print() | |
| patterns = [ | |
| ("π¦ __new__ Double-Checked", NewDoubleChecked), | |
| ("𧬠Metaclass Double-Checked", MetaSingleton), | |
| ] | |
| results = [] | |
| ops_counts = [1000, 5000, 10000, 50000, 100000] | |
| for ops in ops_counts: | |
| print(f"\nπ Testing with {ops:,} operations...") | |
| for name, cls in patterns: | |
| bench = Benchmark(name, cls, num_threads=50, num_ops=ops) | |
| result = bench.run() | |
| results.append(result) | |
| print(f" {name}: {result['ops_per_sec']:>10,.0f} ops/sec") | |
| return results | |
| def plot_part2(results): | |
| """Visualize Part 2 results""" | |
| fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5)) | |
| colors = { | |
| 'π¦ __new__ Double-Checked': 'blue', | |
| '𧬠Metaclass Double-Checked': 'purple' | |
| } | |
| # Throughput plot | |
| for name in set(r['name'] for r in results): | |
| data = [r for r in results if r['name'] == name] | |
| x = [r['ops'] for r in data] | |
| y = [r['ops_per_sec'] for r in data] | |
| ax1.plot(x, y, marker='o', label=name, color=colors.get(name), linewidth=2, markersize=8) | |
| ax1.set_xlabel('Number of Operations', fontsize=12) | |
| ax1.set_ylabel('Operations per Second', fontsize=12) | |
| ax1.set_title('Throughput (Higher is Better)', fontsize=14, fontweight='bold') | |
| ax1.legend(loc='best') | |
| ax1.grid(True, alpha=0.3) | |
| ax1.set_xscale('log') | |
| # Difference percentage | |
| new_data = [r for r in results if r['name'] == 'π¦ __new__ Double-Checked'] | |
| meta_data = [r for r in results if r['name'] == '𧬠Metaclass Double-Checked'] | |
| x = [r['ops'] for r in new_data] | |
| y = [(meta_data[i]['ops_per_sec'] - new_data[i]['ops_per_sec']) / new_data[i]['ops_per_sec'] * 100 | |
| for i in range(len(new_data))] | |
| ax2.bar(range(len(x)), y, tick_label=x, color='purple', alpha=0.7) | |
| ax2.axhline(y=0, color='gray', linestyle='-', alpha=0.5) | |
| ax2.set_xlabel('Number of Operations', fontsize=12) | |
| ax2.set_ylabel('Metaclass Performance Difference (%)', fontsize=12) | |
| ax2.set_title('Metaclass vs __new__ Performance', fontsize=14, fontweight='bold') | |
| ax2.grid(True, alpha=0.3, axis='y') | |
| # Add value labels on bars | |
| for i, v in enumerate(y): | |
| ax2.text(i, v + (1 if v >= 0 else -3), f'{v:+.1f}%', ha='center', fontsize=10) | |
| plt.suptitle('Part 2: __new__ vs Metaclass Double-Checked Locking', fontsize=16, fontweight='bold') | |
| plt.tight_layout() | |
| plt.savefig('part2_metaclass_comparison.png', dpi=300, bbox_inches='tight') | |
| print("\nπ Part 2 results saved as 'part2_metaclass_comparison.png'") | |
| plt.show() | |
| # ============================================================================ | |
| # Summary | |
| # ============================================================================ | |
| def print_summary(part1_results, part2_results): | |
| """Print a comprehensive summary""" | |
| print("\n" + "=" * 70) | |
| print("π COMPREHENSIVE SUMMARY") | |
| print("=" * 70) | |
| # Part 1 Summary | |
| print("\nπ PART 1: Basic Approaches") | |
| print("-" * 40) | |
| # Get results for 10,000 operations | |
| p1_10k = [r for r in part1_results if r['ops'] == 10000] | |
| if p1_10k: | |
| unsafe = next((r for r in p1_10k if 'No Lock' in r['name']), None) | |
| always = next((r for r in p1_10k if 'Always Lock' in r['name']), None) | |
| dcl = next((r for r in p1_10k if 'Double-Checked' in r['name'] and 'Metaclass' not in r['name']), None) | |
| if unsafe and always and dcl: | |
| print(f" β No Lock (Unsafe): {unsafe['ops_per_sec']:>8,.0f} ops/sec (FASTEST but UNSAFE)") | |
| print(f" π Always Lock: {always['ops_per_sec']:>8,.0f} ops/sec (SAFE but SLOW)") | |
| print(f" β‘ Double-Checked: {dcl['ops_per_sec']:>8,.0f} ops/sec (SAFE and FAST)") | |
| print(f"\n π Speedup: Double-Checked is {dcl['ops_per_sec'] / always['ops_per_sec']:.1f}x faster than Always Lock") | |
| print(f" β οΈ No Lock is {unsafe['ops_per_sec'] / dcl['ops_per_sec']:.1f}x faster, but NOT thread-safe!") | |
| # Part 2 Summary | |
| print("\nπ PART 2: __new__ vs Metaclass") | |
| print("-" * 40) | |
| p2_10k = [r for r in part2_results if r['ops'] == 10000] | |
| if p2_10k: | |
| new = next((r for r in p2_10k if '__new__' in r['name']), None) | |
| meta = next((r for r in p2_10k if 'Metaclass' in r['name']), None) | |
| if new and meta: | |
| diff = (meta['ops_per_sec'] - new['ops_per_sec']) / new['ops_per_sec'] * 100 | |
| print(f" π¦ __new__ Double-Checked: {new['ops_per_sec']:>8,.0f} ops/sec") | |
| print(f" 𧬠Metaclass Double-Checked: {meta['ops_per_sec']:>8,.0f} ops/sec") | |
| print(f"\n π Metaclass is {abs(diff):.1f}% {'faster' if diff > 0 else 'slower'} than __new__ approach") | |
| print("\n" + "=" * 70) | |
| print("β BENCHMARK COMPLETE!") | |
| print(" Check the generated plots for visualizations.") | |
| print("=" * 70) | |
| # ============================================================================ | |
| # Main Runner | |
| # ============================================================================ | |
| def main(): | |
| print("\n" + "=" * 70) | |
| print("π¬ DOUBLE-CHECKED LOCKING BENCHMARK SUITE") | |
| print("=" * 70) | |
| print(f"Python: {__import__('sys').version}") | |
| print(f"Threads: {threading.active_count()} active") | |
| print() | |
| # Run Part 1: Basic comparison | |
| part1_results = run_part1() | |
| plot_part1(part1_results) | |
| # Run Part 2: Metaclass comparison | |
| part2_results = run_part2() | |
| plot_part2(part2_results) | |
| # Print summary | |
| print_summary(part1_results, part2_results) | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment