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

Resolve race on neterr variable

Browse Source 
Former-commit-id: 1512274f4c51dc4e8f4e3d11df2f35e051c0c079
This commit is contained in:
John Sully 2019-07-18 17:11:28 -04:00
parent 857f778b38
commit bd3cc0bb95
4 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/cluster.cpp
View File

@ -643,11 +643,11 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
if (listLength(g_pserver->masters) == 0 && g_pserver->loading) return; if (listLength(g_pserver->masters) == 0 && g_pserver->loading) return;
while(max--) { while(max--) {
cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport); cfd = anetTcpAccept(serverTL->neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) { if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK) if (errno != EWOULDBLOCK)
serverLog(LL_VERBOSE, serverLog(LL_VERBOSE,
"Error accepting cluster node: %s", g_pserver->neterr); "Error accepting cluster node: %s", serverTL->neterr);
return; return;
} }
anetNonBlock(NULL,cfd); anetNonBlock(NULL,cfd);
@ -3407,7 +3407,7 @@ void clusterCron(void) {
mstime_t old_ping_sent; mstime_t old_ping_sent;
clusterLink *link; clusterLink *link;
fd = anetTcpNonBlockBindConnect(g_pserver->neterr, node->ip, fd = anetTcpNonBlockBindConnect(serverTL->neterr, node->ip,
node->cport, NET_FIRST_BIND_ADDR); node->cport, NET_FIRST_BIND_ADDR);
if (fd == -1) { if (fd == -1) {
/* We got a synchronous error from connect before /* We got a synchronous error from connect before
@ -3418,7 +3418,7 @@ void clusterCron(void) {
if (node->ping_sent == 0) node->ping_sent = mstime(); if (node->ping_sent == 0) node->ping_sent = mstime();
serverLog(LL_DEBUG, "Unable to connect to " serverLog(LL_DEBUG, "Unable to connect to "
"Cluster Node [%s]:%d -> %s", node->ip, "Cluster Node [%s]:%d -> %s", node->ip,
node->cport, g_pserver->neterr); node->cport, serverTL->neterr);
continue; continue;
} }
link = createClusterLink(node); link = createClusterLink(node);
@ -5010,15 +5010,15 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
} }
/* Create the socket */ /* Create the socket */
fd = anetTcpNonBlockConnect(g_pserver->neterr,szFromObj(c->argv[1]), fd = anetTcpNonBlockConnect(serverTL->neterr,szFromObj(c->argv[1]),
atoi(szFromObj(c->argv[2]))); atoi(szFromObj(c->argv[2])));
if (fd == -1) { if (fd == -1) {
sdsfree(name); sdsfree(name);
addReplyErrorFormat(c,"Can't connect to target node: %s", addReplyErrorFormat(c,"Can't connect to target node: %s",
g_pserver->neterr); serverTL->neterr);
return NULL; return NULL;
} }
anetEnableTcpNoDelay(g_pserver->neterr,fd); anetEnableTcpNoDelay(serverTL->neterr,fd);
/* Check if it connects within the specified timeout. */ /* Check if it connects within the specified timeout. */
if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) { if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {

8
src/networking.cpp
View File

@ -1087,11 +1087,11 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
UNUSED(privdata); UNUSED(privdata);
while(max--) { while(max--) {
cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport); cfd = anetTcpAccept(serverTL->neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) { if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK) if (errno != EWOULDBLOCK)
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Accepting client connection: %s", g_pserver->neterr); "Accepting client connection: %s", serverTL->neterr);
return; return;
} }
serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport); serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
@ -1131,11 +1131,11 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
UNUSED(privdata); UNUSED(privdata);
while(max--) { while(max--) {
cfd = anetUnixAccept(g_pserver->neterr, fd); cfd = anetUnixAccept(serverTL->neterr, fd);
if (cfd == ANET_ERR) { if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK) if (errno != EWOULDBLOCK)
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Accepting client connection: %s", g_pserver->neterr); "Accepting client connection: %s", serverTL->neterr);
return; return;
} }
int ielCur = ielFromEventLoop(el); int ielCur = ielFromEventLoop(el);

14
src/server.cpp
View File

@ -2712,7 +2712,7 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
int unsupported = 0; int unsupported = 0;
/* Bind * for both IPv6 and IPv4, we enter here only if /* Bind * for both IPv6 and IPv4, we enter here only if
* g_pserver->bindaddr_count == 0. */ * g_pserver->bindaddr_count == 0. */
fds[*count] = anetTcp6Server(g_pserver->neterr,port,NULL, fds[*count] = anetTcp6Server(serverTL->neterr,port,NULL,
g_pserver->tcp_backlog, fReusePort); g_pserver->tcp_backlog, fReusePort);
if (fds[*count] != ANET_ERR) { if (fds[*count] != ANET_ERR) {
anetNonBlock(NULL,fds[*count]); anetNonBlock(NULL,fds[*count]);
@ -2724,7 +2724,7 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
if (*count == 1 || unsupported) { if (*count == 1 || unsupported) {
/* Bind the IPv4 address as well. */ /* Bind the IPv4 address as well. */
fds[*count] = anetTcpServer(g_pserver->neterr,port,NULL, fds[*count] = anetTcpServer(serverTL->neterr,port,NULL,
g_pserver->tcp_backlog, fReusePort); g_pserver->tcp_backlog, fReusePort);
if (fds[*count] != ANET_ERR) { if (fds[*count] != ANET_ERR) {
anetNonBlock(NULL,fds[*count]); anetNonBlock(NULL,fds[*count]);
@ -2740,18 +2740,18 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
if (*count + unsupported == 2) break; if (*count + unsupported == 2) break;
} else if (strchr(g_pserver->bindaddr[j],':')) { } else if (strchr(g_pserver->bindaddr[j],':')) {
/* Bind IPv6 address. */ /* Bind IPv6 address. */
fds[*count] = anetTcp6Server(g_pserver->neterr,port,g_pserver->bindaddr[j], fds[*count] = anetTcp6Server(serverTL->neterr,port,g_pserver->bindaddr[j],
g_pserver->tcp_backlog, fReusePort); g_pserver->tcp_backlog, fReusePort);
} else { } else {
/* Bind IPv4 address. */ /* Bind IPv4 address. */
fds[*count] = anetTcpServer(g_pserver->neterr,port,g_pserver->bindaddr[j], fds[*count] = anetTcpServer(serverTL->neterr,port,g_pserver->bindaddr[j],
g_pserver->tcp_backlog, fReusePort); g_pserver->tcp_backlog, fReusePort);
} }
if (fds[*count] == ANET_ERR) { if (fds[*count] == ANET_ERR) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Could not create server TCP listening socket %s:%d: %s", "Could not create server TCP listening socket %s:%d: %s",
g_pserver->bindaddr[j] ? g_pserver->bindaddr[j] : "*", g_pserver->bindaddr[j] ? g_pserver->bindaddr[j] : "*",
port, g_pserver->neterr); port, serverTL->neterr);
if (errno == ENOPROTOOPT || errno == EPROTONOSUPPORT || if (errno == ENOPROTOOPT || errno == EPROTONOSUPPORT ||
errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT || errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||
errno == EAFNOSUPPORT || errno == EADDRNOTAVAIL) errno == EAFNOSUPPORT || errno == EADDRNOTAVAIL)
@ -2837,10 +2837,10 @@ static void initNetworking(int fReusePort)
/* Open the listening Unix domain socket. */ /* Open the listening Unix domain socket. */
if (g_pserver->unixsocket != NULL) { if (g_pserver->unixsocket != NULL) {
unlink(g_pserver->unixsocket); /* don't care if this fails */ unlink(g_pserver->unixsocket); /* don't care if this fails */
g_pserver->sofd = anetUnixServer(g_pserver->neterr,g_pserver->unixsocket, g_pserver->sofd = anetUnixServer(serverTL->neterr,g_pserver->unixsocket,
g_pserver->unixsocketperm, g_pserver->tcp_backlog); g_pserver->unixsocketperm, g_pserver->tcp_backlog);
if (g_pserver->sofd == ANET_ERR) { if (g_pserver->sofd == ANET_ERR) {
serverLog(LL_WARNING, "Opening Unix socket: %s", g_pserver->neterr); serverLog(LL_WARNING, "Opening Unix socket: %s", serverTL->neterr);
exit(1); exit(1);
} }
anetNonBlock(NULL,g_pserver->sofd); anetNonBlock(NULL,g_pserver->sofd);

2
src/server.h
View File

@ -1158,6 +1158,7 @@ struct redisServerThreadVars {
to be processed. */ to be processed. */
client *lua_client = nullptr; /* The "fake client" to query Redis from Lua */ client *lua_client = nullptr; /* The "fake client" to query Redis from Lua */
struct fastlock lockPendingWrite; struct fastlock lockPendingWrite;
char neterr[ANET_ERR_LEN]; /* Error buffer for anet.c */
}; };
struct redisMaster { struct redisMaster {
@ -1275,7 +1276,6 @@ struct redisServer {
rax *clients_index; /* Active clients dictionary by client ID. */ rax *clients_index; /* Active clients dictionary by client ID. */
int clients_paused; /* True if clients are currently paused */ int clients_paused; /* True if clients are currently paused */
mstime_t clients_pause_end_time; /* Time when we undo clients_paused */ mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
char neterr[ANET_ERR_LEN]; /* Error buffer for anet.c */
dict *migrate_cached_sockets;/* MIGRATE cached sockets */ dict *migrate_cached_sockets;/* MIGRATE cached sockets */
std::atomic<uint64_t> next_client_id; /* Next client unique ID. Incremental. */ std::atomic<uint64_t> next_client_id; /* Next client unique ID. Incremental. */
int protected_mode; /* Don't accept external connections. */ int protected_mode; /* Don't accept external connections. */