Compare commits

...

4 Commits

Author SHA1 Message Date
genxium
7b878ff947 Fixes for backend Golang select-multi-channel implementation. 2023-02-21 22:07:48 +08:00
genxium
c78c480f99 Enhanced UDP session resource management. 2023-02-21 15:21:15 +08:00
genxium
b50874f5c4 Enhanced RecvRingBuff in cpp. 2023-02-21 11:54:06 +08:00
genxium
f1db2972fd Updates for RecvRingBuff. 2023-02-20 08:53:06 +08:00
8 changed files with 97 additions and 37 deletions

View File

@@ -25,9 +25,16 @@ import (
"go.uber.org/zap"
"net"
// _ "net/http/pprof"
)
func main() {
/*
// Only used for profiling
go func() {
http.ListenAndServe("0.0.0.0:6060", nil)
}()
*/
MustParseConfig()
MustParseConstants()
storage.Init()

View File

@@ -483,7 +483,7 @@ func (pR *Room) StartBattle() {
*/
totalElapsedNanos := utils.UnixtimeNano() - battleStartedAt
nextRenderFrameId := int32((totalElapsedNanos + pR.dilutedRollbackEstimatedDtNanos - 1) / pR.dilutedRollbackEstimatedDtNanos) // fast ceiling
toSleepNanos := int64(0)
toSleepNanos := int64(pR.dilutedRollbackEstimatedDtNanos >> 1) // Sleep half-frame time by default
if nextRenderFrameId > pR.RenderFrameId {
if 0 == pR.RenderFrameId {
// It's important to send kickoff frame iff "0 == pR.RenderFrameId && nextRenderFrameId > pR.RenderFrameId", otherwise it might send duplicate kickoff frames
@@ -515,7 +515,7 @@ func (pR *Room) StartBattle() {
pR.LastRenderFrameIdTriggeredAt = utils.UnixtimeNano()
elapsedInCalculation := (utils.UnixtimeNano() - stCalculation)
toSleepNanos = pR.dilutedRollbackEstimatedDtNanos - elapsedInCalculation // don't sleep if "nextRenderFrame == pR.RenderFrameId"
toSleepNanos = pR.dilutedRollbackEstimatedDtNanos - elapsedInCalculation
if elapsedInCalculation > pR.RollbackEstimatedDtNanos {
Logger.Warn(fmt.Sprintf("SLOW FRAME! Elapsed time statistics: roomId=%v, room.RenderFrameId=%v, elapsedInCalculation=%v ns, dynamicsDuration=%v ns, RollbackEstimatedDtNanos=%v, dilutedRollbackEstimatedDtNanos=%v", pR.Id, pR.RenderFrameId, elapsedInCalculation, dynamicsDuration, pR.RollbackEstimatedDtNanos, pR.dilutedRollbackEstimatedDtNanos))
}
@@ -544,13 +544,13 @@ func (pR *Room) StartBattle() {
}
select {
// [WARNING] DON'T put a "default" block here! Otherwise "for { select {... default: } }" pattern would NEVER block on empty channel and thus consume a lot of CPU time unnecessarily!
case inputsBufferSnapshot := <-playerDownsyncChan:
pR.downsyncToSinglePlayer(playerId, player, inputsBufferSnapshot.RefRenderFrameId, inputsBufferSnapshot.UnconfirmedMask, inputsBufferSnapshot.ToSendInputFrameDownsyncs, inputsBufferSnapshot.ShouldForceResync)
//Logger.Info(fmt.Sprintf("Sent inputsBufferSnapshot(refRenderFrameId:%d, unconfirmedMask:%v) to for (roomId: %d, playerId:%d)#2", inputsBufferSnapshot.RefRenderFrameId, inputsBufferSnapshot.UnconfirmedMask, pR.Id, playerId))
case inputsBufferSnapshot2 := <-playerSecondaryDownsyncChan:
pR.downsyncPeerInputFrameUpsyncToSinglePlayer(playerId, player, inputsBufferSnapshot2.ToSendInputFrameDownsyncs, inputsBufferSnapshot2.PeerJoinIndex)
//Logger.Info(fmt.Sprintf("Sent secondary inputsBufferSnapshot to for (roomId: %d, playerId:%d)#2", pR.Id, playerId))
default:
}
}
}

View File

@@ -461,7 +461,7 @@
"array": [
0,
0,
216.50635094610968,
210.43877906529718,
0,
0,
0,

View File

@@ -271,6 +271,13 @@ window.initPersistentSessionClient = function(onopenCb, expectedRoomId) {
clientSession.onclose = function(evt) {
// [WARNING] The callback "onclose" might be called AFTER the webpage is refreshed with "1001 == evt.code".
console.warn(`The WS clientSession is closed: evt=${JSON.stringify(evt)}, evt.code=${evt.code}`);
if (cc.sys.isNative) {
if (mapIns.frameDataLoggingEnabled) {
console.warn(`${mapIns._stringifyRdfIdToActuallyUsedInput()}
`);
}
DelayNoMore.UdpSession.closeUdpSession();
}
switch (evt.code) {
case constants.RET_CODE.CLIENT_MISMATCHED_RENDER_FRAME:
break;
@@ -312,15 +319,13 @@ window.initPersistentSessionClient = function(onopenCb, expectedRoomId) {
default:
if (cc.sys.isNative) {
// [WARNING] This could be a BUG in CocosCreator JSB implementation of WebSocket client, the "evt.code" is always "undefined" in the "onclose" callback!
if (mapIns.frameDataLoggingEnabled) {
console.warn(`${mapIns._stringifyRdfIdToActuallyUsedInput()}
`);
}
DelayNoMore.UdpSession.closeUdpSession();
if (window.ALL_BATTLE_STATES.IN_SETTLEMENT != mapIns.battleState && window.ALL_BATTLE_STATES.IN_DISMISSAL != mapIns.battleState) {
mapIns.popupSimplePressToGo("Disconnected unexpectedly, please retry", false, () => {
window.clearLocalStorageAndBackToLoginScene(true);
});
}
}
break;
}
};

File diff suppressed because one or more lines are too long

View File

@@ -32,20 +32,43 @@ SendWork* SendRingBuff::pop() {
}
// Recving
bool isFullWithLoadedVals(int n, int oldCnt, int oldSt, int oldEd) {
return (n <= oldCnt && oldEd == oldSt) || (n > oldCnt && 0 < oldCnt && oldEd == oldSt);
}
void RecvRingBuff::put(char* newBytes, size_t newBytesLen) {
RecvWork* slotEle = (&eles[ed.load()]); // Save for later update
// "RecvRingBuff.ed" is only accessed in "UvRecvThread", thus the order of it relative to the other two is not important.
int oldEd = ed.load();
// We want to increase the success rate of "pop()" if it's being executed by "GameThread/pollUdpRecvRingBuff", thus the below order of loading is IMPORTANT, i.e. load "cnt" first because it's decremented earlier than "st" being incremented.
int oldCnt = cnt.load();
/*
[WARNING]
Note that "RecvRingBuff.st" might have decremented in "GameThread" by a successful "pop()" between "cnt.load()" and "st.load()" here in "UvRecvThread"! Therefore "n <= oldCnt" doesn't necessarily imply "oldEd == oldSt"!
*/
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) {
/*
1. When "n <= oldCnt", it might still be true "oldEd != oldSt" (see the note above);
2. When "n > oldCnt", it might still be true that "oldEd == oldSt" if "pop()" hasn't successfully incremented "st" due to any reason;
3. When "oldEd == oldSt", it doesn't imply anything useful, because any of the following could be true
- a. "n <= oldCnt", i.e. the ringbuff is full
- b. "n > oldCnt && 0 < oldCnt" during the execution of "pop()", i.e. the ringbuff is still effectively full
- c. "n > oldCnt && 0 == oldCnt", i.e. the ringbuff is empty
*/
bool isFull = isFullWithLoadedVals(n, oldCnt, oldSt, oldEd);
while (isFull && 3 > tried) {
// Make room for the new element
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();
isFull = isFullWithLoadedVals(n, oldCnt, oldSt, oldEd);
++tried;
}
if (n <= oldCnt && !ed.compare_exchange_weak(oldSt, oldSt) && 3 == tried) {
if (isFull && 3 == tried) {
// Failed silently, UDP packet can be dropped.
return;
}
@@ -56,11 +79,13 @@ void RecvRingBuff::put(char* newBytes, size_t newBytesLen) {
}
// No need to compare-and-swap, only "UvRecvThread" will access "RecvRingBuff.ed".
ed++;
if (ed >= n) {
ed -= n; // Deliberately not using "%" operator for performance concern
int newEd = oldEd+1;
if (newEd >= n) {
newEd -= n; // Deliberately not using "%" operator for performance concern
}
ed.compare_exchange_weak(oldEd, newEd); // Definitely succeeds because "RecvRingBuff.ed" is only accessed in "UvRecvThread"
// Only increment cnt when the putting of new element is fully done.
cnt++;
}

View File

@@ -160,7 +160,6 @@ void startRecvLoop(void* arg) {
int uvCloseRet = uv_loop_close(l);
CCLOG("UDP recv loop is closed in UvRecvThread, uvCloseRet=%d", uvCloseRet);
uv_mutex_destroy(&recvRingBuffLock);
}
void startSendLoop(void* arg) {
@@ -174,7 +173,6 @@ void startSendLoop(void* arg) {
int uvCloseRet = uv_loop_close(l);
CCLOG("UDP send loop is closed in UvSendThread, uvCloseRet=%d", uvCloseRet);
uv_mutex_destroy(&sendRingBuffLock);
}
int initSendLoop(struct sockaddr const* pUdpAddr) {
@@ -189,9 +187,6 @@ int initSendLoop(struct sockaddr const* pUdpAddr) {
uv_mutex_init(&sendRingBuffLock);
sendRingBuff = new SendRingBuff(maxBuffedMsgs);
uv_mutex_init(&recvRingBuffLock);
recvRingBuff = new RecvRingBuff(maxBuffedMsgs);
uv_async_init(sendLoop, &uvSendLoopStopSig, _onUvStopSig);
uv_async_init(sendLoop, &uvSendLoopTriggerSig, _onUvSthNewToSend);
@@ -208,6 +203,9 @@ bool initRecvLoop(struct sockaddr const* pUdpAddr) {
CCLOGERROR("Failed to bind recv; recvSockInitRes=%d, recvbindRes=%d, reason=%s", recvSockInitRes, recvbindRes, uv_strerror(recvbindRes));
exit(-1);
}
uv_mutex_init(&recvRingBuffLock);
recvRingBuff = new RecvRingBuff(maxBuffedMsgs);
uv_udp_recv_start(udpRecvSocket, _allocBuffer, _onRead);
uv_async_init(recvLoop, &uvRecvLoopStopSig, _onUvStopSig);
@@ -249,6 +247,12 @@ bool DelayNoMore::UdpSession::openUdpSession(int port) {
bool DelayNoMore::UdpSession::closeUdpSession() {
CCLOG("About to close udp session and dealloc all resources...");
/*
[WARNING] It's possible that "closeUdpSession" is called when "openUdpSession" was NEVER CALLED, thus we have to avoid program crash in this case.
In general one shouldn't just check the state of "sendTid" by whether or not "NULL == sendLoop", but in this particular game, both "openUdpSession" and "closeUdpSession" are only called from "GameThread", no thread-safety concern here, i.e. if "openUdpSession" was ever called earlier, then "sendLoop" wouldn't be NULL when "closeUdpSession" is later called.
*/
if (NULL != sendLoop) {
uv_async_send(&uvSendLoopStopSig);
CCLOG("Signaling UvSendThread to end in GameThread...");
uv_thread_join(&sendTid);
@@ -256,6 +260,14 @@ bool DelayNoMore::UdpSession::closeUdpSession() {
free(sendLoop);
delete sendRingBuff;
udpSendSocket = NULL;
sendLoop = NULL;
sendRingBuff = NULL;
uv_mutex_destroy(&sendRingBuffLock);
}
if (NULL != recvLoop) {
uv_async_send(&uvRecvLoopStopSig); // The few if not only guaranteed thread safe utility of libuv :) See http://docs.libuv.org/en/v1.x/async.html#c.uv_async_send
CCLOG("Signaling UvRecvThread to end in GameThread...");
uv_thread_join(&recvTid);
@@ -263,6 +275,13 @@ bool DelayNoMore::UdpSession::closeUdpSession() {
free(recvLoop);
delete recvRingBuff;
udpRecvSocket = NULL;
recvLoop = NULL;
recvRingBuff = NULL;
uv_mutex_destroy(&recvRingBuffLock);
}
CCLOG("Closed udp session and dealloc all resources in GameThread...");
return true;
@@ -356,8 +375,10 @@ bool DelayNoMore::UdpSession::pollUdpRecvRingBuff() {
while (true) {
RecvWork f;
bool res = recvRingBuff->pop(&f);
if (!res) return false;
if (!res) {
// Deliberately returning "true" here to prevent "jswrapper" from printing "Failed to invoke Xxx..." too frequently
return true;
}
// [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"!

View File

@@ -698,6 +698,8 @@ func ApplyInputFrameDownsyncDynamicsOnSingleRenderFrame(inputsBuffer *resolv.Rin
} else if stoppingFromWalking {
thatPlayerInNextFrame.FramesToRecover = chConfig.InertiaFramesToRecover
} else {
// Updates CharacterState and thus the animation to make user see graphical feedback asap.
thatPlayerInNextFrame.CharacterState = ATK_CHARACTER_STATE_WALKING
thatPlayerInNextFrame.FramesToRecover = ((chConfig.InertiaFramesToRecover >> 1) + (chConfig.InertiaFramesToRecover >> 2))
}
} else {