statementKey that corresponds to the key passed
! * to {@link #beforeQuery(java.lang.Object)}.
! *! * If the query is not currently executing or waiting for execution, ! * this method is a no-op. If the query is waiting for the connection ! * to become free and has not started execution, it is immediately ! * cancelled. If the query is currently executing, the backend is asked ! * to cancel the query; this may or may not cause the query to terminate ! * with an error, depending on exact timing. ! *
! * If the backend is asked to cancel a query, further query execution is
! * blocked until the cancellation completes to avoid a race condition
! * where the wrong query could be cancelled.
! *
! * @throws SQLException if the backend needed to be contacted to do a
! * query cancellation, but something went wrong during that process.
! */
! public void cancelQuery(Object statementKey) throws SQLException;
public Statement createStatement() throws SQLException;
public BaseResultSet execSQL(String s) throws SQLException;
public boolean getAutoCommit() throws SQLException;
***************
*** 43,47 ****
public void setAutoCommit(boolean autoCommit) throws SQLException;
public void setCursorName(String cursor) throws SQLException;
! }
--- 63,86 ----
public void setAutoCommit(boolean autoCommit) throws SQLException;
public void setCursorName(String cursor) throws SQLException;
! /**
! * Called by QueryExecutor before a query is sent to the backend.
! * Performs query ordering, cancelling, & mutual exclusion; only one
! * thread is allowed within a beforeQuery .. afterQuery block at once.
! *
! * @param statementKey the statement key to pass to cancelQuery() to cancel this query.
! * @throws SQLException if the query is cancelled or the calling thread is interrupted
! * while waiting for other queries on this conneciton to complete.
! */
! public void beforeQuery(Object statementKey) throws SQLException;
+ /**
+ * Called by QueryExecutor after a query has completed. Every successful execution
+ * of beforeQuery() should be followed by an execution of afterQuery().
+ *
+ * @param statementKey the statement key passed to the corresponding beforeQuery() call.
+ * @throws IllegalStateException if statementKey does not match the currently
+ * executing statement key.
+ */
+ public void afterQuery(Object statementKey) throws IllegalStateException;
+ }
Index: src/interfaces/jdbc/org/postgresql/core/PGStream.java
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/core/PGStream.java,v
retrieving revision 1.3
diff -u -c -r1.3 PGStream.java
*** src/interfaces/jdbc/org/postgresql/core/PGStream.java 8 Sep 2003 17:30:22 -0000 1.3
--- src/interfaces/jdbc/org/postgresql/core/PGStream.java 16 Sep 2003 07:45:42 -0000
***************
*** 152,157 ****
--- 152,173 ----
}
/*
+ * Consume an expected EOF from the backend
+ * @exception SQLException if we get something other than an EOF
+ */
+ public void ReceiveEOF() throws SQLException
+ {
+ try {
+ int c = pg_input.read();
+ if (c < 0)
+ return;
+ throw new PSQLException("postgresql.stream.toomuch", PSQLState.COMMUNICATION_ERROR);
+ } catch (IOException e) {
+ throw new PSQLException("postgresql.stream.ioerror", PSQLState.COMMUNICATION_ERROR);
+ }
+ }
+
+ /*
* Receives a single character from the backend
*
* @return the character received
Index: src/interfaces/jdbc/org/postgresql/core/QueryExecutor.java
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/core/QueryExecutor.java,v
retrieving revision 1.26
diff -u -c -r1.26 QueryExecutor.java
*** src/interfaces/jdbc/org/postgresql/core/QueryExecutor.java 13 Sep 2003 04:02:13 -0000 1.26
--- src/interfaces/jdbc/org/postgresql/core/QueryExecutor.java 16 Sep 2003 07:45:42 -0000
***************
*** 95,108 ****
*/
private BaseResultSet execute() throws SQLException
{
! if (connection.getPGProtocolVersionMajor() == 3) {
! if (Driver.logDebug)
! Driver.debug("Using Protocol Version3 to send query");
! return executeV3();
! } else {
! if (Driver.logDebug)
! Driver.debug("Using Protocol Version2 to send query");
! return executeV2();
}
}
--- 95,119 ----
*/
private BaseResultSet execute() throws SQLException
{
! if (pgStream == null)
! {
! throw new PSQLException("postgresql.con.closed", PSQLState.CONNECTION_DOES_NOT_EXIST);
! }
!
! connection.beforeQuery(statement); // Might throw.
!
! try {
! if (connection.getPGProtocolVersionMajor() == 3) {
! if (Driver.logDebug)
! Driver.debug("Using Protocol Version3 to send query");
! return executeV3();
! } else {
! if (Driver.logDebug)
! Driver.debug("Using Protocol Version2 to send query");
! return executeV2();
! }
! } finally {
! connection.afterQuery(statement);
}
}
***************
*** 111,216 ****
PSQLException error = null;
! if (pgStream == null)
! {
! throw new PSQLException("postgresql.con.closed", PSQLState.CONNECTION_DOES_NOT_EXIST);
! }
! synchronized (pgStream)
{
!
! sendQueryV3();
!
! int c;
! boolean l_endQuery = false;
! while (!l_endQuery)
{
! c = pgStream.ReceiveChar();
! switch (c)
! {
! case 'A': // Asynchronous Notify
! int pid = pgStream.ReceiveInteger(4);
! String msg = pgStream.ReceiveString(connection.getEncoding());
! connection.addNotification(new org.postgresql.core.Notification(msg, pid));
! break;
! case 'B': // Binary Data Transfer
! receiveTupleV3(true);
! break;
! case 'C': // Command Status
! receiveCommandStatusV3();
! break;
! case 'D': // Text Data Transfer
! receiveTupleV3(false);
! break;
! case 'E': // Error Message
!
! // it's possible to get more than one error message for a query
! // see libpq comments wrt backend closing a connection
! // so, append messages to a string buffer and keep processing
! // check at the bottom to see if we need to throw an exception
!
! int l_elen = pgStream.ReceiveIntegerR(4);
! String totalMessage = connection.getEncoding().decode(pgStream.Receive(l_elen-4));
! PSQLException l_error = PSQLException.parseServerError(totalMessage);
!
! if (error != null) {
! error.setNextException(l_error);
! } else {
! error = l_error;
! }
!
! // keep processing
! break;
! case 'I': // Empty Query
! int t = pgStream.ReceiveChar();
! break;
! case 'N': // Error Notification
! int l_nlen = pgStream.ReceiveIntegerR(4);
! statement.addWarning(connection.getEncoding().decode(pgStream.Receive(l_nlen-4)));
! break;
! case 'P': // Portal Name
! String pname = pgStream.ReceiveString(connection.getEncoding());
! break;
! case 'S':
! //TODO: handle parameter status messages
! int l_len = pgStream.ReceiveIntegerR(4);
! String l_pStatus = connection.getEncoding().decode(pgStream.Receive(l_len-4));
! if (Driver.logDebug)
! Driver.debug("ParameterStatus="+ l_pStatus);
! break;
! case 'T': // MetaData Field Description
! receiveFieldsV3();
! break;
! case 'Z':
! // read ReadyForQuery
! //TODO: use size better
! if (pgStream.ReceiveIntegerR(4) != 5) throw new PSQLException("postgresql.con.setup", PSQLState.CONNECTION_UNABLE_TO_CONNECT);
! //TODO: handle transaction status
! char l_tStatus = (char)pgStream.ReceiveChar();
! l_endQuery = true;
! break;
! default:
! throw new PSQLException("postgresql.con.type", PSQLState.CONNECTION_FAILURE, new Character((char) c));
}
!
! }
!
! // did we get an error during this query?
! if ( error != null )
! throw error;
!
! //if an existing result set was passed in reuse it, else
! //create a new one
! if (rs != null)
! {
! rs.reInit(fields, tuples, status, update_count, insert_oid, binaryCursor);
}
! else
{
rs = statement.createResultSet(fields, tuples, status, update_count, insert_oid, binaryCursor);
}
! return rs;
! }
}
private BaseResultSet executeV2() throws SQLException
--- 122,218 ----
PSQLException error = null;
! sendQueryV3();
! int c;
! boolean l_endQuery = false;
! while (!l_endQuery)
{
! c = pgStream.ReceiveChar();
! switch (c)
{
! case 'A': // Asynchronous Notify
! int pid = pgStream.ReceiveInteger(4);
! String msg = pgStream.ReceiveString(connection.getEncoding());
! connection.addNotification(new org.postgresql.core.Notification(msg, pid));
! break;
! case 'B': // Binary Data Transfer
! receiveTupleV3(true);
! break;
! case 'C': // Command Status
! receiveCommandStatusV3();
! break;
! case 'D': // Text Data Transfer
! receiveTupleV3(false);
! break;
! case 'E': // Error Message
!
! // it's possible to get more than one error message for a query
! // see libpq comments wrt backend closing a connection
! // so, append messages to a string buffer and keep processing
! // check at the bottom to see if we need to throw an exception
!
! int l_elen = pgStream.ReceiveIntegerR(4);
! String totalMessage = connection.getEncoding().decode(pgStream.Receive(l_elen-4));
! PSQLException l_error = PSQLException.parseServerError(totalMessage);
!
! if (error != null) {
! error.setNextException(l_error);
! } else {
! error = l_error;
}
!
! // keep processing
! break;
! case 'I': // Empty Query
! int t = pgStream.ReceiveChar();
! break;
! case 'N': // Error Notification
! int l_nlen = pgStream.ReceiveIntegerR(4);
! statement.addWarning(connection.getEncoding().decode(pgStream.Receive(l_nlen-4)));
! break;
! case 'P': // Portal Name
! String pname = pgStream.ReceiveString(connection.getEncoding());
! break;
! case 'S':
! //TODO: handle parameter status messages
! int l_len = pgStream.ReceiveIntegerR(4);
! String l_pStatus = connection.getEncoding().decode(pgStream.Receive(l_len-4));
! if (Driver.logDebug)
! Driver.debug("ParameterStatus="+ l_pStatus);
! break;
! case 'T': // MetaData Field Description
! receiveFieldsV3();
! break;
! case 'Z':
! // read ReadyForQuery
! //TODO: use size better
! if (pgStream.ReceiveIntegerR(4) != 5) throw new PSQLException("postgresql.con.setup", PSQLState.CONNECTION_UNABLE_TO_CONNECT);
! //TODO: handle transaction status
! char l_tStatus = (char)pgStream.ReceiveChar();
! l_endQuery = true;
! break;
! default:
! throw new PSQLException("postgresql.con.type", PSQLState.CONNECTION_FAILURE, new Character((char) c));
}
!
! }
!
! // did we get an error during this query?
! if ( error != null )
! throw error;
!
! //if an existing result set was passed in reuse it, else
! //create a new one
! if (rs != null)
! {
! rs.reInit(fields, tuples, status, update_count, insert_oid, binaryCursor);
! }
! else
{
rs = statement.createResultSet(fields, tuples, status, update_count, insert_oid, binaryCursor);
}
! return rs;
}
private BaseResultSet executeV2() throws SQLException
***************
*** 218,306 ****
StringBuffer errorMessage = null;
! if (pgStream == null)
! {
! throw new PSQLException("postgresql.con.closed", PSQLState.CONNECTION_DOES_NOT_EXIST);
! }
! synchronized (pgStream)
{
!
! sendQueryV2();
!
! int c;
! boolean l_endQuery = false;
! while (!l_endQuery)
! {
! c = pgStream.ReceiveChar();
!
! switch (c)
! {
! case 'A': // Asynchronous Notify
! int pid = pgStream.ReceiveInteger(4);
! String msg = pgStream.ReceiveString(connection.getEncoding());
! connection.addNotification(new org.postgresql.core.Notification(msg, pid));
! break;
! case 'B': // Binary Data Transfer
! receiveTupleV2(true);
! break;
! case 'C': // Command Status
! receiveCommandStatusV2();
! break;
! case 'D': // Text Data Transfer
! receiveTupleV2(false);
! break;
! case 'E': // Error Message
!
! // it's possible to get more than one error message for a query
! // see libpq comments wrt backend closing a connection
! // so, append messages to a string buffer and keep processing
! // check at the bottom to see if we need to throw an exception
!
! if ( errorMessage == null )
! errorMessage = new StringBuffer();
!
! errorMessage.append(pgStream.ReceiveString(connection.getEncoding()));
! // keep processing
! break;
! case 'I': // Empty Query
! int t = pgStream.ReceiveChar();
! break;
! case 'N': // Error Notification
! statement.addWarning(pgStream.ReceiveString(connection.getEncoding()));
! break;
! case 'P': // Portal Name
! String pname = pgStream.ReceiveString(connection.getEncoding());
! break;
! case 'T': // MetaData Field Description
! receiveFieldsV2();
! break;
! case 'Z':
! l_endQuery = true;
! break;
! default:
! throw new PSQLException("postgresql.con.type", PSQLState.CONNECTION_FAILURE, new Character((char) c));
! }
!
}
!
! // did we get an error during this query?
! if ( errorMessage != null )
! throw new SQLException( errorMessage.toString().trim() );
!
!
! //if an existing result set was passed in reuse it, else
! //create a new one
! if (rs != null)
! {
! rs.reInit(fields, tuples, status, update_count, insert_oid, binaryCursor);
! }
! else
! {
! rs = statement.createResultSet(fields, tuples, status, update_count, insert_oid, binaryCursor);
! }
! return rs;
}
}
/*
--- 220,299 ----
StringBuffer errorMessage = null;
! sendQueryV2();
! int c;
! boolean l_endQuery = false;
! while (!l_endQuery)
{
! c = pgStream.ReceiveChar();
!
! switch (c)
! {
! case 'A': // Asynchronous Notify
! int pid = pgStream.ReceiveInteger(4);
! String msg = pgStream.ReceiveString(connection.getEncoding());
! connection.addNotification(new org.postgresql.core.Notification(msg, pid));
! break;
! case 'B': // Binary Data Transfer
! receiveTupleV2(true);
! break;
! case 'C': // Command Status
! receiveCommandStatusV2();
! break;
! case 'D': // Text Data Transfer
! receiveTupleV2(false);
! break;
! case 'E': // Error Message
!
! // it's possible to get more than one error message for a query
! // see libpq comments wrt backend closing a connection
! // so, append messages to a string buffer and keep processing
! // check at the bottom to see if we need to throw an exception
!
! if ( errorMessage == null )
! errorMessage = new StringBuffer();
!
! errorMessage.append(pgStream.ReceiveString(connection.getEncoding()));
! // keep processing
! break;
! case 'I': // Empty Query
! int t = pgStream.ReceiveChar();
! break;
! case 'N': // Error Notification
! statement.addWarning(pgStream.ReceiveString(connection.getEncoding()));
! break;
! case 'P': // Portal Name
! String pname = pgStream.ReceiveString(connection.getEncoding());
! break;
! case 'T': // MetaData Field Description
! receiveFieldsV2();
! break;
! case 'Z':
! l_endQuery = true;
! break;
! default:
! throw new PSQLException("postgresql.con.type", PSQLState.CONNECTION_FAILURE, new Character((char) c));
}
!
! }
!
! // did we get an error during this query?
! if ( errorMessage != null )
! throw new SQLException( errorMessage.toString().trim() );
!
!
! //if an existing result set was passed in reuse it, else
! //create a new one
! if (rs != null)
! {
! rs.reInit(fields, tuples, status, update_count, insert_oid, binaryCursor);
! }
! else
! {
! rs = statement.createResultSet(fields, tuples, status, update_count, insert_oid, binaryCursor);
}
+ return rs;
}
/*
Index: src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Connection.java
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Connection.java,v
retrieving revision 1.26
diff -u -c -r1.26 AbstractJdbc1Connection.java
*** src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Connection.java 13 Sep 2003 04:02:15 -0000 1.26
--- src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Connection.java 16 Sep 2003 07:45:43 -0000
***************
*** 104,109 ****
--- 104,113 ----
*/
private int isolationLevel = Connection.TRANSACTION_READ_COMMITTED;
+ // Query queue and cancellation infrastructure.
+ private Object currentQuery;
+ private final Vector queryQueue = new Vector(); // Better as a LinkedList, but that's not in 1.1.
+ private boolean cancelPending;
public abstract Statement createStatement() throws SQLException;
public abstract DatabaseMetaData getMetaData() throws SQLException;
***************
*** 1775,1786 ****
Types.TIMESTAMP, Types.TIMESTAMP, Types.TIMESTAMP
};
! public void cancelQuery() throws SQLException
{
org.postgresql.core.PGStream cancelStream = null;
try
{
cancelStream = new org.postgresql.core.PGStream(PG_HOST, PG_PORT);
}
catch (ConnectException cex)
{
--- 1779,1824 ----
Types.TIMESTAMP, Types.TIMESTAMP, Types.TIMESTAMP
};
! public void cancelQuery(Object statement) throws SQLException
{
+ synchronized (queryQueue) {
+ if (queryQueue.removeElement(statement)) {
+ // Query hadn't started executing yet. Remove it and wake up
+ // the thread that's waiting to start so it notices the
+ // cancellation.
+ queryQueue.notifyAll();
+ return;
+ }
+
+ if (currentQuery != statement) {
+ // Not executing this statement at all. Don't do anything.
+ return;
+ }
+
+ if (cancelPending) {
+ // We are already doing a cancel for the currently running statement.
+ // Don't do it twice.
+ return;
+ }
+
+ // Need to do a real cancel. Make sure new queries don't start while we do this.
+ cancelPending = true;
+ }
+
org.postgresql.core.PGStream cancelStream = null;
try
{
cancelStream = new org.postgresql.core.PGStream(PG_HOST, PG_PORT);
+
+ // Now we need to construct and send a cancel packet
+ cancelStream.SendInteger(16, 4);
+ cancelStream.SendInteger(80877102, 4);
+ cancelStream.SendInteger(pid, 4);
+ cancelStream.SendInteger(ckey, 4);
+ cancelStream.flush();
+
+ // Wait for the backend to close the connection to avoid races.
+ cancelStream.ReceiveEOF();
}
catch (ConnectException cex)
{
***************
*** 1793,1814 ****
{
throw new PSQLException ("postgresql.con.failed", PSQLState.CONNECTION_UNABLE_TO_CONNECT, e);
}
-
- // Now we need to construct and send a cancel packet
- try
- {
- cancelStream.SendInteger(16, 4);
- cancelStream.SendInteger(80877102, 4);
- cancelStream.SendInteger(pid, 4);
- cancelStream.SendInteger(ckey, 4);
- cancelStream.flush();
- }
- catch (IOException e)
- {
- throw new PSQLException("postgresql.con.failed", PSQLState.CONNECTION_UNABLE_TO_CONNECT, e);
- }
finally
{
try
{
if (cancelStream != null)
--- 1831,1844 ----
{
throw new PSQLException ("postgresql.con.failed", PSQLState.CONNECTION_UNABLE_TO_CONNECT, e);
}
finally
{
+ synchronized (queryQueue) {
+ // Cancellation is done, it's safe to start new queries again.
+ cancelPending = false;
+ queryQueue.notifyAll();
+ }
+
try
{
if (cancelStream != null)
***************
*** 1819,1824 ****
--- 1849,1913 ----
}
}
+ public void beforeQuery(Object statement) throws SQLException {
+ synchronized (queryQueue) {
+ // Fastpath for the common case where we have no contention.
+ if (!cancelPending && currentQuery == null) {
+ currentQuery = statement;
+ return;
+ }
+
+ // Ok, we have multiple queries and/or cancels involved, do the full logic.
+ queryQueue.addElement(statement);
+
+ // Wait for no currently running query and no currently running cancellation.
+ // Once both of these are true, we can start executing.
+ // If this statement is cancelled while still on the query queue,
+ // this is signalled by removing the statement from the queue.
+ while (true) {
+ if (!queryQueue.contains(statement)) {
+ // We were cancelled.
+ throw new PSQLException("postgresql.cancelled", PSQLState.SYSTEM_ERROR);
+ }
+
+ if (!cancelPending && currentQuery == null) {
+ // We can start.
+ queryQueue.removeElement(statement);
+ currentQuery = statement;
+ break;
+ }
+
+ // invariants here:
+ // queryQueue.contains(statement)
+ // currentQuery != statement
+ // currentQuery != null || cancelPending
+
+ try {
+ queryQueue.wait();
+ } catch (InterruptedException ie) {
+ // As it's hard for a caller to detect this case via
+ // the thrown exception, reset interruption state too.
+ Thread.currentThread().interrupt();
+ throw new PSQLException("postgresql.interrupted", PSQLState.SYSTEM_ERROR);
+ }
+ }
+
+ // invariants here:
+ // currentQuery == statement
+ // !cancelPending
+ // !queryQueue.contains(statement)
+ }
+ }
+
+ public void afterQuery(Object statement) {
+ synchronized (queryQueue) {
+ if (statement != currentQuery)
+ throw new IllegalStateException("bad beforeQuery()/afterQuery() ordering");
+
+ currentQuery = null;
+ queryQueue.notifyAll(); // Someone can wake up now.
+ }
+ }
//Methods to support postgres notifications
public void addNotification(org.postgresql.PGNotification p_notification)
Index: src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Statement.java
===================================================================
RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Statement.java,v
retrieving revision 1.36
diff -u -c -r1.36 AbstractJdbc1Statement.java
*** src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Statement.java 13 Sep 2003 04:02:15 -0000 1.36
--- src/interfaces/jdbc/org/postgresql/jdbc1/AbstractJdbc1Statement.java 16 Sep 2003 07:45:43 -0000
***************
*** 670,683 ****
/*
* Cancel can be used by one thread to cancel a statement that
* is being executed by another thread.
- *
- * Not implemented, this method is a no-op. * * @exception SQLException only because thats the spec. */ public void cancel() throws SQLException { ! throw new PSQLException("postgresql.unimplemented", PSQLState.NOT_IMPLEMENTED); } /* --- 670,681 ---- /* * Cancel can be used by one thread to cancel a statement that * is being executed by another thread. * * @exception SQLException only because thats the spec. */ public void cancel() throws SQLException { ! connection.cancelQuery(this); } /* Index: src/interfaces/jdbc/org/postgresql/jdbc2/AbstractJdbc2Statement.java =================================================================== RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/jdbc2/AbstractJdbc2Statement.java,v retrieving revision 1.17 diff -u -c -r1.17 AbstractJdbc2Statement.java *** src/interfaces/jdbc/org/postgresql/jdbc2/AbstractJdbc2Statement.java 9 Sep 2003 10:49:16 -0000 1.17 --- src/interfaces/jdbc/org/postgresql/jdbc2/AbstractJdbc2Statement.java 16 Sep 2003 07:45:43 -0000 *************** *** 118,128 **** return result; } - public void cancel() throws SQLException - { - connection.cancelQuery(); - } - public Connection getConnection() throws SQLException { return (Connection) connection; --- 118,123 ---- Index: src/interfaces/jdbc/org/postgresql/test/jdbc2/MiscTest.java =================================================================== RCS file: /projects/cvsroot/pgsql-server/src/interfaces/jdbc/org/postgresql/test/jdbc2/MiscTest.java,v retrieving revision 1.10 diff -u -c -r1.10 MiscTest.java *** src/interfaces/jdbc/org/postgresql/test/jdbc2/MiscTest.java 29 May 2003 04:39:48 -0000 1.10 --- src/interfaces/jdbc/org/postgresql/test/jdbc2/MiscTest.java 16 Sep 2003 07:45:43 -0000 *************** *** 99,102 **** --- 99,200 ---- fail( ex.getMessage() ); } } + + public void testCancel() throws Exception { + // + // Try to make sure there's no cancel() race where we'd cancel a not-yet-executed query. + // + + Connection con = TestUtil.openDB(); + Statement stmt = con.createStatement(); + for (int i = 0; i < 10000; ++i) { + stmt.executeQuery("select version()").close(); + stmt.cancel(); + } + } + + private class QueryRunnable implements Runnable { + QueryRunnable(Statement stmt, boolean ignoreErrors, int testValue) { + this.stmt = stmt; + this.ignoreErrors = ignoreErrors; + this.testValue = testValue; + } + + public void run() { + while (!kill) { + try { + // Generate a reasonably sized resultset to try to slow the backend + // down a bit, so we have a chance of the cancel actually hitting us. + ResultSet rs = stmt.executeQuery("select " + testValue + " from pg_type t1, pg_type t2"); + int count = 0; + while (rs.next()) { + assertEquals(testValue, rs.getInt(1)); + ++count; + } + assertTrue(count > 0); + rs.close(); + } catch (SQLException e) { + //e.printStackTrace(); + if (!ignoreErrors) { + this.error = e; + break; + } + } catch (Throwable t) { + this.error = t; + break; + } + } + + synchronized (this) { + stopped = true; + notifyAll(); + } + } + + public void stop() throws Exception { + kill = true; + synchronized (this) { + while (!stopped) { + try { this.wait(); } + catch (InterruptedException e) {} + } + } + + if (error != null) { + if (error instanceof Exception) + throw (Exception)error; + else + throw (Error)error; + } + } + + private final Statement stmt; + private final boolean ignoreErrors; + private final int testValue; + + private Throwable error; + private volatile boolean kill = false; + private boolean stopped = false; + } + + public void testThreadedCancel() throws Exception { + Connection con = TestUtil.openDB(); + Statement stmt1 = con.createStatement(); + QueryRunnable qr1 = new QueryRunnable(stmt1, true, 1); // ignore errors; we will repeatedly cancel this one + Statement stmt2 = con.createStatement(); + QueryRunnable qr2 = new QueryRunnable(stmt2, false, 2); // report errors; it should never be cancelled. + + new Thread(qr1).start(); + new Thread(qr2).start(); + + Thread.sleep(1000); // Give them time to start. + + for (int i = 0; i < 500; ++i) { + stmt1.cancel(); + Thread.sleep(10); + } + + qr1.stop(); + qr2.stop(); + } }