gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'redis_6_merge' into keydbpro

Browse Source 
Former-commit-id: 4bd4159e2d8a7cdd0af9719776dca3e7d161d166
This commit is contained in:
John Sully 2020-02-11 03:48:24 -05:00
commit 2af2e5c730
1 changed files with 13 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/networking.cpp
View File

@ -1037,29 +1037,23 @@ int clientHasPendingReplies(client *c) {
return (c->bufpos || listLength(c->reply)) && !(c->flags & CLIENT_CLOSE_ASAP); return (c->bufpos || listLength(c->reply)) && !(c->flags & CLIENT_CLOSE_ASAP);
} }
static std::atomic<int> rgacceptsInFlight[MAX_EVENT_LOOPS];
int chooseBestThreadForAccept() int chooseBestThreadForAccept()
{ {
listIter li;
listNode *ln;
int rgcclients[MAX_EVENT_LOOPS] = {0};
listRewind(g_pserver->clients, &li);
while ((ln = listNext(&li)) != nullptr)
{
client *c = (client*)listNodeValue(ln);
if (c->iel < 0)
continue;
rgcclients[c->iel]++;
}
int ielMinLoad = 0; int ielMinLoad = 0;
int cclientsMin = INT_MAX;
for (int iel = 0; iel < cserver.cthreads; ++iel) for (int iel = 0; iel < cserver.cthreads; ++iel)
{ {
if (rgcclients[iel] < cserver.thread_min_client_threshold) int cclientsThread;
atomicGet(g_pserver->rgthreadvar[iel].cclients, cclientsThread);
cclientsThread += rgacceptsInFlight[iel].load(std::memory_order_relaxed);
if (cclientsThread < cserver.thread_min_client_threshold)
return iel; return iel;
if (rgcclients[iel] < rgcclients[ielMinLoad]) if (cclientsThread < cclientsMin)
{
cclientsMin = cclientsThread;
ielMinLoad = iel; ielMinLoad = iel;
}
} }
return ielMinLoad; return ielMinLoad;
} }
@ -1207,18 +1201,21 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
{ {
{ {
int ielTarget = chooseBestThreadForAccept(); int ielTarget = chooseBestThreadForAccept();
rgacceptsInFlight[ielTarget].fetch_add(1, std::memory_order_relaxed);
if (ielTarget != ielCur) if (ielTarget != ielCur)
{ {
char *szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL); char *szT = (char*)zmalloc(NET_IP_STR_LEN, MALLOC_LOCAL);
memcpy(szT, cip, NET_IP_STR_LEN); memcpy(szT, cip, NET_IP_STR_LEN);
int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [cfd, ielTarget, szT]{ int res = aePostFunction(g_pserver->rgthreadvar[ielTarget].el, [cfd, ielTarget, szT]{
acceptCommonHandler(connCreateAcceptedSocket(cfd),0,szT,ielTarget); acceptCommonHandler(connCreateAcceptedSocket(cfd),0,szT,ielTarget);
rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
zfree(szT); zfree(szT);
}); });
if (res == AE_OK) if (res == AE_OK)
continue; continue;
} }
rgacceptsInFlight[ielTarget].fetch_sub(1, std::memory_order_relaxed);
} }
LLocalThread: LLocalThread: