Skip to content

Commit

Permalink
Add support for ignore lock wait timeout
Browse files Browse the repository at this point in the history
  • Loading branch information
@mirromutth
mirromutth committed Mar 12, 2024
1 parent 9cd2417 commit b8bc0f5
Show file tree
Hide file tree
Showing 3 changed files with 121 additions and 69 deletions.
18 changes: 18 additions & 0 deletions r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/ConnectionContext.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ public final class ConnectionContext implements CodecContext {
@Nullable
private ZoneId timeZone;

private boolean lockWaitTimeoutSupported = false;

/**
* Assume that the auto commit is always turned on, it will be set after handshake V10 request message, or
* OK message which means handshake V9 completed.
Expand Down Expand Up @@ -162,6 +164,22 @@ public int getLocalInfileBufferSize() {
return localInfileBufferSize;
}

/**
* Checks if the server supports lock wait timeout.
*
* @return if the server supports lock wait timeout.
*/
public boolean isLockWaitTimeoutSupported() {
return lockWaitTimeoutSupported;
}

/**
* Enables lock wait timeout supported when loading session variables.
*/
public void enableLockWaitTimeoutSupported() {
this.lockWaitTimeoutSupported = true;
}

/**
* Get the bitmap of server statuses.
*
Expand Down
82 changes: 53 additions & 29 deletions r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/MySqlSimpleConnection.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@ public Mono<Void> beginTransaction() {

@Override
public Mono<Void> beginTransaction(TransactionDefinition definition) {
return Mono.defer(() -> QueryFlow.beginTransaction(client, this, batchSupported, definition));
return Mono.defer(() -> QueryFlow.beginTransaction(client, this, batchSupported, definition, context));
}

@Override
Expand Down Expand Up @@ -306,8 +306,8 @@ public MySqlConnectionMetadata getMetadata() {
}

/**
* MySQL does not have any way to query the isolation level of the current transaction, only inferred from
* past statements, so driver can not make sure the result is right.
* MySQL does not have any way to query the isolation level of the current transaction, only inferred from past
* statements, so driver can not make sure the result is right.
* <p>
* See <a href="https://bugs.mysql.com/bug.php?id=53341">MySQL Bug 53341</a>
* <p>
Expand Down Expand Up @@ -424,6 +424,11 @@ public boolean isInTransaction() {
public Mono<Void> setLockWaitTimeout(Duration timeout) {
requireNonNull(timeout, "timeout must not be null");

if (!context.isLockWaitTimeoutSupported()) {
logger.warn("Lock wait timeout is not supported by server, setLockWaitTimeout operation is ignored");
return Mono.empty();
}

long timeoutSeconds = timeout.getSeconds();
return QueryFlow.executeVoid(client, "SET innodb_lock_wait_timeout=" + timeoutSeconds)
.doOnSuccess(ignored -> this.lockWaitTimeout = this.currentLockWaitTimeout = timeoutSeconds);
Expand Down Expand Up @@ -484,13 +489,20 @@ static Mono<MySqlConnection> init(
) {
Mono<MySqlConnection> connection = initSessionVariables(client, sessionVariables)
.then(loadSessionVariables(client, codecs, context))
.flatMap(data -> loadInnoDbEngineStatus(data, client, codecs, context))
.map(data -> {
ZoneId timeZone = data.timeZone;
if (timeZone != null) {
logger.debug("Got server time zone {} from loading session variables", timeZone);
context.setTimeZone(timeZone);
}

if (data.lockWaitTimeoutSupported) {
context.enableLockWaitTimeoutSupported();
} else {
logger.info("Lock wait timeout is not supported by server, all related operations will be ignored");
}

return new MySqlSimpleConnection(client, context, codecs, data.level, data.lockWaitTimeout,
queryCache, prepareCache, data.product, prepare);
});
Expand Down Expand Up @@ -534,10 +546,10 @@ private static Mono<Void> initSessionVariables(Client client, List<String> sessi
private static Mono<SessionData> loadSessionVariables(
Client client, Codecs codecs, ConnectionContext context
) {
StringBuilder query = new StringBuilder(160)
StringBuilder query = new StringBuilder(128)
.append("SELECT ")
.append(transactionIsolationColumn(context))
.append(",@@innodb_lock_wait_timeout AS l,@@version_comment AS v");
.append(",@@version_comment AS v");

Function<MySqlResult, Flux<SessionData>> handler;

Expand All @@ -554,6 +566,24 @@ private static Mono<SessionData> loadSessionVariables(
.last();
}

private static Mono<SessionData> loadInnoDbEngineStatus(
SessionData data, Client client, Codecs codecs, ConnectionContext context
) {
return new TextSimpleStatement(client, codecs, context,
"SHOW VARIABLES LIKE 'innodb\\\\_lock\\\\_wait\\\\_timeout'")
.execute()
.flatMap(r -> r.map(readable -> {
String value = readable.get(1, String.class);

if (value == null || value.isEmpty()) {
return data;
} else {
return data.lockWaitTimeout(Long.parseLong(value));
}
}))
.single(data);
}

private static Mono<Void> initDatabase(Client client, String database) {
return client.exchange(new InitDbMessage(database), INIT_DB)
.last()
Expand All @@ -572,16 +602,15 @@ private static Mono<Void> initDatabase(Client client, String database) {
private static Flux<SessionData> convertSessionData(MySqlResult r, boolean timeZone) {
return r.map(readable -> {
IsolationLevel level = convertIsolationLevel(readable.get(0, String.class));
long lockWaitTimeout = convertLockWaitTimeout(readable.get(1, Long.class));
String product = readable.get(2, String.class);
String product = readable.get(1, String.class);

return new SessionData(level, lockWaitTimeout, product, timeZone ? readZoneId(readable) : null);
return new SessionData(level, product, timeZone ? readZoneId(readable) : null);
});
}

private static ZoneId readZoneId(Readable readable) {
String systemTimeZone = readable.get(3, String.class);
String timeZone = readable.get(4, String.class);
String systemTimeZone = readable.get(2, String.class);
String timeZone = readable.get(3, String.class);

if (timeZone == null || timeZone.isEmpty() || "SYSTEM".equalsIgnoreCase(timeZone)) {
if (systemTimeZone == null || systemTimeZone.isEmpty()) {
Expand Down Expand Up @@ -628,24 +657,13 @@ private static IsolationLevel convertIsolationLevel(@Nullable String name) {
return IsolationLevel.REPEATABLE_READ;
}

private static long convertLockWaitTimeout(@Nullable Long timeout) {
if (timeout == null) {
logger.error("Lock wait timeout is null, fallback to " + DEFAULT_LOCK_WAIT_TIMEOUT + " seconds");

return DEFAULT_LOCK_WAIT_TIMEOUT;
}

return timeout;
}

/**
* Resolves the column of session isolation level, the {@literal @@tx_isolation} has been marked as
* deprecated.
* Resolves the column of session isolation level, the {@literal @@tx_isolation} has been marked as deprecated.
* <p>
* If server is MariaDB, {@literal @@transaction_isolation} is used starting from {@literal 11.1.1}.
* <p>
* If the server is MySQL, use {@literal @@transaction_isolation} starting from {@literal 8.0.3}, or
* between {@literal 5.7.20} and {@literal 8.0.0} (exclusive).
* If the server is MySQL, use {@literal @@transaction_isolation} starting from {@literal 8.0.3}, or between
* {@literal 5.7.20} and {@literal 8.0.0} (exclusive).
*/
private static String transactionIsolationColumn(ConnectionContext context) {
ServerVersion version = context.getServerVersion();
Expand All @@ -664,20 +682,26 @@ private static final class SessionData {

private final IsolationLevel level;

private final long lockWaitTimeout;

@Nullable
private final String product;

@Nullable
private final ZoneId timeZone;

private SessionData(IsolationLevel level, long lockWaitTimeout, @Nullable String product,
@Nullable ZoneId timeZone) {
private long lockWaitTimeout = -1;

private boolean lockWaitTimeoutSupported;

private SessionData(IsolationLevel level, @Nullable String product, @Nullable ZoneId timeZone) {
this.level = level;
this.lockWaitTimeout = lockWaitTimeout;
this.product = product;
this.timeZone = timeZone;
}

SessionData lockWaitTimeout(long timeout) {
this.lockWaitTimeoutSupported = true;
this.lockWaitTimeout = timeout;
return this;
}
}
}
Loading

0 comments on commit b8bc0f5

Please sign in to comment.