refactoring, relax locks in refresh_pathman_relation_info() and find_inheritance_children_array()

funbringer · funbringer · commit f775ebb84e04 · 2016-10-21T20:37:48.000+03:00
diff --git a/src/init.c b/src/init.c
@@ -466,8 +466,12 @@ fill_prel_with_partitions(const Oid *partitions,
  *
  * borrowed from pg_inherits.c
  */
-Oid *
-find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size)
+find_children_status
+find_inheritance_children_array(Oid parentrelId,
+								LOCKMODE lockmode,
+								bool nowait,
+								uint32 *children_size,	/* ret value #1 */
+								Oid **children)			/* ret value #2 */
 {
 	Relation	relation;
 	SysScanDesc scan;
@@ -485,8 +489,12 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 	 */
 	if (!has_subclass(parentrelId))
 	{
-		*size = 0;
-		return NULL;
+		/* Init return values */
+		*children_size = 0;
+		children = NULL;
+
+		/* Ok, could not find any children */
+		return FCS_NO_CHILDREN;
 	}
 
 	/*
@@ -540,7 +548,25 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 		if (lockmode != NoLock)
 		{
 			/* Get the lock to synchronize against concurrent drop */
-			LockRelationOid(inhrelid, lockmode);
+			if (nowait)
+			{
+				if (!ConditionalLockRelationOid(inhrelid, lockmode))
+				{
+					uint32 j;
+
+					/* Unlock all previously locked children */
+					for (j = 0; j < i; j++)
+						UnlockRelationOid(oidarr[j], lockmode);
+
+					/* Init return values */
+					*children_size = numoids;
+					*children = oidarr;
+
+					/* We couldn't lock this child, retreat! */
+					return FCS_COULD_NOT_LOCK;
+				}
+			}
+			else LockRelationOid(inhrelid, lockmode);
 
 			/*
 			 * Now that we have the lock, double-check to see if the relation
@@ -557,8 +583,12 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 		}
 	}
 
-	*size = numoids;
-	return oidarr;
+	/* Init return values */
+	*children_size = numoids;
+	*children = oidarr;
+
+	/* Ok, we have children */
+	return FCS_FOUND;
 }
 
 /*
diff --git a/src/init.h b/src/init.h
@@ -109,9 +109,19 @@ void fill_prel_with_partitions(const Oid *partitions,
 							   const uint32 parts_count,
 							   PartRelationInfo *prel);
 
-Oid *find_inheritance_children_array(Oid parentrelId,
-									 LOCKMODE lockmode,
-									 uint32 *size);
+/* Result of find_inheritance_children_array() */
+typedef enum
+{
+	FCS_NO_CHILDREN = 0,	/* could not find any children (GOOD) */
+	FCS_COULD_NOT_LOCK,		/* could not lock one of the children */
+	FCS_FOUND				/* found some children (GOOD) */
+} find_children_status;
+
+find_children_status find_inheritance_children_array(Oid parentrelId,
+													 LOCKMODE lockmode,
+													 bool nowait,
+													 uint32 *children_size,
+													 Oid **children);
 
 char *build_check_constraint_name_internal(Oid relid,
 										   AttrNumber attno);
diff --git a/src/relation_info.c b/src/relation_info.c
@@ -75,16 +75,16 @@ refresh_pathman_relation_info(Oid relid,
 	Oid					   *prel_children;
 	uint32					prel_children_count = 0,
 							i;
-	bool					found;
+	bool					found_entry;
 	PartRelationInfo	   *prel;
 	Datum					param_values[Natts_pathman_config_params];
 	bool					param_isnull[Natts_pathman_config_params];
 
 	prel = (PartRelationInfo *) hash_search(partitioned_rels,
 											(const void *) &relid,
-											HASH_ENTER, &found);
+											HASH_ENTER, &found_entry);
 	elog(DEBUG2,
-		 found ?
+		 found_entry ?
 			 "Refreshing record for relation %u in pg_pathman's cache [%u]" :
 			 "Creating new record for relation %u in pg_pathman's cache [%u]",
 		 relid, MyProcPid);
@@ -96,7 +96,7 @@ refresh_pathman_relation_info(Oid relid,
 	prel->cmp_proc = InvalidOid;
 
 	/* Clear outdated resources */
-	if (found && PrelIsValid(prel))
+	if (found_entry && PrelIsValid(prel))
 	{
 		/* Free these arrays iff they're not NULL */
 		FreeChildrenArray(prel);
@@ -136,16 +136,38 @@ refresh_pathman_relation_info(Oid relid,
 	prel->cmp_proc	= typcache->cmp_proc;
 	prel->hash_proc	= typcache->hash_proc;
 
-	LockRelationOid(relid, lockmode);
-	prel_children = find_inheritance_children_array(relid, lockmode,
-													&prel_children_count);
-	UnlockRelationOid(relid, lockmode);
+	/* Try locking parent, exit fast if 'allow_incomplete' */
+	if (allow_incomplete)
+	{
+		if (!ConditionalLockRelationOid(relid, lockmode))
+			return NULL; /* leave an invalid entry */
+	}
+	else LockRelationOid(relid, lockmode);
 
-	/* If there's no children at all, remove this entry */
-	if (prel_children_count == 0)
+	/* Try searching for children (don't wait if we can't lock) */
+	switch (find_inheritance_children_array(relid, lockmode, true,
+											&prel_children_count,
+											&prel_children))
 	{
-		remove_pathman_relation_info(relid);
-		return NULL;
+		/* If there's no children at all, remove this entry */
+		case FCS_NO_CHILDREN:
+			UnlockRelationOid(relid, lockmode);
+			remove_pathman_relation_info(relid);
+			return NULL; /* exit */
+
+		/* If can't lock children, leave an invalid entry */
+		case FCS_COULD_NOT_LOCK:
+			UnlockRelationOid(relid, lockmode);
+			return NULL; /* exit */
+
+		/* Found some children, just unlock parent */
+		case FCS_FOUND:
+			UnlockRelationOid(relid, lockmode);
+			break; /* continue */
+
+		/* Error: unknown result code */
+		default:
+			elog(ERROR, "error in " CppAsString(find_inheritance_children_array));
 	}
 
 	/*
@@ -156,7 +178,7 @@ refresh_pathman_relation_info(Oid relid,
 	 */
 	fill_prel_with_partitions(prel_children, prel_children_count, prel);
 
-	/* Add "partition+parent" tuple to cache */
+	/* Add "partition+parent" pair to cache */
 	for (i = 0; i < prel_children_count; i++)
 		cache_parent_of_partition(prel_children[i], relid);
 
