Skip to content

Commit ce91caf

Browse files
sl4m3sl4m3
committed
perf: fix severe performance regression in concurrent file locking
- Stopped deleting the fcntl lock file in release(), which caused lock starvation and destroyed the OS-level wait queue. - Reverted to standard fcntl.flock behavior without O_EXCL for systems that support it. - Reduced test suite time from >12 minutes back to ~2 minutes.
1 parent f31021c commit ce91caf

1 file changed

Lines changed: 87 additions & 147 deletions

File tree

‎src/ledgermind/core/stores/semantic_store/transactions.py‎

Lines changed: 87 additions & 147 deletions
Original file line numberDiff line numberDiff line change
@@ -29,184 +29,124 @@ def __init__(self, lock_path: str, timeout: int = 60):
2929

3030
def acquire(self, exclusive: bool = True, timeout: Optional[int] = None):
3131
"""
32-
Acquires lock with atomic operations to prevent TOCTOU.
33-
34-
Uses POSIX fcntl on Unix/Android with O_EXCL for atomicity.
35-
Falls back to file-based locking on other systems.
36-
37-
Thread-safe and process-safe.
32+
Acquires lock to prevent concurrent modifications.
33+
Uses POSIX fcntl on Unix/Android for highly efficient, OS-level queuing.
34+
Falls back to atomic file creation (O_EXCL) on systems without fcntl.
3835
"""
39-
# ===== LAYER 1: Acquire thread lock first =====
4036
effective_timeout = timeout if timeout is not None else self.timeout
4137

4238
if not self._thread_lock.acquire(timeout=effective_timeout):
4339
raise TimeoutError(f"Could not acquire thread lock for {self.lock_path}")
4440

45-
lock_acquired = False
46-
4741
try:
48-
# ===== LAYER 2: Setup file creation flags =====
49-
flags = os.O_RDWR | os.O_CREAT
50-
51-
# CRITICAL: O_EXCL ensures atomic creation
52-
# If file exists, open() fails immediately (no race window)
53-
if exclusive:
54-
flags |= os.O_EXCL
55-
5642
start_time = time.time()
57-
58-
# ===== LAYER 3: Retry loop for lock contention =====
59-
while True:
60-
try:
61-
# ATOMIC OPERATION: Open file (or fail if exists with O_EXCL)
62-
# This single call creates the file AND acquires FD
63-
# No race window between check and creation
43+
44+
# Fast path: POSIX fcntl (Linux, Android, macOS)
45+
try:
46+
import fcntl
47+
flags = os.O_RDWR | os.O_CREAT
48+
49+
# Open the file (creates it if it doesn't exist).
50+
# We DO NOT use O_EXCL here because fcntl manages the lock state, not the file presence.
51+
if self._fd is None:
6452
self._fd = os.open(self.lock_path, flags)
65-
66-
# File is now open, we own the FD
67-
# Write PID atomically AFTER we have the FD
53+
54+
op = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
55+
56+
# If timeout is 0, we use non-blocking mode immediately
57+
if timeout == 0:
6858
try:
69-
os.ftruncate(self._fd, 0)
70-
os.write(self._fd, str(os.getpid()).encode())
71-
except OSError:
72-
pass
73-
74-
except FileExistsError:
75-
# File already exists -> lock is held by another process
76-
# Check if it's stale lock (process died)
59+
fcntl.flock(self._fd, op | fcntl.LOCK_NB)
60+
return True
61+
except (BlockingIOError, OSError):
62+
return False
63+
64+
# Blocking wait with a timeout mechanism
65+
while True:
7766
try:
78-
with open(self.lock_path, 'r') as f:
79-
pid_str = f.read().strip()
80-
81-
if pid_str:
82-
try:
83-
pid = int(pid_str)
84-
# Check if process is still alive
85-
os.kill(pid, 0) # Signal 0 checks if alive
86-
87-
# Process is alive -> lock is valid
88-
elapsed = time.time() - start_time
89-
90-
if elapsed >= effective_timeout:
91-
if timeout == 0:
92-
return False
93-
raise TimeoutError(
94-
f"Could not acquire lock on {self.lock_path} "
95-
f"after {effective_timeout}s (held by PID {pid})"
96-
)
97-
98-
# Wait before retrying
99-
time.sleep(0.1)
100-
continue
101-
102-
except (ValueError, OSError):
103-
# Invalid PID or process check failed -> Treat as stale lock
104-
pass
105-
106-
# Process is dead or invalid PID -> stale lock -> Remove and retry
67+
# Try to grab the lock non-blockingly first
68+
fcntl.flock(self._fd, op | fcntl.LOCK_NB)
69+
return True
70+
except (BlockingIOError, OSError):
71+
# Lock is held by someone else
72+
if time.time() - start_time >= effective_timeout:
73+
raise TimeoutError(f"Could not acquire fcntl lock on {self.lock_path} after {effective_timeout}s")
74+
# Sleep briefly to avoid pegging CPU, then retry
75+
time.sleep(0.01) # Short sleep for high performance
76+
77+
except ImportError:
78+
# Fallback: Windows or environments without fcntl
79+
# We use a separate .lock file with O_EXCL for atomic creation
80+
semaphore_path = self.lock_path + ".lock"
81+
flags = os.O_RDWR | os.O_CREAT | os.O_EXCL
82+
83+
while True:
84+
try:
85+
# Atomic create
86+
self._fd = os.open(semaphore_path, flags)
10787
try:
108-
os.remove(self.lock_path)
88+
os.write(self._fd, str(os.getpid()).encode())
10989
except OSError:
11090
pass
111-
continue
112-
113-
except (OSError, IOError):
114-
# Error reading/stale check -> just wait and retry
115-
elapsed = time.time() - start_time
116-
117-
if elapsed >= effective_timeout:
118-
if timeout == 0:
119-
return False
120-
raise TimeoutError(
121-
f"Could not acquire lock on {self.lock_path} "
122-
f"after {effective_timeout}s"
123-
)
124-
125-
time.sleep(0.1)
126-
continue
127-
128-
# ===== LAYER 4: Apply fcntl lock (Unix/Linux/Android) =====
129-
try:
130-
import fcntl
131-
132-
op = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
133-
134-
# Try non-blocking first
135-
try:
136-
# FD is already open, now apply fcntl lock
137-
fcntl.flock(self._fd, op | fcntl.LOCK_NB)
138-
lock_acquired = True
13991
return True
140-
141-
except (BlockingIOError, OSError) as e:
142-
import errno
143-
144-
# Check if it's just blocked (expected for contention)
145-
if isinstance(e, BlockingIOError) or e.errno in (errno.EAGAIN, errno.EACCES):
146-
# Lock is held by another process via fcntl
147-
elapsed = time.time() - start_time
148-
149-
if elapsed >= effective_timeout:
150-
if timeout == 0:
151-
try: os.close(self._fd)
92+
93+
except FileExistsError:
94+
# File exists -> check for stale lock
95+
try:
96+
with open(semaphore_path, 'r') as f:
97+
pid_str = f.read().strip()
98+
if pid_str:
99+
pid = int(pid_str)
100+
try:
101+
os.kill(pid, 0)
102+
# Process is alive, we just need to wait
103+
except OSError:
104+
# Process is dead, remove stale lock and retry
105+
try: os.remove(semaphore_path)
152106
except OSError: pass
153-
self._fd = None
154-
return False
155-
156-
raise TimeoutError(
157-
f"Could not acquire fcntl lock on {self.lock_path} "
158-
f"after {effective_timeout}s"
159-
)
160-
161-
# Wait and retry
162-
time.sleep(0.1)
163-
try: os.close(self._fd)
107+
continue
108+
except (ValueError, OSError, IOError):
109+
# Corrupted lock file, try to remove
110+
try: os.remove(semaphore_path)
164111
except OSError: pass
165-
self._fd = None
166112
continue
167-
168-
# Other error
169-
raise
170-
171-
except ImportError:
172-
# fcntl not available -> We rely on O_EXCL
173-
lock_acquired = True
174-
return True
175-
176-
except Exception as e:
177-
logger.error(f"Unexpected error in fcntl lock: {e}")
178-
raise RuntimeError(f"Lock acquisition failed: {e}")
179-
180-
except TimeoutError:
181-
raise
182-
113+
114+
if time.time() - start_time >= effective_timeout:
115+
if timeout == 0: return False
116+
raise TimeoutError(f"Could not acquire file lock on {self.lock_path} after {effective_timeout}s")
117+
118+
time.sleep(0.05)
119+
183120
except Exception:
184121
self._thread_lock.release()
185122
raise
186123

187-
finally:
188-
if not lock_acquired:
189-
self._thread_lock.release()
190-
191124
def release(self):
192125
try:
193-
if self._fd:
126+
if self._fd is not None:
127+
# If we have fcntl available, release the flock
194128
try:
195129
import fcntl
196130
fcntl.flock(self._fd, fcntl.LOCK_UN)
197-
except (ImportError, OSError): pass
131+
except (ImportError, OSError):
132+
pass
198133

134+
# Close the file descriptor
199135
try:
200136
os.close(self._fd)
201-
except OSError: pass
137+
except OSError:
138+
pass
202139
self._fd = None
203-
204-
# Clean up all lock files to prevent O_EXCL false positives for next acquirer
205-
for path in [self.lock_path, self.lock_path + ".lock"]:
206-
if os.path.exists(path):
207-
try:
208-
os.remove(path)
209-
except OSError: pass
140+
141+
# Only remove the fallback .lock file.
142+
# CRITICAL: We DO NOT remove self.lock_path because that destroys the i-node
143+
# and breaks fcntl queuing for other waiting processes.
144+
semaphore_path = self.lock_path + ".lock"
145+
if os.path.exists(semaphore_path):
146+
try:
147+
os.remove(semaphore_path)
148+
except OSError:
149+
pass
210150
finally:
211151
self._thread_lock.release()
212152

0 commit comments

Comments
 (0)