Skip to content

H
h2database
提交 27252c09 authored 作者: Thomas Mueller's avatar Thomas Mueller
浏览文件
操作

More bugs in the server-less multi-connection mode have been fixed.

上级 1d4b9c7a
显示空白字符变更
内嵌 并排
正在显示 包含 323 行增加 和 122 行删除
h2/src/main/org/h2/constant/SysProperties.java
浏览文件 @ 27252c09
......@@ -463,13 +463,13 @@ public class SysProperties {
public static final boolean RECOMPILE_ALWAYS = getBooleanSetting("h2.recompileAlways", false);
/**
* System property <code>h2.reconnectCheckDelay</code> (default: 250).<br />
* System property <code>h2.reconnectCheckDelay</code> (default: 100).<br />
* Check the .lock.db file every this many milliseconds to detect that the
* database was changed. The process writing to the database must first
* notify a change in the .lock.db file, then wait twice this many
* milliseconds before updating the database.
*/
public static final int RECONNECT_CHECK_DELAY = getIntSetting("h2.reconnectCheckDelay", 250);
public static final int RECONNECT_CHECK_DELAY = getIntSetting("h2.reconnectCheckDelay", 100);
/**
* System property <code>h2.redoBufferSize</code> (default: 262144).<br />
......
h2/src/main/org/h2/engine/Database.java
浏览文件 @ 27252c09
......@@ -22,6 +22,7 @@ import org.h2.command.dml.SetTypes;
import org.h2.constant.ErrorCode;
import org.h2.constant.SysProperties;
import org.h2.constraint.Constraint;
import org.h2.index.BtreeIndex;
import org.h2.index.Cursor;
import org.h2.index.Index;
import org.h2.index.IndexType;
......@@ -310,32 +311,61 @@ public class Database implements DataHandler {
return modificationDataId;
}
private void reconnectModified(boolean pending) {
if (readOnly || lock == null) {
return;
/**
* Set or reset the pending change flag in the .lock.db file.
*
* @param pending the new value of the flag
* @return true if the call was successful,
* false if another connection was faster
*/
synchronized boolean reconnectModified(boolean pending) {
if (readOnly || lock == null || fileLockMethod != FileLock.LOCK_SERIALIZED) {
return true;
}
try {
if (pending == reconnectChangePending) {
long now = System.currentTimeMillis();
if (now > reconnectCheckNext) {
if (pending) {
String pos = log == null ? null : log.getWritePos();
lock.setProperty("logPos", pos);
lock.save();
}
reconnectCheckNext = now + SysProperties.RECONNECT_CHECK_DELAY;
}
return;
return true;
}
Properties old = lock.load();
if (pending) {
getTrace().debug("wait before writing");
Thread.sleep((int) (SysProperties.RECONNECT_CHECK_DELAY * 1.1));
Properties now = lock.load();
if (!now.equals(old)) {
// somebody else was faster
return false;
}
lock.setProperty("modificationDataId", Long.toString(modificationDataId));
lock.setProperty("modificationMetaId", Long.toString(modificationMetaId));
}
String pos = log == null ? null : log.getWritePos();
lock.setProperty("logPos", pos);
lock.setProperty("changePending", pending ? "true" : null);
lock.save();
reconnectLastLock = lock.load();
old = lock.save();
if (pending) {
getTrace().debug("wait before writing again");
Thread.sleep((int) (SysProperties.RECONNECT_CHECK_DELAY * 1.1));
Properties now = lock.load();
if (!now.equals(old)) {
// somebody else was faster
return false;
}
}
reconnectLastLock = old;
reconnectChangePending = pending;
reconnectCheckNext = System.currentTimeMillis() + SysProperties.RECONNECT_CHECK_DELAY;
return true;
} catch (Exception e) {
getTrace().error("pending:"+ pending, e);
return false;
}
}
......@@ -533,7 +563,9 @@ public class Database implements DataHandler {
}
// wait until pending changes are written
isReconnectNeeded();
beforeWriting();
while (!beforeWriting()) {
// until we can write (file are not open - no need to re-connect)
}
if (SysProperties.PAGE_STORE) {
starting = true;
getPageStore();
......@@ -635,7 +667,6 @@ public class Database implements DataHandler {
}
systemSession.commit(true);
traceSystem.getTrace(Trace.DATABASE).info("opened " + databaseName);
afterWriting();
}
public Schema getMainSchema() {
......@@ -1055,9 +1086,11 @@ public class Database implements DataHandler {
if (closing) {
return;
}
if (isReconnectNeeded()) {
// another connection wrote - don't write anything
if (fileLockMethod == FileLock.LOCK_SERIALIZED && !reconnectChangePending) {
// another connection may have written something - don't write
try {
// make sure the log doesn't try to write
log.setReadOnly(true);
closeOpenFilesAndUnlock(false);
} catch (SQLException e) {
// ignore
......@@ -1194,6 +1227,7 @@ public class Database implements DataHandler {
pageStore.checkpoint();
}
}
reconnectModified(false);
closeFiles();
if (persistent && lock == null && fileLockMethod != FileLock.LOCK_NO) {
// everything already closed (maybe in checkPowerOff)
......@@ -1736,6 +1770,11 @@ public class Database implements DataHandler {
if (noDiskSpace) {
throw Message.getSQLException(ErrorCode.NO_DISK_SPACE_AVAILABLE);
}
if (fileLockMethod == FileLock.LOCK_SERIALIZED) {
if (!reconnectChangePending) {
throw Message.getSQLException(ErrorCode.DATABASE_IS_READ_ONLY);
}
}
}
public boolean getReadOnly() {
......@@ -2183,10 +2222,21 @@ public class Database implements DataHandler {
return null;
}
/**
* Check if the contents of the database was changed and therefore it is
* required to re-connect. This method waits until pending changes are
* completed. If a pending change takes too long (more than 2 seconds), the
* pending change is broken.
*
* @return true if reconnecting is required
*/
public boolean isReconnectNeeded() {
if (fileLockMethod != FileLock.LOCK_SERIALIZED) {
return false;
}
if (reconnectChangePending) {
return false;
}
long now = System.currentTimeMillis();
if (now < reconnectCheckNext) {
return false;
......@@ -2195,25 +2245,27 @@ public class Database implements DataHandler {
if (lock == null) {
lock = new FileLock(traceSystem, databaseName + Constants.SUFFIX_LOCK_FILE, Constants.LOCK_SLEEP);
}
Properties prop;
try {
Properties prop = lock.load(), first = prop;
while (true) {
prop = lock.load();
if (prop.equals(reconnectLastLock)) {
return false;
}
if (prop.getProperty("changePending", null) == null) {
break;
}
if (System.currentTimeMillis() > now + SysProperties.RECONNECT_CHECK_DELAY * 4) {
if (System.currentTimeMillis() > now + SysProperties.RECONNECT_CHECK_DELAY * 10) {
if (first.equals(prop)) {
// the writing process didn't update the file -
// it may have terminated
lock.setProperty("changePending", null);
lock.save();
break;
}
}
getTrace().debug("delay (change pending)");
Thread.sleep(SysProperties.RECONNECT_CHECK_DELAY);
prop = lock.load();
}
reconnectLastLock = prop;
} catch (Exception e) {
......@@ -2223,32 +2275,56 @@ public class Database implements DataHandler {
return true;
}
/**
* This method is called after writing to the database.
*/
public void afterWriting() throws SQLException {
if (fileLockMethod != FileLock.LOCK_SERIALIZED || readOnly) {
return;
}
reconnectCheckNext = System.currentTimeMillis() + 1;
}
/**
* Flush all changes when using the serialized mode, and if there are
* pending changes.
* pending changes, and some time has passed. This switches to a new
* transaction log and resets the change pending flag in
* the .lock.db file.
*/
public void checkpointIfRequired() throws SQLException {
if (fileLockMethod != FileLock.LOCK_SERIALIZED || readOnly || !reconnectChangePending) {
return;
}
long now = System.currentTimeMillis();
if (now > reconnectCheckNext) {
if (now > reconnectCheckNext + SysProperties.RECONNECT_CHECK_DELAY) {
getTrace().debug("checkpoint");
flushIndexes(0);
checkpoint();
reconnectModified(false);
}
}
/**
* Flush the indexes that were last changed prior to some time.
*
* @param maxLastChange indexes that were changed
* afterwards are not flushed; 0 to flush all indexes
*/
public synchronized void flushIndexes(long maxLastChange) {
ObjectArray array = getAllSchemaObjects(DbObject.INDEX);
for (int i = 0; i < array.size(); i++) {
DbObject obj = (DbObject) array.get(i);
if (obj instanceof BtreeIndex) {
BtreeIndex idx = (BtreeIndex) obj;
if (idx.getLastChange() == 0) {
continue;
}
Table tab = idx.getTable();
if (tab.isLockedExclusively()) {
continue;
}
if (maxLastChange != 0 && idx.getLastChange() > maxLastChange) {
continue;
}
try {
idx.flush(systemSession);
} catch (SQLException e) {
getTrace().error("flush index " + idx.getName(), e);
}
}
}
}
/**
* Flush all changes and open a new log file.
*/
......@@ -2262,11 +2338,15 @@ public class Database implements DataHandler {
/**
* This method is called before writing to the log file.
*
* @return true if the call was successful,
* false if another connection was faster
*/
public void beforeWriting() {
public boolean beforeWriting() {
if (fileLockMethod == FileLock.LOCK_SERIALIZED) {
reconnectModified(true);
return reconnectModified(true);
}
return true;
}
/**
......
h2/src/main/org/h2/engine/Session.java
浏览文件 @ 27252c09
......@@ -634,7 +634,6 @@ public class Session extends SessionWithState {
Message.throwInternalError();
}
}
database.afterWriting();
if (locks.size() > 0) {
synchronized (database) {
for (int i = 0; i < locks.size(); i++) {
......@@ -1112,8 +1111,20 @@ public class Session extends SessionWithState {
return modificationId;
}
public boolean isReconnectNeeded() {
return database.isReconnectNeeded();
public boolean isReconnectNeeded(boolean write) {
while (true) {
boolean reconnect = database.isReconnectNeeded();
if (reconnect) {
return true;
}
if (write) {
if (database.beforeWriting()) {
return false;
}
} else {
return false;
}
}
}
public SessionInterface reconnect() throws SQLException {
......
h2/src/main/org/h2/engine/SessionInterface.java
浏览文件 @ 27252c09
......@@ -76,9 +76,10 @@ public interface SessionInterface {
/**
* Check if the database changed and therefore reconnecting is required.
*
* @param write if the next operation may be writing
* @return true if reconnecting is required
*/
boolean isReconnectNeeded();
boolean isReconnectNeeded(boolean write);
/**
* Close the connection and open a new connection.
......
h2/src/main/org/h2/engine/SessionRemote.java
浏览文件 @ 27252c09
......@@ -652,7 +652,7 @@ public class SessionRemote extends SessionWithState implements SessionFactory, D
return TempFileDeleter.getInstance();
}
public boolean isReconnectNeeded() {
public boolean isReconnectNeeded(boolean write) {
return false;
}
......
h2/src/main/org/h2/jdbc/JdbcConnection.java
浏览文件 @ 27252c09
......@@ -400,7 +400,7 @@ public class JdbcConnection extends TraceObject implements Connection {
public synchronized void commit() throws SQLException {
try {
debugCodeCall("commit");
checkClosed();
checkClosedForWrite();
commit = prepareCommand("COMMIT", commit);
commit.executeUpdate();
} catch (Exception e) {
......@@ -418,7 +418,7 @@ public class JdbcConnection extends TraceObject implements Connection {
public synchronized void rollback() throws SQLException {
try {
debugCodeCall("rollback");
checkClosed();
checkClosedForWrite();
rollbackInternal();
} catch (Exception e) {
throw logAndConvert(e);
......@@ -641,6 +641,8 @@ public class JdbcConnection extends TraceObject implements Connection {
*/
public void setQueryTimeout(int seconds) throws SQLException {
try {
debugCodeCall("setQueryTimeout", seconds);
checkClosed();
setQueryTimeout = prepareCommand("SET QUERY_TIMEOUT ?", setQueryTimeout);
((ParameterInterface) setQueryTimeout.getParameters().get(0)).setValue(ValueInt.get(seconds * 1000), false);
setQueryTimeout.executeUpdate();
......@@ -654,6 +656,8 @@ public class JdbcConnection extends TraceObject implements Connection {
*/
public int getQueryTimeout() throws SQLException {
try {
debugCodeCall("getQueryTimeout");
checkClosed();
getQueryTimeout = prepareCommand("SELECT VALUE FROM INFORMATION_SCHEMA.SETTINGS WHERE NAME=?", getQueryTimeout);
((ParameterInterface) getQueryTimeout.getParameters().get(0)).setValue(ValueString.get("QUERY_TIMEOUT"), false);
ResultInterface result = getQueryTimeout.executeQuery(0, false);
......@@ -668,7 +672,6 @@ public class JdbcConnection extends TraceObject implements Connection {
} catch (Exception e) {
throw logAndConvert(e);
}
}
/**
......@@ -903,7 +906,7 @@ public class JdbcConnection extends TraceObject implements Connection {
try {
JdbcSavepoint sp = convertSavepoint(savepoint);
debugCode("rollback(" + sp.getTraceObjectName() + ");");
checkClosed();
checkClosedForWrite();
sp.rollback();
} catch (Exception e) {
throw logAndConvert(e);
......@@ -1249,20 +1252,43 @@ public class JdbcConnection extends TraceObject implements Connection {
}
/**
* INTERNAL
* Check if this connection is closed.
* The next operation is a read request.
*
* @return true if the session was re-connected
* @throws SQLException if the connection or session is closed
*/
protected boolean checkClosed() throws SQLException {
return checkClosed(false);
}
/**
* Check if this connection is closed.
* The next operation may be a write request.
*
* @return true if the session was re-connected
* @throws SQLException if the connection or session is closed
*/
private boolean checkClosedForWrite() throws SQLException {
return checkClosed(true);
}
/**
* INTERNAL
* Check if this connection is closed.
*
* @param write if the next operation is possibly writing
* @return true if the session was re-connected
* @throws SQLException if the connection or session is closed
*/
protected boolean checkClosed(boolean write) throws SQLException {
if (session == null) {
throw Message.getSQLException(ErrorCode.OBJECT_CLOSED);
}
if (session.isClosed()) {
throw Message.getSQLException(ErrorCode.DATABASE_CALLED_AT_SHUTDOWN);
}
if (session.isReconnectNeeded()) {
if (session.isReconnectNeeded(write)) {
trace.debug("reconnect");
session = session.reconnect();
setTrace(session.getTrace());
......@@ -1340,7 +1366,7 @@ public class JdbcConnection extends TraceObject implements Connection {
try {
int id = getNextId(TraceObject.CLOB);
debugCodeAssign("Clob", TraceObject.CLOB, id, "createClob()");
checkClosed();
checkClosedForWrite();
ValueLob v = ValueLob.createSmallLob(Value.CLOB, new byte[0]);
return new JdbcClob(this, v, id);
} catch (Exception e) {
......@@ -1357,7 +1383,7 @@ public class JdbcConnection extends TraceObject implements Connection {
try {
int id = getNextId(TraceObject.BLOB);
debugCodeAssign("Blob", TraceObject.BLOB, id, "createClob()");
checkClosed();
checkClosedForWrite();
ValueLob v = ValueLob.createSmallLob(Value.BLOB, new byte[0]);
return new JdbcBlob(this, v, id);
} catch (Exception e) {
......@@ -1375,7 +1401,7 @@ public class JdbcConnection extends TraceObject implements Connection {
try {
int id = getNextId(TraceObject.CLOB);
debugCodeAssign("NClob", TraceObject.CLOB, id, "createNClob()");
checkClosed();
checkClosedForWrite();
ValueLob v = ValueLob.createSmallLob(Value.CLOB, new byte[0]);
return new JdbcClob(this, v, id);
} catch (Exception e) {
......
h2/src/main/org/h2/jdbc/JdbcPreparedStatement.java
浏览文件 @ 27252c09
......@@ -124,7 +124,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
public int executeUpdate() throws SQLException {
try {
debugCodeCall("executeUpdate");
checkClosed();
checkClosedForWrite();
return executeUpdateInternal();
} catch (Exception e) {
throw logAndConvert(e);
......@@ -159,7 +159,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCodeCall("execute");
}
checkClosed();
checkClosedForWrite();
closeOldResultSet();
boolean returnsResultSet;
synchronized (conn.getSession()) {
......@@ -703,7 +703,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setBlob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v;
if (x == null) {
v = ValueNull.INSTANCE;
......@@ -728,7 +728,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setBlob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createBlob(x, -1);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -748,7 +748,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setClob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v;
if (x == null) {
v = ValueNull.INSTANCE;
......@@ -773,7 +773,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setClob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v;
if (x == null) {
v = ValueNull.INSTANCE;
......@@ -832,7 +832,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setBinaryStream("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createBlob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -888,7 +888,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setAsciiStream("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(IOUtils.getAsciiReader(x), length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -943,7 +943,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setCharacterStream("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1031,7 +1031,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
public int[] executeBatch() throws SQLException {
try {
debugCodeCall("executeBatch");
checkClosed();
checkClosedForWrite();
if (batchParameters == null) {
// TODO batch: check what other database do if no parameters are set
batchParameters = new ObjectArray();
......@@ -1081,7 +1081,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
public void addBatch() throws SQLException {
try {
debugCodeCall("addBatch");
checkClosed();
checkClosedForWrite();
ObjectArray parameters = command.getParameters();
Value[] set = new Value[parameters.size()];
for (int i = 0; i < parameters.size(); i++) {
......@@ -1271,7 +1271,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setNCharacterStream("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1303,7 +1303,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setNClob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v;
if (x == null) {
v = ValueNull.INSTANCE;
......@@ -1329,7 +1329,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setNClob("+parameterIndex+", x);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(x, -1);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1349,7 +1349,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setClob("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1369,7 +1369,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setBlob("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createBlob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1389,7 +1389,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
if (isDebugEnabled()) {
debugCode("setNClob("+parameterIndex+", x, "+length+"L);");
}
checkClosed();
checkClosedForWrite();
Value v = conn.createClob(x, length);
setParameter(parameterIndex, v);
} catch (Exception e) {
......@@ -1413,10 +1413,20 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
return getTraceObjectName() + ": " + command;
}
boolean checkClosed() throws SQLException {
if (super.checkClosed()) {
protected boolean checkClosed(boolean write) throws SQLException {
if (super.checkClosed(write)) {
// if the session was re-connected, re-prepare the statement
ObjectArray oldParams = command.getParameters();
command = conn.prepareCommand(sql, fetchSize);
ObjectArray newParams = command.getParameters();
for (int i = 0; i < oldParams.size(); i++) {
ParameterInterface old = (ParameterInterface) oldParams.get(i);
Value value = old.getParamValue();
if (value != null) {
ParameterInterface n = (ParameterInterface) newParams.get(i);
n.setValue(value, false);
}
}
return true;
}
return false;
......
h2/src/main/org/h2/jdbc/JdbcStatement.java
浏览文件 @ 27252c09
......@@ -105,7 +105,7 @@ public class JdbcStatement extends TraceObject implements Statement {
public int executeUpdate(String sql) throws SQLException {
try {
debugCodeCall("executeUpdate", sql);
checkClosed();
checkClosedForWrite();
closeOldResultSet();
if (escapeProcessing) {
sql = conn.translateSQL(sql);
......@@ -144,7 +144,7 @@ public class JdbcStatement extends TraceObject implements Statement {
if (isDebugEnabled()) {
debugCodeCall("execute", sql);
}
checkClosed();
checkClosedForWrite();
closeOldResultSet();
if (escapeProcessing) {
sql = conn.translateSQL(sql);
......@@ -610,7 +610,7 @@ public class JdbcStatement extends TraceObject implements Statement {
public int[] executeBatch() throws SQLException {
try {
debugCodeCall("executeBatch");
checkClosed();
checkClosedForWrite();
if (batchCommands == null) {
// TODO batch: check what other database do if no commands are set
batchCommands = new ObjectArray();
......@@ -843,23 +843,48 @@ public class JdbcStatement extends TraceObject implements Statement {
// =============================================================
/**
* Check if this connection is closed.
* The next operation is a read request.
*
* @return true if the session was re-connected
* @throws SQLException if the connection or session is closed
*/
boolean checkClosed() throws SQLException {
return checkClosed(false);
}
/**
* Check if this connection is closed.
* The next operation may be a write request.
*
* @return true if the session was re-connected
* @throws SQLException if the connection or session is closed
*/
boolean checkClosedForWrite() throws SQLException {
return checkClosed(true);
}
/**
* Check if the statement is closed.
*
* @param write if the next operation is possibly writing
* @return true if a reconnect was required
* @throws SQLException if it is closed
*/
boolean checkClosed() throws SQLException {
protected boolean checkClosed(boolean write) throws SQLException {
if (conn == null) {
throw Message.getSQLException(ErrorCode.OBJECT_CLOSED);
}
if (conn.checkClosed()) {
if (!conn.checkClosed(write)) {
return false;
}
do {
session = conn.getSession();
setTrace(session.getTrace());
} while (conn.checkClosed(write));
return true;
}
return false;
}
/**
* Close and old result set if there is still one open.
......
h2/src/main/org/h2/jdbcx/JdbcXAConnection.java
浏览文件 @ 27252c09
......@@ -494,11 +494,11 @@ implements XAConnection, XAResource
return isClosed || super.isClosed();
}
protected synchronized boolean checkClosed() throws SQLException {
protected synchronized boolean checkClosed(boolean write) throws SQLException {
if (isClosed) {
throw Message.getSQLException(ErrorCode.OBJECT_CLOSED);
}
return super.checkClosed();
return super.checkClosed(write);
}
}
......
h2/src/main/org/h2/log/LogSystem.java
浏览文件 @ 27252c09
......@@ -240,14 +240,16 @@ public class LogSystem {
/**
* Roll back any uncommitted transactions if required, and apply committed
* changed to the data files.
*
* @return if recovery was needed
*/
public void recover() throws SQLException {
public boolean recover() throws SQLException {
if (database == null) {
return;
return false;
}
synchronized (database) {
if (closed) {
return;
return false;
}
undo = new ObjectArray();
for (int i = 0; i < activeLogs.size(); i++) {
......@@ -282,7 +284,7 @@ public class LogSystem {
if (!readOnly && fileChanged && !containsInDoubtTransactions()) {
checkpoint();
}
return;
return fileChanged;
}
}
......@@ -514,7 +516,6 @@ public class LogSystem {
return;
}
database.checkWritingAllowed();
database.beforeWriting();
if (!file.isDataFile()) {
storageId = -storageId;
}
......@@ -541,7 +542,6 @@ public class LogSystem {
return;
}
database.checkWritingAllowed();
database.beforeWriting();
int storageId = record.getStorageId();
if (!file.isDataFile()) {
storageId = -storageId;
......@@ -569,6 +569,7 @@ public class LogSystem {
if (closed || disabled) {
return;
}
database.checkWritingAllowed();
flushAndCloseUnused();
currentLog = new LogFile(this, currentLog.getId() + 1, fileNamePrefix);
activeLogs.add(currentLog);
......@@ -668,7 +669,7 @@ public class LogSystem {
*
* @param readOnly the new value
*/
void setReadOnly(boolean readOnly) {
public void setReadOnly(boolean readOnly) {
this.readOnly = readOnly;
}
......@@ -706,4 +707,13 @@ public class LogSystem {
return accessMode;
}
/**
* Get the write position.
*
* @return the write position
*/
public String getWritePos() {
return currentLog.getId() + "/" + currentLog.getPos();
}
}
h2/src/main/org/h2/store/FileLock.java
浏览文件 @ 27252c09
......@@ -200,8 +200,10 @@ public class FileLock {
/**
* Save the lock file.
*
* @return the saved properties
*/
public void save() throws SQLException {
public Properties save() throws SQLException {
try {
OutputStream out = fs.openFileOutputStream(fileName, false);
try {
......@@ -213,6 +215,7 @@ public class FileLock {
if (trace.isDebugEnabled()) {
trace.debug("save " + properties);
}
return properties;
} catch (IOException e) {
throw getExceptionFatal("Could not save properties " + fileName, e);
}
......@@ -301,9 +304,21 @@ public class FileLock {
private void lockSerialized() throws SQLException {
method = SERIALIZED;
if (fs.createNewFile(fileName)) {
properties = new SortedProperties();
properties.setProperty("method", String.valueOf(method));
setUniqueId();
save();
} else {
while (true) {
try {
properties = load();
} catch (SQLException e) {
// ignore
}
return;
}
}
}
private void lockFile() throws SQLException {
......
h2/src/main/org/h2/store/WriterThread.java
浏览文件 @ 27252c09
......@@ -8,18 +8,13 @@ package org.h2.store;
import java.lang.ref.WeakReference;
import java.sql.SQLException;
import org.h2.constant.SysProperties;
import org.h2.engine.Constants;
import org.h2.engine.Database;
import org.h2.engine.DbObject;
import org.h2.index.BtreeIndex;
import org.h2.log.LogSystem;
import org.h2.message.Trace;
import org.h2.message.TraceSystem;
import org.h2.table.Table;
import org.h2.util.FileUtils;
import org.h2.util.ObjectArray;
/**
* The writer thread is responsible to flush the transaction log file from time
......@@ -91,35 +86,12 @@ public class WriterThread implements Runnable {
return log;
}
private void flushIndexes(Database database) {
private void flushIndexesIfRequired(Database database) {
long time = System.currentTimeMillis();
if (lastIndexFlush + Constants.FLUSH_INDEX_DELAY > time) {
return;
}
synchronized (database) {
ObjectArray array = database.getAllSchemaObjects(DbObject.INDEX);
for (int i = 0; i < array.size(); i++) {
DbObject obj = (DbObject) array.get(i);
if (obj instanceof BtreeIndex) {
BtreeIndex idx = (BtreeIndex) obj;
if (idx.getLastChange() == 0) {
continue;
}
Table tab = idx.getTable();
if (tab.isLockedExclusively()) {
continue;
}
if (idx.getLastChange() + Constants.FLUSH_INDEX_DELAY > time) {
continue;
}
try {
idx.flush(database.getSystemSession());
} catch (SQLException e) {
database.getTrace(Trace.DATABASE).error("flush index " + idx.getName(), e);
}
}
}
}
database.flushIndexes(time - Constants.FLUSH_INDEX_DELAY);
lastIndexFlush = time;
}
......@@ -141,7 +113,7 @@ public class WriterThread implements Runnable {
break;
}
if (Constants.FLUSH_INDEX_DELAY != 0) {
flushIndexes(database);
flushIndexesIfRequired(database);
}
// checkpoint if required
......
h2/src/test/org/h2/test/unit/TestFileLockSerialized.java
浏览文件 @ 27252c09
......@@ -34,6 +34,8 @@ public class TestFileLockSerialized extends TestBase {
public void test() throws Exception {
Class.forName("org.h2.Driver");
testThreeMostlyReaders(true);
testThreeMostlyReaders(false);
testTwoReaders();
testTwoWriters();
testPendingWrite();
......@@ -41,6 +43,55 @@ public class TestFileLockSerialized extends TestBase {
testConcurrentReadWrite();
}
private void testThreeMostlyReaders(final boolean write) throws Exception {
deleteDb("fileLockSerialized");
String url = "jdbc:h2:" + baseDir + "/fileLockSerialized;FILE_LOCK=SERIALIZED;OPEN_NEW=TRUE";
int len = 3;
final Exception[] ex = new Exception[1];
final Connection[] conn = new Connection[len];
final boolean[] stop = new boolean[1];
Thread[] threads = new Thread[len];
for (int i = 0; i < len; i++) {
final Connection c = DriverManager.getConnection(url);
conn[i] = c;
if (i == 0) {
conn[i].createStatement().execute("create table test(id int) as select 1");
}
Thread t = new Thread(new Runnable() {
public void run() {
try {
PreparedStatement p = c.prepareStatement("select * from test where id = ?");
while (!stop[0]) {
if (write) {
if (Math.random() > 0.9) {
c.createStatement().execute("update test set id = id");
}
}
p.setInt(1, 1);
Thread.sleep(10);
p.executeQuery();
p.clearParameters();
}
c.close();
} catch (Exception e) {
ex[0] = e;
}
}
});
t.start();
threads[i] = t;
}
Thread.sleep(1000);
stop[0] = true;
for (int i = 0; i < len; i++) {
threads[i].join();
}
if (ex[0] != null) {
throw ex[0];
}
DriverManager.getConnection(url).close();
}
private void testTwoReaders() throws Exception {
deleteDb("fileLockSerialized");
String url = "jdbc:h2:" + baseDir + "/fileLockSerialized;FILE_LOCK=SERIALIZED;OPEN_NEW=TRUE";
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论