diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index 93da50d..74d76f3 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -165,3 +165,7 @@ lo_lseek64 162 lo_tell64 163 lo_truncate64 164 PQconninfo 165 +PQgetFirstQuery 166 +PQgetLastQuery 167 +PQgetNextQuery 168 +PQgetQueryCommand 169 diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 3af222b..31fa437 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -2893,8 +2893,6 @@ freePGconn(PGconn *conn) free(conn->gsslib); #endif /* Note that conn->Pfdebug is not ours to close or free */ - if (conn->last_query) - free(conn->last_query); if (conn->inBuffer) free(conn->inBuffer); if (conn->outBuffer) @@ -2956,6 +2954,29 @@ closePGconn(PGconn *conn) * absent */ conn->asyncStatus = PGASYNC_IDLE; pqClearAsyncResult(conn); /* deallocate result */ + + /* + * Link active queries into the free list so we can free them + */ + if (conn->queryTail) + { + conn->queryTail->next = conn->queryFree; + conn->queryFree = conn->queryHead; + } + conn->queryHead = conn->queryTail = NULL; + + /* + * Free all query objects + */ + while (conn->queryFree) + { + PGquery * prev = conn->queryFree; + conn->queryFree = prev->next; + if (prev->querycmd) + free(prev->querycmd); + free(prev); + } + resetPQExpBuffer(&conn->errorMessage); pg_freeaddrinfo_all(conn->addrlist_family, conn->addrlist); conn->addrlist = NULL; @@ -3135,7 +3156,7 @@ PQresetPoll(PGconn *conn) } /* - * PQcancelGet: get a PGcancel structure corresponding to a connection. + * PQgetCancel: get a PGcancel structure corresponding to a connection. * * A copy is needed to be able to cancel a running query from a different * thread. If the same structure is used all structure members would have diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 4075e51..2a421a8 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1020,7 +1020,7 @@ pqRowProcessor(PGconn *conn, const char **errmsgp) * row; the original conn->result is left unchanged so that it can be used * again as the template for future rows. */ - if (conn->singleRowMode) + if (conn->queryHead && conn->queryHead->singleRowMode) { /* Copy everything that should be in the result at this point */ res = PQcopyResult(res, @@ -1080,7 +1080,7 @@ pqRowProcessor(PGconn *conn, const char **errmsgp) * Success. In single-row mode, make the result available to the client * immediately. */ - if (conn->singleRowMode) + if (conn->queryHead && conn->queryHead->singleRowMode) { /* Change result status to special single-row value */ res->resultStatus = PGRES_SINGLE_TUPLE; @@ -1088,6 +1088,7 @@ pqRowProcessor(PGconn *conn, const char **errmsgp) conn->next_result = conn->result; conn->result = res; /* And mark the result ready to return */ + /* TODO: Still correct ? */ conn->asyncStatus = PGASYNC_READY; } @@ -1132,14 +1133,12 @@ PQsendQuery(PGconn *conn, const char *query) } /* remember we are using simple query protocol */ - conn->queryclass = PGQUERY_SIMPLE; - + conn->queryTail->queryclass = PGQUERY_SIMPLE; + /* and remember the query text too, if possible */ - /* if insufficient memory, last_query just winds up NULL */ - if (conn->last_query) - free(conn->last_query); - conn->last_query = strdup(query); - + /* if insufficient memory, querycmd just winds up NULL */ + conn->queryTail->querycmd = strdup(query); + /* * Give the data a push. In nonblock mode, don't complain if we're unable * to send it all; PQgetResult() will do any additional flushing needed. @@ -1151,7 +1150,9 @@ PQsendQuery(PGconn *conn, const char *query) } /* OK, it's launched! */ - conn->asyncStatus = PGASYNC_BUSY; + if( conn->asyncStatus == PGASYNC_IDLE ) + conn->asyncStatus = PGASYNC_BUSY; + return 1; } @@ -1272,13 +1273,11 @@ PQsendPrepare(PGconn *conn, goto sendFailed; /* remember we are doing just a Parse */ - conn->queryclass = PGQUERY_PREPARE; + conn->queryTail->queryclass = PGQUERY_PREPARE; /* and remember the query text too, if possible */ - /* if insufficient memory, last_query just winds up NULL */ - if (conn->last_query) - free(conn->last_query); - conn->last_query = strdup(query); + /* if insufficient memory, querycmd just winds up NULL */ + conn->queryTail->querycmd = strdup(query); /* * Give the data a push. In nonblock mode, don't complain if we're unable @@ -1288,6 +1287,7 @@ PQsendPrepare(PGconn *conn, goto sendFailed; /* OK, it's launched! */ + /* TODO: Check status first! */ conn->asyncStatus = PGASYNC_BUSY; return 1; @@ -1344,6 +1344,8 @@ PQsendQueryPrepared(PGconn *conn, static bool PQsendQueryStart(PGconn *conn) { + PGquery * query; + if (!conn) return false; @@ -1357,21 +1359,46 @@ 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 in copy mode, either. */ + switch (conn->asyncStatus) { - printfPQExpBuffer(&conn->errorMessage, + case PGASYNC_IDLE: + case PGASYNC_BUSY: + case PGASYNC_READY: + break; + case PGASYNC_COPY_IN: + case PGASYNC_COPY_OUT: + case PGASYNC_COPY_BOTH: + printfPQExpBuffer(&conn->errorMessage, libpq_gettext("another command is already in progress\n")); - return false; + return false; } - /* initialize async result-accumulation state */ - conn->result = NULL; - conn->next_result = NULL; - - /* reset single-row processing mode */ - conn->singleRowMode = false; - + if( !conn->queryFree ) + { + query = (PGquery*) malloc(sizeof(PGquery)); + query->querycmd = 0; + query->singleRowMode = false; + query->next = 0; + } + else + { + query = conn->queryFree; + conn->queryFree = query->next; + if (query->querycmd) + free(query->querycmd); + query->querycmd = NULL; + query->next = NULL; + } + + if( conn->queryTail ) + conn->queryTail->next = query; + else + conn->queryHead = query; + + conn->queryTail = query; + /* ready to send command message */ return true; } @@ -1522,16 +1549,12 @@ PQsendQueryGuts(PGconn *conn, goto sendFailed; /* remember we are using extended query protocol */ - conn->queryclass = PGQUERY_EXTENDED; + conn->queryTail->queryclass = PGQUERY_EXTENDED; /* and remember the query text too, if possible */ - /* if insufficient memory, last_query just winds up NULL */ - if (conn->last_query) - free(conn->last_query); + /* if insufficient memory, querycmd just winds up NULL */ if (command) - conn->last_query = strdup(command); - else - conn->last_query = NULL; + conn->queryTail->querycmd = strdup(command); /* * Give the data a push. In nonblock mode, don't complain if we're unable @@ -1541,6 +1564,7 @@ PQsendQueryGuts(PGconn *conn, goto sendFailed; /* OK, it's launched! */ + /* TODO: Check status first! */ conn->asyncStatus = PGASYNC_BUSY; return 1; @@ -1576,7 +1600,7 @@ pqHandleSendFailure(PGconn *conn) } /* - * Select row-by-row processing mode + * Select row-by-row processing mode for the last launched query */ int PQsetSingleRowMode(PGconn *conn) @@ -1585,18 +1609,16 @@ PQsetSingleRowMode(PGconn *conn) * Only allow setting the flag when we have launched a query and not yet * received any results. */ - if (!conn) - return 0; - if (conn->asyncStatus != PGASYNC_BUSY) + if (!conn || !conn->queryTail) return 0; - if (conn->queryclass != PGQUERY_SIMPLE && - conn->queryclass != PGQUERY_EXTENDED) + if (conn->asyncStatus != PGASYNC_BUSY && conn->queryTail == conn->queryHead) return 0; - if (conn->result) + if (conn->queryTail->queryclass != PGQUERY_SIMPLE && + conn->queryTail->queryclass != PGQUERY_EXTENDED) return 0; /* OK, set flag */ - conn->singleRowMode = true; + conn->queryTail->singleRowMode = true; return 1; } @@ -1670,6 +1692,51 @@ PQisBusy(PGconn *conn) /* + * PQgetQueryCommand + */ +const char * +PQgetQueryCommand(PGquery *query) +{ + if (!query) + return NULL; + return query->querycmd; +} + +/* + * PQgetFirstQuery + */ +PGquery * +PQgetFirstQuery(PGconn *conn) +{ + if (!conn) + return 0; + + return conn->queryHead; +} + +/* + * PQgetLastQuery + */ +PGquery * +PQgetLastQuery(PGconn *conn) +{ + if (!conn) + return 0; + return conn->queryTail; +} + +/* + * PQgetNextQuery + */ +PGquery * +PQgetNextQuery(PGquery *query) +{ + if (!query) + return 0; + return query->next; +} + +/* * PQgetResult * Get the next PGresult produced by a query. Returns NULL if no * query work remains or an error has occurred (e.g. out of @@ -2132,14 +2199,7 @@ PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target) goto sendFailed; /* remember we are doing a Describe */ - conn->queryclass = PGQUERY_DESCRIBE; - - /* reset last-query string (not relevant now) */ - if (conn->last_query) - { - free(conn->last_query); - conn->last_query = NULL; - } + conn->queryTail->queryclass = PGQUERY_DESCRIBE; /* * Give the data a push. In nonblock mode, don't complain if we're unable @@ -2301,7 +2361,7 @@ PQputCopyEnd(PGconn *conn, const char *errormsg) * If we sent the COPY command in extended-query mode, we must issue a * Sync as well. */ - if (conn->queryclass != PGQUERY_SIMPLE) + if (conn->queryHead->queryclass != PGQUERY_SIMPLE) { if (pqPutMsgStart('S', false, conn) < 0 || pqPutMsgEnd(conn) < 0) diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index c514ca5..f8a262c 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -55,7 +55,26 @@ static void reportErrorPosition(PQExpBuffer msg, const char *query, int loc, int encoding); static int build_startup_packet(const PGconn *conn, char *packet, const PQEnvironmentOption *options); +static void pqQueryAdvance(PGconn *conn); +void +pqQueryAdvance(PGconn *conn) +{ + PGquery * query; + + query = conn->queryHead; + if (query == NULL) + return; + + /* Advance queryHead */ + conn->queryHead = query->next; + /* Push last query onto free stack */ + query->next = conn->queryFree; + conn->queryFree = query; + + if (conn->queryHead == NULL) + conn->queryTail = NULL; +} /* * parseInput: if appropriate, parse input data from backend @@ -68,7 +87,7 @@ pqParseInput3(PGconn *conn) char id; int msgLength; int avail; - + /* * Loop to parse successive complete messages available in the buffer. */ @@ -218,7 +237,15 @@ pqParseInput3(PGconn *conn) case 'Z': /* backend is ready for new query */ if (getReadyForQuery(conn)) return; - conn->asyncStatus = PGASYNC_IDLE; + + pqQueryAdvance(conn); + /* initialize async result-accumulation state */ + conn->result = NULL; + conn->next_result = NULL; + if( conn->queryHead != NULL ) + conn->asyncStatus = PGASYNC_BUSY; + else + conn->asyncStatus = PGASYNC_IDLE; break; case 'I': /* empty query */ if (conn->result == NULL) @@ -232,7 +259,7 @@ pqParseInput3(PGconn *conn) break; case '1': /* Parse Complete */ /* If we're doing PQprepare, we're done; else ignore */ - if (conn->queryclass == PGQUERY_PREPARE) + if (conn->queryHead->queryclass == PGQUERY_PREPARE) { if (conn->result == NULL) { @@ -266,7 +293,7 @@ pqParseInput3(PGconn *conn) break; case 'T': /* Row Description */ if (conn->result == NULL || - conn->queryclass == PGQUERY_DESCRIBE) + conn->queryHead->queryclass == PGQUERY_DESCRIBE) { /* First 'T' in a query sequence */ if (getRowDescriptions(conn, msgLength)) @@ -299,7 +326,7 @@ pqParseInput3(PGconn *conn) * instead of TUPLES_OK. Otherwise we can just ignore * this message. */ - if (conn->queryclass == PGQUERY_DESCRIBE) + if (conn->queryHead && conn->queryHead->queryclass == PGQUERY_DESCRIBE) { if (conn->result == NULL) { @@ -422,6 +449,8 @@ pqParseInput3(PGconn *conn) static void handleSyncLoss(PGconn *conn, char id, int msgLength) { + PGquery * query; + printfPQExpBuffer(&conn->errorMessage, libpq_gettext( "lost synchronization with server: got message type \"%c\", length %d\n"), @@ -430,6 +459,15 @@ handleSyncLoss(PGconn *conn, char id, int msgLength) pqSaveErrorResult(conn); conn->asyncStatus = PGASYNC_READY; /* drop out of GetResult wait loop */ + /* All queries are canceled, move them to the free list and free the query commands */ + while ((query = conn->queryHead) != NULL) + { + free(query->querycmd); + query->querycmd = NULL; + conn->queryHead = query->next; + query->next = conn->queryFree; + } + pqDropConnection(conn); conn->status = CONNECTION_BAD; /* No more connection to backend */ } @@ -455,7 +493,7 @@ getRowDescriptions(PGconn *conn, int msgLength) * PGresult created by getParamDescriptions, and we should fill data into * that. Otherwise, create a new, empty PGresult. */ - if (conn->queryclass == PGQUERY_DESCRIBE) + if (conn->queryHead->queryclass == PGQUERY_DESCRIBE) { if (conn->result) result = conn->result; @@ -562,7 +600,7 @@ getRowDescriptions(PGconn *conn, int msgLength) * If we're doing a Describe, we're done, and ready to pass the result * back to the client. */ - if (conn->queryclass == PGQUERY_DESCRIBE) + if (conn->queryHead->queryclass == PGQUERY_DESCRIBE) { conn->asyncStatus = PGASYNC_READY; return 0; @@ -865,10 +903,10 @@ pqGetErrorNotice3(PGconn *conn, bool isError) val = PQresultErrorField(res, PG_DIAG_STATEMENT_POSITION); if (val) { - if (conn->verbosity != PQERRORS_TERSE && conn->last_query != NULL) + if (conn->verbosity != PQERRORS_TERSE && conn->queryHead->querycmd != NULL) { /* emit position as a syntax cursor display */ - querytext = conn->last_query; + querytext = conn->queryHead->querycmd; querypos = atoi(val); } else @@ -1696,7 +1734,7 @@ pqEndcopy3(PGconn *conn) * If we sent the COPY command in extended-query mode, we must issue a * Sync as well. */ - if (conn->queryclass != PGQUERY_SIMPLE) + if (conn->queryHead->queryclass != PGQUERY_SIMPLE) { if (pqPutMsgStart('S', false, conn) < 0 || pqPutMsgEnd(conn) < 0) diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index b81dc16..163a5e0 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -141,6 +141,13 @@ typedef struct pg_result PGresult; */ typedef struct pg_cancel PGcancel; +/* PGquery encapsulates the progress of a single query command issued + * to the async api functions + * The contents of this struct are not supposed to be known to applications. + */ +typedef struct pg_query PGquery; + + /* PGnotify represents the occurrence of a NOTIFY message. * Ideally this would be an opaque typedef, but it's so simple that it's * unlikely to change. @@ -404,6 +411,11 @@ extern PGresult *PQgetResult(PGconn *conn); extern int PQisBusy(PGconn *conn); extern int PQconsumeInput(PGconn *conn); +extern const char * PQgetQueryCommand(PGquery *query); +extern PGquery *PQgetFirstQuery(PGconn *conn); +extern PGquery *PQgetLastQuery(PGconn *conn); +extern PGquery *PQgetNextQuery(PGquery *query); + /* LISTEN/NOTIFY support */ extern PGnotify *PQnotifies(PGconn *conn); diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 4ef46ff..fb9bd61 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -291,6 +291,16 @@ typedef struct pgDataValue const char *value; /* data value, without zero-termination */ } PGdataValue; +typedef struct pg_query +{ + PGQueryClass queryclass; + char *querycmd; /* last SQL command, or NULL if unknown */ + bool singleRowMode; /* return query result row-by-row? */ + struct pg_query * next; + void *userptr; /* convenience for the user */ +} PGquery; + + /* * PGconn stores all the state data associated with a single connection * to a backend. @@ -350,13 +360,19 @@ struct pg_conn ConnStatusType status; PGAsyncStatusType asyncStatus; PGTransactionStatusType xactStatus; /* never changes to ACTIVE */ - PGQueryClass queryclass; - char *last_query; /* last SQL command, or NULL if unknown */ + + /* queryHead and queryTail form a FIFO representing queries sent + * to the backend. queryHead is the first query sent, and is the + * query we are receiving results from, or have received results from */ + PGquery *queryHead; + PGquery *queryTail; + PGquery *queryFree; /* Reuse PGQuery allocations */ + int nQueries; + char last_sqlstate[6]; /* last reported SQLSTATE */ bool options_valid; /* true if OK to attempt connection */ bool nonblocking; /* whether this connection is using nonblock * sending semantics */ - bool singleRowMode; /* return current query result row-by-row? */ char copy_is_binary; /* 1 = copy binary, 0 = copy text */ int copy_already_done; /* # bytes already returned in COPY * OUT */ diff --git a/src/test/examples/Makefile b/src/test/examples/Makefile index aee5c04..3996760 100644 --- a/src/test/examples/Makefile +++ b/src/test/examples/Makefile @@ -14,7 +14,7 @@ override CPPFLAGS := -I$(libpq_srcdir) $(CPPFLAGS) override LDLIBS := $(libpq_pgport) $(LDLIBS) -PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo testlo64 +PROGS = testlibpq testlibpq2 testlibpq3 testlibpq4 testlo testlo64 testlibpqpipeline testlibpqpipeline2 all: $(PROGS) diff --git a/src/test/examples/testlibpqpipeline.c b/src/test/examples/testlibpqpipeline.c index 725aad5..da0466f 100644 --- a/src/test/examples/testlibpqpipeline.c +++ b/src/test/examples/testlibpqpipeline.c @@ -101,7 +101,7 @@ int testPipelinedSeries( PGconn * conn, int totalQueries, int totalQueued, int b #define TEST_P(q) \ if( (result = testPipelined(conn,totalQueries,totalQueued,q)) != 0 ) \ return result; - TEST_P("INSERT INTO test() VALUES ()"); + TEST_P("INSERT INTO test(id) VALUES (DEFAULT)"); TEST_P("SELECT * FROM test LIMIT 1"); TEST_P("SELECT * FROM test"); TEST_P("DELETE FROM test"); @@ -137,7 +137,7 @@ main(int argc, char **argv) PQsetnonblocking(conn,1); - PQexec(conn,"CREATE TABLE test ( id PRIMARY KEY AUTOINCREMENT )"); + PQclear(PQexec(conn,"CREATE TABLE test ( id SERIAL PRIMARY KEY )")); baseline = testPipelinedSeries(conn,10,1,0); testPipelinedSeries(conn,10,3,baseline); @@ -150,7 +150,7 @@ main(int argc, char **argv) testPipelinedSeries(conn,100,50,baseline); testPipelinedSeries(conn,100,100,baseline); - PQexec(conn,"DROP TABLE test"); + PQclear(PQexec(conn,"DROP TABLE test")); return 0; } \ No newline at end of file