在慢速网络上实现更好POSTGRESQL性能的管道模式

#include <libpq-fe.h>
#include <stdio.h>


/*
 * Receive and check a statement result.
 * If "res" is NULL, we expect a NULL result and
 * print the message if we get anything else.
 * If "res" is not NULL, the result is stored there.
 * In that case, if the result status is different
 * from "expected_status", print the message.
 */
static int checkResult(PGconn *conn,
                       PGresult **res,
                       ExecStatusType expected_status,
                       char * const message)
{
    PGresult *r;


    if (res == NULL)
    {
        if ((r = PQgetResult(conn)) == NULL)
            return 0;


        PQclear(r);
        fprintf(stderr, "%s: unexpected result\n",
                message);
        return 1;
    }


    if ((*res = PQgetResult(conn)) == NULL)
    {
        fprintf(stderr, "%s: missing result\n",
                message);
        return 1;
    }


    if (PQresultStatus(*res) == expected_status)
        return 0;


    fprintf(stderr, "%s: %s\n",
            message,
            PQresultErrorMessage(*res));
    PQclear(*res);
    return 1;
}


/* transfer "amount" from "from_acct" to "to_acct" */
static int transfer(PGconn *conn,
                    int from_acct,
                    int to_acct,
                    double amount)
{
    PGresult *res;
    int rc;
    char acct[100], amt[100];   * will fit a number */
    char * const values[] = { acct, amt };  * parameters */


    * enter pipeline mode */
    if (!PQenterPipelineMode(conn))
    {
        fprintf(stderr, "Cannot enter pipeline mode: %s\n",
                PQerrorMessage(conn));
        return 1;
    }


    * send query to subtract amount from the first account */
    snprintf(values[0], 100, "%d", from_acct);
    snprintf(values[1], 100, "%.2f", -amount);
    if (!PQsendQueryPrepared(conn,
                             "stmt", * statement name */
                             2,      * parameter count */
                             (const char * const *) values,
                             NULL,   * parameter lengths */
                             NULL,   * text parameters */
                             0))     * text result */
    {
        fprintf(stderr, "Error queuing first update: %s\n",
                PQerrorMessage(conn));
        rc = 1;
    }


    *
     * Tell the server that it should start returning results
     * right now rather than wait and gather the results for
     * the whole pipeline in a single packet.
     * There is no great benefit for short statements like these,
     * but it can reduce the time until we get the first result.
     */
    if (rc == 0 && PQsendFlushRequest(conn) == 0)
    {
        fprintf(stderr, "Error queuing flush request\n");
        rc = 1;
    }


    *
     * Dispatch pipelined commands to the server.
     * There is no great benefit for short statements like these,
     * but it can reduce the time until we get the first result.
     */
    if (rc == 0 && PQflush(conn) == -1)
    {
        fprintf(stderr,
                "Error flushing data to the server: %s\n",
                PQerrorMessage(conn));
        rc = 1;
    }


    * send query to add amount to the second account */
    snprintf(values[0], 100, "%d", to_acct);
    snprintf(values[1], 100, "%.2f", amount);
    if (rc == 0
        && !PQsendQueryPrepared(conn,
                                "stmt", * statement name */
                                2,      * parameter count */
                                (const char * const *) values,
                                NULL,   * parameter lengths */
                                NULL,   * text parameters */
                                0))     * text result */
    {
        fprintf(stderr,
                "Error queuing second update: %s\n",
                PQerrorMessage(conn));
        rc = 1;
    }


    /*---
     * Send a "sync" request:
     * - flush the remaining statements
     * - end the transaction
     * - wait for results
     */
    if (PQpipelineSync(conn) == 0)
    {
        fprintf(stderr, "Error sending \"sync\" request: %s\n",
                PQerrorMessage(conn));
        rc = 1;
    }


    /* consume the first statement result */
    if (checkResult(conn, &res, PGRES_TUPLES_OK, "first update"))
        rc = 1;
    else
        printf("Account %d now has %s\n",
               from_acct,
               PQgetvalue(res, 0, 0));
    if (res != NULL)
        PQclear(res);


    /* the next call must return nothing */
    if (checkResult(conn, NULL, -1, "end of first result set"))
        rc = 1;


    /* consume the second statement result */
    if (checkResult(conn, &res, PGRES_TUPLES_OK, "second update"))
        rc = 1;
    else
        printf("Account %d now has %s\n",
               to_acct,
               PQgetvalue(res, 0, 0));
    if (res != NULL)
        PQclear(res);


    /* the next call must return nothing */
    if (checkResult(conn, NULL, -1, "end of second result set"))
        rc = 1;


    /* consume the "ReadyForQuery" response */
    if (checkResult(conn, &res, PGRES_PIPELINE_SYNC, "sync result"))
        rc = 1;
    else if (res != NULL)
        PQclear(res);


    /* exit pipeline mode */
    if (PQexitPipelineMode(conn) == 0)
    {
        fprintf(stderr, "error ending pipeline mode: %s\n",
                PQresultErrorMessage(res));
        rc = 1;
    }
    return rc;
}
复制

|                | 没有管道 (8次网络延迟) | 管道(2次网络延迟) |
| -------------- | ---------------------- | ----------------- |
| first attempt  | 406 ms                 | 111 ms            |
| second attempt | 414 ms                 | 104 ms            |
| third attempt  | 414 ms                 | 103 ms            |
复制

CREATE PROCEDURE transfer(
   p_from_acct bigint,
   p_to_acct bigint,
   p_amount numeric
) LANGUAGE plpgsql AS
$$BEGIN
   UPDATE account
   SET amount = amount - p_amount
   WHERE id = p_from_acct;


   UPDATE account
   SET amount = amount + p_amount
   WHERE id = p_to_acct;
END;$$;
复制