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

error checking, older windows, fixes

Browse Source
This commit is contained in:
Henry Rawas
2012-02-09 16:27:19 -08:00
committed by Igor Zinkovsky
parent ac1cdea778
commit 387330d753
13 changed files with 114 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files
msvs/RedisCheckAof/RedisCheckAof.vcxproj
+2
View File
@@ -48,6 +48,7 @@
<Optimization>Disabled</Optimization>
<PreprocessorDefinitions>WIN32;_DEBUG;_CONSOLE;PTW32_STATIC_LIB;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4996</DisableSpecificWarnings>
<RuntimeLibrary>MultiThreadedDebug</RuntimeLibrary>
</ClCompile>
<Link>
<GenerateDebugInformation>true</GenerateDebugInformation>
@@ -63,6 +64,7 @@
<IntrinsicFunctions>true</IntrinsicFunctions>
<PreprocessorDefinitions>WIN32;_CONSOLE;PTW32_STATIC_LIB;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4996</DisableSpecificWarnings>
<RuntimeLibrary>MultiThreaded</RuntimeLibrary>
</ClCompile>
<Link>
<GenerateDebugInformation>true</GenerateDebugInformation>
src/ae_wsiocp.c
+45 -11
View File
@@ -33,8 +33,19 @@
#include <mswsock.h>
#include <Guiddef.h>
/* 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 {
}
}
}
src/anet.c
+2 -3
View File
@@ -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 {
src/dict.c
-4
View File
@@ -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)
src/redis-benchmark.c
+2
View File
@@ -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");
src/redis-check-dump.c
+4
View File
@@ -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;
src/redis.c
+1 -1
View File
@@ -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
src/replication.c
+5
View File
@@ -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. */
}
}
src/win32_wsiocp.c
+7 -2
View File
@@ -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;
src/win32fixes.c
+7 -226
View File
@@ -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 */
src/win32fixes.h
-19
View File
@@ -30,19 +30,8 @@
#include <process.h>
#include <sys/types.h>
//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
tests/integration/replication.tcl
+32 -3
View File
@@ -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]
tests/unit/protocol.tcl
+7 -4
View File
@@ -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 {