Use annotations to reduce instability of isolation-test results.

We've long contended with isolation test results that aren't entirely
stable.  Some test scripts insert long delays to try to force stable
results, which is not terribly desirable; but other erratic failure
modes remain, causing unrepeatable buildfarm failures.  I've spent a
fair amount of time trying to solve this by improving the server-side
support code, without much success: that way is fundamentally unable
to cope with diffs that stem from chance ordering of arrival of
messages from different server processes.

We can improve matters on the client side, however, by annotating
the test scripts themselves to show the desired reporting order
of events that might occur in different orders.  This patch adds
three types of annotations to deal with (a) test steps that might or
might not complete their waits before the isolationtester can see them
waiting; (b) test steps in different sessions that can legitimately
complete in either order; and (c) NOTIFY messages that might arrive
before or after the completion of a step in another session.  We might
need more annotation types later, but this seems to be enough to deal
with the instabilities we've seen in the buildfarm.  It also lets us
get rid of all the long delays that were previously used, cutting more
than a minute off the runtime of the isolation tests.

Back-patch to all supported branches, because the buildfarm
instabilities affect all the branches, and because it seems desirable
to keep isolationtester's capabilities the same across all branches
to simplify possible future back-patching of tests.

Discussion: https://postgr.es/m/327948.1623725828@sss.pgh.pa.us

1 files changed, 483 insertions, 301 deletions

diff --git a/src/test/isolation/isolationtester.c b/src/test/isolation/isolationtester.c
index 7b916a94230..c46a4c3b7f1 100644
--- a/src/test/isolation/isolationtester.c
+++ b/src/test/isolation/isolationtester.c
@@ -26,32 +26,55 @@
 
 /*
  * conns[0] is the global setup, teardown, and watchdog connection.  Additional
- * connections represent spec-defined sessions.  We also track the backend
- * PID, in numeric and string formats, for each connection.
+ * connections represent spec-defined sessions.
  */
-static PGconn **conns = NULL;
-static int *backend_pids = NULL;
-static const char **backend_pid_strs = NULL;
+typedef struct IsoConnInfo
+{
+	/* The libpq connection object for this connection. */
+	PGconn	   *conn;
+	/* The backend PID, in numeric and string formats. */
+	int			backend_pid;
+	const char *backend_pid_str;
+	/* Name of the associated session. */
+	const char *sessionname;
+	/* Active step on this connection, or NULL if idle. */
+	PermutationStep *active_step;
+	/* Number of NOTICE messages received from connection. */
+	int			total_notices;
+} IsoConnInfo;
+
+static IsoConnInfo *conns = NULL;
 static int	nconns = 0;
 
+/* Flag indicating some new NOTICE has arrived */
+static bool any_new_notice = false;
+
 /* Maximum time to wait before giving up on a step (in usec) */
 static int64 max_step_wait = 300 * USECS_PER_SEC;
 
 
+static void check_testspec(TestSpec *testspec);
 static void run_testspec(TestSpec *testspec);
 static void run_all_permutations(TestSpec *testspec);
 static void run_all_permutations_recurse(TestSpec *testspec, int nsteps,
-							 Step **steps);
+										 PermutationStep **steps);
 static void run_named_permutations(TestSpec *testspec);
-static void run_permutation(TestSpec *testspec, int nsteps, Step **steps);
+static void run_permutation(TestSpec *testspec, int nsteps,
+							PermutationStep **steps);
 
-#define STEP_NONBLOCK	0x1		/* return 0 as soon as cmd waits for a lock */
+/* Flag bits for try_complete_step(s) */
+#define STEP_NONBLOCK	0x1		/* return as soon as cmd waits for a lock */
 #define STEP_RETRY		0x2		/* this is a retry of a previously-waiting cmd */
-static bool try_complete_step(TestSpec *testspec, Step *step, int flags);
+
+static int	try_complete_steps(TestSpec *testspec, PermutationStep **waiting,
+							   int nwaiting, int flags);
+static bool try_complete_step(TestSpec *testspec, PermutationStep *pstep,
+							  int flags);
 
 static int	step_qsort_cmp(const void *a, const void *b);
 static int	step_bsearch_cmp(const void *a, const void *b);
 
+static bool step_has_blocker(PermutationStep *pstep);
 static void printResultSet(PGresult *res);
 static void isotesterNoticeProcessor(void *arg, const char *message);
 static void blackholeNoticeProcessor(void *arg, const char *message);
@@ -63,7 +86,8 @@ exit_nicely(void)
 	int			i;
 
 	for (i = 0; i < nconns; i++)
-		PQfinish(conns[i]);
+		if (conns[i].conn)
+			PQfinish(conns[i].conn);
 	exit(1);
 }
 
@@ -73,14 +97,10 @@ main(int argc, char **argv)
 	const char *conninfo;
 	const char *env_wait;
 	TestSpec   *testspec;
-	int			i,
-				j;
-	int			n;
 	PGresult   *res;
 	PQExpBufferData wait_query;
 	int			opt;
-	int			nallsteps;
-	Step	  **allsteps;
+	int			i;
 
 	while ((opt = getopt(argc, argv, "V")) != -1)
 	{
@@ -125,35 +145,8 @@ main(int argc, char **argv)
 	spec_yyparse();
 	testspec = &parseresult;
 
-	/* Create a lookup table of all steps. */
-	nallsteps = 0;
-	for (i = 0; i < testspec->nsessions; i++)
-		nallsteps += testspec->sessions[i]->nsteps;
-
-	allsteps = malloc(nallsteps * sizeof(Step *));
-
-	n = 0;
-	for (i = 0; i < testspec->nsessions; i++)
-	{
-		for (j = 0; j < testspec->sessions[i]->nsteps; j++)
-			allsteps[n++] = testspec->sessions[i]->steps[j];
-	}
-
-	qsort(allsteps, nallsteps, sizeof(Step *), &step_qsort_cmp);
-	testspec->nallsteps = nallsteps;
-	testspec->allsteps = allsteps;
-
-	/* Verify that all step names are unique */
-	for (i = 1; i < testspec->nallsteps; i++)
-	{
-		if (strcmp(testspec->allsteps[i - 1]->name,
-				   testspec->allsteps[i]->name) == 0)
-		{
-			fprintf(stderr, "duplicate step name: %s\n",
-					testspec->allsteps[i]->name);
-			exit_nicely();
-		}
-	}
+	/* Perform post-parse checking, and fill in linking fields */
+	check_testspec(testspec);
 
 	printf("Parsed test spec with %d sessions\n", testspec->nsessions);
 
@@ -162,17 +155,20 @@ main(int argc, char **argv)
 	 * extra for lock wait detection and global work.
 	 */
 	nconns = 1 + testspec->nsessions;
-	conns = (PGconn **) pg_malloc0(nconns * sizeof(PGconn *));
-	backend_pids = pg_malloc0(nconns * sizeof(*backend_pids));
-	backend_pid_strs = pg_malloc0(nconns * sizeof(*backend_pid_strs));
+	conns = (IsoConnInfo *) pg_malloc0(nconns * sizeof(IsoConnInfo));
 
 	for (i = 0; i < nconns; i++)
 	{
-		conns[i] = PQconnectdb(conninfo);
-		if (PQstatus(conns[i]) != CONNECTION_OK)
+		if (i == 0)
+			conns[i].sessionname = "control connection";
+		else
+			conns[i].sessionname = testspec->sessions[i - 1]->name;
+
+		conns[i].conn = PQconnectdb(conninfo);
+		if (PQstatus(conns[i].conn) != CONNECTION_OK)
 		{
-			fprintf(stderr, "Connection %d to database failed: %s",
-					i, PQerrorMessage(conns[i]));
+			fprintf(stderr, "Connection %d failed: %s",
+					i, PQerrorMessage(conns[i].conn));
 			exit_nicely();
 		}
 
@@ -183,27 +179,17 @@ main(int argc, char **argv)
 		 * messages).
 		 */
 		if (i != 0)
-			PQsetNoticeProcessor(conns[i],
+			PQsetNoticeProcessor(conns[i].conn,
 								 isotesterNoticeProcessor,
-								 (void *) (testspec->sessions[i - 1]->name));
+								 (void *) &conns[i]);
 		else
-			PQsetNoticeProcessor(conns[i],
+			PQsetNoticeProcessor(conns[i].conn,
 								 blackholeNoticeProcessor,
 								 NULL);
 
 		/* Save each connection's backend PID for subsequent use. */
-		backend_pids[i] = PQbackendPID(conns[i]);
-		backend_pid_strs[i] = psprintf("%d", backend_pids[i]);
-	}
-
-	/* Set the session index fields in steps. */
-	for (i = 0; i < testspec->nsessions; i++)
-	{
-		Session    *session = testspec->sessions[i];
-		int			stepindex;
-
-		for (stepindex = 0; stepindex < session->nsteps; stepindex++)
-			session->steps[stepindex]->session = i;
+		conns[i].backend_pid = PQbackendPID(conns[i].conn);
+		conns[i].backend_pid_str = psprintf("%d", conns[i].backend_pid);
 	}
 
 	/*
@@ -218,16 +204,16 @@ main(int argc, char **argv)
 	appendPQExpBufferStr(&wait_query,
 						 "SELECT pg_catalog.pg_blocking_pids($1) && '{");
 	/* The spec syntax requires at least one session; assume that here. */
-	appendPQExpBufferStr(&wait_query, backend_pid_strs[1]);
+	appendPQExpBufferStr(&wait_query, conns[1].backend_pid_str);
 	for (i = 2; i < nconns; i++)
-		appendPQExpBuffer(&wait_query, ",%s", backend_pid_strs[i]);
+		appendPQExpBuffer(&wait_query, ",%s", conns[i].backend_pid_str);
 	appendPQExpBufferStr(&wait_query, "}'::integer[]");
 
-	res = PQprepare(conns[0], PREP_WAITING, wait_query.data, 0, NULL);
+	res = PQprepare(conns[0].conn, PREP_WAITING, wait_query.data, 0, NULL);
 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 	{
 		fprintf(stderr, "prepare of lock wait query failed: %s",
-				PQerrorMessage(conns[0]));
+				PQerrorMessage(conns[0].conn));
 		exit_nicely();
 	}
 	PQclear(res);
@@ -241,10 +227,149 @@ main(int argc, char **argv)
 
 	/* Clean up and exit */
 	for (i = 0; i < nconns; i++)
-		PQfinish(conns[i]);
+		PQfinish(conns[i].conn);
 	return 0;
 }
 
+/*
+ * Validity-check the test spec and fill in cross-links between nodes.
+ */
+static void
+check_testspec(TestSpec *testspec)
+{
+	int			nallsteps;
+	Step	  **allsteps;
+	int			i,
+				j,
+				k;
+
+	/* Create a sorted lookup table of all steps. */
+	nallsteps = 0;
+	for (i = 0; i < testspec->nsessions; i++)
+		nallsteps += testspec->sessions[i]->nsteps;
+
+	allsteps = pg_malloc(nallsteps * sizeof(Step *));
+
+	k = 0;
+	for (i = 0; i < testspec->nsessions; i++)
+	{
+		for (j = 0; j < testspec->sessions[i]->nsteps; j++)
+			allsteps[k++] = testspec->sessions[i]->steps[j];
+	}
+
+	qsort(allsteps, nallsteps, sizeof(Step *), step_qsort_cmp);
+
+	/* Verify that all step names are unique. */
+	for (i = 1; i < nallsteps; i++)
+	{
+		if (strcmp(allsteps[i - 1]->name,
+				   allsteps[i]->name) == 0)
+		{
+			fprintf(stderr, "duplicate step name: %s\n",
+					allsteps[i]->name);
+			exit_nicely();
+		}
+	}
+
+	/* Set the session index fields in steps. */
+	for (i = 0; i < testspec->nsessions; i++)
+	{
+		Session    *session = testspec->sessions[i];
+
+		for (j = 0; j < session->nsteps; j++)
+			session->steps[j]->session = i;
+	}
+
+	/*
+	 * If we have manually-specified permutations, link PermutationSteps to
+	 * Steps, and fill in blocker links.
+	 */
+	for (i = 0; i < testspec->npermutations; i++)
+	{
+		Permutation *p = testspec->permutations[i];
+
+		for (j = 0; j < p->nsteps; j++)
+		{
+			PermutationStep *pstep = p->steps[j];
+			Step	  **this = (Step **) bsearch(pstep->name,
+												 allsteps,
+												 nallsteps,
+												 sizeof(Step *),
+												 step_bsearch_cmp);
+
+			if (this == NULL)
+			{
+				fprintf(stderr, "undefined step \"%s\" specified in permutation\n",
+						pstep->name);
+				exit_nicely();
+			}
+			pstep->step = *this;
+
+			/* Mark the step used, for check below */
+			pstep->step->used = true;
+		}
+
+		/*
+		 * Identify any blocker steps.  We search only the current
+		 * permutation, since steps not used there couldn't be concurrent.
+		 * Note that it's OK to reference later permutation steps, so this
+		 * can't be combined with the previous loop.
+		 */
+		for (j = 0; j < p->nsteps; j++)
+		{
+			PermutationStep *pstep = p->steps[j];
+
+			for (k = 0; k < pstep->nblockers; k++)
+			{
+				PermutationStepBlocker *blocker = pstep->blockers[k];
+				int			n;
+
+				if (blocker->blocktype == PSB_ONCE)
+					continue;	/* nothing to link to */
+
+				blocker->step = NULL;
+				for (n = 0; n < p->nsteps; n++)
+				{
+					PermutationStep *otherp = p->steps[n];
+
+					if (strcmp(otherp->name, blocker->stepname) == 0)
+					{
+						blocker->step = otherp->step;
+						break;
+					}
+				}
+				if (blocker->step == NULL)
+				{
+					fprintf(stderr, "undefined blocking step \"%s\" referenced in permutation step \"%s\"\n",
+							blocker->stepname, pstep->name);
+					exit_nicely();
+				}
+				/* can't block on completion of step of own session */
+				if (blocker->step->session == pstep->step->session)
+				{
+					fprintf(stderr, "permutation step \"%s\" cannot block on its own session\n",
+							pstep->name);
+					exit_nicely();
+				}
+			}
+		}
+	}
+
+	/*
+	 * If we have manually-specified permutations, verify that all steps have
+	 * been used, warning about anything defined but not used.  We can skip
+	 * this when using automatically-generated permutations.
+	 */
+	if (testspec->permutations)
+	{
+		for (i = 0; i < nallsteps; i++)
+		{
+			if (!allsteps[i]->used)
+				fprintf(stderr, "unused step name: %s\n", allsteps[i]->name);
+		}
+	}
+}
+
 static int *piles;
 
 /*
@@ -254,23 +379,10 @@ static int *piles;
 static void
 run_testspec(TestSpec *testspec)
 {
-	int			i;
-
 	if (testspec->permutations)
 		run_named_permutations(testspec);
 	else
 		run_all_permutations(testspec);
-
-	/*
-	 * Verify that all steps have been used, complaining about anything
-	 * defined but not used.
-	 */
-	for (i = 0; i < testspec->nallsteps; i++)
-	{
-		if (!testspec->allsteps[i]->used)
-			fprintf(stderr, "unused step name: %s\n",
-					testspec->allsteps[i]->name);
-	}
 }
 
 /*
@@ -281,14 +393,19 @@ run_all_permutations(TestSpec *testspec)
 {
 	int			nsteps;
 	int			i;
-	Step	  **steps;
+	PermutationStep *steps;
+	PermutationStep **stepptrs;
 
 	/* Count the total number of steps in all sessions */
 	nsteps = 0;
 	for (i = 0; i < testspec->nsessions; i++)
 		nsteps += testspec->sessions[i]->nsteps;
 
-	steps = malloc(sizeof(Step *) * nsteps);
+	/* Create PermutationStep workspace array */
+	steps = (PermutationStep *) pg_malloc0(sizeof(PermutationStep) * nsteps);
+	stepptrs = (PermutationStep **) pg_malloc(sizeof(PermutationStep *) * nsteps);
+	for (i = 0; i < nsteps; i++)
+		stepptrs[i] = steps + i;
 
 	/*
 	 * To generate the permutations, we conceptually put the steps of each
@@ -299,32 +416,41 @@ run_all_permutations(TestSpec *testspec)
 	 * A pile is actually just an integer which tells how many steps we've
 	 * already picked from this pile.
 	 */
-	piles = malloc(sizeof(int) * testspec->nsessions);
+	piles = pg_malloc(sizeof(int) * testspec->nsessions);
 	for (i = 0; i < testspec->nsessions; i++)
 		piles[i] = 0;
 
-	run_all_permutations_recurse(testspec, 0, steps);
+	run_all_permutations_recurse(testspec, 0, stepptrs);
 }
 
 static void
-run_all_permutations_recurse(TestSpec *testspec, int nsteps, Step **steps)
+run_all_permutations_recurse(TestSpec *testspec, int nsteps, PermutationStep **steps)
 {
 	int			i;
-	int			found = 0;
+	bool		found = false;
 
 	for (i = 0; i < testspec->nsessions; i++)
 	{
 		/* If there's any more steps in this pile, pick it and recurse */
 		if (piles[i] < testspec->sessions[i]->nsteps)
 		{
-			steps[nsteps] = testspec->sessions[i]->steps[piles[i]];
+			Step	   *newstep = testspec->sessions[i]->steps[piles[i]];
+
+			/*
+			 * These automatically-generated PermutationSteps never have
+			 * blocker conditions.  So we need only fill these fields, relying
+			 * on run_all_permutations() to have zeroed the rest:
+			 */
+			steps[nsteps]->name = newstep->name;
+			steps[nsteps]->step = newstep;
+
 			piles[i]++;
 
 			run_all_permutations_recurse(testspec, nsteps + 1, steps);
 
 			piles[i]--;
 
-			found = 1;
+			found = true;
 		}
 	}
 
@@ -339,38 +465,13 @@ run_all_permutations_recurse(TestSpec *testspec, int nsteps, Step **steps)
 static void
 run_named_permutations(TestSpec *testspec)
 {
-	int			i,
-				j;
+	int			i;
 
 	for (i = 0; i < testspec->npermutations; i++)
 	{
 		Permutation *p = testspec->permutations[i];
-		Step	  **steps;
-
-		steps = malloc(p->nsteps * sizeof(Step *));
-
-		/* Find all the named steps using the lookup table */
-		for (j = 0; j < p->nsteps; j++)
-		{
-			Step	  **this = (Step **) bsearch(p->stepnames[j],
-												 testspec->allsteps,
-												 testspec->nallsteps,
-												 sizeof(Step *),
-												 &step_bsearch_cmp);
-
-			if (this == NULL)
-			{
-				fprintf(stderr, "undefined step \"%s\" specified in permutation\n",
-						p->stepnames[j]);
-				exit_nicely();
-			}
-			steps[j] = *this;
-		}
 
-		/* And run them */
-		run_permutation(testspec, p->nsteps, steps);
-
-		free(steps);
+		run_permutation(testspec, p->nsteps, p->steps);
 	}
 }
 
@@ -393,101 +494,34 @@ step_bsearch_cmp(const void *a, const void *b)
 }
 
 /*
- * If a step caused an error to be reported, print it out and clear it.
- */
-static void
-report_error_message(Step *step)
-{
-	if (step->errormsg)
-	{
-		fprintf(stdout, "%s\n", step->errormsg);
-		free(step->errormsg);
-		step->errormsg = NULL;
-	}
-}
-
-/*
- * As above, but reports messages possibly emitted by multiple steps.  This is
- * useful when we have a blocked command awakened by another one; we want to
- * report all messages identically, for the case where we don't care which
- * one fails due to a timeout such as deadlock timeout.
- */
-static void
-report_multiple_error_messages(Step *step, int nextra, Step **extrastep)
-{
-	PQExpBufferData buffer;
-	int			n;
-
-	if (nextra == 0)
-	{
-		report_error_message(step);
-		return;
-	}
-
-	initPQExpBuffer(&buffer);
-	appendPQExpBufferStr(&buffer, step->name);
-
-	for (n = 0; n < nextra; ++n)
-		appendPQExpBuffer(&buffer, " %s", extrastep[n]->name);
-
-	if (step->errormsg)
-	{
-		fprintf(stdout, "error in steps %s: %s\n", buffer.data,
-				step->errormsg);
-		free(step->errormsg);
-		step->errormsg = NULL;
-	}
-
-	for (n = 0; n < nextra; ++n)
-	{
-		if (extrastep[n]->errormsg == NULL)
-			continue;
-		fprintf(stdout, "error in steps %s: %s\n",
-				buffer.data, extrastep[n]->errormsg);
-		free(extrastep[n]->errormsg);
-		extrastep[n]->errormsg = NULL;
-	}
-
-	termPQExpBuffer(&buffer);
-}
-
-/*
  * Run one permutation
  */
 static void
-run_permutation(TestSpec *testspec, int nsteps, Step **steps)
+run_permutation(TestSpec *testspec, int nsteps, PermutationStep **steps)
 {
 	PGresult   *res;
 	int			i;
-	int			w;
 	int			nwaiting = 0;
-	int			nerrorstep = 0;
-	Step	  **waiting;
-	Step	  **errorstep;
+	PermutationStep **waiting;
 
-	waiting = malloc(sizeof(Step *) * testspec->nsessions);
-	errorstep = malloc(sizeof(Step *) * testspec->nsessions);
+	waiting = pg_malloc(sizeof(PermutationStep *) * testspec->nsessions);
 
 	printf("\nstarting permutation:");
 	for (i = 0; i < nsteps; i++)
-	{
-		/* Track the permutation as in-use */
-		steps[i]->used = true;
 		printf(" %s", steps[i]->name);
-	}
 	printf("\n");
 
 	/* Perform setup */
 	for (i = 0; i < testspec->nsetupsqls; i++)
 	{
-		res = PQexec(conns[0], testspec->setupsqls[i]);
+		res = PQexec(conns[0].conn, testspec->setupsqls[i]);
 		if (PQresultStatus(res) == PGRES_TUPLES_OK)
 		{
 			printResultSet(res);
 		}
 		else if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
-			fprintf(stderr, "setup failed: %s", PQerrorMessage(conns[0]));
+			fprintf(stderr, "setup failed: %s", PQerrorMessage(conns[0].conn));
 			exit_nicely();
 		}
 		PQclear(res);
@@ -498,7 +532,7 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 	{
 		if (testspec->sessions[i]->setupsql)
 		{
-			res = PQexec(conns[i + 1], testspec->sessions[i]->setupsql);
+			res = PQexec(conns[i + 1].conn, testspec->sessions[i]->setupsql);
 			if (PQresultStatus(res) == PGRES_TUPLES_OK)
 			{
 				printResultSet(res);
@@ -506,8 +540,8 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 			else if (PQresultStatus(res) != PGRES_COMMAND_OK)
 			{
 				fprintf(stderr, "setup of session %s failed: %s",
-						testspec->sessions[i]->name,
-						PQerrorMessage(conns[i + 1]));
+						conns[i + 1].sessionname,
+						PQerrorMessage(conns[i + 1].conn));
 				exit_nicely();
 			}
 			PQclear(res);
@@ -517,73 +551,96 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 	/* Perform steps */
 	for (i = 0; i < nsteps; i++)
 	{
-		Step	   *step = steps[i];
-		PGconn	   *conn = conns[1 + step->session];
-		Step	   *oldstep = NULL;
+		PermutationStep *pstep = steps[i];
+		Step	   *step = pstep->step;
+		IsoConnInfo *iconn = &conns[1 + step->session];
+		PGconn	   *conn = iconn->conn;
 		bool		mustwait;
+		int			j;
 
 		/*
 		 * Check whether the session that needs to perform the next step is
 		 * still blocked on an earlier step.  If so, wait for it to finish.
-		 *
-		 * (In older versions of this tool, we allowed precisely one session
-		 * to be waiting at a time.  If we reached a step that required that
-		 * session to execute the next command, we would declare the whole
-		 * permutation invalid, cancel everything, and move on to the next
-		 * one.  Unfortunately, that made it impossible to test the deadlock
-		 * detector using this framework, unless the number of processes
-		 * involved in the deadlock was precisely two.  We now assume that if
-		 * we reach a step that is still blocked, we need to wait for it to
-		 * unblock itself.)
 		 */
-		for (w = 0; w < nwaiting; ++w)
+		if (iconn->active_step != NULL)
 		{
-			if (step->session == waiting[w]->session)
-			{
-				oldstep = waiting[w];
+			struct timeval start_time;
 
-				/* Wait for previous step on this connection. */
-				try_complete_step(testspec, oldstep, STEP_RETRY);
+			gettimeofday(&start_time, NULL);
 
-				/* Remove that step from the waiting[] array. */
-				if (w + 1 < nwaiting)
-					memmove(&waiting[w], &waiting[w + 1],
-							(nwaiting - (w + 1)) * sizeof(Step *));
-				nwaiting--;
-
-				break;
-			}
-		}
-		if (oldstep != NULL)
-		{
-			/*
-			 * Check for completion of any steps that were previously waiting.
-			 * Remove any that have completed from waiting[], and include them
-			 * in the list for report_multiple_error_messages().
-			 */
-			w = 0;
-			nerrorstep = 0;
-			while (w < nwaiting)
+			while (iconn->active_step != NULL)
 			{
-				if (try_complete_step(testspec, waiting[w],
-									  STEP_NONBLOCK | STEP_RETRY))
-				{
-					/* Still blocked on a lock, leave it alone. */
-					w++;
-				}
-				else
+				PermutationStep *oldstep = iconn->active_step;
+
+				/*
+				 * Wait for oldstep.  But even though we don't use
+				 * STEP_NONBLOCK, it might not complete because of blocker
+				 * conditions.
+				 */
+				if (!try_complete_step(testspec, oldstep, STEP_RETRY))
 				{
-					/* This one finished, too! */
-					errorstep[nerrorstep++] = waiting[w];
+					/* Done, so remove oldstep from the waiting[] array. */
+					int			w;
+
+					for (w = 0; w < nwaiting; w++)
+					{
+						if (oldstep == waiting[w])
+							break;
+					}
+					if (w >= nwaiting)
+						abort();	/* can't happen */
 					if (w + 1 < nwaiting)
 						memmove(&waiting[w], &waiting[w + 1],
-								(nwaiting - (w + 1)) * sizeof(Step *));
+								(nwaiting - (w + 1)) * sizeof(PermutationStep *));
 					nwaiting--;
 				}
-			}
 
-			/* Report all errors together. */
-			report_multiple_error_messages(oldstep, nerrorstep, errorstep);
+				/*
+				 * Check for other steps that have finished.  We should do
+				 * this if oldstep completed, as it might have unblocked
+				 * something.  On the other hand, if oldstep hasn't completed,
+				 * we must poll all the active steps in hopes of unblocking
+				 * oldstep.  So either way, poll them.
+				 */
+				nwaiting = try_complete_steps(testspec, waiting, nwaiting,
+											  STEP_NONBLOCK | STEP_RETRY);
+
+				/*
+				 * If the target session is still busy, apply a timeout to
+				 * keep from hanging indefinitely, which could happen with
+				 * incorrect blocker annotations.  Use the same 2 *
+				 * max_step_wait limit as try_complete_step does for deciding
+				 * to die.  (We don't bother with trying to cancel anything,
+				 * since it's unclear what to cancel in this case.)
+				 */
+				if (iconn->active_step != NULL)
+				{
+					struct timeval current_time;
+					int64		td;
+
+					gettimeofday(&current_time, NULL);
+					td = (int64) current_time.tv_sec - (int64) start_time.tv_sec;
+					td *= USECS_PER_SEC;
+					td += (int64) current_time.tv_usec - (int64) start_time.tv_usec;
+					if (td > 2 * max_step_wait)
+					{
+						fprintf(stderr, "step %s timed out after %d seconds\n",
+								iconn->active_step->name,
+								(int) (td / USECS_PER_SEC));
+						fprintf(stderr, "active steps are:");
+						for (j = 1; j < nconns; j++)
+						{
+							IsoConnInfo *oconn = &conns[j];
+
+							if (oconn->active_step != NULL)
+								fprintf(stderr, " %s",
+										oconn->active_step->name);
+						}
+						fprintf(stderr, "\n");
+						exit_nicely();
+					}
+				}
+			}
 		}
 
 		/* Send the query for this step. */
@@ -594,40 +651,37 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 			exit_nicely();
 		}
 
-		/* Try to complete this step without blocking.  */
-		mustwait = try_complete_step(testspec, step, STEP_NONBLOCK);
+		/* Remember we launched a step. */
+		iconn->active_step = pstep;
 
-		/* Check for completion of any steps that were previously waiting. */
-		w = 0;
-		nerrorstep = 0;
-		while (w < nwaiting)
+		/* Remember target number of NOTICEs for any blocker conditions. */
+		for (j = 0; j < pstep->nblockers; j++)
 		{
-			if (try_complete_step(testspec, waiting[w],
-								  STEP_NONBLOCK | STEP_RETRY))
-				w++;
-			else
-			{
-				errorstep[nerrorstep++] = waiting[w];
-				if (w + 1 < nwaiting)
-					memmove(&waiting[w], &waiting[w + 1],
-							(nwaiting - (w + 1)) * sizeof(Step *));
-				nwaiting--;
-			}
+			PermutationStepBlocker *blocker = pstep->blockers[j];
+
+			if (blocker->blocktype == PSB_NUM_NOTICES)
+				blocker->target_notices = blocker->num_notices +
+					conns[blocker->step->session + 1].total_notices;
 		}
 
-		/* Report any error from this step, and any steps that it unblocked. */
-		report_multiple_error_messages(step, nerrorstep, errorstep);
+		/* Try to complete this step without blocking.  */
+		mustwait = try_complete_step(testspec, pstep, STEP_NONBLOCK);
+
+		/* Check for completion of any steps that were previously waiting. */
+		nwaiting = try_complete_steps(testspec, waiting, nwaiting,
+									  STEP_NONBLOCK | STEP_RETRY);
 
 		/* If this step is waiting, add it to the array of waiters. */
 		if (mustwait)
-			waiting[nwaiting++] = step;
+			waiting[nwaiting++] = pstep;
 	}
 
 	/* Wait for any remaining queries. */
-	for (w = 0; w < nwaiting; ++w)
+	nwaiting = try_complete_steps(testspec, waiting, nwaiting, STEP_RETRY);
+	if (nwaiting != 0)
 	{
-		try_complete_step(testspec, waiting[w], STEP_RETRY);
-		report_error_message(waiting[w]);
+		fprintf(stderr, "failed to complete permutation due to mutually-blocking steps\n");
+		exit_nicely();
 	}
 
 	/* Perform per-session teardown */
@@ -635,7 +689,7 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 	{
 		if (testspec->sessions[i]->teardownsql)
 		{
-			res = PQexec(conns[i + 1], testspec->sessions[i]->teardownsql);
+			res = PQexec(conns[i + 1].conn, testspec->sessions[i]->teardownsql);
 			if (PQresultStatus(res) == PGRES_TUPLES_OK)
 			{
 				printResultSet(res);
@@ -643,8 +697,8 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 			else if (PQresultStatus(res) != PGRES_COMMAND_OK)
 			{
 				fprintf(stderr, "teardown of session %s failed: %s",
-						testspec->sessions[i]->name,
-						PQerrorMessage(conns[i + 1]));
+						conns[i + 1].sessionname,
+						PQerrorMessage(conns[i + 1].conn));
 				/* don't exit on teardown failure */
 			}
 			PQclear(res);
@@ -654,7 +708,7 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 	/* Perform teardown */
 	if (testspec->teardownsql)
 	{
-		res = PQexec(conns[0], testspec->teardownsql);
+		res = PQexec(conns[0].conn, testspec->teardownsql);
 		if (PQresultStatus(res) == PGRES_TUPLES_OK)
 		{
 			printResultSet(res);
@@ -662,36 +716,90 @@ run_permutation(TestSpec *testspec, int nsteps, Step **steps)
 		else if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		{
 			fprintf(stderr, "teardown failed: %s",
-					PQerrorMessage(conns[0]));
+					PQerrorMessage(conns[0].conn));
 			/* don't exit on teardown failure */
 		}
 		PQclear(res);
 	}
 
 	free(waiting);
-	free(errorstep);
+}
+
+/*
+ * Check for completion of any waiting step(s).
+ * Remove completed ones from the waiting[] array,
+ * and return the new value of nwaiting.
+ * See try_complete_step for the meaning of the flags.
+ */
+static int
+try_complete_steps(TestSpec *testspec, PermutationStep **waiting,
+				   int nwaiting, int flags)
+{
+	int			old_nwaiting;
+	bool		have_blocker;
+
+	do
+	{
+		int			w = 0;
+
+		/* Reset latch; we only care about notices received within loop. */
+		any_new_notice = false;
+
+		/* Likewise, these variables reset for each retry. */
+		old_nwaiting = nwaiting;
+		have_blocker = false;
+
+		/* Scan the array, try to complete steps. */
+		while (w < nwaiting)
+		{
+			if (try_complete_step(testspec, waiting[w], flags))
+			{
+				/* Still blocked, leave it alone. */
+				if (waiting[w]->nblockers > 0)
+					have_blocker = true;
+				w++;
+			}
+			else
+			{
+				/* Done, remove it from array. */
+				if (w + 1 < nwaiting)
+					memmove(&waiting[w], &waiting[w + 1],
+							(nwaiting - (w + 1)) * sizeof(PermutationStep *));
+				nwaiting--;
+			}
+		}
+
+		/*
+		 * If any of the still-waiting steps have blocker conditions attached,
+		 * it's possible that one of the steps we examined afterwards has
+		 * released them (either by completing, or by sending a NOTICE).  If
+		 * any step completions or NOTICEs happened, repeat the loop until
+		 * none occurs.  Without this provision, completion timing could vary
+		 * depending on the order in which the steps appear in the array.
+		 */
+	} while (have_blocker && (nwaiting < old_nwaiting || any_new_notice));
+	return nwaiting;
 }
 
 /*
  * Our caller already sent the query associated with this step.  Wait for it
- * to either complete or (if given the STEP_NONBLOCK flag) to block while
- * waiting for a lock.  We assume that any lock wait will persist until we
- * have executed additional steps in the permutation.
+ * to either complete, or hit a blocking condition.
  *
  * When calling this function on behalf of a given step for a second or later
- * time, pass the STEP_RETRY flag.  This only affects the messages printed.
+ * time, pass the STEP_RETRY flag.  Do not pass it on the first call.
  *
- * If the query returns an error, the message is saved in step->errormsg.
- * Caller should call report_error_message shortly after this, to have it
- * printed and cleared.
- *
- * If the STEP_NONBLOCK flag was specified and the query is waiting to acquire
- * a lock, returns true.  Otherwise, returns false.
+ * Returns true if the step was *not* completed, false if it was completed.
+ * Reasons for non-completion are (a) the STEP_NONBLOCK flag was specified
+ * and the query is waiting to acquire a lock, or (b) the step has an
+ * unsatisfied blocker condition.  When STEP_NONBLOCK is given, we assume
+ * that any lock wait will persist until we have executed additional steps.
  */
 static bool
-try_complete_step(TestSpec *testspec, Step *step, int flags)
+try_complete_step(TestSpec *testspec, PermutationStep *pstep, int flags)
 {
-	PGconn	   *conn = conns[1 + step->session];
+	Step	   *step = pstep->step;
+	IsoConnInfo *iconn = &conns[1 + step->session];
+	PGconn	   *conn = iconn->conn;
 	fd_set		read_set;
 	struct timeval start_time;
 	struct timeval timeout;
@@ -701,6 +809,28 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 	PGnotify   *notify;
 	bool		canceled = false;
 
+	/*
+	 * If the step is annotated with (*), then on the first call, force it to
+	 * wait.  This is useful for ensuring consistent output when the step
+	 * might or might not complete so fast that we don't observe it waiting.
+	 */
+	if (!(flags & STEP_RETRY))
+	{
+		int			i;
+
+		for (i = 0; i < pstep->nblockers; i++)
+		{
+			PermutationStepBlocker *blocker = pstep->blockers[i];
+
+			if (blocker->blocktype == PSB_ONCE)
+			{
+				printf("step %s: %s <waiting ...>\n",
+					   step->name, step->sql);
+				return true;
+			}
+		}
+	}
+
 	if (sock < 0)
 	{
 		fprintf(stderr, "invalid socket: %s", PQerrorMessage(conn));
@@ -734,14 +864,14 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 			{
 				bool		waiting;
 
-				res = PQexecPrepared(conns[0], PREP_WAITING, 1,
-									 &backend_pid_strs[step->session + 1],
+				res = PQexecPrepared(conns[0].conn, PREP_WAITING, 1,
+									 &conns[step->session + 1].backend_pid_str,
 									 NULL, NULL, 0);
 				if (PQresultStatus(res) != PGRES_TUPLES_OK ||
 					PQntuples(res) != 1)
 				{
 					fprintf(stderr, "lock wait query failed: %s",
-							PQerrorMessage(conns[0]));
+							PQerrorMessage(conns[0].conn));
 					exit_nicely();
 				}
 				waiting = ((PQgetvalue(res, 0, 0))[0] == 't');
@@ -762,7 +892,7 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 					{
 						fprintf(stderr, "PQconsumeInput failed: %s\n",
 								PQerrorMessage(conn));
-						exit(1);
+						exit_nicely();
 					}
 					if (!PQisBusy(conn))
 						break;
@@ -840,6 +970,19 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 		}
 	}
 
+	/*
+	 * The step is done, but we won't report it as complete so long as there
+	 * are blockers.
+	 */
+	if (step_has_blocker(pstep))
+	{
+		if (!(flags & STEP_RETRY))
+			printf("step %s: %s <waiting ...>\n",
+				   step->name, step->sql);
+		return true;
+	}
+
+	/* Otherwise, go ahead and complete it. */
 	if (flags & STEP_RETRY)
 		printf("step %s: <... completed>\n", step->name);
 	else
@@ -850,16 +993,12 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 		switch (PQresultStatus(res))
 		{
 			case PGRES_COMMAND_OK:
+			case PGRES_EMPTY_QUERY:
 				break;
 			case PGRES_TUPLES_OK:
 				printResultSet(res);
 				break;
 			case PGRES_FATAL_ERROR:
-				if (step->errormsg != NULL)
-				{
-					printf("WARNING: this step had a leftover error message\n");
-					printf("%s\n", step->errormsg);
-				}
 
 				/*
 				 * Detail may contain XID values, so we want to just show
@@ -873,9 +1012,9 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 													PG_DIAG_MESSAGE_PRIMARY);
 
 					if (sev && msg)
-						step->errormsg = psprintf("%s:  %s", sev, msg);
+						printf("%s:  %s\n", sev, msg);
 					else
-						step->errormsg = pg_strdup(PQresultErrorMessage(res));
+						printf("%s\n", PQresultErrorMessage(res));
 				}
 				break;
 			default:
@@ -896,9 +1035,9 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 
 		for (i = 0; i < testspec->nsessions; i++)
 		{
-			if (notify->be_pid == backend_pids[i + 1])
+			if (notify->be_pid == conns[i + 1].backend_pid)
 			{
-				sendername = testspec->sessions[i]->name;
+				sendername = conns[i + 1].sessionname;
 				break;
 			}
 		}
@@ -915,6 +1054,43 @@ try_complete_step(TestSpec *testspec, Step *step, int flags)
 		PQconsumeInput(conn);
 	}
 
+	/* Connection is now idle. */
+	iconn->active_step = NULL;
+
+	return false;
+}
+
+/* Detect whether a step has any unsatisfied blocker conditions */
+static bool
+step_has_blocker(PermutationStep *pstep)
+{
+	int			i;
+
+	for (i = 0; i < pstep->nblockers; i++)
+	{
+		PermutationStepBlocker *blocker = pstep->blockers[i];
+		IsoConnInfo *iconn;
+
+		switch (blocker->blocktype)
+		{
+			case PSB_ONCE:
+				/* Ignore; try_complete_step handles this specially */
+				break;
+			case PSB_OTHER_STEP:
+				/* Block if referenced step is active */
+				iconn = &conns[1 + blocker->step->session];
+				if (iconn->active_step &&
+					iconn->active_step->step == blocker->step)
+					return true;
+				break;
+			case PSB_NUM_NOTICES:
+				/* Block if not enough notices received yet */
+				iconn = &conns[1 + blocker->step->session];
+				if (iconn->total_notices < blocker->target_notices)
+					return true;
+				break;
+		}
+	}
 	return false;
 }
 
@@ -940,11 +1116,17 @@ printResultSet(PGresult *res)
 	}
 }
 
-/* notice processor, prefixes each message with the session name */
+/* notice processor for regular user sessions */
 static void
 isotesterNoticeProcessor(void *arg, const char *message)
 {
-	printf("%s: %s", (char *) arg, message);
+	IsoConnInfo *myconn = (IsoConnInfo *) arg;
+
+	/* Prefix the backend's message with the session name. */
+	printf("%s: %s", myconn->sessionname, message);
+	/* Record notices, since we may need this to decide to unblock a step. */
+	myconn->total_notices++;
+	any_new_notice = true;
 }
 
 /* notice processor, hides the message */