From f23bdda3240d3aa511c843a4f91870d17b53b970 Mon Sep 17 00:00:00 2001 From: Tom Lane Date: Tue, 12 May 2009 16:43:32 +0000 Subject: [PATCH] Fix LOCK TABLE to eliminate the race condition that could make it give weird errors when tables are concurrently dropped. To do this we must take lock on each relation before we check its privileges. The old code was trying to do that the other way around, which is a bit pointless when there are lots of other commands that lock relations before checking privileges. I did keep it checking each relation's privilege before locking the next relation, which is a detail that ALTER TABLE isn't too picky about. --- src/backend/access/heap/heapam.c | 52 +------------- src/backend/commands/lockcmds.c | 151 +++++++++++++++++++++++++++++---------- src/include/access/heapam.h | 3 +- 3 files changed, 114 insertions(+), 92 deletions(-) diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 43af0c7e04..272fb32c58 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -8,7 +8,7 @@ * * * IDENTIFICATION - * $PostgreSQL: pgsql/src/backend/access/heap/heapam.c,v 1.274 2009/01/20 18:59:36 heikki Exp $ + * $PostgreSQL: pgsql/src/backend/access/heap/heapam.c,v 1.275 2009/05/12 16:43:32 tgl Exp $ * * * INTERFACE ROUTINES @@ -948,56 +948,6 @@ try_relation_open(Oid relationId, LOCKMODE lockmode) } /* ---------------- - * relation_open_nowait - open but don't wait for lock - * - * Same as relation_open, except throw an error instead of waiting - * when the requested lock is not immediately obtainable. - * ---------------- - */ -Relation -relation_open_nowait(Oid relationId, LOCKMODE lockmode) -{ - Relation r; - - Assert(lockmode >= NoLock && lockmode < MAX_LOCKMODES); - - /* Get the lock before trying to open the relcache entry */ - if (lockmode != NoLock) - { - if (!ConditionalLockRelationOid(relationId, lockmode)) - { - /* try to throw error by name; relation could be deleted... */ - char *relname = get_rel_name(relationId); - - if (relname) - ereport(ERROR, - (errcode(ERRCODE_LOCK_NOT_AVAILABLE), - errmsg("could not obtain lock on relation \"%s\"", - relname))); - else - ereport(ERROR, - (errcode(ERRCODE_LOCK_NOT_AVAILABLE), - errmsg("could not obtain lock on relation with OID %u", - relationId))); - } - } - - /* The relcache does all the real work... */ - r = RelationIdGetRelation(relationId); - - if (!RelationIsValid(r)) - elog(ERROR, "could not open relation with OID %u", relationId); - - /* Make note that we've accessed a temporary relation */ - if (r->rd_istemp) - MyXactAccessedTempRel = true; - - pgstat_initstats(r); - - return r; -} - -/* ---------------- * relation_openrv - open any relation specified by a RangeVar * * Same as relation_open, but the relation is specified by a RangeVar. diff --git a/src/backend/commands/lockcmds.c b/src/backend/commands/lockcmds.c index 84b152a200..1e9b5dd961 100644 --- a/src/backend/commands/lockcmds.c +++ b/src/backend/commands/lockcmds.c @@ -1,14 +1,14 @@ /*------------------------------------------------------------------------- * * lockcmds.c - * Lock command support code + * LOCK command support code * * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * * IDENTIFICATION - * $PostgreSQL: pgsql/src/backend/commands/lockcmds.c,v 1.23 2009/05/12 03:11:01 tgl Exp $ + * $PostgreSQL: pgsql/src/backend/commands/lockcmds.c,v 1.24 2009/05/12 16:43:32 tgl Exp $ * *------------------------------------------------------------------------- */ @@ -20,9 +20,12 @@ #include "commands/lockcmds.h" #include "miscadmin.h" #include "parser/parse_clause.h" +#include "storage/lmgr.h" #include "utils/acl.h" #include "utils/lsyscache.h" -#include "utils/rel.h" + +static void LockTableRecurse(Oid reloid, RangeVar *rv, + LOCKMODE lockmode, bool nowait, bool recurse); /* @@ -34,57 +37,127 @@ LockTableCommand(LockStmt *lockstmt) ListCell *p; /* - * Iterate over the list and open, lock, and close the relations one at a - * time + * Iterate over the list and process the named relations one at a time */ - foreach(p, lockstmt->relations) { - RangeVar *relation = lfirst(p); - Oid reloid; + RangeVar *relation = (RangeVar *) lfirst(p); bool recurse = interpretInhOption(relation->inhOpt); - List *children_and_self; - ListCell *child; + Oid reloid; reloid = RangeVarGetRelid(relation, false); - /* XXX NoLock here is not really a good idea */ - if (recurse) - children_and_self = find_all_inheritors(reloid, NoLock); - else - children_and_self = list_make1_oid(reloid); + LockTableRecurse(reloid, relation, + lockstmt->mode, lockstmt->nowait, recurse); + } +} - foreach(child, children_and_self) +/* + * Apply LOCK TABLE recursively over an inheritance tree + * + * At top level, "rv" is the original command argument; we use it to throw + * an appropriate error message if the relation isn't there. Below top level, + * "rv" is NULL and we should just silently ignore any dropped child rel. + */ +static void +LockTableRecurse(Oid reloid, RangeVar *rv, + LOCKMODE lockmode, bool nowait, bool recurse) +{ + Relation rel; + AclResult aclresult; + + /* + * Acquire the lock. We must do this first to protect against + * concurrent drops. Note that a lock against an already-dropped + * relation's OID won't fail. + */ + if (nowait) + { + if (!ConditionalLockRelationOid(reloid, lockmode)) { - Oid childreloid = lfirst_oid(child); - Relation rel; - AclResult aclresult; - - /* We don't want to open the relation until we've checked privilege. */ - if (lockstmt->mode == AccessShareLock) - aclresult = pg_class_aclcheck(childreloid, GetUserId(), - ACL_SELECT); + /* try to throw error by name; relation could be deleted... */ + char *relname = rv ? rv->relname : get_rel_name(reloid); + + if (relname) + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on relation \"%s\"", + relname))); else - aclresult = pg_class_aclcheck(childreloid, GetUserId(), - ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE); + ereport(ERROR, + (errcode(ERRCODE_LOCK_NOT_AVAILABLE), + errmsg("could not obtain lock on relation with OID %u", + reloid))); + } + } + else + LockRelationOid(reloid, lockmode); - if (aclresult != ACLCHECK_OK) - aclcheck_error(aclresult, ACL_KIND_CLASS, - get_rel_name(childreloid)); + /* + * Now that we have the lock, check to see if the relation really exists + * or not. + */ + rel = try_relation_open(reloid, NoLock); - if (lockstmt->nowait) - rel = relation_open_nowait(childreloid, lockstmt->mode); - else - rel = relation_open(childreloid, lockstmt->mode); + if (!rel) + { + /* Release useless lock */ + UnlockRelationOid(reloid, lockmode); - /* Currently, we only allow plain tables to be locked */ - if (rel->rd_rel->relkind != RELKIND_RELATION) + /* At top level, throw error; otherwise, ignore this child rel */ + if (rv) + { + if (rv->schemaname) ereport(ERROR, - (errcode(ERRCODE_WRONG_OBJECT_TYPE), - errmsg("\"%s\" is not a table", - get_rel_name(childreloid)))); + (errcode(ERRCODE_UNDEFINED_TABLE), + errmsg("relation \"%s.%s\" does not exist", + rv->schemaname, rv->relname))); + else + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_TABLE), + errmsg("relation \"%s\" does not exist", + rv->relname))); + } + + return; + } - relation_close(rel, NoLock); /* close rel, keep lock */ + /* Verify adequate privilege */ + if (lockmode == AccessShareLock) + aclresult = pg_class_aclcheck(reloid, GetUserId(), + ACL_SELECT); + else + aclresult = pg_class_aclcheck(reloid, GetUserId(), + ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE); + if (aclresult != ACLCHECK_OK) + aclcheck_error(aclresult, ACL_KIND_CLASS, + RelationGetRelationName(rel)); + + /* Currently, we only allow plain tables to be locked */ + if (rel->rd_rel->relkind != RELKIND_RELATION) + ereport(ERROR, + (errcode(ERRCODE_WRONG_OBJECT_TYPE), + errmsg("\"%s\" is not a table", + RelationGetRelationName(rel)))); + + /* + * If requested, recurse to children. We use find_inheritance_children + * not find_all_inheritors to avoid taking locks far in advance of + * checking privileges. This means we'll visit multiply-inheriting + * children more than once, but that's no problem. + */ + if (recurse) + { + List *children = find_inheritance_children(reloid, NoLock); + ListCell *lc; + + foreach(lc, children) + { + Oid childreloid = lfirst_oid(lc); + + LockTableRecurse(childreloid, NULL, lockmode, nowait, recurse); } } + + relation_close(rel, NoLock); /* close rel, keep lock */ } diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index d773652d0d..ec49192cac 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -7,7 +7,7 @@ * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * - * $PostgreSQL: pgsql/src/include/access/heapam.h,v 1.141 2009/01/01 17:23:56 momjian Exp $ + * $PostgreSQL: pgsql/src/include/access/heapam.h,v 1.142 2009/05/12 16:43:32 tgl Exp $ * *------------------------------------------------------------------------- */ @@ -49,7 +49,6 @@ typedef enum /* in heap/heapam.c */ extern Relation relation_open(Oid relationId, LOCKMODE lockmode); extern Relation try_relation_open(Oid relationId, LOCKMODE lockmode); -extern Relation relation_open_nowait(Oid relationId, LOCKMODE lockmode); extern Relation relation_openrv(const RangeVar *relation, LOCKMODE lockmode); extern Relation try_relation_openrv(const RangeVar *relation, LOCKMODE lockmode); extern void relation_close(Relation relation, LOCKMODE lockmode); -- 2.11.0