Enhanced UDP message callback handling.

frontend/build-templates/jsb-link/frameworks/runtime-src/Classes/send_ring_buff.cpp

@@ -33,31 +33,70 @@ SendWork* SendRingBuff::pop() {
 
 // Recving
 void RecvRingBuff::put(char* newBytes, size_t newBytesLen) {
-    while (0 < cnt && cnt >= n) {
+    RecvWork* slotEle = (&eles[ed.load()]); // Save for later update
+
+    int oldCnt = cnt.load();
+    int oldSt = st.load(); // Used to guard against "cnt decremented in pop(...), but st not yet incremented and thus return value not yet copied to avoid contamination"
+    int tried = 0;
+    while (n <= oldCnt && !ed.compare_exchange_weak(oldSt, oldSt) && 3 > tried) {
         // Make room for the new element
-        this->pop();
+        this->pop(NULL);
+        oldCnt = cnt.load(); // If "pop()" above failed, it'd only be due to concurrent calls to "pop()", either way the updated "cnt" should be good to go
+        oldSt = st.load();
+        ++tried;
     }
-    eles[ed].bytesLen = newBytesLen;
-    memset(eles[ed].ui8Arr, 0, sizeof eles[ed].ui8Arr);
+    if (n <= oldCnt && !ed.compare_exchange_weak(oldSt, oldSt) && 3 == tried) {
+        // Failed silently, UDP packet can be dropped.
+        return;
+    }
+    slotEle->bytesLen = newBytesLen;
+    memset(slotEle->ui8Arr, 0, sizeof slotEle->ui8Arr);
     for (int i = 0; i < newBytesLen; i++) {
-        *(eles[ed].ui8Arr + i) = *(newBytes + i);
+        *(slotEle->ui8Arr + i) = *(newBytes + i);
     }
+
+    // No need to compare-and-swap, only "UvRecvThread" will access "RecvRingBuff.ed".
     ed++;
-    cnt++;
     if (ed >= n) {
         ed -= n; // Deliberately not using "%" operator for performance concern
     }
+
+    // Only increment cnt when the putting of new element is fully done.
+    cnt++;
 }
 
-RecvWork* RecvRingBuff::pop() {
-    if (0 == cnt) {
+bool RecvRingBuff::pop(RecvWork* out) {
+    int oldCnt = std::atomic_fetch_sub(&cnt, 1);
+    /*
+    [WARNING] 
+
+    After here, two cases should be taken care of.
+    1. If "n == oldCnt", we need guard against "put" to avoid contaminating "ret" by the "putting".
+    2. If "0 >= oldCnt", we need guard against another "pop" to avoid over-popping. 
+    */
+    if (0 >= oldCnt) {
+        // "pop" could be accessed by either "GameThread/pollUdpRecvRingBuff" or "UvRecvThread/put", thus we should be proactively guard against concurrent popping while "1 == cnt"
+        ++cnt;
         return NULL;
     }
-    RecvWork* ret = &(eles[st]);
-    cnt--;
-    st++;
-    if (st >= n) {
-        st -= n;
+
+    // When concurrent "pop"s reach here, over-popping is definitely avoided.
+    int oldSt = st.load();
+    if (out) {
+        RecvWork* src = (&eles[oldSt]);
+        memset(out->ui8Arr, 0, sizeof out->ui8Arr);
+        memcpy(out->ui8Arr, src, src->bytesLen);
+        out->bytesLen = src->bytesLen;
+    }
+    int newSt = oldSt + 1;
+    if (newSt >= n) {
+        newSt -= n;
+    }
+    if (st.compare_exchange_weak(oldSt, newSt)) {
+        return true;
+    } else {
+        // Failed concurrent access should recover the "cnt"
+        ++cnt;
+        return false;
     }
-    return ret;
 }

frontend/build-templates/jsb-link/frameworks/runtime-src/Classes/send_ring_buff.hpp

@@ -4,6 +4,8 @@
 #include "uv/uv.h"
 #define __SSIZE_T // Otherwise "ssize_t" would have conflicting macros error that stops compiling
 
+#include <atomic>
+
 int const RING_BUFF_CONSECUTIVE_SET = 0;
 int const RING_BUFF_NON_CONSECUTIVE_SET = 1;
 int const RING_BUFF_FAILED_TO_SET = 2;
@@ -48,10 +50,17 @@ public:
 	size_t bytesLen;
 };
 
-// [WARNING] This class is specific to "RecvWork"
+/*
+[WARNING] This class is specific to "RecvWork"; its "put" and "pop" methods are designed to be thread-safe & lock-free for our particular case, i.e. only concurrent access from "UvRecvThread" & "GameThread", in a sense more sophisticated than the Golang or JavaScript versions. 
+
+There's yet no plan to support thread-safe & lock-free "getByFrameId/setByFrameId" -- being thread-safe is easy by use of mutex, which is very SLOWWWWW when used in 60fps race-conditions.
+
+The generic "thread-safe, lock-free ring buffer or circular buffer" is a big problem, widely discussed over the internet and in literatures, search "lock-free circular buffer" for more information.
+*/
 class RecvRingBuff {
 public:
-	int ed, st, n, cnt;
+	int n;
+	std::atomic_int ed, st, cnt;
 	RecvWork eles[maxBuffedMsgs]; // preallocated on stack to save heap alloc/dealloc time
 	RecvRingBuff(int newN) {
 		this->n = newN;
@@ -60,6 +69,6 @@ public:
 
 	void put(char* newBytes, size_t newBytesLen);
 
-	RecvWork* pop();
+	bool pop(RecvWork* out);
 };
 #endif

frontend/build-templates/jsb-link/frameworks/runtime-src/Classes/udp_session.cpp

@@ -72,9 +72,9 @@ void _onRead(uv_udp_t* req, ssize_t nread, uv_buf_t const* buf, struct sockaddr
 #if defined(COCOS2D_DEBUG) && (COCOS2D_DEBUG > 0)
         CCLOG("UDP received %u bytes inputFrameUpsync from %s:%d", nread, ip, port);
 #endif
-        uv_mutex_lock(&recvRingBuffLock);
+        //uv_mutex_lock(&recvRingBuffLock);
         recvRingBuff->put(buf->base, nread);
-        uv_mutex_unlock(&recvRingBuffLock);
+        //uv_mutex_unlock(&recvRingBuffLock);
     }
     free(buf->base);
 
@@ -352,15 +352,12 @@ bool DelayNoMore::UdpSession::broadcastInputFrameUpsync(BYTEC* const bytes, size
 bool DelayNoMore::UdpSession::pollUdpRecvRingBuff() {
     // This function is called by GameThread 60 fps.
 
-    if (0 >= recvRingBuff->cnt) {
-        // This check is NOT thread-safe, but as "pollUdpRecvRingBuff" is called by GameThread, we want it to lock as few as possible.
-        return true;
-    }
+    //uv_mutex_lock(&recvRingBuffLock);
+    while (true) {
+        RecvWork f;
+        bool res = recvRingBuff->pop(&f); 
+        if (!res) return false;
 
-    uv_mutex_lock(&recvRingBuffLock);
-    RecvWork* f = NULL;
-    while (0 < recvRingBuff->cnt) {
-        f = recvRingBuff->pop();        
         // [WARNING] Declaring "AutoHandleScope" is critical here, otherwise "onUdpMessageCb.toObject()" wouldn't be recognized as a function of the ScriptEngine!
         se::AutoHandleScope hs;
         // [WARNING] Use of the "ScriptEngine" is only allowed in "GameThread a.k.a. CocosThread"!
@@ -368,7 +365,7 @@ bool DelayNoMore::UdpSession::pollUdpRecvRingBuff() {
         se::ScriptEngine::getInstance()->getGlobalObject()->getProperty("onUdpMessage", &onUdpMessageCb);
         if (onUdpMessageCb.isObject() && onUdpMessageCb.toObject()->isFunction()) {
             //CCLOG("UDP received %d bytes upsync -- 1", nread);
-            se::Object* const gameThreadMsg = se::Object::createTypedArray(se::Object::TypedArrayType::UINT8, f->ui8Arr, f->bytesLen);
+            se::Object* const gameThreadMsg = se::Object::createTypedArray(se::Object::TypedArrayType::UINT8, f.ui8Arr, f.bytesLen);
             //CCLOG("UDP received %d bytes upsync -- 2", nread);
             se::ValueArray args = { se::Value(gameThreadMsg) };
 
@@ -382,6 +379,6 @@ bool DelayNoMore::UdpSession::pollUdpRecvRingBuff() {
             //CCLOG("UDP received %d bytes upsync -- 4", nread);
         }
     }
-    uv_mutex_unlock(&recvRingBuffLock);
+    //uv_mutex_unlock(&recvRingBuffLock);
     return true;
 }

jsexport/battle/battle.go

@@ -755,7 +755,7 @@ func ApplyInputFrameDownsyncDynamicsOnSingleRenderFrame(inputsBuffer *RingBuffer
 		// Reset playerCollider position from the "virtual grid position"
 		newVx, newVy := currPlayerDownsync.VirtualGridX+currPlayerDownsync.VelX, currPlayerDownsync.VirtualGridY+currPlayerDownsync.VelY
 		if 0 >= thatPlayerInNextFrame.Hp && 0 == thatPlayerInNextFrame.FramesToRecover {
-			// Revive
+			// Revive from Dying
 			newVx, newVy = currPlayerDownsync.RevivalVirtualGridX, currPlayerDownsync.RevivalVirtualGridY
 			thatPlayerInNextFrame.CharacterState = ATK_CHARACTER_STATE_IDLE1
 			thatPlayerInNextFrame.Hp = currPlayerDownsync.MaxHp
@@ -945,7 +945,8 @@ func ApplyInputFrameDownsyncDynamicsOnSingleRenderFrame(inputsBuffer *RingBuffer
 			if fallStopping {
 				thatPlayerInNextFrame.VelY = 0
 				thatPlayerInNextFrame.VelX = 0
-				if ATK_CHARACTER_STATE_BLOWN_UP1 == thatPlayerInNextFrame.CharacterState {
+				if ATK_CHARACTER_STATE_DYING == thatPlayerInNextFrame.CharacterState {
+				} else if ATK_CHARACTER_STATE_BLOWN_UP1 == thatPlayerInNextFrame.CharacterState {
 					thatPlayerInNextFrame.CharacterState = ATK_CHARACTER_STATE_LAY_DOWN1
 					thatPlayerInNextFrame.FramesToRecover = chConfig.LayDownFramesToRecover
 				} else {
@@ -960,9 +961,11 @@ func ApplyInputFrameDownsyncDynamicsOnSingleRenderFrame(inputsBuffer *RingBuffer
 					thatPlayerInNextFrame.FramesToRecover = 0
 				}
 			} else {
-				// landedOnGravityPushback not fallStopping, could be in LayDown or GetUp
+				// landedOnGravityPushback not fallStopping, could be in LayDown or GetUp or Dying
 				if _, existent := nonAttackingSet[thatPlayerInNextFrame.CharacterState]; existent {
-					if ATK_CHARACTER_STATE_LAY_DOWN1 == thatPlayerInNextFrame.CharacterState {
+					if ATK_CHARACTER_STATE_DYING == thatPlayerInNextFrame.CharacterState {
+						// No update needed for Dying
+					} else if ATK_CHARACTER_STATE_LAY_DOWN1 == thatPlayerInNextFrame.CharacterState {
 						if 0 == thatPlayerInNextFrame.FramesToRecover {
 							thatPlayerInNextFrame.CharacterState = ATK_CHARACTER_STATE_GET_UP1
 							thatPlayerInNextFrame.FramesToRecover = chConfig.GetUpFramesToRecover