Re: rfc - libpq extensions

First, consider a motivating example. A single-threaded event-driven
application server serves two concurrent clients, client-1 and client-2,
using only a single "pooled" connection.

client-1 and client-2 connect to the application server.

client-1 issues an app specific request to the app server. The app
server, which is sitting in an event loop, services client-1 by
acquiring the single available connection and issuing an asynchronous
query. The server then registers the database connection's FD for
I/O events and returns to the event loop.

client-2 issues a request. The app server wakes from the event loop and
tries to service client-2. At this point, the server cannot acquire the
database connection because it's in use (and it can't issue the query
because libpq doesn't support queued queries). Simply creating a new
database connection doesn't solve the problem since we may wish to
support a very large number of concurrent clients (which is the reason
we're using an event-driven server and connection pooling).

To get around this I thought the following extension API (implemented
using the libpq API) would help. Here's a highly simplified pseudo-code
example with error handling omitted:

==========

// Create a cache with a maximum of 1 "hard" connections
PGXconnCache* pCache = PQXcreateConnCache(1);

// Create two "soft" (or virtual) connections and associate them
// with the connection cache from which they will draw "hard"
// connections when necessary.
//

PGXconn* pConn1 = PQXconnect(pCache, connString);
PGXconn* pConn2 = PQXconnect(pCache, connString);

// Create 2 queries. Issue one on each connection.
//
// Upon calling PQXconnExecute, pConn1 goes to the "hard"
// connection cache and obtains the single available
// connection and asynchronously issues the query on it.
// The descriptor returned for PQXconnFd and registered
// with our event handling method is the descriptor
// of the underlying "hard" connection's socket.

PGXquery* pQuery1 = PQXcreateQuery(queryString1);
PQXconnExecute(pConn1, pQuery1);
register_event(PQXconnFd(pConn1), EV_READ, event_handler, pConn1);

// At this point, the single connection is in use. When
// pConn2 goes to the cache it will find it empty. So
// pConn2 queues its query and registers itself with
// the cache for a notification as soon as a "hard"
// connection becomes available.
//
// In this case, the descriptor returned by PQXconnFd is
// the descriptor for the read end of a pipe. The write
// end is what pConn2 registered with the cache. As soon
// as pConn1 returns its "hard" conn to the cache, the
// cache will write a single byte to pConn2's notification
// pipe. From an application developer's perspective, however
// this detail is transparent.

PGXquery* pQuery2 = PQXcreateQuery(queryString2);
PQXconnExecute(pConn2, pQuery2);
register_event(PQXconnFd(pConn2), EV_READ, event_handler, pConn2);

while(1)
{
    event_dispatch();
    if (PQXgetQueryState(pQuery1) == PGX_QUERY_COMPLETE))
    {
        // process results
    }
    if (PQXgetQueryState(pQuery2) == PGX_QUERY_COMPLETE))
    {
        // process results
    }

}

==========

void
event_handler(fd, void* arg)
{
    PGXconn* pConn = (PGXconn*)(arg);
    // For pConn1, the call to connEvent will result in
    // processing of events coming over the database socket.
    // Once pConn1 is done processing its current query, it
    // will return its "hard" connection to the cache. The
    // cache will then notify the waiting pConn2 pipe (by
    // writing to pConn2's pipe).
    // When the thread of control returns to the event loop
    // it will pick that event up and we'll end up back here
    // but with an event for pConn2.
    // pConn2 can then go to the cache, get the newly
    // available "hard" connection and asynchronously issue
    // its query.
    PQXconnEvent(pConn);
}

==========

// Here's a preliminary list of methods that I've started
// to implement. The descriptions/semantics are preliminary, too.

/**
 * Create hard connection cache with a maximum of maxCn
 * hard connections for any given connection string. Returns
 * a valid pointer on success, 0 otherwise.
 *
 */
extern PGXconnCache* PQXcreateConnCache(unsigned int maxCn);

/**
 * Make hard connection allocations synchronous. Subsequent
 * operations on this cache will be thread safe (TBD ?)
 *
 */
extern void PQXsetConnCacheSync(PGXconnCache* pCC);

/**
 * Release hard connection cache pCC. Any soft connections
 * are closed and their respective pending queries are cancelled.
 *
 */
extern void PQXfreeConnCache(PGXconnCache* pCC);

/**
 * Create a soft connection associated with the hard
 * connection cache pCC. The format of the connection
 * string is as described in PQconnectdb().
 *
 */
extern PGXconn* PQXconnect(PGXconnCache* pCC, const char* conninfo);

/**
 * Returns the event descriptor for the given soft connection
 * suitable for use with poll() or select().
 *
 */
extern int PQXconnFd(const PGXconn* pCn);

/**
 * Use soft connection pCn to execute the query pQ. If the
 * soft connection is associated with an actual connection
 * then query goes out immediately, otherwise it gets
 * queued for later execution. Returns 1 if query was issued,
 * 0 if it was queued, and -1 on error.
 *
 */
extern int PQXconnExecute(PGXconn* pCn, PGXquery* pQ);

/**
 * Call this method whenever there's activity on
 * the descriptor for this soft connection.
 *
 */
extern int PQXconnEvent(PGXconn* pCn);

/**
 * Closes the soft connection pCn. Any held hard connections
 * are returned to the cache from which the soft connection
 * sprang.
 *
 */
extern void PQXconnClose(PGXconn* pCn);

/**
 * Releases all resources associated with pCn. If the connection
 * is open, it is closed before being freed.
 *
 */
extern void PQXconnFree(PGXconn* pCn);

/**
 * Returns the soft connection state. If you're asynchronously
 * connecting you should register pCn's event descriptor for
 * writing when this method returns PGX_CONN_CONNECTING. Note
 * that you should always check this method after calling
 * PQXconnect since connections can sometime happen synchronously
 * when the TCP peer is local (this will save you from having
 * to register the FD for write events).
 */
extern PGXconnStateEnum PQXconnState(PGXconn* pCn);

/**
 * Returns the error message associated with a soft connection,
 * if any.
 *
 */
extern const char* PQXconnError(PGXconn* pCn);

/**
 * Create a SQL query from the given text. The
 * text is NOT checked for syntax locally but on the
 * backend server.
 *
 */
extern PGXquery* PQXcreateQuery(const char* pText);

/**
 * Releases the given query. If the query is pending
 * execution it is cancelled. If it is currently executing
 * a cancel request is issued (and the response is
 * ignored).
 *
 */
extern void PQXfreeQuery(PGXquery* pQ);

/**
 * Cancels the query - if the query is pending execution
 * then its cancelled immediately. If it's been shipped
 * off then the query state becomes CANCEL_PENDING.
 *
 * Returns 1 on success, 0 on failure.
 */
extern int PQXcancelQuery(PGXquery* pQ);

/**
 * Get a query's result object (if any)
 *
 */
extern PGresult* PQXgetQueryResult(PGXquery* pQ);

/**
 * Returns the state of the given query.
 *
 */
extern PGXqueryStateEnum PQXgetQueryState(PGXquery* pQ);

/**
 * Sets the text for query pQ provided pQ is not currently
 * executing. The query buffer for pQ is reused or expanded
 * to accomodate pText as necessary. Returns 1 on success,
 * 0 otherwise.
 *
 */
extern int PQXsetQueryText(PGXquery* pQ, const char* pText);


On Fri, 10 Jan 2003 17:15:39 +0000 (UTC)
neilc@samurai.com (Neil Conway) wrote:

> On Fri, 2003-01-10 at 12:10, Iker Arizmendi wrote:
> > As things stand now you have to wait for each query to finish before
> > issuing a new one.
>
> How exactly are you planning to fix this?
>
> Cheers,
>
> Neil
> --
> Neil Conway <neilc@samurai.com> || PGP Key ID: DB3C29FC
>
>
>
>
> ---------------------------(end of
> broadcast)--------------------------- TIP 5: Have you checked our
> extensive FAQ?
>
> http://www.postgresql.org/users-lounge/docs/faq.html