Created at: 2024-12-05
From the docs:
PostgreSQL provides a means for creating locks that have application-defined meanings. These are called advisory locks, because the system does not enforce their use — it is up to the application to use them correctly. Advisory locks can be useful for locking strategies that are an awkward fit for the MVCC model.
For example, a common use of advisory locks is to emulate pessimistic locking strategies typical of so-called “flat file” data management systems. While a flag stored in a table could be used for the same purpose, advisory locks are faster, avoid table bloat, and are automatically cleaned up by the server at the end of the session.
There are two ways to acquire an advisory lock in PostgreSQL: at session level or at transaction level. Unlike standard lock requests, session-level advisory lock requests do not honor transaction semantics: a lock acquired during a transaction that is later rolled back will still be held following the rollback. Transaction-level lock requests, on the other hand, behave more like regular lock requests: they are automatically released at the end of the transaction, and there is no explicit unlock operation. This behavior is often more convenient than the session-level behavior for short-term usage of an advisory lock. Session-level and transaction-level lock requests for the same advisory lock identifier will block each other in the expected way
SELECT pg_advisory_lock(id) FROM foo WHERE id = 12345; -- ok
SELECT pg_advisory_lock(id) FROM foo WHERE id > 12345 LIMIT 100; -- danger!
SELECT pg_advisory_lock(q.id) FROM
(
SELECT id FROM foo WHERE id > 12345 LIMIT 100
) q; -- ok
The second form is dangerous because the LIMIT is not guaranteed to be applied before the locking function is executed. This might cause some locks to be acquired that the application was not expecting, and hence would fail to release (until it ends the session).
-- session-level advisory lock
SELECT pg_advisory_lock(1234);
-- session-level advisory lock, but return immediately if lock is already
-- acquired.
SELECT pg_try_advisory_lock(1234);
-- Release the lock explicitly (optional, as it will automatically release when
-- the session ends)
SELECT pg_advisory_unlock(1234);
-- exclusive transaction-level advisory lock. No unlock option available.
-- Docs say "there is no provision for manual release".
SELECT pg_advisory_xact_lock(1234);
If you need to know if the application code already acquired the advisory lock you are after, you can perform this query:
BEGIN;
SELECT pg_advisory_xact_lock(-8732361555977846743);
-- This is the same as the above in 32bit.
SELECT pg_advisory_xact_lock(1302970409);
-- This is the max 32 bit
SELECT pg_advisory_xact_lock(4294967295);
-- This is the max 32 bit + 1
SELECT pg_advisory_xact_lock(4294967296);
SELECT * FROM pg_locks
WHERE (
locktype = 'advisory'
AND classid = (-8732361555977846743 >> 32)
-- 4294967295 is 0xFFFFFFFF because our id is too large to fit in 32bits
AND objid = (-8732361555977846743 & 4294967295)
AND pid != pg_backend_pid()
);
-- This doesn't work, the transformation needs to be done in the application
-- code
SELECT * FROM pg_locks
WHERE (
locktype = 'advisory'
AND classid = (4294967296 >> 32)
AND objid = (4294967296 & 4294967295)
AND pid != pg_backend_pid()
);
SELECT pg_advisory_xact_lock(1302970409);
ROLLBACK;
In python:
with transaction.atomic():
with connection.cursor() as cursor:
lock_id = 4294967296
high_part = lock_id >> 32
low_part = lock_id & 0xFFFFFFFF
query = (
"""
-- This is the max 32 bit + 1
SELECT pg_advisory_xact_lock(4294967296);
SELECT pg_advisory_xact_lock(4294967295);
"""
)
cursor.execute(query)
lock_query = (
f"SELECT * FROM pg_locks WHERE locktype = 'advisory' "
f"AND classid = {high_part} "
f"AND objid = {low_part} "
f"AND pid = pg_backend_pid()"
)
print(lock_query)
cursor.execute(lock_query)
print(cursor.fetchall())
import time
import psycopg2
# Change these before running locally.
DB_NAME = "advisory_lock"
USER = "marcelo.fernandes"
PASSWORD = ""
HOST = "localhost"
PORT = 5415
def get_cursor_and_connection():
conn = psycopg2.connect(
dbname=DB_NAME,
user=USER,
password=PASSWORD,
host=HOST,
port=PORT,
)
conn.autocommit = True
return conn.cursor(), conn
def maybe_get_advisory_session_lock(cursor):
print("- Acquiring advisory lock...")
cursor.execute("SELECT pg_try_advisory_lock(12345);")
result = cursor.fetchone()[0]
print(f"- result: {result}")
if result is True:
time.sleep(60)
return
assert result is False
def run():
cursor, conn = get_cursor_and_connection()
maybe_get_advisory_session_lock(cursor)
cursor.close()
conn.close()
if __name__ == "__main__":
run()