Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add support for ignore lock wait timeout #259

Merged
merged 1 commit into from
Mar 13, 2024
Merged

Add support for ignore lock wait timeout #259

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions r2dbc-mysql/src/main/java/io/asyncer/r2dbc/mysql/ConnectionContext.java
View file
Open in desktop
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
Open in desktop
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
Loading