wisplite/distant-horizons-core-sharded
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace pooled buffers with unpooled

Browse Source
This commit is contained in:
s809
2025-08-07 17:55:22 +05:00
parent a05bd307f9
commit 674fc30e77
8 changed files with 47 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/main/java/com/seibel/distanthorizons/core/level/AbstractDhServerLevel.java
+18 -20
View File
@@ -261,29 +261,27 @@ public abstract class AbstractDhServerLevel extends AbstractDhLevel implements I
}
LodUtil.assertTrue(this.beaconBeamRepo != null, "beaconBeamRepo should not be null");
try (FullDataPayload payload = new FullDataPayload(data, this.beaconBeamRepo.getAllBeamsForPos(data.getPos())))
FullDataPayload payload = new FullDataPayload(data, this.beaconBeamRepo.getAllBeamsForPos(data.getPos()));
for (ServerPlayerState serverPlayerState : this.serverPlayerStateManager.getReadyPlayers())
{
for (ServerPlayerState serverPlayerState : this.serverPlayerStateManager.getReadyPlayers())
if (serverPlayerState.getServerPlayer().getLevel() != this.serverLevelWrapper)
{
if (serverPlayerState.getServerPlayer().getLevel() != this.serverLevelWrapper)
continue;
}
if (!serverPlayerState.sessionConfig.isRealTimeUpdatesEnabled())
{
continue;
}
Vec3d playerPosition = serverPlayerState.getServerPlayer().getPosition();
int distanceFromPlayer = DhSectionPos.getChebyshevSignedBlockDistance(data.getPos(), new DhBlockPos2D((int) playerPosition.x, (int) playerPosition.z)) / 16;
if (distanceFromPlayer <= serverPlayerState.sessionConfig.getMaxUpdateDistanceRadius())
{
serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () ->
{
continue;
}
if (!serverPlayerState.sessionConfig.isRealTimeUpdatesEnabled())
{
continue;
}
Vec3d playerPosition = serverPlayerState.getServerPlayer().getPosition();
int distanceFromPlayer = DhSectionPos.getChebyshevSignedBlockDistance(data.getPos(), new DhBlockPos2D((int) playerPosition.x, (int) playerPosition.z)) / 16;
if (distanceFromPlayer <= serverPlayerState.sessionConfig.getMaxUpdateDistanceRadius())
{
serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () ->
{
serverPlayerState.networkSession.sendMessage(new FullDataPartialUpdateMessage(this.serverLevelWrapper, payload));
});
}
serverPlayerState.networkSession.sendMessage(new FullDataPartialUpdateMessage(this.serverLevelWrapper, payload));
});
}
}
});
core/src/main/java/com/seibel/distanthorizons/core/level/DhClientLevel.java
+1 -1
View File
@@ -164,7 +164,7 @@ public class DhClientLevel extends AbstractDhLevel implements IDhClientLevel
}
try(FullDataSourceV2DTO dataSourceDto = this.networkState.fullDataPayloadReceiver.decodeDataSourceAndReleaseBuffer(message.payload))
try (FullDataSourceV2DTO dataSourceDto = this.networkState.fullDataPayloadReceiver.decodeDataSource(message.payload))
{
boolean isSameLevel = message.isSameLevelAs(this.levelWrapper);
NETWORK_LOGGER.debug("Buffer {} isSameLevel: {}", message.payload.dtoBufferId, isSameLevel);
core/src/main/java/com/seibel/distanthorizons/core/multiplayer/client/AbstractFullDataNetworkRequestQueue.java
+1 -1
View File
@@ -256,7 +256,7 @@ public abstract class AbstractFullDataNetworkRequestQueue implements IDebugRende
if (response.payload != null)
{
FullDataSourceV2DTO dataSourceDto = this.networkState.fullDataPayloadReceiver.decodeDataSourceAndReleaseBuffer(response.payload);
FullDataSourceV2DTO dataSourceDto = this.networkState.fullDataPayloadReceiver.decodeDataSource(response.payload);
// set application flags based on the received detail level,
// this is needed so the data sources propagate correctly
core/src/main/java/com/seibel/distanthorizons/core/multiplayer/fullData/FullDataPayload.java
+3 -10
View File
@@ -9,18 +9,17 @@ import com.seibel.distanthorizons.core.network.messages.fullData.FullDataSplitMe
import com.seibel.distanthorizons.core.sql.dto.BeaconBeamDTO;
import com.seibel.distanthorizons.core.sql.dto.FullDataSourceV2DTO;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
/**
* @see FullDataSplitMessage
*/
public class FullDataPayload implements INetworkObject, AutoCloseable
public class FullDataPayload implements INetworkObject
{
private static final AtomicInteger lastBufferId = new AtomicInteger();
@@ -47,7 +46,7 @@ public class FullDataPayload implements INetworkObject, AutoCloseable
EDhApiDataCompressionMode compressionMode = Config.Common.LodBuilding.dataCompression.get();
try (FullDataSourceV2DTO dataSourceDto = FullDataSourceV2DTO.CreateFromDataSource(fullDataSource, compressionMode))
{
this.dtoBuffer = ByteBufAllocator.DEFAULT.buffer();
this.dtoBuffer = Unpooled.buffer();
dataSourceDto.encode(this.dtoBuffer);
}
}
@@ -85,12 +84,6 @@ public class FullDataPayload implements INetworkObject, AutoCloseable
// base overrides //
//================//
@Override
public void close()
{
this.dtoBuffer.release();
}
@Override
public String toString()
{
core/src/main/java/com/seibel/distanthorizons/core/multiplayer/fullData/FullDataPayloadReceiver.java
+4 -18
View File
@@ -9,8 +9,8 @@ import com.seibel.distanthorizons.core.network.INetworkObject;
import com.seibel.distanthorizons.core.network.messages.fullData.FullDataSplitMessage;
import com.seibel.distanthorizons.core.sql.dto.FullDataSourceV2DTO;
import com.seibel.distanthorizons.core.util.LodUtil;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.CompositeByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.logging.log4j.LogManager;
import java.util.Objects;
@@ -24,15 +24,7 @@ public class FullDataPayloadReceiver implements AutoCloseable
private final ConcurrentMap<Integer, CompositeByteBuf> buffersById = CacheBuilder.newBuilder()
.expireAfterAccess(10, TimeUnit.SECONDS)
.removalListener((RemovalNotification<Integer, CompositeByteBuf> notification) ->
{
// If an entry was replaced without removing, the buffer has to be released manually
if (notification.getCause() != RemovalCause.REPLACED)
{
Objects.requireNonNull(notification.getValue()).release();
}
})
.build().asMap();
.<Integer, CompositeByteBuf>build().asMap();
@Override
public void close()
@@ -46,13 +38,7 @@ public class FullDataPayloadReceiver implements AutoCloseable
{
if (message.isFirst)
{
if (composite != null)
{
composite.release();
LOGGER.debug("Released existing full data buffer [" + message.bufferId + "]");
}
composite = ByteBufAllocator.DEFAULT.compositeBuffer();
composite = Unpooled.compositeBuffer();
LOGGER.debug("Created new full data buffer [" + message.bufferId + "]: [" + composite + "]");
}
else if (composite == null)
@@ -67,7 +53,7 @@ public class FullDataPayloadReceiver implements AutoCloseable
});
}
public FullDataSourceV2DTO decodeDataSourceAndReleaseBuffer(FullDataPayload payload)
public FullDataSourceV2DTO decodeDataSource(FullDataPayload payload)
{
CompositeByteBuf compositeByteBuffer = this.buffersById.get(payload.dtoBufferId);
LodUtil.assertTrue(compositeByteBuffer != null);
core/src/main/java/com/seibel/distanthorizons/core/multiplayer/fullData/FullDataPayloadSender.java
+2 -19
View File
@@ -38,12 +38,6 @@ public class FullDataPayloadSender implements AutoCloseable
public void close()
{
this.tickTimerTask.cancel();
PendingTransfer pendingTransfer;
while ((pendingTransfer = this.transferQueue.poll()) != null)
{
pendingTransfer.close();
}
}
@@ -78,36 +72,25 @@ public class FullDataPayloadSender implements AutoCloseable
if (pendingTransfer.buffer.readableBytes() == 0)
{
pendingTransfer.sendFinalMessage.run();
pendingTransfer.close();
this.transferQueue.poll();
}
}
}
private static class PendingTransfer implements AutoCloseable
private static class PendingTransfer
{
public final int bufferId;
public final ByteBuf buffer;
public final Runnable sendFinalMessage;
private final AtomicBoolean isClosed = new AtomicBoolean();
private PendingTransfer(FullDataPayload payload, Runnable sendFinalMessage)
{
this.bufferId = payload.dtoBufferId;
this.buffer = payload.dtoBuffer.retainedDuplicate().readerIndex(0);
this.buffer = payload.dtoBuffer.duplicate().readerIndex(0);
this.sendFinalMessage = sendFinalMessage;
}
@Override
public void close()
{
if (this.isClosed.compareAndSet(false, true))
{
this.buffer.release();
}
}
}
}
core/src/main/java/com/seibel/distanthorizons/core/multiplayer/server/FullDataSourceRequestHandler.java
+16 -20
View File
@@ -119,16 +119,14 @@ public class FullDataSourceRequestHandler
}
// send the found data source to client
try (FullDataPayload payload = new FullDataPayload(fullDataSource, this.getAllBeamsForPos(message.sectionPos)))
FullDataPayload payload = new FullDataPayload(fullDataSource, this.getAllBeamsForPos(message.sectionPos));
fullDataSource.close();
serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () ->
{
fullDataSource.close();
serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () ->
{
message.sendResponse(new FullDataSourceResponseMessage(payload));
rateLimiterSet.syncOnLoginRateLimiter.release();
});
}
message.sendResponse(new FullDataSourceResponseMessage(payload));
rateLimiterSet.syncOnLoginRateLimiter.release();
});
}
catch (Exception e)
{
@@ -245,19 +243,17 @@ public class FullDataSourceRequestHandler
}
CompletableFuture.runAsync(() ->
{
try (FullDataPayload payload = new FullDataPayload(requestGroup.fullDataSource, this.getAllBeamsForPos(entry.getKey())))
FullDataPayload payload = new FullDataPayload(requestGroup.fullDataSource, this.getAllBeamsForPos(entry.getKey()));
requestGroup.fullDataSource.close();
for (DataSourceRequestGroup.RequestData requestData : requestGroup.requestMessages.values())
{
requestGroup.fullDataSource.close();
this.requestGroupsByFutureId.remove(requestData.futureId());
for (DataSourceRequestGroup.RequestData requestData : requestGroup.requestMessages.values())
{
this.requestGroupsByFutureId.remove(requestData.futureId());
requestData.serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () -> {
requestData.message.sendResponse(new FullDataSourceResponseMessage(payload));
requestData.rateLimiterSet.generationRequestRateLimiter.release();
});
}
requestData.serverPlayerState.fullDataPayloadSender.sendInChunks(payload, () -> {
requestData.message.sendResponse(new FullDataSourceResponseMessage(payload));
requestData.rateLimiterSet.generationRequestRateLimiter.release();
});
}
}, executor);
}
core/src/main/java/com/seibel/distanthorizons/core/network/messages/fullData/FullDataSplitMessage.java
+2 -14
View File
@@ -22,8 +22,8 @@ package com.seibel.distanthorizons.core.network.messages.fullData;
import com.google.common.base.MoreObjects;
import com.seibel.distanthorizons.core.multiplayer.fullData.FullDataPayload;
import com.seibel.distanthorizons.core.network.messages.AbstractNetworkMessage;
import com.seibel.distanthorizons.core.util.TimerUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import java.util.Timer;
@@ -34,16 +34,10 @@ import java.util.Timer;
*/
public class FullDataSplitMessage extends AbstractNetworkMessage
{
private static final long BUFFER_RELEASE_DELAY_MS = 5000L;
public int bufferId;
public ByteBuf buffer;
public boolean isFirst;
// Reference counting is unreliable here for some reason so this is a "fix"
private static final Timer bufferReleaseTimer = TimerUtil.CreateTimer("FullDataBufferCleanupTimer");
private boolean releaseScheduled = false;
//==============//
// constructors //
@@ -72,12 +66,6 @@ public class FullDataSplitMessage extends AbstractNetworkMessage
out.writeBytes(this.buffer.readerIndex(0));
out.writeBoolean(this.isFirst);
if (!this.releaseScheduled)
{
bufferReleaseTimer.schedule(TimerUtil.createTimerTask(this.buffer::release), BUFFER_RELEASE_DELAY_MS);
this.releaseScheduled = true;
}
}
@Override
@@ -86,7 +74,7 @@ public class FullDataSplitMessage extends AbstractNetworkMessage
this.bufferId = in.readInt();
int bufferSize = in.readInt();
this.buffer = in.readBytes(bufferSize);
this.buffer = Unpooled.copiedBuffer(in.readSlice(bufferSize));
this.isFirst = in.readBoolean();
}