use unreliable datachannels and custom netcode to handle throttles and dropped moves

src/ProfilerUI.hx

@@ -104,6 +104,15 @@ class ProfilerUI {
 					+ 'Last Ack Move: ${Net.isClient ? @:privateAccess Net.clientConnection.moveManager.lastAckMoveId : 0}\n'
 					+ 'Move Ack RTT: ${Net.isClient ? @:privateAccess Net.clientConnection.moveManager.ackRTT : 0}';
 			}
+			if (Net.isHost) {
+				var strs = [];
+				strs.push('World Ticks: ${MarbleGame.instance.world.timeState.ticks}');
+				for (dc => cc in Net.clients) {
+					strs.push('${cc.id} move: sz ${@:privateAccess cc.moveManager.getQueueSize()} avg ${@:privateAccess cc.moveManager.serverAvgMoveListSize}');
+				}
+
+				instance.networkStats.text = strs.join('\n');
+			}
 		} else {
 			instance.networkStats.text = "";
 		}

src/net/ClientConnection.hx

@@ -24,16 +24,18 @@ enum abstract NetPlatform(Int) from Int to Int {
 class ClientConnection extends GameConnection {
 	var socket:RTCPeerConnection;
 	var datachannel:RTCDataChannel;
+	var datachannelUnreliable:RTCDataChannel;
 	var rtt:Float;
 	var pingSendTime:Float;
 	var _rttRecords:Array<Float> = [];
 	var lastRecvTime:Float;
 	var didWarnTimeout:Bool = false;
 
-	public function new(id:Int, socket:RTCPeerConnection, datachannel:RTCDataChannel) {
+	public function new(id:Int, socket:RTCPeerConnection, datachannel:RTCDataChannel, datachannelUnreliable:RTCDataChannel) {
 		super(id);
 		this.socket = socket;
 		this.datachannel = datachannel;
+		this.datachannelUnreliable = datachannelUnreliable;
 		this.state = GameplayState.LOBBY;
 		this.rtt = 0;
 		this.name = "Unknown";
@@ -43,6 +45,10 @@ class ClientConnection extends GameConnection {
 		datachannel.sendBytes(b);
 	}
 
+	override function sendBytesUnreliable(b:Bytes) {
+		datachannelUnreliable.sendBytes(b);
+	}
+
 	public inline function needsTimeoutWarn(t:Float) {
 		return (t - lastRecvTime) > 10 && !didWarnTimeout;
 	}
@@ -111,6 +117,8 @@ abstract class GameConnection {
 
 	public function sendBytes(b:haxe.io.Bytes) {}
 
+	public function sendBytesUnreliable(b:haxe.io.Bytes) {}
+
 	public inline function getName() {
 		return name;
 	}

src/net/MoveManager.hx

@@ -45,14 +45,17 @@ class MoveManager {
 	var ackRTT:Int = -1;
 
 	var maxMoves = 45;
+	var maxSendMoveListSize = 30;
 
-	var serverTargetMoveListSize = 3;
+	var serverTargetMoveListSize = 4;
 	var serverMaxMoveListSize = 8;
-	var serverAvgMoveListSize = 3.0;
+	var serverAvgMoveListSize = 4.0;
 	var serverSmoothMoveAvg = 0.15;
-	var serverMoveListSizeSlack = 1.0;
-	var serverDefaultMinTargetMoveListSize = 3;
+	var serverMoveListSizeSlack = 1.5;
+	var serverDefaultMinTargetMoveListSize = 4;
 	var serverAbnormalMoveCount = 0;
+	var serverLastRecvMove = 0;
+	var serverLastAckMove = 0;
 
 	public var stall = false;
 
@@ -119,15 +122,19 @@ class MoveManager {
 		if (nextMoveId >= 65535) // 65535 is reserved for null move
 			nextMoveId = 0;
 
+		var moveStartIdx = queuedMoves.length - maxSendMoveListSize;
+		if (moveStartIdx < 0)
+			moveStartIdx = 0;
+
 		var b = new OutputBitStream();
 		var movePacket = new MarbleMovePacket();
 		movePacket.clientId = Net.clientId;
-		movePacket.move = netMove;
+		movePacket.moves = queuedMoves.slice(moveStartIdx);
 		movePacket.clientTicks = timeState.ticks;
 		b.writeByte(NetPacketType.MarbleMove);
 		movePacket.serialize(b);
 
-		Net.sendPacketToHost(b);
+		Net.sendPacketToHostUnreliable(b);
 
 		return netMove;
 	}
@@ -168,7 +175,17 @@ class MoveManager {
 	}
 
 	public inline function queueMove(m:NetMove) {
-		queuedMoves.push(m);
+		if (serverLastRecvMove < m.id && serverLastAckMove < m.id) {
+			queuedMoves.push(m);
+			serverLastRecvMove = m.id;
+		}
+		// if (queuedMoves.length != 0) {
+		// 	var lastQueuedMove = queuedMoves[queuedMoves.length - 1];
+		// 	if (lastQueuedMove.id < m.id)
+		// 		queuedMoves.push(m);
+		// } else if (lastMove == null || lastMove.id < m.id) {
+		// 	queuedMoves.push(m);
+		// }
 	}
 
 	public function getNextMove() {
@@ -214,6 +231,7 @@ class MoveManager {
 		} else {
 			lastMove = queuedMoves[0];
 			queuedMoves.shift();
+			lastAckMoveId = lastMove.id;
 			return lastMove;
 		}
 	}

src/net/Net.hx

@@ -62,6 +62,7 @@ class ServerInfo {
 class Net {
 	static var client:RTCPeerConnection;
 	static var clientDatachannel:RTCDataChannel;
+	static var clientDatachannelUnreliable:RTCDataChannel;
 
 	public static var isMP:Bool;
 	public static var isHost:Bool;
@@ -113,8 +114,22 @@ class Net {
 				}));
 			}
 		}
+		var reliable:datachannel.RTCDataChannel = null;
+		var unreliable:datachannel.RTCDataChannel = null;
 		peer.onDataChannel = (dc:datachannel.RTCDataChannel) -> {
-			onClientConnect(peer, dc);
+			if (dc.name == "mp")
+				reliable = dc;
+			if (dc.name == "unreliable") {
+				unreliable = dc;
+				switch (dc.reliability) {
+					case Reliable:
+						Console.log("Error opening unreliable datachannel!");
+					case Unreliable(maxRetransmits, maxLifetime):
+						Console.log("Opened unreliable datachannel: " + maxRetransmits + " " + maxLifetime);
+				}
+			}
+			if (reliable != null && unreliable != null)
+				onClientConnect(peer, reliable, unreliable);
 		}
 	}
 
@@ -146,22 +161,34 @@ class Net {
 			}
 
 			clientDatachannel = client.createDatachannel("mp");
-			clientDatachannel.onOpen = (n) -> {
-				var loadGui:MultiplayerLoadingGui = cast MarbleGame.canvas.content;
-				if (loadGui != null) {
-					loadGui.setLoadingStatus("Handshaking");
+			clientDatachannelUnreliable = client.createDatachannelWithOptions("unreliable", false, 0, 600);
+
+			var closing = false;
+			var openFlags = 0;
+
+			var onDatachannelOpen = (idx:Int) -> {
+				openFlags |= idx;
+				if (openFlags == 3) {
+					var loadGui:MultiplayerLoadingGui = cast MarbleGame.canvas.content;
+					if (loadGui != null) {
+						loadGui.setLoadingStatus("Handshaking");
+					}
+					Console.log("Successfully connected!");
+					clients.set(client, new ClientConnection(0, client, clientDatachannel, clientDatachannelUnreliable)); // host is always 0
+					clientIdMap[0] = clients[client];
+					clientConnection = cast clients[client];
+					onConnectedToServer();
+					haxe.Timer.delay(() -> connectedCb(), 1500); // 1.5 second delay to do the RTT calculation
 				}
-				Console.log("Successfully connected!");
-				clients.set(client, new ClientConnection(0, client, clientDatachannel)); // host is always 0
-				clientIdMap[0] = clients[client];
-				clientConnection = cast clients[client];
-				onConnectedToServer();
-				haxe.Timer.delay(() -> connectedCb(), 1500); // 1.5 second delay to do the RTT calculation
 			}
-			clientDatachannel.onMessage = (b) -> {
+			var onDatachannelMessage = (dc:RTCDataChannel, b:haxe.io.Bytes) -> {
 				onPacketReceived(clientConnection, client, clientDatachannel, new InputBitStream(b));
 			}
-			clientDatachannel.onClosed = () -> {
+
+			var onDatachannelClose = (dc:RTCDataChannel) -> {
+				if (closing)
+					return;
+				closing = true;
 				var weLeftOurselves = !Net.isClient; // If we left ourselves, this would be set to false due to order of ops, disconnect being called first, and then the datachannel closing
 				disconnect();
 				if (MarbleGame.instance.world != null) {
@@ -175,7 +202,11 @@ class Net {
 					}
 				}
 			}
-			clientDatachannel.onError = (msg) -> {
+
+			var onDatachannelError = (msg:String) -> {
+				if (closing)
+					return;
+				closing = true;
 				Console.log('Errored out due to ${msg}');
 				disconnect();
 				if (MarbleGame.instance.world != null) {
@@ -186,6 +217,31 @@ class Net {
 				loadGui.setErrorStatus("Connection error");
 			}
 
+			clientDatachannel.onOpen = (n) -> {
+				onDatachannelOpen(1);
+			}
+			clientDatachannel.onMessage = (b) -> {
+				onDatachannelMessage(clientDatachannel, b);
+			}
+			clientDatachannel.onClosed = () -> {
+				onDatachannelClose(clientDatachannel);
+			}
+			clientDatachannel.onError = (msg) -> {
+				onDatachannelError(msg);
+			}
+			clientDatachannelUnreliable.onOpen = (n) -> {
+				onDatachannelOpen(2);
+			}
+			clientDatachannelUnreliable.onMessage = (b) -> {
+				onDatachannelMessage(clientDatachannelUnreliable, b);
+			}
+			clientDatachannelUnreliable.onClosed = () -> {
+				onDatachannelClose(clientDatachannelUnreliable);
+			}
+			clientDatachannelUnreliable.onError = (msg) -> {
+				onDatachannelError(msg);
+			}
+
 			isMP = true;
 			isHost = false;
 			isClient = true;
@@ -263,23 +319,54 @@ class Net {
 		}
 	}
 
-	static function onClientConnect(c:RTCPeerConnection, dc:RTCDataChannel) {
+	static function onClientConnect(c:RTCPeerConnection, dc:RTCDataChannel, dcu:RTCDataChannel) {
 		clientId += 1;
-		var cc = new ClientConnection(clientId, c, dc);
+		var cc = new ClientConnection(clientId, c, dc, dcu);
 		clients.set(c, cc);
 		clientIdMap[clientId] = clients[c];
-		dc.onMessage = (msgBytes) -> {
+
+		var closing = false;
+
+		var onMessage = (dc:RTCDataChannel, msgBytes:haxe.io.Bytes) -> {
 			onPacketReceived(cc, c, dc, new InputBitStream(msgBytes));
 		}
-		dc.onClosed = () -> {
+		var onClosed = () -> {
+			if (closing)
+				return;
+			closing = true;
 			clients.remove(c);
 			onClientLeave(cc);
 		}
-		dc.onError = (msg) -> {
+
+		var onError = (msg:String) -> {
+			if (closing)
+				return;
+			closing = true;
 			clients.remove(c);
 			Console.log('Client ${cc.id} errored out due to: ${msg}');
 			onClientLeave(cc);
 		}
+
+		dc.onMessage = (msgBytes) -> {
+			onMessage(dc, msgBytes);
+		}
+		dc.onClosed = () -> {
+			onClosed();
+		}
+		dc.onError = (msg) -> {
+			onError(msg);
+		}
+
+		dcu.onMessage = (msgBytes) -> {
+			onMessage(dcu, msgBytes);
+		}
+		dcu.onClosed = () -> {
+			onClosed();
+		}
+		dcu.onError = (msg) -> {
+			onError(msg);
+		}
+
 		var b = haxe.io.Bytes.alloc(2);
 		b.set(0, ClientIdAssign);
 		b.set(1, clientId);
@@ -444,7 +531,8 @@ class Net {
 				movePacket.deserialize(input);
 				var cc = clientIdMap[movePacket.clientId];
 				if (cc.state == GAME)
-					cc.queueMove(movePacket.move);
+					for (move in movePacket.moves)
+						cc.queueMove(move);
 
 			case PowerupPickup:
 				var powerupPickupPacket = new PowerupPickupPacket();
@@ -532,6 +620,13 @@ class Net {
 		}
 	}
 
+	public static function sendPacketToHostUnreliable(packetData:OutputBitStream) {
+		if (clientDatachannelUnreliable.state == Open) {
+			var bytes = packetData.getBytes();
+			clientDatachannelUnreliable.sendBytes(bytes);
+		}
+	}
+
 	public static function sendPacketToClient(client:GameConnection, packetData:OutputBitStream) {
 		var bytes = packetData.getBytes();
 		client.sendBytes(bytes);

src/net/NetPacket.hx

@@ -14,20 +14,28 @@ interface NetPacket {
 class MarbleMovePacket implements NetPacket {
 	var clientId:Int;
 	var clientTicks:Int;
-	var move:NetMove;
+	var moves:Array<NetMove>;
 
-	public function new() {}
+	public function new() {
+		moves = [];
+	}
 
 	public inline function deserialize(b:InputBitStream) {
 		clientId = b.readByte();
 		clientTicks = b.readUInt16();
-		move = MoveManager.unpackMove(b);
+		var count = b.readInt(5);
+		moves = [];
+		for (i in 0...count) {
+			moves.push(MoveManager.unpackMove(b));
+		}
 	}
 
 	public inline function serialize(b:OutputBitStream) {
 		b.writeByte(clientId);
 		b.writeUInt16(clientTicks);
-		MoveManager.packMove(move, b);
+		b.writeInt(moves.length, 5);
+		for (move in moves)
+			MoveManager.packMove(move, b);
 	}
 }