import errno
import fcntl
import os
import time
class AlreadyLocked(Exception):
pass
class FileLock:
def __init__(self, path, timeout=0, poll_interval=0.1):
self.path = path
self.timeout = timeout
self.poll_interval = poll_interval
self._fd = None
def acquire(self):
self._fd = os.open(self.path, os.O_RDWR | os.O_CREAT, 0o644)
deadline = time.monotonic() + self.timeout
while True:
try:
fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except OSError as exc:
if exc.errno not in (errno.EAGAIN, errno.EACCES):
os.close(self._fd)
self._fd = None
raise
if self.timeout <= 0 or time.monotonic() >= deadline:
os.close(self._fd)
self._fd = None
raise AlreadyLocked(self.path)
time.sleep(self.poll_interval)
os.ftruncate(self._fd, 0)
os.write(self._fd, "{} {}\n".format(os.getpid(), int(time.time())).encode())
return self
def release(self):
if self._fd is None:
return
try:
fcntl.flock(self._fd, fcntl.LOCK_UN)
finally:
os.close(self._fd)
self._fd = None
def __enter__(self):
return self.acquire()
def __exit__(self, exc_type, exc, tb):
self.release()
return False
import functools
from filelock import FileLock, AlreadyLocked
_BUSY = object()
def single_instance(lock_path, timeout=0, on_busy="raise"):
if on_busy not in ("raise", "skip"):
raise ValueError("on_busy must be 'raise' or 'skip'")
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
lock = FileLock(lock_path, timeout=timeout)
try:
lock.acquire()
except AlreadyLocked:
if on_busy == "raise":
raise
return _BUSY
try:
return func(*args, **kwargs)
finally:
lock.release()
wrapper.busy_sentinel = _BUSY
return wrapper
return decorator
import logging
import sys
import time
from single_instance import single_instance
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
log = logging.getLogger("report_job")
LOCK_PATH = "/tmp/nightly_report.lock"
@single_instance(LOCK_PATH, on_busy="skip")
def generate_report():
log.info("starting report generation")
time.sleep(30) # stand-in for heavy aggregation work
log.info("report written")
return "ok"
def main():
result = generate_report()
if result is generate_report.busy_sentinel:
log.info("previous run still active, skipping this tick")
return 0
return 0
if __name__ == "__main__":
sys.exit(main())
This snippet shows how to stop a script from running twice at the same time using a POSIX file lock, a common need for cron jobs and periodic workers that must not overlap. The core idea is that flock(2) gives the kernel ownership of the lock: as long as the process holds the file descriptor open, the advisory lock is held, and when the process dies for any reason the OS releases it automatically. That auto-release is the key advantage over a plain "lock file exists" check, which leaves stale locks behind after a crash.
In filelock.py, FileLock is a context manager that opens a lock path with os.open and requests fcntl.flock with LOCK_EX | LOCK_NB. The non-blocking flag means a second instance does not wait — it immediately gets an OSError with EAGAIN/EACCES, which is translated into a custom AlreadyLocked exception so callers can distinguish contention from real I/O errors. The optional timeout path uses LOCK_EX in a short poll loop so callers can choose to wait a bounded time instead of failing fast. On __exit__ the code calls flock(..., LOCK_UN) and closes the descriptor; it deliberately does not delete the lock file, because unlinking a file another process may already have open reintroduces the classic lock-file race.
The writes into the file (PID and timestamp) are purely diagnostic — the lock itself is enforced by flock, not by the file contents. This matters because advisory locks only work between processes that both call flock; unrelated writers are not blocked.
single_instance.py wraps the context manager in a decorator so a whole function can be guarded declaratively. @single_instance("/tmp/report.lock") acquires the lock around the call and, on contention, either raises or returns a sentinel depending on on_busy. The job.py tab is a runnable example: a cron entry invokes it every minute, and if a previous run is still churning the new invocation exits cleanly with a log line rather than doubling the work.
A caveat worth noting: flock locks are tied to open file descriptions, so behavior across fork and NFS can differ, and for cross-host coordination a database or Redis lock is more appropriate. For single-host scheduling, this approach is simple and crash-safe.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
use clap::Parser;
#[derive(Parser, Debug)]
#[command(author, version, about)]
struct Args {
#[arg(short, long)]clap for CLI argument parsing with derive macros
Share this code
Here's the card — post it anywhere.
