MCPcopy Create free account
hub / github.com/FastLED/FastLED / check_lock

Function check_lock

ci/util/lock_admin.py:143–189  ·  view source on GitHub ↗

Check status of a specific lock.

(lock_name: str)

Source from the content-addressed store, hash-verified

141
142
143def check_lock(lock_name: str) -> int:
144 """Check status of a specific lock."""
145 print(f"\n🔍 LOCK INFORMATION: {lock_name}")
146 print("=" * 80)
147
148 db = get_lock_database()
149 holders = db.get_lock_info(lock_name)
150
151 if not holders:
152 print(f"❌ No lock found with name: {lock_name}")
153 return 1
154
155 for holder in holders:
156 pid = holder["owner_pid"]
157 operation = holder["operation"]
158 hostname = holder["hostname"]
159 mode = holder["lock_mode"]
160 acquired_at = holder["acquired_at"]
161
162 print(f"PID: {pid}")
163 print(f"Operation: {operation}")
164 print(f"Mode: {mode}")
165 print(f"Acquired at: {acquired_at}")
166 print(f"Hostname: {hostname}")
167
168 if is_process_alive(pid):
169 print(f"Process status: 🟢 ALIVE (PID {pid} is running)")
170 print("Lock status: ACTIVE")
171 else:
172 print(f"Process status: 🔴 DEAD (PID {pid} not found)")
173 print("Lock status: STALE")
174
175 print("=" * 80)
176
177 # Offer to break stale lock
178 stale_holders = [h for h in holders if not is_process_alive(h["owner_pid"])]
179 if stale_holders:
180 response = input("\nStale holder(s) found. Break them? (yes/no): ")
181 if response.lower() == "yes":
182 if db.break_stale_lock(lock_name):
183 print("✅ Stale lock(s) broken successfully")
184 return 0
185 else:
186 print("❌ Failed to break stale lock(s)")
187 return 1
188
189 return 0
190
191
192def main() -> int:

Callers 1

mainFunction · 0.85

Calls 5

get_lock_databaseFunction · 0.90
is_process_aliveFunction · 0.90
get_lock_infoMethod · 0.80
break_stale_lockMethod · 0.80
printFunction · 0.50

Tested by

no test coverage detected