Astronand/Mindustry
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Partial 7.0 merge - API preview

Browse Source
This commit is contained in:
Anuken
2021-06-02 11:08:08 -04:00
parent ea75a357ca
commit 28b235ef07
531 changed files with 12356 additions and 6286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/src/mindustry/net/ArcNetProvider.java
+130 -25
View File
@@ -2,14 +2,17 @@ package mindustry.net;
import arc.*;
import arc.func.*;
import arc.math.*;
import arc.net.*;
import arc.net.FrameworkMessage.*;
import arc.struct.*;
import arc.util.*;
import arc.util.Log.*;
import arc.util.async.*;
import arc.util.pooling.*;
import arc.util.io.*;
import mindustry.net.Net.*;
import mindustry.net.Packets.*;
import net.jpountz.lz4.*;
import java.io.*;
import java.net.*;
@@ -28,8 +31,15 @@ public class ArcNetProvider implements NetProvider{
final CopyOnWriteArrayList<ArcConnection> connections = new CopyOnWriteArrayList<>();
Thread serverThread;
private static final LZ4FastDecompressor decompressor = LZ4Factory.fastestInstance().fastDecompressor();
private static final LZ4Compressor compressor = LZ4Factory.fastestInstance().fastCompressor();
public ArcNetProvider(){
ArcNet.errorHandler = e -> Log.debug(Strings.getStackTrace(e));
ArcNet.errorHandler = e -> {
if(Log.level == LogLevel.debug){
Log.debug(Strings.getStackTrace(e));
}
};
client = new Client(8192, 8192, new PacketSerializer());
client.setDiscoveryPacket(packetSupplier);
@@ -56,11 +66,11 @@ public class ArcNetProvider implements NetProvider{
@Override
public void received(Connection connection, Object object){
if(object instanceof FrameworkMessage) return;
if(!(object instanceof Packet p)) return;
Core.app.post(() -> {
try{
net.handleClientReceived(object);
net.handleClientReceived(p);
}catch(Throwable e){
net.handleException(e);
}
@@ -111,13 +121,13 @@ public class ArcNetProvider implements NetProvider{
@Override
public void received(Connection connection, Object object){
ArcConnection k = getByArcID(connection.getID());
if(object instanceof FrameworkMessage || k == null) return;
if(!(object instanceof Packet pack) || k == null) return;
Core.app.post(() -> {
try{
net.handleServerReceived(k, object);
net.handleServerReceived(k, pack);
}catch(Throwable e){
e.printStackTrace();
Log.err(e);
}
});
}
@@ -163,9 +173,9 @@ public class ArcNetProvider implements NetProvider{
}
@Override
public void sendClient(Object object, SendMode mode){
public void sendClient(Object object, boolean reliable){
try{
if(mode == SendMode.tcp){
if(reliable){
client.sendTCP(object);
}else{
client.sendUDP(object);
@@ -174,8 +184,6 @@ public class ArcNetProvider implements NetProvider{
}catch(BufferOverflowException | BufferUnderflowException e){
net.showError(e);
}
Pools.free(object);
}
@Override
@@ -293,7 +301,7 @@ public class ArcNetProvider implements NetProvider{
//send an object so the receiving side knows how to handle the following chunks
StreamBegin begin = new StreamBegin();
begin.total = stream.stream.available();
begin.type = Registrator.getID(stream.getClass());
begin.type = Net.getPacketId(stream);
connection.sendTCP(begin);
id = begin.id;
}
@@ -309,9 +317,9 @@ public class ArcNetProvider implements NetProvider{
}
@Override
public void send(Object object, SendMode mode){
public void send(Object object, boolean reliable){
try{
if(mode == SendMode.tcp){
if(reliable){
connection.sendTCP(object);
}else{
connection.sendUDP(object);
@@ -332,32 +340,129 @@ public class ArcNetProvider implements NetProvider{
}
}
@SuppressWarnings("unchecked")
public static class PacketSerializer implements NetSerializer{
//for debugging total read/write speeds
private static final boolean debug = false;
ThreadLocal<ByteBuffer> decompressBuffer = new ThreadLocal<>(){
@Override
protected ByteBuffer initialValue(){
return ByteBuffer.allocate(32768);
}
};
ThreadLocal<Reads> reads = new ThreadLocal<>(){
@Override
protected Reads initialValue(){
return new Reads(new ByteBufferInput(decompressBuffer.get()));
}
};
ThreadLocal<Writes> writes = new ThreadLocal<>(){
@Override
protected Writes initialValue(){
return new Writes(new ByteBufferOutput(decompressBuffer.get()));
}
};
//for debugging network write counts
static WindowedMean upload = new WindowedMean(5), download = new WindowedMean(5);
static long lastUpload, lastDownload, uploadAccum, downloadAccum;
static int lastPos;
@Override
public Object read(ByteBuffer byteBuffer){
if(debug){
if(Time.timeSinceMillis(lastDownload) >= 1000){
lastDownload = Time.millis();
download.add(downloadAccum);
downloadAccum = 0;
Log.info("Download: @ b/s", download.mean());
}
downloadAccum += byteBuffer.remaining();
}
byte id = byteBuffer.get();
if(id == -2){
return readFramework(byteBuffer);
}else{
Packet packet = Pools.obtain((Class<Packet>)Registrator.getByID(id).type, (Prov<Packet>)Registrator.getByID(id).constructor);
packet.read(byteBuffer);
//read length int, followed by compressed lz4 data
//TODO not thread safe!!!
Packet packet = Net.newPacket(id);
var buffer = decompressBuffer.get();
int length = byteBuffer.getShort() & 0xffff;
byte compression = byteBuffer.get();
//no compression, copy over buffer
if(compression == 0){
buffer.position(0).limit(length);
buffer.put(byteBuffer.array(), byteBuffer.position(), length);
buffer.position(0);
packet.read(reads.get());
//move read packets forward
byteBuffer.position(byteBuffer.position() + buffer.position());
}else{
//decompress otherwise
int read = decompressor.decompress(byteBuffer, byteBuffer.position(), buffer, 0, length);
buffer.position(0);
buffer.limit(length);
packet.read(reads.get());
//move buffer forward based on bytes read by decompressor
byteBuffer.position(byteBuffer.position() + read);
}
return packet;
}
}
@Override
public void write(ByteBuffer byteBuffer, Object o){
if(o instanceof FrameworkMessage){
if(debug){
lastPos = byteBuffer.position();
}
//write raw buffer
if(o instanceof ByteBuffer raw){
byteBuffer.put(raw);
}else if(o instanceof FrameworkMessage msg){
byteBuffer.put((byte)-2); //code for framework message
writeFramework(byteBuffer, (FrameworkMessage)o);
writeFramework(byteBuffer, msg);
}else{
if(!(o instanceof Packet)) throw new RuntimeException("All sent objects must implement be Packets! Class: " + o.getClass());
byte id = Registrator.getID(o.getClass());
if(id == -1) throw new RuntimeException("Unregistered class: " + o.getClass());
if(!(o instanceof Packet pack)) throw new RuntimeException("All sent objects must implement be Packets! Class: " + o.getClass());
byte id = Net.getPacketId(pack);
byteBuffer.put(id);
((Packet)o).write(byteBuffer);
var temp = decompressBuffer.get();
temp.position(0);
temp.limit(temp.capacity());
pack.write(writes.get());
short length = (short)temp.position();
//write length, uncompressed
byteBuffer.putShort(length);
//don't bother with small packets
if(length < 36 || pack instanceof StreamChunk){
//write direct contents...
byteBuffer.put((byte)0); //0 = no compression
byteBuffer.put(temp.array(), 0, length);
}else{
byteBuffer.put((byte)1); //1 = compression
//write compressed data; this does not modify position!
int written = compressor.compress(temp, 0, temp.position(), byteBuffer, byteBuffer.position(), byteBuffer.remaining());
//skip to indicate the written, compressed data
byteBuffer.position(byteBuffer.position() + written);
}
}
if(debug){
if(Time.timeSinceMillis(lastUpload) >= 1000){
lastUpload = Time.millis();
upload.add(uploadAccum);
uploadAccum = 0;
Log.info("Upload: @ b/s", upload.mean());
}
uploadAccum += byteBuffer.position() - lastPos;
}
}
@@ -388,9 +493,9 @@ public class ArcNetProvider implements NetProvider{
p.isReply = buffer.get() == 1;
return p;
}else if(id == 1){
return new DiscoverHost();
return FrameworkMessage.discoverHost;
}else if(id == 2){
return new KeepAlive();
return FrameworkMessage.keepAlive;
}else if(id == 3){
RegisterUDP p = new RegisterUDP();
p.connectionID = buffer.getInt();