Skip to content

Commit d93e72c

Browse files
Manwe03Manwe03
committed
- Handle transfer with neo events
- MSPT logger - Worker health
1 parent fc8ff3f commit d93e72c

15 files changed

Lines changed: 325 additions & 30 deletions

‎src/main/java/com/manwe/dsl/DistributedServerLevels.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
package com.manwe.dsl;
22

33
import com.manwe.dsl.config.DSLServerConfigs;
4+
import com.manwe.dsl.dedicatedServer.worker.LocalProfiler;
5+
import com.manwe.dsl.dedicatedServer.worker.MetricFileLogger;
46
import com.manwe.dsl.dedicatedServer.worker.WorkerBoundaryListener;
57
import org.slf4j.Logger;
68

@@ -48,6 +50,8 @@ public DistributedServerLevels(IEventBus modEventBus, ModContainer modContainer)
4850
// Register our mod's ModConfigSpec so that FML can create and load the config file for us
4951
modContainer.registerConfig(ModConfig.Type.COMMON, DSLServerConfigs.SPEC, "dsl-server.toml");
5052
NeoForge.EVENT_BUS.register(new WorkerBoundaryListener());
53+
NeoForge.EVENT_BUS.register(new LocalProfiler());
54+
NeoForge.EVENT_BUS.register(MetricFileLogger.class);
5155

5256
}
5357

‎src/main/java/com/manwe/dsl/config/DSLServerConfigs.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ public class DSLServerConfigs {
1616
public static final ModConfigSpec.ConfigValue<Integer> REGION_SIZE;
1717
public static final ModConfigSpec.ConfigValue<Integer> WORKER_SIZE;
1818

19+
public static final ModConfigSpec.ConfigValue<Boolean> LOG_TICK_TIME;
20+
1921
public static final ModConfigSpec.ConfigValue<Boolean> USE_ARBITER;
2022
public static final ModConfigSpec.ConfigValue<List<? extends String>> CONNECTIONS;
2123
public static final ModConfigSpec.ConfigValue<Integer> PORT;
@@ -24,6 +26,7 @@ public class DSLServerConfigs {
2426
BUILDER.push("Server");
2527
IS_PROXY = BUILDER.comment("Is this node a proxy? (only one can be the proxy)").define("is_proxy",true);
2628
WORKER_ID = BUILDER.comment("If false which worker id?").define("worker_id",1);
29+
LOG_TICK_TIME = BUILDER.comment("Log every tick duration of every worker").define("log_tick_time",false);
2730
BUILDER.pop();
2831

2932
BUILDER.push("Nodes").comment("This values need to be equal in all nodes");

‎src/main/java/com/manwe/dsl/connectionRouting/RegionRouter.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -155,7 +155,7 @@ public static int computeWorkerId(int blockX, int blockZ){
155155
* @param chunkZ chunk coordinates
156156
* @return The ID of the server allocated to this position
157157
*/
158-
private static int computeWorkerIdChunk(int chunkX, int chunkZ) {
158+
public static int computeWorkerIdChunk(int chunkX, int chunkZ) {
159159
int regionX = chunkX >> regionSize;
160160
int regionZ = chunkZ >> regionSize;
161161
return Math.min(Math.max((regionX < 0 ? (-regionX) - 1 : regionX), (regionZ < 0 ? (-regionZ) - 1 : regionZ)) + 1, nWorkers);

‎src/main/java/com/manwe/dsl/dedicatedServer/CustomDedicatedServer.java‎

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,9 +51,10 @@
5151
public class CustomDedicatedServer extends DedicatedServer {
5252

5353
URI arbiterUri = URI.create(DSLServerConfigs.ARBITER_ADDR.get());
54-
boolean isProxy = DSLServerConfigs.IS_PROXY.get();
55-
int workerSize = DSLServerConfigs.WORKER_SIZE.get();
56-
int workerId = DSLServerConfigs.WORKER_ID.get();
54+
private final boolean isProxy = DSLServerConfigs.IS_PROXY.get();
55+
private final int workerSize = DSLServerConfigs.WORKER_SIZE.get();
56+
public int workerId = DSLServerConfigs.WORKER_ID.get();
57+
public boolean logTiks = DSLServerConfigs.LOG_TICK_TIME.get();
5758
ArbiterClient arbiterClient = new ArbiterClient(arbiterUri);
5859
private ArbiterClient.ArbiterRes topology;
5960
private static final int SHARD_MAX_ENTITIES = 1_000_000;
@@ -62,6 +63,9 @@ public class CustomDedicatedServer extends DedicatedServer {
6263
private final AtomicInteger timeSugestions = new AtomicInteger();
6364
private RegionRouter router = null;
6465

66+
public Map<Integer, Float> workersMSPT = new ConcurrentHashMap<>();
67+
public Map<Integer, List<Long>> workersNanoTicks = new HashMap<>();
68+
6569
public CustomDedicatedServer(Thread pServerThread, LevelStorageSource.LevelStorageAccess pStorageSource, PackRepository pPackRepository, WorldStem pWorldStem, DedicatedServerSettings pSettings, DataFixer pFixerUpper, Services pServices, ChunkProgressListenerFactory pProgressListenerFactory) {
6670
super(pServerThread, pStorageSource, pPackRepository, pWorldStem, pSettings, pFixerUpper, pServices, pProgressListenerFactory);
6771

‎src/main/java/com/manwe/dsl/dedicatedServer/InternalGameProtocols.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@ public class InternalGameProtocols {
6969
.addPacket(InternalPacketTypes.WORKER_PROXY_FAKE_PLAYER_DISCONNECT, ProxyBoundFakePlayerDisconnectPacket.STREAM_CODEC)
7070

7171
.addPacket(InternalPacketTypes.WORKER_PROXY_SYNC_TIME, ProxyBoundSyncTimePacket.STREAM_CODEC)
72+
.addPacket(InternalPacketTypes.WORKER_PROXY_HEALTH, ProxyBoundHealthPacket.STREAM_CODEC)
7273
);
7374

7475
public static final ProtocolInfo<ProxyListener> CLIENTBOUND = CLIENTBOUND_TEMPLATE.bind(FriendlyByteBuf::new);

‎src/main/java/com/manwe/dsl/dedicatedServer/InternalPacketTypes.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ private static <T extends Packet<WorkerListener>> PacketType<T> createServerboun
6767
public static final PacketType<ProxyBoundFakePlayerDisconnectPacket> WORKER_PROXY_FAKE_PLAYER_DISCONNECT = createClientbound("proxy_fake_player_disconnect");
6868

6969
public static final PacketType<ProxyBoundSyncTimePacket> WORKER_PROXY_SYNC_TIME = createClientbound("proxy_sync_time");
70+
public static final PacketType<ProxyBoundHealthPacket> WORKER_PROXY_HEALTH = createClientbound("proxy_health");
7071

7172
private static <T extends Packet<ProxyListener>> PacketType<T> createClientbound(String pId) {
7273
return new PacketType<>(PacketFlow.CLIENTBOUND, ResourceLocation.withDefaultNamespace(pId));

‎src/main/java/com/manwe/dsl/dedicatedServer/proxy/back/listeners/ProxyListener.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,4 +47,6 @@ public interface ProxyListener extends ClientboundPacketListener {
4747
* Sends message to be compared with other workers
4848
*/
4949
void handleSyncTime(ProxyBoundSyncTimePacket packet);
50+
51+
void handleWorkerHealth(ProxyBoundHealthPacket packet);
5052
}

‎src/main/java/com/manwe/dsl/dedicatedServer/proxy/back/listeners/ProxyListenerImpl.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,4 +181,23 @@ public void handleEntityTrasnfer(ProxyBoundEntityTransferPacket packet) {
181181
public void handleSyncTime(ProxyBoundSyncTimePacket packet) {
182182
router.server.addSugestedLevelTime(packet.levelTime);
183183
}
184+
185+
@Override
186+
public void handleWorkerHealth(ProxyBoundHealthPacket packet) {
187+
long[] tickData = packet.getTickTime();
188+
long avgMST = 0;
189+
for (long nanoTickTime : tickData){
190+
avgMST += nanoTickTime;
191+
}
192+
router.server.workersMSPT.put(packet.getWorkerSource(), ((float) avgMST / (float) tickData.length) / 1_000_000F);
193+
194+
if(router.server.logTiks){ //Save raw data
195+
DistributedServerLevels.LOGGER.info("Worker: "+packet.getWorkerSource()+" Avg MSPT "+ ((float) avgMST/ (float) tickData.length)/ (float) 1_000_000 );
196+
197+
List<Long> nanoTick = router.server.workersNanoTicks.computeIfAbsent(packet.getWorkerSource(), integer -> new ArrayList<>());
198+
for (long tickTime : packet.getTickTime()){
199+
nanoTick.add(tickTime);
200+
}
201+
}
202+
}
184203
}

‎src/main/java/com/manwe/dsl/dedicatedServer/proxy/back/packets/ProxyBoundHealthPacket.java‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
1+
package com.manwe.dsl.dedicatedServer.proxy.back.packets;
2+
3+
import com.manwe.dsl.dedicatedServer.InternalPacketTypes;
4+
import com.manwe.dsl.dedicatedServer.proxy.back.listeners.ProxyListener;
5+
import net.minecraft.network.FriendlyByteBuf;
6+
import net.minecraft.network.codec.StreamCodec;
7+
import net.minecraft.network.protocol.Packet;
8+
import net.minecraft.network.protocol.PacketType;
9+
import org.jetbrains.annotations.NotNull;
10+
11+
public class ProxyBoundHealthPacket implements Packet<ProxyListener> {
12+
13+
public static final StreamCodec<FriendlyByteBuf, ProxyBoundHealthPacket> STREAM_CODEC = Packet.codec(
14+
ProxyBoundHealthPacket::write, ProxyBoundHealthPacket::new
15+
);
16+
17+
private final long[] tickTime;
18+
private final int workerId;
19+
20+
public ProxyBoundHealthPacket(long[] tickTime, int workerId) {
21+
this.tickTime = tickTime;
22+
this.workerId = workerId;
23+
}
24+
25+
private ProxyBoundHealthPacket(FriendlyByteBuf buf) {
26+
this.tickTime = buf.readLongArray();
27+
this.workerId = buf.readInt();
28+
}
29+
30+
private void write(FriendlyByteBuf buf) {
31+
buf.writeLongArray(this.tickTime);
32+
buf.writeInt(this.workerId);
33+
}
34+
35+
public int getWorkerSource() {
36+
return workerId;
37+
}
38+
39+
public long[] getTickTime() {
40+
return tickTime;
41+
}
42+
43+
@Override
44+
public @NotNull PacketType<? extends Packet<ProxyListener>> type() {
45+
return InternalPacketTypes.WORKER_PROXY_HEALTH;
46+
}
47+
48+
/**
49+
* Passes this Packet on to the PacketListener for processing.
50+
*/
51+
@Override
52+
public void handle(ProxyListener pHandler) {
53+
pHandler.handleWorkerHealth(this);
54+
}
55+
}

‎src/main/java/com/manwe/dsl/dedicatedServer/worker/LocalProfiler.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
package com.manwe.dsl.dedicatedServer.worker;
2+
3+
import com.manwe.dsl.DistributedServerLevels;
4+
import com.manwe.dsl.dedicatedServer.CustomDedicatedServer;
5+
import com.manwe.dsl.dedicatedServer.proxy.back.packets.ProxyBoundHealthPacket;
6+
import com.manwe.dsl.dedicatedServer.worker.listeners.WorkerListenerImpl;
7+
import net.neoforged.bus.api.SubscribeEvent;
8+
import net.neoforged.neoforge.event.tick.ServerTickEvent;
9+
10+
import java.util.NoSuchElementException;
11+
12+
public class LocalProfiler {
13+
14+
private static final int SAMPLE_SIZE = 100;
15+
16+
private static final long[] history = new long[SAMPLE_SIZE];
17+
private static int index = 0;
18+
19+
private static long tickStartNanos;
20+
21+
@SubscribeEvent
22+
public void onServerTick(ServerTickEvent.Pre e) {
23+
if(e.getServer() instanceof CustomDedicatedServer customDedicatedServer && customDedicatedServer.isProxy()) return;
24+
tickStartNanos = System.nanoTime();
25+
}
26+
27+
@SubscribeEvent
28+
public void onServerTick(ServerTickEvent.Post e) {
29+
if(e.getServer() instanceof CustomDedicatedServer customDedicatedServer && customDedicatedServer.isProxy()) return;
30+
long duration = System.nanoTime() - tickStartNanos; // ns que tardó el tick
31+
if(++index == SAMPLE_SIZE){
32+
try {
33+
if(!(e.getServer() instanceof CustomDedicatedServer customDedicatedServer && customDedicatedServer.getConnection().getConnections().getFirst() instanceof WorkerConnection workerConnection && workerConnection.getPacketListener() instanceof WorkerListenerImpl workerListener)){
34+
return;
35+
}
36+
workerListener.send(new ProxyBoundHealthPacket(history, customDedicatedServer.workerId));
37+
} catch (NoSuchElementException exception){
38+
//Ignore
39+
DistributedServerLevels.LOGGER.info("Tried to send mspt info, Proxy not yet connected");
40+
}
41+
index = 0;
42+
}
43+
history[index] = duration;
44+
}
45+
}

0 commit comments

Comments
 (0)