forked from public-mirrors/HuskSync
Introduce new lockstep syncing system, modularize sync modes (#178)
* Start work on modular sync systems * Add experimental lockstep sync system, close #69 * Refactor RedisMessageType enum * Fixup lockstep syncing * Bump to 3.1 * Update docs with details about the new Sync Modes * Sync mode config key is `mode` instead of `type` * Add server to data snapshot overview * API: Add API for setting data syncers * Fixup weird statistic matching logicfeat/data-edit-commands
William
1 year ago
committed by
GitHub
parent
03ca335293
commit
cae17f6e68
@ -0,0 +1,77 @@
|
||||
/*
|
||||
* This file is part of HuskSync, licensed under the Apache License 2.0.
|
||||
*
|
||||
* Copyright (c) William278 <will27528@gmail.com>
|
||||
* Copyright (c) contributors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.william278.husksync.config;
|
||||
|
||||
import net.william278.annotaml.YamlFile;
|
||||
import net.william278.annotaml.YamlKey;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.nio.file.Path;
|
||||
|
||||
/**
|
||||
* Represents a server on a proxied network.
|
||||
*/
|
||||
@YamlFile(header = """
|
||||
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
|
||||
┃ HuskSync Server ID config ┃
|
||||
┃ Developed by William278 ┃
|
||||
┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
|
||||
┣╸ This file should contain the ID of this server as defined in your proxy config.
|
||||
┗╸ If you join it using /server alpha, then set it to 'alpha' (case-sensitive)""")
|
||||
public class Server {
|
||||
|
||||
/**
|
||||
* Default server identifier.
|
||||
*/
|
||||
@NotNull
|
||||
public static String getDefaultServerName() {
|
||||
try {
|
||||
final Path serverDirectory = Path.of(System.getProperty("user.dir"));
|
||||
return serverDirectory.getFileName().toString().trim();
|
||||
} catch (Exception e) {
|
||||
return "server";
|
||||
}
|
||||
}
|
||||
|
||||
@YamlKey("name")
|
||||
private String serverName = getDefaultServerName();
|
||||
|
||||
@SuppressWarnings("unused")
|
||||
private Server() {
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(@NotNull Object other) {
|
||||
// If the name of this server matches another, the servers are the same.
|
||||
if (other instanceof Server server) {
|
||||
return server.getName().equalsIgnoreCase(this.getName());
|
||||
}
|
||||
return super.equals(other);
|
||||
}
|
||||
|
||||
/**
|
||||
* Proxy-defined name of this server.
|
||||
*/
|
||||
@NotNull
|
||||
public String getName() {
|
||||
return serverName;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,50 +0,0 @@
|
||||
/*
|
||||
* This file is part of HuskSync, licensed under the Apache License 2.0.
|
||||
*
|
||||
* Copyright (c) William278 <will27528@gmail.com>
|
||||
* Copyright (c) contributors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.william278.husksync.redis;
|
||||
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.Locale;
|
||||
import java.util.Optional;
|
||||
|
||||
public enum RedisMessageType {
|
||||
|
||||
UPDATE_USER_DATA,
|
||||
REQUEST_USER_DATA,
|
||||
RETURN_USER_DATA;
|
||||
|
||||
@NotNull
|
||||
public String getMessageChannel(@NotNull String clusterId) {
|
||||
return String.format(
|
||||
"%s:%s:%s",
|
||||
RedisManager.KEY_NAMESPACE.toLowerCase(Locale.ENGLISH),
|
||||
clusterId.toLowerCase(Locale.ENGLISH),
|
||||
name().toLowerCase(Locale.ENGLISH)
|
||||
);
|
||||
}
|
||||
|
||||
public static Optional<RedisMessageType> getTypeFromChannel(@NotNull String channel, @NotNull String clusterId) {
|
||||
return Arrays.stream(values())
|
||||
.filter(messageType -> messageType.getMessageChannel(clusterId).equalsIgnoreCase(channel))
|
||||
.findFirst();
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,152 @@
|
||||
/*
|
||||
* This file is part of HuskSync, licensed under the Apache License 2.0.
|
||||
*
|
||||
* Copyright (c) William278 <will27528@gmail.com>
|
||||
* Copyright (c) contributors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.william278.husksync.sync;
|
||||
|
||||
import net.william278.husksync.HuskSync;
|
||||
import net.william278.husksync.api.HuskSyncAPI;
|
||||
import net.william278.husksync.data.DataSnapshot;
|
||||
import net.william278.husksync.user.OnlineUser;
|
||||
import net.william278.husksync.util.Task;
|
||||
import org.jetbrains.annotations.ApiStatus;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* Handles the synchronization of data when a player changes servers or logs in
|
||||
*
|
||||
* @since 3.1
|
||||
*/
|
||||
public abstract class DataSyncer {
|
||||
private static final long BASE_LISTEN_ATTEMPTS = 16;
|
||||
private static final long LISTEN_DELAY = 10;
|
||||
|
||||
protected final HuskSync plugin;
|
||||
private final long maxListenAttempts;
|
||||
|
||||
@ApiStatus.Internal
|
||||
protected DataSyncer(@NotNull HuskSync plugin) {
|
||||
this.plugin = plugin;
|
||||
this.maxListenAttempts = getMaxListenAttempts();
|
||||
}
|
||||
|
||||
/**
|
||||
* API-exposed constructor for a {@link DataSyncer}
|
||||
*
|
||||
* @param api instance of the {@link HuskSyncAPI}
|
||||
*/
|
||||
@SuppressWarnings("unused")
|
||||
public DataSyncer(@NotNull HuskSyncAPI api) {
|
||||
this(api.getPlugin());
|
||||
}
|
||||
|
||||
/**
|
||||
* Called when the plugin is enabled
|
||||
*/
|
||||
public void initialize() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Called when the plugin is disabled
|
||||
*/
|
||||
public void terminate() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Called when a user's data should be fetched and applied to them
|
||||
*
|
||||
* @param user the user to fetch data for
|
||||
*/
|
||||
public abstract void setUserData(@NotNull OnlineUser user);
|
||||
|
||||
/**
|
||||
* Called when a user's data should be serialized and saved
|
||||
*
|
||||
* @param user the user to save
|
||||
*/
|
||||
public abstract void saveUserData(@NotNull OnlineUser user);
|
||||
|
||||
// Calculates the max attempts the system should listen for user data for based on the latency value
|
||||
private long getMaxListenAttempts() {
|
||||
return BASE_LISTEN_ATTEMPTS + (
|
||||
(Math.max(100, plugin.getSettings().getNetworkLatencyMilliseconds()) / 1000) * 20 / LISTEN_DELAY
|
||||
);
|
||||
}
|
||||
|
||||
// Set a user's data from the database, or set them as a new user
|
||||
@ApiStatus.Internal
|
||||
protected void setUserFromDatabase(@NotNull OnlineUser user) {
|
||||
plugin.getDatabase().getLatestSnapshot(user).ifPresentOrElse(
|
||||
snapshot -> user.applySnapshot(snapshot, DataSnapshot.UpdateCause.SYNCHRONIZED),
|
||||
() -> user.completeSync(true, DataSnapshot.UpdateCause.NEW_USER, plugin)
|
||||
);
|
||||
}
|
||||
|
||||
// Continuously listen for data from Redis
|
||||
@ApiStatus.Internal
|
||||
protected void listenForRedisData(@NotNull OnlineUser user, @NotNull Supplier<Boolean> completionSupplier) {
|
||||
final AtomicLong timesRun = new AtomicLong(0L);
|
||||
final AtomicReference<Task.Repeating> task = new AtomicReference<>();
|
||||
final Runnable runnable = () -> {
|
||||
if (user.isOffline()) {
|
||||
task.get().cancel();
|
||||
return;
|
||||
}
|
||||
if (plugin.isDisabling() || timesRun.getAndIncrement() > maxListenAttempts) {
|
||||
task.get().cancel();
|
||||
setUserFromDatabase(user);
|
||||
return;
|
||||
}
|
||||
|
||||
if (completionSupplier.get()) {
|
||||
task.get().cancel();
|
||||
}
|
||||
};
|
||||
task.set(plugin.getRepeatingTask(runnable, LISTEN_DELAY));
|
||||
task.get().run();
|
||||
}
|
||||
|
||||
/**
|
||||
* Represents the different available default modes of {@link DataSyncer}
|
||||
*
|
||||
* @since 3.1
|
||||
*/
|
||||
public enum Mode {
|
||||
DELAY(DelayDataSyncer::new),
|
||||
LOCKSTEP(LockstepDataSyncer::new);
|
||||
|
||||
private final Function<HuskSync, ? extends DataSyncer> supplier;
|
||||
|
||||
Mode(@NotNull Function<HuskSync, ? extends DataSyncer> supplier) {
|
||||
this.supplier = supplier;
|
||||
}
|
||||
|
||||
@NotNull
|
||||
public DataSyncer create(@NotNull HuskSync plugin) {
|
||||
return supplier.apply(plugin);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
/*
|
||||
* This file is part of HuskSync, licensed under the Apache License 2.0.
|
||||
*
|
||||
* Copyright (c) William278 <will27528@gmail.com>
|
||||
* Copyright (c) contributors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.william278.husksync.sync;
|
||||
|
||||
import net.william278.husksync.HuskSync;
|
||||
import net.william278.husksync.data.DataSnapshot;
|
||||
import net.william278.husksync.user.OnlineUser;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
/**
|
||||
* A data syncer which applies a network delay before checking the presence of user data
|
||||
*/
|
||||
public class DelayDataSyncer extends DataSyncer {
|
||||
|
||||
public DelayDataSyncer(@NotNull HuskSync plugin) {
|
||||
super(plugin);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setUserData(@NotNull OnlineUser user) {
|
||||
plugin.runAsyncDelayed(
|
||||
() -> {
|
||||
// Fetch from the database if the user isn't changing servers
|
||||
if (!plugin.getRedisManager().getUserServerSwitch(user)) {
|
||||
this.setUserFromDatabase(user);
|
||||
return;
|
||||
}
|
||||
|
||||
// Listen for the data to be updated
|
||||
this.listenForRedisData(
|
||||
user,
|
||||
() -> plugin.getRedisManager().getUserData(user).map(data -> {
|
||||
user.applySnapshot(data, DataSnapshot.UpdateCause.SYNCHRONIZED);
|
||||
return true;
|
||||
}).orElse(false)
|
||||
);
|
||||
},
|
||||
Math.max(0, plugin.getSettings().getNetworkLatencyMilliseconds() / 50L)
|
||||
);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void saveUserData(@NotNull OnlineUser user) {
|
||||
plugin.runAsync(() -> {
|
||||
plugin.getRedisManager().setUserServerSwitch(user);
|
||||
final DataSnapshot.Packed data = user.createSnapshot(DataSnapshot.SaveCause.DISCONNECT);
|
||||
plugin.getRedisManager().setUserData(user, data);
|
||||
plugin.getDatabase().addSnapshot(user, data);
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,69 @@
|
||||
/*
|
||||
* This file is part of HuskSync, licensed under the Apache License 2.0.
|
||||
*
|
||||
* Copyright (c) William278 <will27528@gmail.com>
|
||||
* Copyright (c) contributors
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package net.william278.husksync.sync;
|
||||
|
||||
import net.william278.husksync.HuskSync;
|
||||
import net.william278.husksync.data.DataSnapshot;
|
||||
import net.william278.husksync.user.OnlineUser;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
|
||||
public class LockstepDataSyncer extends DataSyncer {
|
||||
|
||||
public LockstepDataSyncer(@NotNull HuskSync plugin) {
|
||||
super(plugin);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void initialize() {
|
||||
plugin.getRedisManager().clearUsersCheckedOutOnServer();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void terminate() {
|
||||
plugin.getRedisManager().clearUsersCheckedOutOnServer();
|
||||
}
|
||||
|
||||
// Consume their data when they are checked in
|
||||
@Override
|
||||
public void setUserData(@NotNull OnlineUser user) {
|
||||
this.listenForRedisData(user, () -> {
|
||||
if (plugin.getRedisManager().getUserCheckedOut(user).isEmpty()) {
|
||||
plugin.getRedisManager().setUserCheckedOut(user, true);
|
||||
plugin.getRedisManager().getUserData(user).ifPresentOrElse(
|
||||
data -> user.applySnapshot(data, DataSnapshot.UpdateCause.SYNCHRONIZED),
|
||||
() -> this.setUserFromDatabase(user)
|
||||
);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void saveUserData(@NotNull OnlineUser user) {
|
||||
plugin.runAsync(() -> {
|
||||
final DataSnapshot.Packed data = user.createSnapshot(DataSnapshot.SaveCause.DISCONNECT);
|
||||
plugin.getRedisManager().setUserData(user, data);
|
||||
plugin.getRedisManager().setUserCheckedOut(user, false);
|
||||
plugin.getDatabase().addSnapshot(user, data);
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,44 @@
|
||||
HuskSync offers two built-in **synchronization modes**. These sync modes change the way data is synced between servers. This page details the two sync modes available and how they work.
|
||||
|
||||
* The `DELAY` sync mode is the default sync mode, that use the `network_latency_miliseconds` value to apply a delay before listening to Redis data
|
||||
* The `LOCKSTEP` sync mode uses a data checkout system to ensure that all servers are in sync regardless of network latency or tick rate fluctuations. This mode was introduced in HuskSync v3.1
|
||||
|
||||
You can change which sync mode you are using by editing the `sync_mode` setting under `synchronization` in `config.yml`.
|
||||
|
||||
> **Warning:** Please note that you must use the same sync mode on all servers (at least within a cluster).
|
||||
|
||||
<details>
|
||||
<summary>Changing the sync mode (config.yml)</summary>
|
||||
|
||||
```yaml
|
||||
synchronization:
|
||||
# The mode of data synchronization to use (DELAY or LOCKSTEP). DELAY should be fine for most networks. Docs: https://william278.net/docs/husksync/sync-modes
|
||||
mode: DELAY
|
||||
```
|
||||
</details>
|
||||
|
||||
## Delay
|
||||
|
||||
|
||||
The `DELAY` sync mode works as described below:
|
||||
* When a user disconnects from a server, a `SERVER_SWITCH` key is immediately set on Redis, followed by a `DATA_UPDATE` key which contains the user's packed and serialized Data Snapshot.
|
||||
* When the user connects to a server, they are marked as locked (unable to break blocks, use containers, etc.)
|
||||
* The server asynchronously waits for the `network_latency_miliseconds` value (default: 500ms) to allow the source server time to serialize & set their key.
|
||||
* After waiting, the server checks for the `SERVER_SWITCH` key.
|
||||
* If present, it will continuously attempt to read for a `DATA_UPDATE` key; when read, their data will be set from the snapshot deserialized from Redis.
|
||||
* If not present, their data will be pulled from the database (as though they joined the network)
|
||||
|
||||
`DELAY` has been the default sync mode since HuskSync v2.0. In HuskSync v3.1, `LOCKSTEP` was introduced. Since the delay mode has been tested and deployed for the longest, it is still the default, though note this may change in the future.
|
||||
|
||||
However, if your network has a fluctuating tick rate or significant latency (especially if you have servers on different hardware/locations), you may wish to use `LOCKSTEP` instead for a more reliable sync system.
|
||||
|
||||
## Lockstep
|
||||
The `LOCKSTEP` sync mode works as described below:
|
||||
* When a user connects to a server, the server will continuously asynchronously check if a `DATA_CHECKOUT` key is present.
|
||||
* If, or when, the key is not present, the plugin will set a new `DATA_CHECKOUT` key.
|
||||
* After this, the plugin will check Redis for the presence of a `DATA_UPDATE` key.
|
||||
* If a `DATA_UPDATE` key is present, the user's data will be set from the snapshot deserialized from Redis contained within that key.
|
||||
* Otherwise, their data will be pulled from the database.
|
||||
* When a user disconnects from a server, their data is serialized and set to Redis with a `DATA_UPDATE` key. After this key has been set, the user's current `DATA_CHECKOUT` key will be removed from Redis.
|
||||
|
||||
Additionally, note that `DATA_CHECKOUT` keys are set with the server ID of the server which "checked out" the data (taken from the `server.yml` config file). On both shutdown and startup, the plugin will clear all `DATA_CHECKOUT` keys for the current server ID (to prevent stale keys in the event of a server crash for instance)
|
||||
Loading…
Reference in New Issue