Changeset: e5beb233c692 for MonetDB URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=e5beb233c692 Modified Files: java/ChangeLog.Dec2011 java/src/nl/cwi/monetdb/jdbc/MonetConnection.java Branch: Dec2011 Log Message:
MonetConnection: serialise access to the connection to the server It seems at some point in time we destroyed the locks around the "server" connection object that was meant to make sure only one thread at a time would do a question-response sequence. Since the lock around the code in the main loop was missing, race-conditions could take please, which lead to unexplainable and odd errors in case of multithreaded use of a single connection. diffs (truncated from 432 to 300 lines): diff --git a/java/ChangeLog.Dec2011 b/java/ChangeLog.Dec2011 --- a/java/ChangeLog.Dec2011 +++ b/java/ChangeLog.Dec2011 @@ -1,6 +1,11 @@ # ChangeLog file for java # This file is updated with Maddlog +* Sat Dec 10 2011 Fabian Groffen <[email protected]> +- Fixed a bug where closing ResultSets and PreparedStatements could lead + to errors on concurrent running queries using the same Connection due + to a race condition. + * Thu Dec 8 2011 Fabian Groffen <[email protected]> - Changed version scheme of JDBC driver and MCL jar to be more standard, from monetdb-X.Y-<thing>.jar to monetdb-<thing>-X.Y.jar, bug #2943 diff --git a/java/src/nl/cwi/monetdb/jdbc/MonetConnection.java b/java/src/nl/cwi/monetdb/jdbc/MonetConnection.java --- a/java/src/nl/cwi/monetdb/jdbc/MonetConnection.java +++ b/java/src/nl/cwi/monetdb/jdbc/MonetConnection.java @@ -2114,216 +2114,218 @@ public class MonetConnection extends Mon String error = null; try { - // make sure we're ready to send query; read data till we - // have the prompt it is possible (and most likely) that we - // already have the prompt and do not have to skip any - // lines. Ignore errors from previous result sets. - in.waitForPrompt(); + synchronized (server) { + // make sure we're ready to send query; read data till we + // have the prompt it is possible (and most likely) that we + // already have the prompt and do not have to skip any + // lines. Ignore errors from previous result sets. + in.waitForPrompt(); - int size; - // {{{ set reply size - /** - * Change the reply size of the server. If the given - * value is the same as the current value known to use, - * then ignore this call. If it is set to 0 we get a - * prompt after the server sent it's header. - */ - size = cachesize == 0 ? DEF_FETCHSIZE : cachesize; - size = maxrows != 0 ? Math.min(maxrows, size) : size; - // don't do work if it's not needed - if (lang == LANG_SQL && size != curReplySize && templ != commandTempl) { - sendControlCommand("reply_size " + size); + int size; + // {{{ set reply size + /** + * Change the reply size of the server. If the given + * value is the same as the current value known to use, + * then ignore this call. If it is set to 0 we get a + * prompt after the server sent it's header. + */ + size = cachesize == 0 ? DEF_FETCHSIZE : cachesize; + size = maxrows != 0 ? Math.min(maxrows, size) : size; + // don't do work if it's not needed + if (lang == LANG_SQL && size != curReplySize && templ != commandTempl) { + sendControlCommand("reply_size " + size); - // store the reply size after a successful change - curReplySize = size; - } - // }}} set reply size + // store the reply size after a successful change + curReplySize = size; + } + // }}} set reply size - // If the query is larger than the TCP buffer size, use a - // special send thread to avoid deadlock with the server due - // to blocking behaviour when the buffer is full. Because - // the server will be writing back results to us, it will - // eventually block as well when its TCP buffer gets full, - // as we are blocking an not consuming from it. The result - // is a state where both client and server want to write, - // but block. - if (query.length() > MapiSocket.BLOCK) { - // get a reference to the send thread - if (sendThread == null) sendThread = new SendThread(out); - // tell it to do some work! - sendThread.runQuery(templ, query); - sendThreadInUse = true; - } else { - // this is a simple call, which is a lot cheaper and will - // always succeed for small queries. - out.writeLine( - (templ[0] == null ? "" : templ[0]) + - query + - (templ[1] == null ? "" : templ[1])); - } + // If the query is larger than the TCP buffer size, use a + // special send thread to avoid deadlock with the server due + // to blocking behaviour when the buffer is full. Because + // the server will be writing back results to us, it will + // eventually block as well when its TCP buffer gets full, + // as we are blocking an not consuming from it. The result + // is a state where both client and server want to write, + // but block. + if (query.length() > MapiSocket.BLOCK) { + // get a reference to the send thread + if (sendThread == null) sendThread = new SendThread(out); + // tell it to do some work! + sendThread.runQuery(templ, query); + sendThreadInUse = true; + } else { + // this is a simple call, which is a lot cheaper and will + // always succeed for small queries. + out.writeLine( + (templ[0] == null ? "" : templ[0]) + + query + + (templ[1] == null ? "" : templ[1])); + } - // go for new results - String tmpLine = in.readLine(); - int linetype = in.getLineType(); - Response res = null; - while (linetype != BufferedMCLReader.PROMPT) { - // each response should start with a start of header - // (or error) - switch (linetype) { - case BufferedMCLReader.SOHEADER: - // make the response object, and fill it - try { - switch (sohp.parse(tmpLine)) { - case StartOfHeaderParser.Q_PARSE: - throw new MCLParseException("Q_PARSE header not allowed here", 1); - case StartOfHeaderParser.Q_TABLE: - case StartOfHeaderParser.Q_PREPARE: { - int id = sohp.getNextAsInt(); - int tuplecount = sohp.getNextAsInt(); - int columncount = sohp.getNextAsInt(); - int rowcount = sohp.getNextAsInt(); - // enforce the maxrows setting - if (maxrows != 0 && tuplecount > maxrows) - tuplecount = maxrows; - res = new ResultSetResponse( - id, - tuplecount, - columncount, - rowcount, - this, - seqnr - ); - // only add this resultset to - // the hashmap if it can possibly - // have an additional datablock - if (rowcount < tuplecount) { - if (rsresponses == null) - rsresponses = new HashMap(); - rsresponses.put( - new Integer(id), - res + // go for new results + String tmpLine = in.readLine(); + int linetype = in.getLineType(); + Response res = null; + while (linetype != BufferedMCLReader.PROMPT) { + // each response should start with a start of header + // (or error) + switch (linetype) { + case BufferedMCLReader.SOHEADER: + // make the response object, and fill it + try { + switch (sohp.parse(tmpLine)) { + case StartOfHeaderParser.Q_PARSE: + throw new MCLParseException("Q_PARSE header not allowed here", 1); + case StartOfHeaderParser.Q_TABLE: + case StartOfHeaderParser.Q_PREPARE: { + int id = sohp.getNextAsInt(); + int tuplecount = sohp.getNextAsInt(); + int columncount = sohp.getNextAsInt(); + int rowcount = sohp.getNextAsInt(); + // enforce the maxrows setting + if (maxrows != 0 && tuplecount > maxrows) + tuplecount = maxrows; + res = new ResultSetResponse( + id, + tuplecount, + columncount, + rowcount, + this, + seqnr ); - } - } break; - case StartOfHeaderParser.Q_UPDATE: - res = new UpdateResponse( - sohp.getNextAsInt(), // count - sohp.getNextAsString() // key-id + // only add this resultset to + // the hashmap if it can possibly + // have an additional datablock + if (rowcount < tuplecount) { + if (rsresponses == null) + rsresponses = new HashMap(); + rsresponses.put( + new Integer(id), + res ); - break; - case StartOfHeaderParser.Q_SCHEMA: - res = new SchemaResponse(); - break; - case StartOfHeaderParser.Q_TRANS: - boolean ac = sohp.getNextAsString().equals("t") ? true : false; - if (autoCommit && ac) { - addWarning("Server enabled auto commit " + - "mode while local state " + - "already was auto commit." + } + } break; + case StartOfHeaderParser.Q_UPDATE: + res = new UpdateResponse( + sohp.getNextAsInt(), // count + sohp.getNextAsString() // key-id ); - } - autoCommit = ac; - res = new AutoCommitResponse(ac); - break; - case StartOfHeaderParser.Q_BLOCK: { - // a new block of results for a - // response... - int id = sohp.getNextAsInt(); - sohp.getNextAsInt(); // columncount - int rowcount = sohp.getNextAsInt(); - int offset = sohp.getNextAsInt(); - ResultSetResponse t = - (ResultSetResponse)rsresponses.get(new Integer(id)); - if (t == null) { - error = "no ResultSetResponse with id " + id + " found"; - break; - } + break; + case StartOfHeaderParser.Q_SCHEMA: + res = new SchemaResponse(); + break; + case StartOfHeaderParser.Q_TRANS: + boolean ac = sohp.getNextAsString().equals("t") ? true : false; + if (autoCommit && ac) { + addWarning("Server enabled auto commit " + + "mode while local state " + + "already was auto commit." + ); + } + autoCommit = ac; + res = new AutoCommitResponse(ac); + break; + case StartOfHeaderParser.Q_BLOCK: { + // a new block of results for a + // response... + int id = sohp.getNextAsInt(); + sohp.getNextAsInt(); // columncount + int rowcount = sohp.getNextAsInt(); + int offset = sohp.getNextAsInt(); + ResultSetResponse t = + (ResultSetResponse)rsresponses.get(new Integer(id)); + if (t == null) { + error = "no ResultSetResponse with id " + id + " found"; + break; + } - DataBlockResponse r = - new DataBlockResponse( - rowcount, // rowcount - t.getRSType() == ResultSet.TYPE_FORWARD_ONLY - ); + DataBlockResponse r = + new DataBlockResponse( + rowcount, // rowcount + t.getRSType() == ResultSet.TYPE_FORWARD_ONLY + ); - t.addDataBlockResponse(offset, r); - res = r; - } break; - } - } catch (MCLParseException e) { - error = "error while parsing start of header:\n" + - e.getMessage() + - " found: '" + tmpLine.charAt(e.getErrorOffset()) + "'" + - " in: \"" + tmpLine + "\"" + - " at pos: " + e.getErrorOffset(); - // flush all the rest - in.waitForPrompt(); - linetype = in.getLineType(); - break; - } - - // immediately handle errors after parsing - // the header (res may be null) - if (error != null) { - in.waitForPrompt(); - linetype = in.getLineType(); - break; - } - _______________________________________________ Checkin-list mailing list [email protected] http://mail.monetdb.org/mailman/listinfo/checkin-list
