isolationtester.c

Go to the documentation of this file.

/*
 * src/test/isolation/isolationtester.c
 *
 * isolationtester.c
 *      Runs an isolation test specified by a spec file.
 */
 
#include "postgres_fe.h"
 
#include <sys/select.h>
#include <sys/time.h>
 
#include "datatype/timestamp.h"
#include "isolationtester.h"
#include "libpq-fe.h"
#include "pg_getopt.h"
#include "pqexpbuffer.h"
 
#define PREP_WAITING "isolationtester_waiting"
 
/*
 * conns[0] is the global setup, teardown, and watchdog connection.  Additional
 * connections represent spec-defined sessions.
 */
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 = 360 * 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 *piles,
                                         int nsteps, PermutationStep **steps);
static void run_named_permutations(TestSpec *testspec);
static void run_permutation(TestSpec *testspec, int nsteps,
                            PermutationStep **steps);
 
/* 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 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);
 
static void
disconnect_atexit(void)
{
    int         i;
 
    for (i = 0; i < nconns; i++)
        if (conns[i].conn)
            PQfinish(conns[i].conn);
}
 
int
main(int argc, char **argv)
{
    const char *conninfo;
    const char *env_wait;
    TestSpec   *testspec;
    PGresult   *res;
    PQExpBufferData wait_query;
    int         opt;
    int         i;
 
    while ((opt = getopt(argc, argv, "V")) != -1)
    {
        switch (opt)
        {
            case 'V':
                puts("isolationtester (PostgreSQL) " PG_VERSION);
                exit(0);
            default:
                fprintf(stderr, "Usage: isolationtester [CONNINFO]\n");
                return EXIT_FAILURE;
        }
    }
 
    /*
     * Make stdout unbuffered to match stderr; and ensure stderr is unbuffered
     * too, which it should already be everywhere except sometimes in Windows.
     */
    setbuf(stdout, NULL);
    setbuf(stderr, NULL);
 
    /*
     * If the user supplies a non-option parameter on the command line, use it
     * as the conninfo string; otherwise default to setting dbname=postgres
     * and using environment variables or defaults for all other connection
     * parameters.
     */
    if (argc > optind)
        conninfo = argv[optind];
    else
        conninfo = "dbname = postgres";
 
    /*
     * If PG_TEST_TIMEOUT_DEFAULT is set, adopt its value (given in seconds)
     * as half the max time to wait for any one step to complete.
     */
    env_wait = getenv("PG_TEST_TIMEOUT_DEFAULT");
    if (env_wait != NULL)
        max_step_wait = 2 * ((int64) atoi(env_wait)) * USECS_PER_SEC;
 
    /* Read the test spec from stdin */
    spec_yyparse();
    testspec = &parseresult;
 
    /* Perform post-parse checking, and fill in linking fields */
    check_testspec(testspec);
 
    printf("Parsed test spec with %d sessions\n", testspec->nsessions);
 
    /*
     * Establish connections to the database, one for each session and an
     * extra for lock wait detection and global work.
     */
    nconns = 1 + testspec->nsessions;
    conns = (IsoConnInfo *) pg_malloc0(nconns * sizeof(IsoConnInfo));
    atexit(disconnect_atexit);
 
    for (i = 0; i < nconns; i++)
    {
        const char *sessionname;
 
        if (i == 0)
            sessionname = "control connection";
        else
            sessionname = testspec->sessions[i - 1]->name;
 
        conns[i].sessionname = sessionname;
 
        conns[i].conn = PQconnectdb(conninfo);
        if (PQstatus(conns[i].conn) != CONNECTION_OK)
        {
            fprintf(stderr, "Connection %d failed: %s",
                    i, PQerrorMessage(conns[i].conn));
            exit(1);
        }
 
        /*
         * Set up notice processors for the user-defined connections, so that
         * messages can get printed prefixed with the session names.  The
         * control connection gets a "blackhole" processor instead (hides all
         * messages).
         */
        if (i != 0)
            PQsetNoticeProcessor(conns[i].conn,
                                 isotesterNoticeProcessor,
                                 &conns[i]);
        else
            PQsetNoticeProcessor(conns[i].conn,
                                 blackholeNoticeProcessor,
                                 NULL);
 
        /*
         * Similarly, append the session name to application_name to make it
         * easier to map spec file sessions to log output and
         * pg_stat_activity. The reason to append instead of just setting the
         * name is that we don't know the name of the test currently running.
         */
        res = PQexecParams(conns[i].conn,
                           "SELECT set_config('application_name',\n"
                           "  current_setting('application_name') || '/' || $1,\n"
                           "  false)",
                           1, NULL,
                           &sessionname,
                           NULL, NULL, 0);
        if (PQresultStatus(res) != PGRES_TUPLES_OK)
        {
            fprintf(stderr, "setting of application name failed: %s",
                    PQerrorMessage(conns[i].conn));
            exit(1);
        }
 
        /* Save each connection's backend PID for subsequent use. */
        conns[i].backend_pid = PQbackendPID(conns[i].conn);
        conns[i].backend_pid_str = psprintf("%d", conns[i].backend_pid);
    }
 
    /*
     * Build the query we'll use to detect lock contention among sessions in
     * the test specification.  Most of the time, we could get away with
     * simply checking whether a session is waiting for *any* lock: we don't
     * exactly expect concurrent use of test tables.  However, autovacuum will
     * occasionally take AccessExclusiveLock to truncate a table, and we must
     * ignore that transient wait.
     */
    initPQExpBuffer(&wait_query);
    appendPQExpBufferStr(&wait_query,
                         "SELECT pg_catalog.pg_isolation_test_session_is_blocked($1, '{");
    /* The spec syntax requires at least one session; assume that here. */
    appendPQExpBufferStr(&wait_query, conns[1].backend_pid_str);
    for (i = 2; i < nconns; i++)
        appendPQExpBuffer(&wait_query, ",%s", conns[i].backend_pid_str);
    appendPQExpBufferStr(&wait_query, "}')");
 
    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].conn));
        exit(1);
    }
    PQclear(res);
    termPQExpBuffer(&wait_query);
 
    /*
     * Run the permutations specified in the spec, or all if none were
     * explicitly specified.
     */
    run_testspec(testspec);
 
    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(1);
        }
    }
 
    /* 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(1);
            }
            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(1);
                }
                /* 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(1);
                }
            }
        }
    }
 
    /*
     * 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);
        }
    }
 
    free(allsteps);
}
 
/*
 * Run the permutations specified in the spec, or all if none were
 * explicitly specified.
 */
static void
run_testspec(TestSpec *testspec)
{
    if (testspec->permutations)
        run_named_permutations(testspec);
    else
        run_all_permutations(testspec);
}
 
/*
 * Run all permutations of the steps and sessions.
 */
static void
run_all_permutations(TestSpec *testspec)
{
    int         nsteps;
    int         i;
    PermutationStep *steps;
    PermutationStep **stepptrs;
    int        *piles;
 
    /* Count the total number of steps in all sessions */
    nsteps = 0;
    for (i = 0; i < testspec->nsessions; i++)
        nsteps += testspec->sessions[i]->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
     * session on a pile. To generate a permutation, we pick steps from the
     * piles until all piles are empty. By picking steps from piles in
     * different order, we get different permutations.
     *
     * A pile is actually just an integer which tells how many steps we've
     * already picked from this pile.
     */
    piles = pg_malloc(sizeof(int) * testspec->nsessions);
    for (i = 0; i < testspec->nsessions; i++)
        piles[i] = 0;
 
    run_all_permutations_recurse(testspec, piles, 0, stepptrs);
 
    free(steps);
    free(stepptrs);
    free(piles);
}
 
static void
run_all_permutations_recurse(TestSpec *testspec, int *piles,
                             int nsteps, PermutationStep **steps)
{
    int         i;
    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)
        {
            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, piles, nsteps + 1, steps);
 
            piles[i]--;
 
            found = true;
        }
    }
 
    /* If all the piles were empty, this permutation is completed. Run it */
    if (!found)
        run_permutation(testspec, nsteps, steps);
}
 
/*
 * Run permutations given in the test spec
 */
static void
run_named_permutations(TestSpec *testspec)
{
    int         i;
 
    for (i = 0; i < testspec->npermutations; i++)
    {
        Permutation *p = testspec->permutations[i];
 
        run_permutation(testspec, p->nsteps, p->steps);
    }
}
 
static int
step_qsort_cmp(const void *a, const void *b)
{
    Step       *stepa = *((Step *const *) a);
    Step       *stepb = *((Step *const *) b);
 
    return strcmp(stepa->name, stepb->name);
}
 
static int
step_bsearch_cmp(const void *a, const void *b)
{
    const char *stepname = (const char *) a;
    Step       *step = *((Step *const *) b);
 
    return strcmp(stepname, step->name);
}
 
/*
 * Run one permutation
 */
static void
run_permutation(TestSpec *testspec, int nsteps, PermutationStep **steps)
{
    PGresult   *res;
    int         i;
    int         nwaiting = 0;
    PermutationStep **waiting;
 
    waiting = pg_malloc(sizeof(PermutationStep *) * testspec->nsessions);
 
    printf("\nstarting permutation:");
    for (i = 0; i < nsteps; i++)
        printf(" %s", steps[i]->name);
    printf("\n");
 
    /* Perform setup */
    for (i = 0; i < testspec->nsetupsqls; 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].conn));
            exit(1);
        }
        PQclear(res);
    }
 
    /* Perform per-session setup */
    for (i = 0; i < testspec->nsessions; i++)
    {
        if (testspec->sessions[i]->setupsql)
        {
            res = PQexec(conns[i + 1].conn, testspec->sessions[i]->setupsql);
            if (PQresultStatus(res) == PGRES_TUPLES_OK)
            {
                printResultSet(res);
            }
            else if (PQresultStatus(res) != PGRES_COMMAND_OK)
            {
                fprintf(stderr, "setup of session %s failed: %s",
                        conns[i + 1].sessionname,
                        PQerrorMessage(conns[i + 1].conn));
                exit(1);
            }
            PQclear(res);
        }
    }
 
    /* Perform steps */
    for (i = 0; i < nsteps; i++)
    {
        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.
         */
        if (iconn->active_step != NULL)
        {
            struct timeval start_time;
 
            gettimeofday(&start_time, NULL);
 
            while (iconn->active_step != NULL)
            {
                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))
                {
                    /* 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(PermutationStep *));
                    nwaiting--;
                }
 
                /*
                 * 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(1);
                    }
                }
            }
        }
 
        /* Send the query for this step. */
        if (!PQsendQuery(conn, step->sql))
        {
            fprintf(stdout, "failed to send query for step %s: %s\n",
                    step->name, PQerrorMessage(conn));
            exit(1);
        }
 
        /* Remember we launched a step. */
        iconn->active_step = pstep;
 
        /* Remember target number of NOTICEs for any blocker conditions. */
        for (j = 0; j < pstep->nblockers; j++)
        {
            PermutationStepBlocker *blocker = pstep->blockers[j];
 
            if (blocker->blocktype == PSB_NUM_NOTICES)
                blocker->target_notices = blocker->num_notices +
                    conns[blocker->step->session + 1].total_notices;
        }
 
        /* 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++] = pstep;
    }
 
    /* Wait for any remaining queries. */
    nwaiting = try_complete_steps(testspec, waiting, nwaiting, STEP_RETRY);
    if (nwaiting != 0)
    {
        fprintf(stderr, "failed to complete permutation due to mutually-blocking steps\n");
        exit(1);
    }
 
    /* Perform per-session teardown */
    for (i = 0; i < testspec->nsessions; i++)
    {
        if (testspec->sessions[i]->teardownsql)
        {
            res = PQexec(conns[i + 1].conn, testspec->sessions[i]->teardownsql);
            if (PQresultStatus(res) == PGRES_TUPLES_OK)
            {
                printResultSet(res);
            }
            else if (PQresultStatus(res) != PGRES_COMMAND_OK)
            {
                fprintf(stderr, "teardown of session %s failed: %s",
                        conns[i + 1].sessionname,
                        PQerrorMessage(conns[i + 1].conn));
                /* don't exit on teardown failure */
            }
            PQclear(res);
        }
    }
 
    /* Perform teardown */
    if (testspec->teardownsql)
    {
        res = PQexec(conns[0].conn, testspec->teardownsql);
        if (PQresultStatus(res) == PGRES_TUPLES_OK)
        {
            printResultSet(res);
        }
        else if (PQresultStatus(res) != PGRES_COMMAND_OK)
        {
            fprintf(stderr, "teardown failed: %s",
                    PQerrorMessage(conns[0].conn));
            /* don't exit on teardown failure */
        }
        PQclear(res);
    }
 
    free(waiting);
}
 
/*
 * 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 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.  Do not pass it on the first call.
 *
 * 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, PermutationStep *pstep, int flags)
{
    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;
    int         sock = PQsocket(conn);
    int         ret;
    PGresult   *res;
    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));
        exit(1);
    }
 
    gettimeofday(&start_time, NULL);
    FD_ZERO(&read_set);
 
    while (PQisBusy(conn))
    {
        FD_SET(sock, &read_set);
        timeout.tv_sec = 0;
        timeout.tv_usec = 10000;    /* Check for lock waits every 10ms. */
 
        ret = select(sock + 1, &read_set, NULL, NULL, &timeout);
        if (ret < 0)            /* error in select() */
        {
            if (errno == EINTR)
                continue;
            fprintf(stderr, "select failed: %m\n");
            exit(1);
        }
        else if (ret == 0)      /* select() timeout: check for lock wait */
        {
            struct timeval current_time;
            int64       td;
 
            /* If it's OK for the step to block, check whether it has. */
            if (flags & STEP_NONBLOCK)
            {
                bool        waiting;
 
                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].conn));
                    exit(1);
                }
                waiting = ((PQgetvalue(res, 0, 0))[0] == 't');
                PQclear(res);
 
                if (waiting)    /* waiting to acquire a lock */
                {
                    /*
                     * Since it takes time to perform the lock-check query,
                     * some data --- notably, NOTICE messages --- might have
                     * arrived since we looked.  We must call PQconsumeInput
                     * and then PQisBusy to collect and process any such
                     * messages.  In the (unlikely) case that PQisBusy then
                     * returns false, we might as well go examine the
                     * available result.
                     */
                    if (!PQconsumeInput(conn))
                    {
                        fprintf(stderr, "PQconsumeInput failed: %s\n",
                                PQerrorMessage(conn));
                        exit(1);
                    }
                    if (!PQisBusy(conn))
                        break;
 
                    /*
                     * conn is still busy, so conclude that the step really is
                     * waiting.
                     */
                    if (!(flags & STEP_RETRY))
                        printf("step %s: %s <waiting ...>\n",
                               step->name, step->sql);
                    return true;
                }
                /* else, not waiting */
            }
 
            /* Figure out how long we've been waiting for this step. */
            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;
 
            /*
             * After max_step_wait microseconds, try to cancel the query.
             *
             * If the user tries to test an invalid permutation, we don't want
             * to hang forever, especially when this is running in the
             * buildfarm.  This will presumably lead to this permutation
             * failing, but remaining permutations and tests should still be
             * OK.
             */
            if (td > max_step_wait && !canceled)
            {
                PGcancelConn *cancel_conn = PQcancelCreate(conn);
 
                if (PQcancelBlocking(cancel_conn))
                {
                    /*
                     * print to stdout not stderr, as this should appear in
                     * the test case's results
                     */
                    printf("isolationtester: canceling step %s after %d seconds\n",
                           step->name, (int) (td / USECS_PER_SEC));
                    canceled = true;
                }
                else
                    fprintf(stderr, "PQcancel failed: %s\n", PQcancelErrorMessage(cancel_conn));
                PQcancelFinish(cancel_conn);
            }
 
            /*
             * After twice max_step_wait, just give up and die.
             *
             * Since cleanup steps won't be run in this case, this may cause
             * later tests to fail.  That stinks, but it's better than waiting
             * forever for the server to respond to the cancel.
             */
            if (td > 2 * max_step_wait)
            {
                fprintf(stderr, "step %s timed out after %d seconds\n",
                        step->name, (int) (td / USECS_PER_SEC));
                exit(1);
            }
        }
        else if (!PQconsumeInput(conn)) /* select(): data available */
        {
            fprintf(stderr, "PQconsumeInput failed: %s\n",
                    PQerrorMessage(conn));
            exit(1);
        }
    }
 
    /*
     * 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
        printf("step %s: %s\n", step->name, step->sql);
 
    while ((res = PQgetResult(conn)))
    {
        switch (PQresultStatus(res))
        {
            case PGRES_COMMAND_OK:
            case PGRES_EMPTY_QUERY:
                break;
            case PGRES_TUPLES_OK:
                printResultSet(res);
                break;
            case PGRES_FATAL_ERROR:
 
                /*
                 * Detail may contain XID values, so we want to just show
                 * primary.  Beware however that libpq-generated error results
                 * may not contain subfields, only an old-style message.
                 */
                {
                    const char *sev = PQresultErrorField(res,
                                                         PG_DIAG_SEVERITY);
                    const char *msg = PQresultErrorField(res,
                                                         PG_DIAG_MESSAGE_PRIMARY);
 
                    if (sev && msg)
                        printf("%s:  %s\n", sev, msg);
                    else
                        printf("%s\n", PQresultErrorMessage(res));
                }
                break;
            default:
                printf("unexpected result status: %s\n",
                       PQresStatus(PQresultStatus(res)));
        }
        PQclear(res);
    }
 
    /* Report any available NOTIFY messages, too */
    PQconsumeInput(conn);
    while ((notify = PQnotifies(conn)) != NULL)
    {
        /* Try to identify which session it came from */
        const char *sendername = NULL;
        char        pidstring[32];
        int         i;
 
        for (i = 0; i < testspec->nsessions; i++)
        {
            if (notify->be_pid == conns[i + 1].backend_pid)
            {
                sendername = conns[i + 1].sessionname;
                break;
            }
        }
        if (sendername == NULL)
        {
            /* Doesn't seem to be any test session, so show the hard way */
            snprintf(pidstring, sizeof(pidstring), "PID %d", notify->be_pid);
            sendername = pidstring;
        }
        printf("%s: NOTIFY \"%s\" with payload \"%s\" from %s\n",
               testspec->sessions[step->session]->name,
               notify->relname, notify->extra, sendername);
        PQfreemem(notify);
        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;
}
 
static void
printResultSet(PGresult *res)
{
    PQprintOpt  popt;
 
    memset(&popt, 0, sizeof(popt));
    popt.header = true;
    popt.align = true;
    popt.fieldSep = "|";
    PQprint(stdout, res, &popt);
}
 
/* notice processor for regular user sessions */
static void
isotesterNoticeProcessor(void *arg, const char *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 */
static void
blackholeNoticeProcessor(void *arg, const char *message)
{
    /* do nothing */
}