Re: [HACKERS] PATCH: Batch/pipelining support for libpq

Hi,

On 2017-08-10 15:23:06 +1000, Vaishnavi Prabakaran wrote:
> Andres Freund wrote :
> > If you were to send a gigabyte of queries, it'd buffer them all up in
> memory... So some
> >more intelligence is going to be needed.
> 
> Firstly, sorry for the delayed response as I got busy with other
> activities.

No worries - development of new features was slowed down anyway, due to
the v10 feature freeze.


> To buffer up the queries before flushing them to the socket, I think
> "conn->outCount>=65536" is ok to use, as 64k is considered to be safe in
> Windows as per comments in pqSendSome() API.
> 
> Attached the code patch to replace pqFlush calls with pqBatchFlush in the
> asynchronous libpq function calls flow.
> Still pqFlush is used in "PQbatchSyncQueue" and
> "PQexitBatchMode" functions.


> Am failing to see the benefit in allowing user to set
> PQBatchAutoFlush(true|false) property? Is it really needed?

I'm inclined not to introduce that for now. If somebody comes up with a
convincing usecase and numbers, we can add it later. Libpq API is set in
stone, so I'd rather not introduce unnecessary stuff...



> +   <para>
> +    Much like asynchronous query mode, there is no performance disadvantage to
> +    using batching and pipelining. It increases client application complexity
> +    and extra caution is required to prevent client/server deadlocks but
> +    can sometimes offer considerable performance improvements.
> +   </para>

That's not necessarily true, is it? Unless you count always doing
batches of exactly size 1.


> +   <para>
> +    Batching is most useful when the server is distant, i.e. network latency
> +    (<quote>ping time</quote>) is high, and when many small operations are being performed in
> +    rapid sequence. There is usually less benefit in using batches when each
> +    query takes many multiples of the client/server round-trip time to execute.
> +    A 100-statement operation run on a server 300ms round-trip-time away would take
> +    30 seconds in network latency alone without batching; with batching it may spend
> +    as little as 0.3s waiting for results from the server.
> +   </para>

I'd add a remark that this is frequently beneficial even in cases of
minimal latency - as e.g. shown by the numbers I presented upthread.


> +   <para>
> +    Use batches when your application does lots of small
> +    <literal>INSERT</literal>, <literal>UPDATE</literal> and
> +    <literal>DELETE</literal> operations that can't easily be transformed into
> +    operations on sets or into a
> +    <link linkend="libpq-copy"><literal>COPY</literal></link> operation.
> +   </para>

Aren't SELECTs also a major beneficiarry of this?


> +   <para>
> +    Batching is less useful when information from one operation is required by the
> +    client before it knows enough to send the next operation.

s/less/not/


> +   <note>
> +    <para>
> +     The batch API was introduced in PostgreSQL 10.0, but clients using PostgresSQL 10.0 version of libpq can
> +     use batches on server versions 8.4 and newer. Batching works on any server
> +     that supports the v3 extended query protocol.
> +    </para>
> +   </note>

Where's the 8.4 coming from?


> +   <para>
> +    The client uses libpq's asynchronous query functions to dispatch work,
> +    marking the end of each batch with <function>PQbatchSyncQueue</function>.
> +    And to get results, it uses <function>PQgetResult</function> and
> +    <function>PQbatchProcessQueue</function>. It may eventually exit
> +    batch mode with <function>PQexitBatchMode</function> once all results are
> +    processed.
> +   </para>
> +
> +   <note>
> +    <para>
> +     It is best to use batch mode with <application>libpq</application> in
> +     <link linkend="libpq-pqsetnonblocking">non-blocking mode</link>. If used in
> +     blocking mode it is possible for a client/server deadlock to occur. The
> +     client will block trying to send queries to the server, but the server will
> +     block trying to send results from queries it has already processed to the
> +     client. This only occurs when the client sends enough queries to fill its
> +     output buffer and the server's receive buffer before switching to
> +     processing input from the server, but it's hard to predict exactly when
> +     that'll happen so it's best to always use non-blocking mode.
> +    </para>
> +   </note>

Mention that nonblocking only actually helps if send/recv is done as
required, and can essentially require unbound memory?  We probably
should either document or implement some smarts about when to signal
read/write readyness. Otherwise we e.g. might be receiving tons of
result data without having sent the next query - or the other way round.


> +   <sect3 id="libpq-batch-sending">
> +    <title>Issuing queries</title>
> +
> +    <para>
> +     After entering batch mode the application dispatches requests
> +     using normal asynchronous <application>libpq</application> functions such as 
> +     <function>PQsendQueryParams</function>, <function>PQsendPrepare</function>,
> +     <function>PQsendQueryPrepared</function>, <function>PQsendDescribePortal</function>,
> +     <function>PQsendDescribePrepared</function>.
> +     The asynchronous requests are followed by a <link
> +     linkend="libpq-PQbatchSyncQueue"><function>PQbatchSyncQueue(conn)</function></link> call to mark
> +     the end of the batch. The client <emphasis>does not</emphasis> need to call
> +     <function>PQgetResult</function> immediately after dispatching each
> +     operation. <link linkend="libpq-batch-results">Result processing</link>
> +     is handled separately.
> +    </para>
> +    
> +    <para>
> +     Batched operations will be executed by the server in the order the client
> +     sends them. The server will send the results in the order the statements
> +     executed. The server may begin executing the batch before all commands
> +     in the batch are queued and the end of batch command is sent. If any
> +     statement encounters an error the server aborts the current transaction and
> +     skips processing the rest of the batch. Query processing resumes after the
> +     end of the failed batch.
> +    </para>

Maybe note that multiple batches can be "in flight"?
I.e. PQbatchSyncQueue() is about error handling, nothing else? Don't
have a great idea, but we might want to rename...


> +    <note>
> +     <para>
> +      The client must not assume that work is committed when it
> +      <emphasis>sends</emphasis> a <literal>COMMIT</literal>, only when the
> +      corresponding result is received to confirm the commit is complete.
> +      Because errors arrive asynchronously the application needs to be able to
> +      restart from the last <emphasis>received</emphasis> committed change and
> +      resend work done after that point if something goes wrong.
> +     </para>
> +    </note>

This seems fairly independent of batching.


> +   </sect3>
> +
> +   <sect3 id="libpq-batch-interleave">
> +    <title>Interleaving result processing and query dispatch</title>
> +
> +    <para>
> +     To avoid deadlocks on large batches the client should be structured around
> +     a nonblocking I/O loop using a function like <function>select</function>,
> +     <function>poll</function>, <function>epoll</function>,
> +     <function>WaitForMultipleObjectEx</function>, etc.
> +    </para>
> +
> +    <para>
> +     The client application should generally maintain a queue of work still to
> +     be dispatched and a queue of work that has been dispatched but not yet had
> +     its results processed.

Hm. Why? If queries are just issued, no such queue is required?


> When the socket is writable it should dispatch more
> +     work. When the socket is readable it should read results and process them,
> +     matching them up to the next entry in its expected results queue. Batches
> +     should be scoped to logical units of work, usually (but not always) one
> +     transaction per batch. There's no need to exit batch mode and re-enter it
> +     between batches or to wait for one batch to finish before sending the next.
> +    </para>

This really needs to take memory usage into account.

> +       </variablelist>
> +
> +     </listitem>
> +    </varlistentry>
> +
> +    <varlistentry id="libpq-PQenterBatchMode">
> +     <term>
> +      <function>PQenterBatchMode</function>
> +      <indexterm>
> +       <primary>PQenterBatchMode</primary>
> +      </indexterm>
> +     </term>
> +
> +     <listitem>
> +      <para>
> +      Causes a connection to enter batch mode if it is currently idle or
> +      already in batch mode.
> +
> +<synopsis>
> +int PQenterBatchMode(PGconn *conn);
> +</synopsis>
> +
> +        </para>
> +        <para>
> +          Returns 1 for success. Returns 0 and has no 
> +          effect if the connection is not currently idle, i.e. it has a result 
> +          ready, is waiting for more input from the server, etc. This function 
> +          does not actually send anything to the server, it just changes the 
> +          <application>libpq</application> connection state.
> +
> +        </para>
> +     </listitem>
> +    </varlistentry>
> +
> +    <varlistentry id="libpq-PQexitBatchMode">
> +     <term>
> +      <function>PQexitBatchMode</function>
> +      <indexterm>
> +       <primary>PQexitBatchMode</primary>
> +      </indexterm>
> +     </term>
> +
> +     <listitem>
> +      <para>
> +      Causes a connection to exit batch mode if it is currently in batch mode
> +      with an empty queue and no pending results.
> +<synopsis>
> +int PQexitBatchMode(PGconn *conn);
> +</synopsis>
> +        </para>
> +        <para>Returns 1 for success.
> +      Returns 1 and takes no action if not in batch mode. If the connection has

"returns 1"?


> +    <varlistentry id="libpq-PQbatchSyncQueue">
> +     <term>
> +      <function>PQbatchSyncQueue</function>
> +      <indexterm>
> +       <primary>PQbatchSyncQueue</primary>
> +      </indexterm>
> +     </term>
> +
> +     <listitem>
> +      <para>
> +      Delimits the end of a set of a batched commands by sending a <link
> +      linkend="protocol-flow-ext-query">sync message</link> and flushing
> +      the send buffer. The end of a batch serves as 
> +      the delimiter of an implicit transaction and
> +      an error recovery point; see <link linkend="libpq-batch-errors">
> +      error handling</link>.

I wonder why this isn't framed as PQbatchIssue/Send/...()? Syncing seems
to mostly make sense from a protocol POV.


> +    <varlistentry id="libpq-PQbatchQueueCount">
> +     <term>
> +      <function>PQbatchQueueCount</function>
> +      <indexterm>
> +       <primary>PQbatchQueueCount</primary>
> +      </indexterm>
> +     </term>
> +
> +     <listitem>
> +      <para>
> +      Returns the number of queries still in the queue for this batch, not
> +      including any query that's currently having results being processed.
> +      This is the number of times <function>PQbatchProcessQueue</function> has to be
> +      called before the query queue is empty again.
> +
> +<synopsis>
> +int PQbatchQueueCount(PGconn *conn);
> +</synopsis>
> +
> +      </para>
> +     </listitem>
> +    </varlistentry>

Given that apps are supposed to track this, I'm not sure why we have
this?


> +/*
> + * PQrecyclePipelinedCommand
> + *    Push a command queue entry onto the freelist. It must be a dangling entry
> + *    with null next pointer and not referenced by any other entry's next pointer.
> + */
> +static void
> +PQrecyclePipelinedCommand(PGconn *conn, PGcommandQueueEntry * entry)
> +{
> +    if (entry == NULL)
> +        return;
> +    if (entry->next != NULL)
> +    {
> +        fprintf(stderr, libpq_gettext("tried to recycle non-dangling command queue entry"));
> +        abort();

Don't think we use abort() in libpq like that. There's some Assert()s
tho.


>  static bool
>  PQsendQueryStart(PGconn *conn)
> @@ -1377,20 +1486,59 @@ PQsendQueryStart(PGconn *conn)
>                            libpq_gettext("no connection to the server\n"));
>          return false;
>      }
> -    /* Can't send while already busy, either. */
> -    if (conn->asyncStatus != PGASYNC_IDLE)
> +    /* Can't send while already busy, either, unless enqueuing for later */
> +    if (conn->asyncStatus != PGASYNC_IDLE && conn->batch_status == PQBATCH_MODE_OFF)
>      {
>          printfPQExpBuffer(&conn->errorMessage,
>                            libpq_gettext("another command is already in progress\n"));
>          return false;
>      }
>  
> -    /* initialize async result-accumulation state */
> -    pqClearAsyncResult(conn);
> +    if (conn->batch_status != PQBATCH_MODE_OFF)
> +    {
> +    /*

Weirdly indented.


> +    if (conn->batch_status != PQBATCH_MODE_OFF)
> +    {
> +        pipeCmd = PQmakePipelinedCommand(conn);
> +
> +        if (pipeCmd == NULL)
> +            return 0;            /* error msg already set */
> +
> +        last_query = &pipeCmd->query;
> +        queryclass = &pipeCmd->queryclass;
> +    }
> +    else
> +    {
> +        last_query = &conn->last_query;
> +        queryclass = &conn->queryclass;
> +    }

The amount of complexity / branches we're adding to all of these is more
than a bit unsightly.


> +/*
> + * PQbatchQueueCount
> + *     Return number of queries currently pending in batch mode
> + */
> +int
> +PQbatchQueueCount(PGconn *conn)
> +{
> +    int            count = 0;
> +    PGcommandQueueEntry *entry;
> +
> +    if (PQbatchStatus(conn) == PQBATCH_MODE_OFF)
> +        return 0;
> +
> +    entry = conn->cmd_queue_head;
> +    while (entry != NULL)
> +    {
> +        ++count;
> +        entry = entry->next;
> +    }
> +    return count;
> +}

Ugh, O(N)? In that case I'd rather just remove this.


> +/*
> + * PQbatchBegin

Mismatched w/ actual function name.


> + *     Put an idle connection in batch mode. Commands submitted after this
> + *     can be pipelined on the connection, there's no requirement to wait for
> + *     one to finish before the next is dispatched.
> + *
> + *     Queuing of new query or syncing during COPY is not allowed.

+"a"?

> + *     A set of commands is terminated by a PQbatchQueueSync. Multiple sets of batched
> + *     commands may be sent while in batch mode. Batch mode can be exited by
> + *     calling PQbatchEnd() once all results are processed.
> + *
> + *     This doesn't actually send anything on the wire, it just puts libpq
> + *     into a state where it can pipeline work.
> + */
> +int
> +PQenterBatchMode(PGconn *conn)
> +{
> +    if (!conn)
> +        return false;

true/false isn't quite in line with int return code.


> +/*
> + * PQbatchEnd

wrong name.

> + *     End batch mode and return to normal command mode.
> + *
> + *     Has no effect unless the client has processed all results
> + *     from all outstanding batches and the connection is idle.

That seems wrong - will lead to hard to diagnose errors.


> + *     Returns true if batch mode ended.
> + */
> +int
> +PQexitBatchMode(PGconn *conn)
> +{
> +    if (!conn)
> +        return false;
> +
> +    if (conn->batch_status == PQBATCH_MODE_OFF)
> +        return true;
> +
> +    switch (conn->asyncStatus)
> +    {
> +        case PGASYNC_IDLE:
> +            printfPQExpBuffer(&conn->errorMessage,
> +                   libpq_gettext_noop("internal error, IDLE in batch mode"));
> +            break;
> +        case PGASYNC_COPY_IN:
> +        case PGASYNC_COPY_OUT:
> +        case PGASYNC_COPY_BOTH:
> +            printfPQExpBuffer(&conn->errorMessage,
> +                   libpq_gettext_noop("internal error, COPY in batch mode"));
> +            break;

So we'll still check the queue in this case, that's a bit weird?



> +/*
> + * PQbatchQueueSync

Out of sync.

> + *     End a batch submission by sending a protocol sync. The connection will
> + *     remain in batch mode and unavailable for new non-batch commands until all
> + *     results from the batch are processed by the client.

"unavailable for new non-batch commands" - that's hard to follow, and
seems pretty redundant with PQendBatchMode (or however it's called).

> + *     It's legal to start submitting another batch immediately, without waiting
> + *     for the results of the current batch. There's no need to end batch mode
> + *     and start it again.
> + *
> + *     If a command in a batch fails, every subsequent command up to and including
> + *     the PQbatchQueueSync command result gets set to PGRES_BATCH_ABORTED state. If the
> + *     whole batch is processed without error, a PGresult with PGRES_BATCH_END is
> + *     produced.

Hm, should probably mention that that's only true for commands since the
last PQbatchQueueSync?



> +/*
> + * PQbatchQueueProcess

Out of sync.


> + * Returns false if the current query isn't done yet, the connection
> + * is not in a batch, or there are no more queries to process.

Last complaint about this - think this forgiving mode is a mistake.


> + */
> +int
> +PQbatchProcessQueue(PGconn *conn)
> +{

> +    /* This command's results will come in immediately.
> +     * Initialize async result-accumulation state */
> +    pqClearAsyncResult(conn);

I'm not following?


>  /*
>   * PQgetResult
> @@ -1749,10 +2228,32 @@ PQgetResult(PGconn *conn)

> +            if (conn->batch_status != PQBATCH_MODE_OFF)
> +            {
> +                /*
> +                 * batched queries aren't followed by a Sync to put us back in
> +                 * PGASYNC_IDLE state, and when we do get a sync we could
> +                 * still have another batch coming after this one.

This needs rephrasing.


> +                 * The connection isn't idle since we can't submit new
> +                 * nonbatched commands. It isn't also busy since the current
> +                 * command is done and we need to process a new one.
> +                 */
> +                conn->asyncStatus = PGASYNC_QUEUED;

Not sure I like the name.


> +    if (conn->asyncStatus == PGASYNC_QUEUED || conn->batch_status != PQBATCH_MODE_OFF)
> +    {
> +        printfPQExpBuffer(&conn->errorMessage,
> +                          libpq_gettext("Synchronous command execution functions are not allowed in batch
mode\n"));
> +        return false;
> +    }

Why do we need the PGASYNC_QUEUED test here?

> +/* pqBatchFlush
> + * In batch mode, data will be flushed only when the out buffer reaches the threshold value.
> + * In non-batch mode, data will be flushed all the time.
> + */
> +static int
> +pqBatchFlush(PGconn *conn)
> +{
> +    if ((conn->batch_status == PQBATCH_MODE_OFF)||(conn->outCount>=65536))
> +        return(pqFlush(conn));
> +    return 0; /* Just to keep compiler quiet */
> +}

This should be defined in a macro or such, rather than hardcoded.


Falling over now. This seems like enough feedback for a bit of work
anyway.

Regards,

Andres