Fixes for UDP use in input prediction.

README.md

@@ -27,7 +27,7 @@ _(how input delay roughly works)_
 
 ![input_delay_intro](./charts/InputDelayIntro.jpg)
 
-_(how rollback-and-chase in this project roughly works, kindly note that by the current implementation, each frontend only maintains a `lastAllConfirmedInputFrameId` for all the other peers, because the backend only downsyncs all-confirmed inputFrames, see [markConfirmationIfApplicable](https://github.com/genxium/DelayNoMore/blob/v0.9.14/battle_srv/models/room.go#L1085) for more information -- if a serverless peer-to-peer communication is seriously needed here, consider porting [markConfirmationIfApplicable](https://github.com/genxium/DelayNoMore/blob/v0.9.14/battle_srv/models/room.go#L1085) into frontend for maintaining `lastAllConfirmedInputFrameId` under chaotic reception order of inputFrames from peers)_
+_(how rollback-and-chase in this project roughly works)_
 
 ![server_clients](./charts/ServerClients.jpg)
 ![rollback_and_chase_intro](./charts/RollbackAndChase.jpg)

battle_srv/models/player.go

@@ -47,10 +47,11 @@ type Player struct {
 	TutorialStage int       `db:"tutorial_stage"`
 
 	// other in-battle info fields
-	LastReceivedInputFrameId int32
+	LastReceivedInputFrameId    int32
-	LastSentInputFrameId     int32
+	LastUdpReceivedInputFrameId int32
-	AckingFrameId            int32
+	LastSentInputFrameId        int32
-	AckingInputFrameId       int32
+	AckingFrameId               int32
+	AckingInputFrameId          int32
 
 	UdpAddr                *PeerUdpAddr
 	BattleUdpTunnelAddr    *net.UDPAddr // This addr is used by backend only, not visible to frontend

battle_srv/models/room.go

@@ -136,7 +136,7 @@ type Room struct {
 	EffectivePlayerCount                   int32
 	DismissalWaitGroup                     sync.WaitGroup
 	InputsBuffer                           *battle.RingBuffer // Indices are STRICTLY consecutive
-	InputsBufferLock                       sync.Mutex         // Guards [InputsBuffer, LatestPlayerUpsyncedInputFrameId, LastAllConfirmedInputFrameId, LastAllConfirmedInputList, LastAllConfirmedInputFrameIdWithChange, LastIndividuallyConfirmedInputList, player.LastReceivedInputFrameId]
+	InputsBufferLock                       sync.Mutex         // Guards [InputsBuffer, LatestPlayerUpsyncedInputFrameId, LastAllConfirmedInputFrameId, LastAllConfirmedInputList, LastAllConfirmedInputFrameIdWithChange, LastIndividuallyConfirmedInputList, player.LastReceivedInputFrameId, player.LastUdpReceivedInputFrameId]
 	RenderFrameBuffer                      *battle.RingBuffer // Indices are STRICTLY consecutive
 	LatestPlayerUpsyncedInputFrameId       int32
 	LastAllConfirmedInputFrameId           int32
@@ -189,6 +189,7 @@ func (pR *Room) AddPlayerIfPossible(pPlayerFromDbInit *Player, session *websocke
 	pPlayerFromDbInit.AckingInputFrameId = -1
 	pPlayerFromDbInit.LastSentInputFrameId = MAGIC_LAST_SENT_INPUT_FRAME_ID_NORMAL_ADDED
 	pPlayerFromDbInit.LastReceivedInputFrameId = MAGIC_LAST_SENT_INPUT_FRAME_ID_NORMAL_ADDED
+	pPlayerFromDbInit.LastUdpReceivedInputFrameId = MAGIC_LAST_SENT_INPUT_FRAME_ID_NORMAL_ADDED
 	pPlayerFromDbInit.BattleState = PlayerBattleStateIns.ADDED_PENDING_BATTLE_COLLIDER_ACK
 
 	pPlayerFromDbInit.ColliderRadius = DEFAULT_PLAYER_RADIUS // Hardcoded
@@ -230,6 +231,7 @@ func (pR *Room) ReAddPlayerIfPossible(pTmpPlayerInstance *Player, session *webso
 	pEffectiveInRoomPlayerInstance.AckingFrameId = -1
 	pEffectiveInRoomPlayerInstance.AckingInputFrameId = -1
 	pEffectiveInRoomPlayerInstance.LastSentInputFrameId = MAGIC_LAST_SENT_INPUT_FRAME_ID_READDED
+	// [WARNING] DON'T reset "player.LastReceivedInputFrameId" & "player.LastUdpReceivedInputFrameId" upon reconnection!
 	pEffectiveInRoomPlayerInstance.BattleState = PlayerBattleStateIns.READDED_PENDING_BATTLE_COLLIDER_ACK
 
 	pEffectiveInRoomPlayerInstance.ColliderRadius = DEFAULT_PLAYER_RADIUS // Hardcoded
@@ -1170,6 +1172,7 @@ func (pR *Room) markConfirmationIfApplicable(inputFrameUpsyncBatch []*pb.InputFr
 			continue
 		}
 		if clientInputFrameId < player.LastReceivedInputFrameId {
+			// [WARNING] It's important for correctness that we use "player.LastReceivedInputFrameId" instead of "player.LastUdpReceivedInputFrameId" here!
 			Logger.Debug(fmt.Sprintf("Omitting obsolete inputFrameUpsync#2: roomId=%v, playerId=%v, clientInputFrameId=%v, playerLastReceivedInputFrameId=%v, InputsBuffer=%v", pR.Id, playerId, clientInputFrameId, player.LastReceivedInputFrameId, pR.InputsBufferString(false)))
 			continue
 		}
@@ -1183,12 +1186,23 @@ func (pR *Room) markConfirmationIfApplicable(inputFrameUpsyncBatch []*pb.InputFr
 		targetInputFrameDownsync.ConfirmedList |= uint64(1 << uint32(player.JoinIndex-1))
 
 		if false == fromUDP {
-			// [WARNING] We have to distinguish whether or not the incoming batch is from UDP here, otherwise "pR.LatestPlayerUpsyncedInputFrameId - pR.LastAllConfirmedInputFrameId" might become unexpectedly large in case of "UDP packet loss + slow ws session"!
+			/*
+						   [WARNING] We have to distinguish whether or not the incoming batch is from UDP here, otherwise "pR.LatestPlayerUpsyncedInputFrameId - pR.LastAllConfirmedInputFrameId" might become unexpectedly large in case of "UDP packet loss + slow ws session"!
+
+						   Moreover, only ws session upsyncs should advance "player.LastReceivedInputFrameId" & "pR.LatestPlayerUpsyncedInputFrameId".
+
+			               Kindly note that the updates of "player.LastReceivedInputFrameId" could be discrete before and after reconnection.
+			*/
 			player.LastReceivedInputFrameId = clientInputFrameId
 			if clientInputFrameId > pR.LatestPlayerUpsyncedInputFrameId {
 				pR.LatestPlayerUpsyncedInputFrameId = clientInputFrameId
 			}
-			// It's safe (in terms of getting an eventually correct "RenderFrameBuffer") to put the following update of "pR.LastIndividuallyConfirmedInputList" which is ONLY used for prediction in "InputsBuffer" out of "false == fromUDP" block, but I'm still putting it in for convenient debugging.
+		}
+
+		if clientInputFrameId > player.LastUdpReceivedInputFrameId {
+			// No need to update "player.LastUdpReceivedInputFrameId" only when "true == fromUDP", we should keep "player.LastUdpReceivedInputFrameId >= player.LastReceivedInputFrameId" at any moment.
+			player.LastUdpReceivedInputFrameId = clientInputFrameId
+			// It's safe (in terms of getting an eventually correct "RenderFrameBuffer") to put the following update of "pR.LastIndividuallyConfirmedInputList" which is ONLY used for prediction in "InputsBuffer" out of "false == fromUDP" block.
 			pR.LastIndividuallyConfirmedInputList[player.JoinIndex-1] = inputFrameUpsync.Encoded
 		}
 	}
@@ -1256,6 +1270,7 @@ func (pR *Room) forceConfirmationIfApplicable(prevRenderFrameId int32) uint64 {
 	totPlayerCnt := uint32(pR.Capacity)
 	allConfirmedMask := uint64((1 << totPlayerCnt) - 1)
 	unconfirmedMask := uint64(0)
+	// As "pR.LastAllConfirmedInputFrameId" can be advanced by UDP but "pR.LatestPlayerUpsyncedInputFrameId" could only be advanced by ws session, when the following condition is met we know that the slow ticker is really in trouble!
 	if pR.LatestPlayerUpsyncedInputFrameId > (pR.LastAllConfirmedInputFrameId + pR.InputFrameUpsyncDelayTolerance + 1) {
 		// Type#1 check whether there's a significantly slow ticker among players
 		oldLastAllConfirmedInputFrameId := pR.LastAllConfirmedInputFrameId

frontend/assets/scripts/Map.js

@@ -46,7 +46,7 @@ window.onUdpMessage = (args) => {
       const renderedInputFrameIdUpper = gopkgs.ConvertToDelayedInputFrameId(self.renderFrameId);
       const peerJoinIndex = req.joinIndex;
       const batch = req.inputFrameUpsyncBatch;
-      self.onPeerInputFrameUpsync(peerJoinIndex, batch);
+      self.onPeerInputFrameUpsync(peerJoinIndex, batch, true);
     }
   }
 };
@@ -166,16 +166,10 @@ cc.Class({
       return [previousSelfInput, existingInputFrame.InputList[joinIndex - 1]];
     }
 
-    const lastAllConfirmedInputFrame = self.recentInputCache.GetByFrameId(self.lastAllConfirmedInputFrameId);
     const prefabbedInputList = new Array(self.playerRichInfoDict.size).fill(0);
     // the returned "gopkgs.NewInputFrameDownsync.InputList" is immutable, thus we can only modify the values in "prefabbedInputList"
     for (let k in prefabbedInputList) {
-      if (null != previousInputFrameDownsync) {
+      prefabbedInputList[k] = self.lastIndividuallyConfirmedInputList[k];
-        prefabbedInputList[k] = previousInputFrameDownsync.InputList[k];
-      }
-      if (0 <= self.lastAllConfirmedInputFrameId && inputFrameId - 1 > self.lastAllConfirmedInputFrameId) {
-        prefabbedInputList[k] = lastAllConfirmedInputFrame.InputList[k];
-      }
       // Don't predict "btnA & btnB"!
       prefabbedInputList[k] = (prefabbedInputList[k] & 15);
     }
@@ -355,6 +349,8 @@ cc.Class({
     self.lastUpsyncInputFrameId = -1;
     self.chaserRenderFrameId = -1; // at any moment, "chaserRenderFrameId <= renderFrameId", but "chaserRenderFrameId" would fluctuate according to "onInputFrameDownsyncBatch"
 
+    self.lastIndividuallyConfirmedInputFrameId = new Array(window.boundRoomCapacity).fill(-1);
+    self.lastIndividuallyConfirmedInputList = new Array(window.boundRoomCapacity).fill(0);
     self.recentRenderCache = new RingBuffer(self.renderCacheSize);
 
     self.recentInputCache = gopkgs.NewRingBufferJs((self.renderCacheSize >> 1) + 1);
@@ -814,6 +810,16 @@ cc.Class({
     return true;
   },
 
+  _markConfirmationIfApplicable() {
+    const self = this;
+    while (self.recentInputCache.StFrameId <= self.lastAllConfirmedInputFrameId && self.lastAllConfirmedInputFrameId < self.recentInputCache.EdFrameId) {
+      const inputFrameDownsync = self.recentInputCache.GetByFrameId(self.lastAllConfirmedInputFrameId);
+      if (null == inputFrameDownsync) break;
+      if (self._allConfirmed(inputFrameDownsync.ConfirmedList)) break;
+      ++self.lastAllConfirmedInputFrameId;
+    }
+  },
+
   onInputFrameDownsyncBatch(batch /* []*pb.InputFrameDownsync */ ) {
     // TODO: find some kind of synchronization mechanism against "getOrPrefabInputFrameUpsync"!
     if (null == batch) {
@@ -835,8 +841,6 @@ cc.Class({
       if (inputFrameDownsyncId <= self.lastAllConfirmedInputFrameId) {
         continue;
       }
-      // [WARNING] Take all "inputFrameDownsync" from backend as all-confirmed, it'll be later checked by "rollbackAndChase". 
-      self.lastAllConfirmedInputFrameId = inputFrameDownsyncId;
       const localInputFrame = self.recentInputCache.GetByFrameId(inputFrameDownsyncId);
       if (null != localInputFrame
         &&
@@ -846,14 +850,23 @@ cc.Class({
       ) {
         firstPredictedYetIncorrectInputFrameId = inputFrameDownsyncId;
       }
+      // [WARNING] Take all "inputFrameDownsync" from backend as all-confirmed, it'll be later checked by "rollbackAndChase". 
       inputFrameDownsync.confirmedList = (1 << self.playerRichInfoDict.size) - 1;
       const inputFrameDownsyncLocal = gopkgs.NewInputFrameDownsync(inputFrameDownsync.inputFrameId, inputFrameDownsync.inputList, inputFrameDownsync.confirmedList); // "battle.InputFrameDownsync" in "jsexport"
+      for (let j in self.playerRichInfoArr) {
+        const jj = parseInt(j);
+        if (inputFrameDownsync.inputFrameId > self.lastIndividuallyConfirmedInputFrameId[jj]) {
+          self.lastIndividuallyConfirmedInputFrameId[jj] = inputFrameDownsync.inputFrameId;
+          self.lastIndividuallyConfirmedInputList[jj] = inputFrameDownsync.inputList[jj];
+        }
+      }
       //console.log(`Confirmed inputFrameId=${inputFrameDownsync.inputFrameId}`);
       const [ret, oldStFrameId, oldEdFrameId] = self.recentInputCache.SetByFrameId(inputFrameDownsyncLocal, inputFrameDownsync.inputFrameId);
       if (window.RING_BUFF_FAILED_TO_SET == ret) {
         throw `Failed to dump input cache (maybe recentInputCache too small)! inputFrameDownsync.inputFrameId=${inputFrameDownsync.inputFrameId}, lastAllConfirmedInputFrameId=${self.lastAllConfirmedInputFrameId}; recentRenderCache=${self._stringifyRecentRenderCache(false)}, recentInputCache=${self._stringifyRecentInputCache(false)}`;
       }
     }
+    self._markConfirmationIfApplicable();
 
     if (null == firstPredictedYetIncorrectInputFrameId) return;
     const renderFrameId1 = gopkgs.ConvertToFirstUsedRenderFrameId(firstPredictedYetIncorrectInputFrameId) - 1;
@@ -879,7 +892,7 @@ batchInputFrameIdRange=[${batch[0].inputFrameId}, ${batch[batch.length - 1].inpu
     self.networkDoctor.logRollbackFrames(self.renderFrameId - self.chaserRenderFrameId);
   },
 
-  onPeerInputFrameUpsync(peerJoinIndex, batch /* []*pb.InputFrameDownsync */ ) {
+  onPeerInputFrameUpsync(peerJoinIndex, batch, fromUDP) {
     // TODO: find some kind of synchronization mechanism against "getOrPrefabInputFrameUpsync"!
     // See `<proj-root>/ConcerningEdgeCases.md` for why this method exists.
     if (null == batch) {
@@ -897,32 +910,39 @@ batchInputFrameIdRange=[${batch[0].inputFrameId}, ${batch[batch.length - 1].inpu
     //console.log(`Received peer inputFrameUpsync batch w/ inputFrameId in [${batch[0].inputFrameId}, ${batch[batch.length - 1].inputFrameId}] for prediction assistance`);
     const renderedInputFrameIdUpper = gopkgs.ConvertToDelayedInputFrameId(self.renderFrameId);
     for (let k in batch) {
-      const inputFrameDownsync = batch[k];
+      const inputFrame = batch[k]; // could be either "pb.InputFrameDownsync" or "pb.InputFrameUpsync", depending on "fromUDP"
-      const inputFrameDownsyncId = inputFrameDownsync.inputFrameId;
+      const inputFrameId = inputFrame.inputFrameId;
-      if (inputFrameDownsyncId < renderedInputFrameIdUpper) {
+      if (inputFrameId < renderedInputFrameIdUpper) {
         // Avoid obfuscating already rendered history
         continue;
       }
-      if (inputFrameDownsyncId <= self.lastAllConfirmedInputFrameId) {
+      if (inputFrameId <= self.lastAllConfirmedInputFrameId) {
+        // [WARNING] Don't reject it by "inputFrameId <= self.lastIndividuallyConfirmedInputFrameId[peerJoinIndex-1]", the arrival of UDP packets might not reserve their sending order!
         continue;
       }
-      self.getOrPrefabInputFrameUpsync(inputFrameDownsyncId); // Make sure that inputFrame exists locally
+      self.getOrPrefabInputFrameUpsync(inputFrameId); // Make sure that inputFrame exists locally
-      const existingInputFrame = self.recentInputCache.GetByFrameId(inputFrameDownsyncId);
+      const existingInputFrame = self.recentInputCache.GetByFrameId(inputFrameId);
       if (0 < (existingInputFrame.ConfirmedList & (1 << (peerJoinIndex - 1)))) {
         continue;
       }
+      const peerEncodedInput = (true == fromUDP ? inputFrame.encoded : inputFrame.inputList[peerJoinIndex - 1]);
+      if (inputFrameId > self.lastIndividuallyConfirmedInputFrameId[peerJoinIndex - 1]) {
+        self.lastIndividuallyConfirmedInputFrameId[peerJoinIndex - 1] = inputFrameId;
+        self.lastIndividuallyConfirmedInputList[peerJoinIndex - 1] = peerEncodedInput;
+      }
       effCnt += 1;
       // the returned "gopkgs.NewInputFrameDownsync.InputList" is immutable, thus we can only modify the values in "newInputList" and "newConfirmedList"!
       let newInputList = new Array(self.playerRichInfoDict.size).fill(0);
       for (let i in existingInputFrame.InputList) {
         newInputList[i] = existingInputFrame.InputList[i];
       }
+      newInputList[peerJoinIndex - 1] = peerEncodedInput;
       let newConfirmedList = (existingInputFrame.confirmedList | (1 << (peerJoinIndex - 1)));
-      // No need to change "lastAllConfirmedInputFrameId", leave it to "onInputFrameDownsyncBatch" -- we're just helping prediction here
+      const newInputFrameDownsyncLocal = gopkgs.NewInputFrameDownsync(inputFrameId, newInputList, newConfirmedList);
-      const newInputFrameDownsyncLocal = gopkgs.NewInputFrameDownsync(inputFrameDownsyncId, newInputList, newConfirmedList);
+      self.recentInputCache.SetByFrameId(newInputFrameDownsyncLocal, inputFrameId);
-      self.recentInputCache.SetByFrameId(newInputFrameDownsyncLocal, inputFrameDownsyncId);
     }
     if (0 < effCnt) {
+      self._markConfirmationIfApplicable();
       self.networkDoctor.logPeerInputFrameUpsync(batch[0].inputFrameId, batch[batch.length - 1].inputFrameId);
     }
   },

frontend/assets/scripts/WsSessionMgr.js

@@ -335,7 +335,7 @@ window.initSecondarySession = function(onopenCb, boundRoomId) {
       //console.log(`Got non-empty onmessage decoded: resp.act=${resp.act}`);
       switch (resp.act) {
         case window.DOWNSYNC_MSG_ACT_PEER_INPUT_BATCH:
-          mapIns.onPeerInputFrameUpsync(resp.peerJoinIndex, resp.inputFrameDownsyncBatch);
+          mapIns.onPeerInputFrameUpsync(resp.peerJoinIndex, resp.inputFrameDownsyncBatch, false);
           break;
         default:
           break;