diff --git a/msvs/RedisCheckAof/RedisCheckAof.vcxproj b/msvs/RedisCheckAof/RedisCheckAof.vcxproj index 2985b0f4..038e7c0a 100644 --- a/msvs/RedisCheckAof/RedisCheckAof.vcxproj +++ b/msvs/RedisCheckAof/RedisCheckAof.vcxproj @@ -48,6 +48,7 @@ Disabled WIN32;_DEBUG;_CONSOLE;PTW32_STATIC_LIB;%(PreprocessorDefinitions) 4996 + MultiThreadedDebug true @@ -63,6 +64,7 @@ true WIN32;_CONSOLE;PTW32_STATIC_LIB;%(PreprocessorDefinitions) 4996 + MultiThreaded true diff --git a/src/ae_wsiocp.c b/src/ae_wsiocp.c index fc2174c9..94756c92 100644 --- a/src/ae_wsiocp.c +++ b/src/ae_wsiocp.c @@ -33,8 +33,19 @@ #include #include - +/* Use GetQueuedCompletionStatusEx if possible. + * Try to load the function pointer dynamically. + * If it is not available, use GetQueuedCompletionStatus */ #define MAX_COMPLETE_PER_POLL 100 +typedef BOOL (WINAPI *sGetQueuedCompletionStatusEx) + (HANDLE CompletionPort, + LPOVERLAPPED_ENTRY lpCompletionPortEntries, + ULONG ulCount, + PULONG ulNumEntriesRemoved, + DWORD dwMilliseconds, + BOOL fAlertable); +sGetQueuedCompletionStatusEx pGetQueuedCompletionStatusEx; + /* structure that keeps state of sockets and Completion port handle */ typedef struct aeApiState { @@ -56,6 +67,7 @@ aeSockState *aeGetSockState(void *apistate, int fd) { /* Called by ae to initialize state */ static int aeApiCreate(aeEventLoop *eventLoop) { + HMODULE kernel32_module; aeApiState *state = (aeApiState *)zmalloc(sizeof(aeApiState)); if (!state) return -1; @@ -72,6 +84,19 @@ static int aeApiCreate(aeEventLoop *eventLoop) { NULL, 0, 1); + if (state->iocp == NULL) { + zfree(state->sockstate); + zfree(state); + return -1; + } + + pGetQueuedCompletionStatusEx = NULL; + kernel32_module = GetModuleHandleA("kernel32.dll"); + if (kernel32_module != NULL) { + pGetQueuedCompletionStatusEx = (sGetQueuedCompletionStatusEx) GetProcAddress( + kernel32_module, + "GetQueuedCompletionStatusEx"); + } state->setsize = AE_SETSIZE; eventLoop->apidata = state; @@ -153,13 +178,24 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { int rc; int mswait = (tvp->tv_sec * 1000) + (tvp->tv_usec / 1000); - /* first get an array of completion notifications */ - rc = GetQueuedCompletionStatusEx(state->iocp, - state->entries, - MAX_COMPLETE_PER_POLL, - &numComplete, - mswait, - FALSE); + if (pGetQueuedCompletionStatusEx != NULL) { + /* first get an array of completion notifications */ + rc = pGetQueuedCompletionStatusEx(state->iocp, + state->entries, + MAX_COMPLETE_PER_POLL, + &numComplete, + mswait, + FALSE); + } else { + /* need to get one at a time. Use first array element */ + rc = GetQueuedCompletionStatus(state->iocp, + &state->entries[0].dwNumberOfBytesTransferred, + &state->entries[0].lpCompletionKey, + &state->entries[0].lpOverlapped, + mswait); + numComplete = 1; + } + if (rc && numComplete > 0) { LPOVERLAPPED_ENTRY entry = state->entries; for (j = 0; j < numComplete && numevents < AE_SETSIZE; j++, entry++) { @@ -181,8 +217,7 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { } } else { /* check if event is read complete (may be 0 length read) */ - if (entry->lpOverlapped == &sockstate->ov_read && - entry->lpOverlapped->Internal != STATUS_PENDING) { + if (entry->lpOverlapped == &sockstate->ov_read) { sockstate->masks &= ~READ_QUEUED; if (sockstate->masks & AE_READABLE) { eventLoop->fired[numevents].fd = sock; @@ -206,7 +241,6 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { eventLoop->fired[numevents].fd = sock; eventLoop->fired[numevents].mask = AE_WRITABLE; numevents++; - } else { } } } diff --git a/src/anet.c b/src/anet.c index a9e0338e..47e48413 100644 --- a/src/anet.c +++ b/src/anet.c @@ -183,8 +183,7 @@ static int anetCreateSocket(char *err, int domain) { #define ANET_CONNECT_NONE 0 #define ANET_CONNECT_NONBLOCK 1 -static int anetTcpGenericConnect(char *err, char *addr, int port, int flags) -{ +static int anetTcpGenericConnect(char *err, char *addr, int port, int flags) { int s; struct sockaddr_in sa; unsigned long inAddress; @@ -519,7 +518,7 @@ static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *l SOCKET fd; while(1) { fd = aeWinAccept((SOCKET)s,sa,len); - if (fd == INVALID_SOCKET) { + if (fd == SOCKET_ERROR) { if (errno == WSAEINTR) continue; else { diff --git a/src/dict.c b/src/dict.c index 5542b845..0bf27b73 100644 --- a/src/dict.c +++ b/src/dict.c @@ -590,11 +590,7 @@ static size_t _dictNextPower(size_t size) { size_t i = DICT_HT_INITIAL_SIZE; -#ifdef _WIN64 - if (size >= LONG_LONG_MAX) return LONG_LONG_MAX; -#else if (size >= LONG_MAX) return LONG_MAX; -#endif while(1) { if (i >= size) diff --git a/src/redis-benchmark.c b/src/redis-benchmark.c index 551aaa7c..df66244f 100644 --- a/src/redis-benchmark.c +++ b/src/redis-benchmark.c @@ -217,7 +217,9 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) { fprintf(stderr,"Error: %s\n",c->context->errstr); exit(1); } +#ifdef _WIN32 aeWinReceiveDone(c->context->fd); +#endif if (reply != NULL) { if (reply == (void*)REDIS_REPLY_ERROR) { fprintf(stderr,"Unexpected error reply, exiting...\n"); diff --git a/src/redis-check-dump.c b/src/redis-check-dump.c index 19f8e79b..e78b1400 100644 --- a/src/redis-check-dump.c +++ b/src/redis-check-dump.c @@ -682,7 +682,11 @@ void process() { #endif int main(int argc, char **argv) { int fd; +#ifdef _WIN32 off size; +#else + off_t size; +#endif struct stat stat; void *data; diff --git a/src/redis.c b/src/redis.c index d931e292..74d86289 100644 --- a/src/redis.c +++ b/src/redis.c @@ -979,7 +979,7 @@ void initServer() { if (server.appendonly) { #ifdef _WIN32 - server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT|_O_BINARY,0); + server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT|_O_BINARY,_S_IREAD|_S_IWRITE); #else server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644); #endif diff --git a/src/replication.c b/src/replication.c index da1b97ef..2509a895 100644 --- a/src/replication.c +++ b/src/replication.c @@ -756,7 +756,12 @@ void replicationCron(void) { * connection last interaction time, and at the same time * we'll be sure that being a single char there are no * short-write problems. */ +#ifdef _WIN32 + if (aeWinSocketSend(slave->fd, "\n", 1, 0, + server.el, NULL, NULL, NULL) == -1) { +#else if (write(slave->fd, "\n", 1) == -1) { +#endif /* Don't worry, it's just a ping. */ } } diff --git a/src/win32_wsiocp.c b/src/win32_wsiocp.c index b40386c2..9b76067a 100644 --- a/src/win32_wsiocp.c +++ b/src/win32_wsiocp.c @@ -170,7 +170,7 @@ int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len) { areq = sockstate->reqs; if (areq == NULL) { errno = WSAEINVAL; - return -1; + return SOCKET_ERROR; } sockstate->reqs = areq->next; @@ -182,6 +182,10 @@ int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len) { SO_UPDATE_ACCEPT_CONTEXT, (char*)&fd, sizeof(fd)); + if (result == SOCKET_ERROR) { + errno = WSAGetLastError(); + return SOCKET_ERROR; + } locallen = *len; getaddrs(areq->buf, @@ -201,7 +205,7 @@ int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len) { /* queue another accept */ if (aeWinQueueAccept(fd) == -1) { - return -1; + return SOCKET_ERROR; } return acceptsock; @@ -349,6 +353,7 @@ int aeWinSocketDetach(int fd, int shutd) { if (shutd == 1) { if (shutdown(fd, SD_SEND) != SOCKET_ERROR) { + /* read data until no more or error to ensure shutdown completed */ while (1) { int rc = recv(fd, rbuf, 100, 0); if (rc == 0 || rc == SOCKET_ERROR) break; diff --git a/src/win32fixes.c b/src/win32fixes.c index c9d9c0f8..3b0e51ea 100644 --- a/src/win32fixes.c +++ b/src/win32fixes.c @@ -17,7 +17,6 @@ /* Redefined here to avoid redis.h so it can be used in other projects */ #define REDIS_NOTUSED(V) ((void) V) -#define REDIS_THREAD_STACK_SIZE (1024*1024*4) /* Winsock requires library initialization on startup */ int w32initWinSock(void) { @@ -164,237 +163,24 @@ int replace_ftruncate(int fd, off64_t length) { /* Rename which works on Windows when file exists */ int replace_rename(const char *src, const char *dst) { /* anti-virus may lock file - error code 5. Retry until it works or get a different error */ - int maxtries = 50; - static unsigned int instance = 1; - while (maxtries-- > 0) { + int retries = 50; + while (1) { if (MoveFileEx(src, dst, MOVEFILE_REPLACE_EXISTING | MOVEFILE_COPY_ALLOWED | MOVEFILE_WRITE_THROUGH)) { return 0; } else { errno = GetLastError(); if (errno != 5) break; + retries--; + if (retries == 0) { + retries = 50; + Sleep(10); + } } } /* On error we will return generic error code without GetLastError() */ return -1; } -#ifndef PTW32_STATIC_LIB -/* Proxy structure to pass fnuc and arg to thread */ -typedef struct thread_params -{ - void *(*func)(void *); - void * arg; -} thread_params; - -/* Proxy function by windows thread requirements */ -static unsigned __stdcall win32_proxy_threadproc(void *arg) { - - thread_params *p = (thread_params *) arg; - p->func(p->arg); - - /* Dealocate params */ - free(p); - - _endthreadex(0); - return 0; -} - -int pthread_create(pthread_t *thread, const void *unused, - void *(*start_routine)(void*), void *arg) { - - HANDLE h; - thread_params *params = malloc(sizeof(thread_params)); - REDIS_NOTUSED(unused); - - params->func = start_routine; - params->arg = arg; - - h =(HANDLE) _beginthreadex(NULL, /* Security not used */ - REDIS_THREAD_STACK_SIZE, /* Set custom stack size */ - win32_proxy_threadproc, /* calls win32 stdcall proxy */ - params, /* real threadproc is passed as paremeter */ - STACK_SIZE_PARAM_IS_A_RESERVATION, /* reserve stack */ - thread /* returned thread id */ - ); - - if (!h) - return errno; - - CloseHandle(h); - return 0; -} - -/* Noop in windows */ -int pthread_detach (pthread_t thread) { - REDIS_NOTUSED(thread); - return 0; /* noop */ - } - -pthread_t pthread_self(void) { - return GetCurrentThreadId(); -} - -int win32_pthread_join(pthread_t *thread, void **value_ptr) { - REDIS_NOTUSED(value_ptr); - int result; - HANDLE h = OpenThread(SYNCHRONIZE, FALSE, *thread); - - switch (WaitForSingleObject(h, INFINITE)) { - case WAIT_OBJECT_0: -// if (value_ptr) -// *value_ptr = thread->arg; - result = 0; - case WAIT_ABANDONED: - result = EINVAL; - default: - result = GetLastError(); - } - - CloseHandle(h); - return result; -} - -int pthread_cond_init(pthread_cond_t *cond, const void *unused) { - REDIS_NOTUSED(unused); - cond->waiters = 0; - cond->was_broadcast = 0; - - InitializeCriticalSection(&cond->waiters_lock); - - cond->sema = CreateSemaphore(NULL, 0, LONG_MAX, NULL); - if (!cond->sema) { - errno = GetLastError(); - return -1; - } - - cond->continue_broadcast = CreateEvent(NULL, /* security */ - FALSE, /* auto-reset */ - FALSE, /* not signaled */ - NULL); /* name */ - if (!cond->continue_broadcast) { - errno = GetLastError(); - return -1; - } - - return 0; -} - -int pthread_cond_destroy(pthread_cond_t *cond) { - CloseHandle(cond->sema); - CloseHandle(cond->continue_broadcast); - DeleteCriticalSection(&cond->waiters_lock); - return 0; -} - -int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex) { - int last_waiter; - - EnterCriticalSection(&cond->waiters_lock); - cond->waiters++; - LeaveCriticalSection(&cond->waiters_lock); - - /* - * Unlock external mutex and wait for signal. - * NOTE: we've held mutex locked long enough to increment - * waiters count above, so there's no problem with - * leaving mutex unlocked before we wait on semaphore. - */ - LeaveCriticalSection(mutex); - - /* let's wait - ignore return value */ - WaitForSingleObject(cond->sema, INFINITE); - - /* - * Decrease waiters count. If we are the last waiter, then we must - * notify the broadcasting thread that it can continue. - * But if we continued due to cond_signal, we do not have to do that - * because the signaling thread knows that only one waiter continued. - */ - EnterCriticalSection(&cond->waiters_lock); - cond->waiters--; - last_waiter = cond->was_broadcast && cond->waiters == 0; - LeaveCriticalSection(&cond->waiters_lock); - - if (last_waiter) { - /* - * cond_broadcast was issued while mutex was held. This means - * that all other waiters have continued, but are contending - * for the mutex at the end of this function because the - * broadcasting thread did not leave cond_broadcast, yet. - * (This is so that it can be sure that each waiter has - * consumed exactly one slice of the semaphor.) - * The last waiter must tell the broadcasting thread that it - * can go on. - */ - SetEvent(cond->continue_broadcast); - /* - * Now we go on to contend with all other waiters for - * the mutex. Auf in den Kampf! - */ - } - /* lock external mutex again */ - EnterCriticalSection(mutex); - - return 0; -} - -/* - * IMPORTANT: This implementation requires that pthread_cond_signal - * is called while the mutex is held that is used in the corresponding - * pthread_cond_wait calls! - */ -int pthread_cond_signal(pthread_cond_t *cond) { - int have_waiters; - - EnterCriticalSection(&cond->waiters_lock); - have_waiters = cond->waiters > 0; - LeaveCriticalSection(&cond->waiters_lock); - - /* - * Signal only when there are waiters - */ - if (have_waiters) - return ReleaseSemaphore(cond->sema, 1, NULL) ? - 0 : GetLastError(); - else - return 0; -} - -/* - * DOUBLY IMPORTANT: This implementation requires that pthread_cond_broadcast - * is called while the mutex is held that is used in the corresponding - * pthread_cond_wait calls! - */ -int pthread_cond_broadcast(pthread_cond_t *cond) -{ - EnterCriticalSection(&cond->waiters_lock); - - if ((cond->was_broadcast = cond->waiters > 0)) { - /* wake up all waiters */ - ReleaseSemaphore(cond->sema, cond->waiters, NULL); - LeaveCriticalSection(&cond->waiters_lock); - /* - * At this point all waiters continue. Each one takes its - * slice of the semaphor. Now it's our turn to wait: Since - * the external mutex is held, no thread can leave cond_wait, - * yet. For this reason, we can be sure that no thread gets - * a chance to eat *more* than one slice. OTOH, it means - * that the last waiter must send us a wake-up. - */ - WaitForSingleObject(cond->continue_broadcast, INFINITE); - /* - * Since the external mutex is held, no thread can enter - * cond_wait, and, hence, it is safe to reset this flag - * without cond->waiters_lock held. - */ - cond->was_broadcast = 0; - } else { - LeaveCriticalSection(&cond->waiters_lock); - } - return 0; -} - -#endif int pthread_sigmask(int how, const sigset_t *set, sigset_t *oset) { REDIS_NOTUSED(set); REDIS_NOTUSED(oset); @@ -419,12 +205,7 @@ int pthread_sigmask(int how, const sigset_t *set, sigset_t *oset) { /* child process will have data snapshot. */ /* Windows has no support for fork(). */ int fork(void) { -#ifdef _WIN32_FORK - /* TODO: Implement fork() for redis background writing */ return -1; -#else - return -1; -#endif } /* Redis CPU GetProcessTimes -> rusage */ diff --git a/src/win32fixes.h b/src/win32fixes.h index 2b34fc06..57083b23 100644 --- a/src/win32fixes.h +++ b/src/win32fixes.h @@ -30,19 +30,8 @@ #include #include -//Misc -#ifdef __STRICT_ANSI__ -#define _exit exit -#define fileno(__F) ((__F)->_file) - -#define strcasecmp lstrcmpiA - -#define fseeko(stream, offset, origin) fseek(stream, offset, origin) -#define ftello(stream) ftell(stream) -#else #define fseeko fseeko64 #define ftello ftello64 -#endif #define inline __inline @@ -209,14 +198,6 @@ struct sigaction { int sigaction(int sig, struct sigaction *in, struct sigaction *out); /* Sockets */ -/* #define EMSGSIZE WSAEMSGSIZE */ -/* #define EAFNOSUPPORT WSAEAFNOSUPPORT */ -/* #define EWOULDBLOCK WSAEWOULDBLOCK */ -/* #define ENOBUFS WSAENOBUFS */ -/* #define EPROTONOSUPPORT WSAEPROTONOSUPPORT */ -/* #define ECONNREFUSED WSAECONNREFUSED */ -/* #define EBADFD WSAENOTSOCK */ -/* #define EOPNOTSUPP WSAEOPNOTSUPP */ #ifndef ECONNRESET #define ECONNRESET WSAECONNRESET diff --git a/tests/integration/replication.tcl b/tests/integration/replication.tcl index e7580157..47c8be7b 100644 --- a/tests/integration/replication.tcl +++ b/tests/integration/replication.tcl @@ -11,6 +11,16 @@ start_server {tags {"repl"}} { $rd brpoplpush a b 5 r lpush a foo after 1000 + set digest [r debug digest] + set retry 10 + while {$retry} { + after 500 + set digest0 [r -1 debug digest] + if {$digest0 eq $digest} { + break + } + incr retry -1 + } assert_equal [r debug digest] [r -1 debug digest] } @@ -21,6 +31,16 @@ start_server {tags {"repl"}} { r lpush c 3 $rd brpoplpush c d 5 after 1000 + set digest [r debug digest] + set retry 10 + while {$retry} { + after 500 + set digest0 [r -1 debug digest] + if {$digest0 eq $digest} { + break + } + incr retry -1 + } assert_equal [r debug digest] [r -1 debug digest] } } @@ -54,13 +74,13 @@ start_server {tags {"repl"}} { test {SET on the master should immediately propagate} { r -1 set mykey bar - if {$::valgrind} {after 2000} + if {$::valgrind} {after 2000} else {after 100} r 0 get mykey } {bar} test {FLUSHALL should replicate} { r -1 flushall - if {$::valgrind} {after 2000} + if {$::valgrind} {after 2000} else {after 100} list [r -1 dbsize] [r 0 dbsize] } {0 0} } @@ -116,7 +136,16 @@ start_server {tags {"repl"}} { stop_write_load $load_handle3 stop_write_load $load_handle4 after 1000 - set digest [$master debug digest] + set retry 20 + while {$retry} { + set digest [$master debug digest] + set digest0 [[lindex $slaves 0] debug digest] + if {$digest0 eq $digest} { + break + } + after 500 + incr retry -1 + } set digest0 [[lindex $slaves 0] debug digest] set digest1 [[lindex $slaves 1] debug digest] set digest2 [[lindex $slaves 2] debug digest] diff --git a/tests/unit/protocol.tcl b/tests/unit/protocol.tcl index 9b4e7049..dd8d41e8 100644 --- a/tests/unit/protocol.tcl +++ b/tests/unit/protocol.tcl @@ -68,16 +68,19 @@ start_server {tags {"protocol"}} { puts -nonewline $s $seq set payload [string repeat A 1024]"\n" set test_start [clock seconds] - set test_time_limit 5 + set test_time_limit 20 while 1 { if {[catch { puts -nonewline $s payload flush $s incr payload_size [string length $payload] }]} { -# temporarily disable reading from closed connection -# set retval [gets $s] - set retval "Protocol error" + if {[catch {set retval [gets $s]}]} { + set retval "" + } + if {$retval == ""} { + set retval "Protocol error" + } close $s break } else {