Redis/redis-For-Windows
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

[Fix] Fork code for background processing fixes and code refactoring.

Browse Source 
- [Fix] BeginForkOperation_Aof()/_Rdb()/_Socket() and BeginForkOperation() code
         refactoring.
 - [Fix] rewriteAppendOnlyFileBackground() code refactoring to minimize the code
         changes for WIN32.
 - [Fix] rewriteAppendOnlyFileBackground() must update the latency monitor, the
         fork stats and replicationScriptCacheFlush().
 - [Fix] rdbSaveBackground() code refactoring to minimize the code changes for
         WIN32.
 - [Fix] rdbSaveBackground() must update the latency monitor and the fork stats.
 - [Fix] memory leak in rdbSaveToSlavesSockets().
 - [Fix] properly releasing resources in rdbSaveToSlavesSockets().
 - [Fix] QForkChildInit() not setting the operationFailed event in case of
         exception.
 - [Fix] QForkChildInit() AV in catch() statement.
This commit is contained in:
Enrico Giordani
2015-07-15 13:07:16 +02:00
parent d82a361542
commit 32aea31d74
6 changed files with 45 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Win32_Interop/Win32_QFork.cpp
+17 -24
View File
@@ -340,20 +340,17 @@ BOOL QForkChildInit(HANDLE QForkConrolMemoryMapHandle, DWORD ParentProcessID) {
catch(std::system_error syserr) {
if (ReportSpecialSystemErrors(syserr.code().value()) == false) {
::redisLog(REDIS_WARNING, "QForkChildInit: system error caught. error code=0x%08x, message=%s\n", syserr.code().value(), syserr.what());
g_pQForkControl = NULL;
if (g_pQForkControl != NULL) {
if (g_pQForkControl->operationFailed != NULL) {
SetEvent(g_pQForkControl->operationFailed);
}
}
return FALSE;
}
}
catch(std::runtime_error runerr) {
::redisLog(REDIS_WARNING, "QForkChildInit: runtime error caught. message=%s\n", runerr.what());
}
if (g_pQForkControl != NULL) {
if (g_pQForkControl->operationFailed != NULL) {
SetEvent(g_pQForkControl->operationFailed);
}
g_pQForkControl = NULL;
SetEvent(g_pQForkControl->operationFailed);
return FALSE;
}
return FALSE;
}
@@ -917,12 +914,13 @@ void CreateChildProcess(PROCESS_INFORMATION *pi, char* logfile, DWORD dwCreation
typedef void (*CHILD_PID_HOOK)(DWORD pid);
BOOL BeginForkOperation(OperationType type, LPVOID globalData, int sizeOfGlobalData, DWORD* childPID, uint32_t dictHashSeed, char* logfile, CHILD_PID_HOOK pidHook = NULL) {
pid_t BeginForkOperation(OperationType type, LPVOID globalData, int sizeOfGlobalData, uint32_t dictHashSeed, char* logfile, CHILD_PID_HOOK pidHook = NULL) {
PROCESS_INFORMATION pi;
try {
pi.hProcess = INVALID_HANDLE_VALUE;
pi.dwProcessId = -1;
if(pidHook != NULL) {
if (pidHook != NULL) {
CreateChildProcess(&pi, logfile, CREATE_SUSPENDED);
pidHook(pi.dwProcessId);
CopyForkOperationData(type, globalData, sizeOfGlobalData, dictHashSeed);
@@ -932,7 +930,6 @@ BOOL BeginForkOperation(OperationType type, LPVOID globalData, int sizeOfGlobalD
CreateChildProcess(&pi, logfile, 0);
}
*childPID = pi.dwProcessId;
CloseHandle(pi.hThread);
// wait for "forked" process to map memory
@@ -943,7 +940,7 @@ BOOL BeginForkOperation(OperationType type, LPVOID globalData, int sizeOfGlobalD
"Forked Process did not respond in a timely manner.");
}
return TRUE;
return pi.dwProcessId;
}
catch(std::system_error syserr) {
::redisLog(REDIS_WARNING, "BeginForkOperation: system error caught. error code=0x%08x, message=%s\n", syserr.code().value(), syserr.what());
@@ -954,34 +951,32 @@ BOOL BeginForkOperation(OperationType type, LPVOID globalData, int sizeOfGlobalD
catch(...) {
::redisLog(REDIS_WARNING, "BeginForkOperation: other exception caught.\n");
}
if(pi.hProcess != INVALID_HANDLE_VALUE) {
if (pi.hProcess != INVALID_HANDLE_VALUE) {
TerminateProcess(pi.hProcess, 1);
}
return FALSE;
return -1;
}
BOOL BeginForkOperation_Rdb(
pid_t BeginForkOperation_Rdb(
char *filename,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile)
{
strcpy_s(g_pQForkControl->globalData.filename, filename);
return BeginForkOperation(otRDB, globalData, sizeOfGlobalData, childPID, dictHashSeed, logfile);
return BeginForkOperation(otRDB, globalData, sizeOfGlobalData, dictHashSeed, logfile);
}
BOOL BeginForkOperation_Aof(
pid_t BeginForkOperation_Aof(
char *filename,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile)
{
strcpy_s(g_pQForkControl->globalData.filename, filename);
return BeginForkOperation(otAOF, globalData, sizeOfGlobalData, childPID, dictHashSeed, logfile);
return BeginForkOperation(otAOF, globalData, sizeOfGlobalData, dictHashSeed, logfile);
}
void BeginForkOperation_Socket_PidHook(DWORD dwProcessId) {
@@ -992,14 +987,13 @@ void BeginForkOperation_Socket_PidHook(DWORD dwProcessId) {
}
}
BOOL BeginForkOperation_Socket(
pid_t BeginForkOperation_Socket(
int *fds,
int numfds,
uint64_t *clientids,
int pipe_write_fd,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile)
{
@@ -1015,7 +1009,6 @@ BOOL BeginForkOperation_Socket(
return BeginForkOperation(otSocket,
globalData,
sizeOfGlobalData,
childPID,
dictHashSeed,
logfile,
BeginForkOperation_Socket_PidHook);
src/Win32_Interop/Win32_QFork.h
+3 -6
View File
@@ -66,30 +66,27 @@ StartupStatus QForkStartup(int argc, char** argv);
BOOL QForkShutdown();
// For parent process use only
BOOL BeginForkOperation_Rdb(
pid_t BeginForkOperation_Rdb(
char* fileName,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile);
BOOL BeginForkOperation_Aof(
pid_t BeginForkOperation_Aof(
char* fileName,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile);
BOOL BeginForkOperation_Socket(
pid_t BeginForkOperation_Socket(
int *fds,
int numfds,
uint64_t *clientids,
int pipe_write_fd,
LPVOID globalData,
int sizeOfGlobalData,
DWORD* childPID,
unsigned __int32 dictHashSeed,
char* logfile);
src/Win32_Interop/win32_types.h
+2 -2
View File
@@ -72,6 +72,6 @@ typedef double PORT_LONGDOUBLE;
#endif
/* The maximum possible size_t value has all bits set */
#define MAX_SIZE_T (~(size_t)0)
#define MAX_SIZE_T (~(size_t)0)
typedef int pid_t;
src/Win32_Interop/win32fixes.h
-1
View File
@@ -178,7 +178,6 @@ int getrusage(int who, struct rusage * rusage);
#endif /*SIG_SETMASK*/
typedef void (*__p_sig_fn_t)(int);
typedef DWORD pid_t;
#ifndef _SIGSET_T_
#define _SIGSET_T_
src/aof.c
+12 -32
View File
@@ -1100,49 +1100,27 @@ werr:
* finally will rename(2) the temp file in the actual file name.
* The the new file is reopened as the new append only file. Profit!
*/
#ifdef _WIN32
int rewriteAppendOnlyFileBackground(void) {
PORT_LONGLONG start;
char tmpfile[256];
if (server.aof_child_pid != -1) return REDIS_ERR;
start = ustime();
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (BeginForkOperation_Aof(tmpfile, &server, sizeof(server), &server.aof_child_pid, dictGetHashFunctionSeed(), server.logfile) == FALSE) {
redisLog(REDIS_WARNING,
"Can't rewrite append only file in background: fork: %s",
strerror(errno));
return REDIS_ERR;
}
server.stat_fork_time = ustime()-start;
redisLog(REDIS_NOTICE,
"Background append only file rewriting started by pid %d",server.aof_child_pid);
server.aof_rewrite_scheduled = 0;
server.aof_rewrite_time_start = time(NULL);
updateDictResizePolicy();
/* We set appendseldb to -1 in order to force the next call to the
* feedAppendOnlyFile() to issue a SELECT command, so the differences
* accumulated by the parent into server.aof_rewrite_buf will start
* with a SELECT statement and it will be safe to merge. */
server.aof_selected_db = -1;
return REDIS_OK;
}
#else
int rewriteAppendOnlyFileBackground(void) {
pid_t childpid;
PORT_LONGLONG start;
if (server.aof_child_pid != -1) return REDIS_ERR;
start = ustime();
#ifndef _WIN32
if ((childpid = fork()) == 0) {
#endif
char tmpfile[256];
/* Child */
#ifndef _WIN32
/* Child */
closeListeningSockets(0);
redisSetProcTitle("redis-aof-rewrite");
#endif
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
#ifdef _WIN32
childpid = BeginForkOperation_Aof(tmpfile, &server, sizeof(server), dictGetHashFunctionSeed(), server.logfile);
#else
if (rewriteAppendOnlyFile(tmpfile) == REDIS_OK) {
size_t private_dirty = zmalloc_get_private_dirty();
@@ -1156,6 +1134,7 @@ int rewriteAppendOnlyFileBackground(void) {
exitFromChild(1);
}
} else {
#endif
/* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
@@ -1179,10 +1158,11 @@ int rewriteAppendOnlyFileBackground(void) {
server.aof_selected_db = -1;
replicationScriptCacheFlush();
return REDIS_OK;
#ifndef _WIN32
}
#endif
return REDIS_OK; /* unreached */
}
#endif
void bgrewriteaofCommand(redisClient *c) {
if (server.aof_child_pid != -1) {
src/rdb.c
+11 -26
View File
@@ -787,24 +787,6 @@ werr:
return REDIS_ERR;
}
#ifdef _WIN32
int rdbSaveBackground(char *filename) {
PORT_LONGLONG start;
start = ustime();
server.dirty_before_bgsave = server.dirty;
if (BeginForkOperation_Rdb(filename, &server, sizeof(server), &server.rdb_child_pid, dictGetHashFunctionSeed(), server.logfile)) {
server.stat_fork_time = ustime()-start;
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return REDIS_OK;
} else {
redisLog(REDIS_WARNING,"Can't save in background: fork: %s", strerror(errno));
return REDIS_ERR;
}
}
#else
int rdbSaveBackground(char *filename) {
pid_t childpid;
PORT_LONGLONG start;
@@ -815,6 +797,9 @@ int rdbSaveBackground(char *filename) {
server.lastbgsave_try = time(NULL);
start = ustime();
#ifdef _WIN32
childpid = BeginForkOperation_Rdb(filename, &server, sizeof(server), dictGetHashFunctionSeed(), server.logfile);
#else
if ((childpid = fork()) == 0) {
int retval;
@@ -833,6 +818,7 @@ int rdbSaveBackground(char *filename) {
}
exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else {
#endif
/* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
@@ -849,10 +835,11 @@ int rdbSaveBackground(char *filename) {
server.rdb_child_type = REDIS_RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return REDIS_OK;
#ifndef _WIN32
}
#endif
return REDIS_OK; /* unreached */
}
#endif
void rdbRemoveTempFile(pid_t childpid) {
char tmpfile[256];
@@ -1482,7 +1469,9 @@ int rdbSaveToSlavesSockets(void) {
/* Create the child process. */
start = ustime();
#ifndef _WIN32
#ifdef _WIN32
childpid = BeginForkOperation_Socket(fds, numfds, clientids, pipefds[1], &server, sizeof(server), dictGetHashFunctionSeed(), server.logfile);
#else
if ((childpid = fork()) == 0) {
/* Child */
int retval;
@@ -1549,12 +1538,6 @@ int rdbSaveToSlavesSockets(void) {
zfree(clientids);
exitFromChild((retval == REDIS_OK) ? 0 : 1);
} else {
#else // #ifndef _WIN32
if (!BeginForkOperation_Socket(fds, numfds, clientids, pipefds[1], &server, sizeof(server), &server.rdb_child_pid, dictGetHashFunctionSeed(), server.logfile)) {
redisLog(REDIS_WARNING,"Can't save in background: fork: %s", strerror(errno));
return REDIS_ERR;
} else {
childpid = server.rdb_child_pid;
#endif
/* Parent */
zfree(clientids); /* Not used by parent. Free ASAP. */
@@ -1576,7 +1559,9 @@ int rdbSaveToSlavesSockets(void) {
updateDictResizePolicy();
zfree(fds);
return REDIS_OK;
#ifndef _WIN32
}
#endif
return REDIS_OK; /* unreached */
}