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

[Cleanup] Code refactoring.

Browse Source 
Renamed functions and variable, removed unused code, changed API replacement
for WSA functions:
 - Changed the aeWin prefix to WSAIOCP_.
 - Changed WSA functions from been remapped to be named with the FDAPI_ prefix.
 - Removed unused API replacements.
Changed comments.
Code formatting.
This commit is contained in:
Enrico Giordani
2015-08-24 17:34:05 +02:00
parent 60899dd617
commit 779942d789
17 changed files with 645 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deps/hiredis/async.c
Vendored
+3 -3
View File
@@ -170,8 +170,8 @@ redisAsyncContext *redisAsyncConnect(const char *ip, int port) {
SOCKADDR_STORAGE ss;
redisContext *c = redisPreConnectNonBlock(ip, port, &ss);
redisAsyncContext *ac = redisAsyncInitialize(c);
if (aeWinSocketConnect(ac->c.fd, &ss) != 0) {
ac->c.err = errno;
if (WSIOCP_SocketConnect(ac->c.fd, &ss) != 0) {
ac->c.err = errno;
strerror_r(errno, ac->c.errstr, sizeof(ac->c.errstr));
}
__redisAsyncCopyError(ac);
@@ -182,7 +182,7 @@ redisAsyncContext *redisAsyncConnectBind(const char *ip, int port, const char *s
SOCKADDR_STORAGE ss;
redisContext *c = redisPreConnectNonBlock(ip, port, &ss);
redisAsyncContext *ac = redisAsyncInitialize(c);
if (aeWinSocketConnectBind(ac->c.fd, &ss, source_addr) != 0) {
if (WSIOCP_SocketConnectBind(ac->c.fd, &ss, source_addr) != 0) {
ac->c.err = errno;
strerror_r(errno, ac->c.errstr, sizeof(ac->c.errstr));
}
deps/hiredis/net.c
Vendored
+9 -10
View File
@@ -160,16 +160,15 @@ int redisKeepAlive(redisContext *c, int interval) {
settings.keepalivetime = interval*1000;
settings.keepaliveinterval = interval*1000/3;
overlapped.hEvent = NULL;
WSAIoctl(
fd,
SIO_KEEPALIVE_VALS,
&settings,
sizeof(struct tcp_keepalive),
NULL,
0,
&bytesReturned,
&overlapped,
NULL);
FDAPI_WSAIoctl(fd,
SIO_KEEPALIVE_VALS,
&settings,
sizeof(struct tcp_keepalive),
NULL,
0,
&bytesReturned,
&overlapped,
NULL);
}
#else
val = interval;
src/Win32_Interop/Win32_Assert.h
+1 -1
View File
@@ -24,7 +24,7 @@
#define WIN32_ASSERT_H
#ifdef _DEBUG
#define ASSERT(condition) { if(!(condition)){ fprintf(stderr, "Assertion failed: %s @ %s::%s (%d)\n", #condition , __FILE__, __FUNCTION__, __LINE__); DebugBreak();} }
#define ASSERT(condition) do { if(!(condition)){ fprintf(stderr, "Assertion failed: %s @ %s::%s (%d)\n", #condition , __FILE__, __FUNCTION__, __LINE__); DebugBreak();} } while(0)
#else
#define ASSERT(condition)
#endif
src/Win32_Interop/Win32_FDAPI.cpp
+383 -384
View File
File diff suppressed because it is too large Load Diff
src/Win32_Interop/Win32_FDAPI.h
+31 -52
View File
@@ -40,18 +40,14 @@ typedef unsigned long nfds_t;
#include <stdio.h>
// the following are required to be defined before WS2tcpip is included.
typedef void (*redis_WSASetLastError)(int iError);
typedef int (*redis_WSAGetLastError)(void);
typedef int (*redis_WSAIoctl)(int rfd,DWORD dwIoControlCode,LPVOID lpvInBuffer,DWORD cbInBuffer,LPVOID lpvOutBuffer,DWORD cbOutBuffer,LPDWORD lpcbBytesReturned,LPWSAOVERLAPPED lpOverlapped,LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
#ifdef __cplusplus
extern "C"
{
#endif
extern redis_WSASetLastError WSASetLastError;
extern redis_WSAGetLastError WSAGetLastError;
extern redis_WSAIoctl WSAIoctl;
#ifdef __cplusplus
}
@@ -135,19 +131,6 @@ typedef struct pollfd {
} WSAPOLLFD, *PWSAPOLLFD, FAR *LPWSAPOLLFD;
#endif
// WinSock APIs used in Win32_wsiocp.cpp
typedef int (*redis_WSASend)(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD lpNumberOfBytesSent, DWORD dwFlags, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
typedef int (*redis_WSARecv)(int rfd,LPWSABUF lpBuffers,DWORD dwBufferCount,LPDWORD lpNumberOfBytesRecvd,LPDWORD lpFlags,LPWSAOVERLAPPED lpOverlapped,LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
typedef int (*redis_WSACleanup)(void);
typedef int (*redis_ioctlsocket)(int rfd,long cmd,u_long *argp );
typedef unsigned long (*redis_inet_addr)(const char *cp);
typedef struct hostent* (*redis_gethostbyname)(const char *name);
typedef char* (*redis_inet_ntoa)(struct in_addr in);
typedef BOOL (*redis_WSAGetOverlappedResult)(int rfd,LPWSAOVERLAPPED lpOverlapped, LPDWORD lpcbTransfer, BOOL fWait, LPDWORD lpdwFlags);
typedef int (*redis_WSADuplicateSocket)(int rfd, DWORD dwProcessId, LPWSAPROTOCOL_INFO lpProtocolInfo);
typedef int (*redis_WSASocket)(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags);
// other API forwards
typedef int (*redis_setmode)(int fd,int mode);
typedef size_t (*redis_fwrite)(const void * _Str, size_t _Size, size_t _Count, FILE * _File);
@@ -172,12 +155,11 @@ typedef int (*_redis_fstat)(int fd, struct __stat64 *buffer);
typedef int (*redis_listen)(int sockfd, int backlog);
typedef int (*redis_ftruncate)(int fd, PORT_LONGLONG length);
typedef int (*redis_bind)(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
typedef int (*redis_shutdown)(int sockfd, int how);
typedef u_short (*redis_htons)(u_short hostshort);
typedef u_long (*redis_htonl)(u_long hostlong);
typedef u_short (*redis_ntohs)(u_short netshort);
typedef int (*redis_getpeername)(int sockfd, struct sockaddr *addr, socklen_t * addrlen);
typedef int (*redis_getsockname)(int sockfd, struct sockaddr* addrsock, int* addrlen );
typedef u_short (*redis_ntohs)(u_short netshort);
typedef void (*redis_freeaddrinfo)(struct addrinfo *ai);
typedef int (*redis_getaddrinfo)(const char *node, const char *service, const struct addrinfo *hints, struct addrinfo **res);
typedef const char* (*redis_inet_ntop)(int af, const void *src, char *dst, size_t size);
@@ -205,17 +187,6 @@ extern "C"
// API replacements
extern redis_pipe pipe;
extern redis_socket socket;
extern redis_WSASend WSASend;
extern redis_WSARecv WSARecv;
extern redis_WSACleanup WSACleanup;
extern redis_ioctlsocket ioctlsocket;
extern redis_inet_addr inet_addr;
extern redis_inet_ntoa inet_ntoa;
extern redis_WSAGetOverlappedResult WSAGetOverlappedResult;
extern redis_WSADuplicateSocket WSADuplicateSocket;
extern redis_WSASocket WSASocket;
extern redis_close fdapi_close;
extern redis_open open;
extern redis_accept accept;
extern redis_setsockopt setsockopt;
@@ -226,48 +197,56 @@ extern redis_connect connect;
extern redis_read read;
extern redis_write write;
extern redis_fsync fsync;
extern _redis_fstat fdapi_fstat64;
extern redis_listen listen;
extern redis_ftruncate ftruncate;
extern redis_bind bind;
extern redis_shutdown shutdown;
extern redis_gethostbyname gethostbyname;
extern redis_htons htons;
extern redis_htonl htonl;
extern redis_getpeername getpeername;
extern redis_getsockname getsockname;
extern redis_ntohs ntohs;
extern redis_setmode fdapi_setmode;
extern redis_fwrite fdapi_fwrite;
extern redis_fclose fdapi_fclose;
extern redis_fileno fdapi_fileno;
extern redis_select select;
extern redis_ntohl ntohl;
extern redis_isatty isatty;
extern redis_access access;
extern redis_lseek64 lseek64;
extern redis_get_osfhandle fdapi_get_osfhandle;
extern redis_open_osfhandle fdapi_open_osfhandle;
extern redis_freeaddrinfo freeaddrinfo;
extern redis_getaddrinfo getaddrinfo;
extern redis_inet_ntop inet_ntop;
// other FD based APIs
BOOL FDAPI_SetFDInformation(int FD, DWORD mask, DWORD flags);
HANDLE FDAPI_CreateIoCompletionPortOnFD(int FD, HANDLE ExistingCompletionPort, ULONG_PTR CompletionKey, DWORD NumberOfConcurrentThreads);
BOOL FDAPI_AcceptEx(int listenFD,int acceptFD,PVOID lpOutputBuffer,DWORD dwReceiveDataLength,DWORD dwLocalAddressLength,DWORD dwRemoteAddressLength,LPDWORD lpdwBytesReceived,LPOVERLAPPED lpOverlapped);
BOOL FDAPI_ConnectEx(int fd,const struct sockaddr *name,int namelen,PVOID lpSendBuffer,DWORD dwSendDataLength,LPDWORD lpdwBytesSent,LPOVERLAPPED lpOverlapped);
void FDAPI_GetAcceptExSockaddrs(int fd, PVOID lpOutputBuffer,DWORD dwReceiveDataLength,DWORD dwLocalAddressLength,DWORD dwRemoteAddressLength,LPSOCKADDR *LocalSockaddr,LPINT LocalSockaddrLength,LPSOCKADDR *RemoteSockaddr,LPINT RemoteSockaddrLength);
int FDAPI_UpdateAcceptContext(int fd);
void FDAPI_SetCloseSocketState(fnWSIOCP_CloseSocketStateRFD* func);
void FDAPI_ClearSocketInfo(int fd);
void** FDAPI_GetSocketStatePtr(int rfd);
extern redis_close fdapi_close;
extern _redis_fstat fdapi_fstat64;
extern redis_setmode fdapi_setmode;
extern redis_fwrite fdapi_fwrite;
extern redis_fclose fdapi_fclose;
extern redis_fileno fdapi_fileno;
extern redis_get_osfhandle fdapi_get_osfhandle;
extern redis_open_osfhandle fdapi_open_osfhandle;
// other networking functions
// Other FD based APIs
BOOL FDAPI_SetFDInformation(int rfd, DWORD mask, DWORD flags);
HANDLE FDAPI_CreateIoCompletionPort(int rfd, HANDLE ExistingCompletionPort, ULONG_PTR CompletionKey, DWORD NumberOfConcurrentThreads);
BOOL FDAPI_AcceptEx(int listenFD,int acceptFD,PVOID lpOutputBuffer,DWORD dwReceiveDataLength,DWORD dwLocalAddressLength,DWORD dwRemoteAddressLength,LPDWORD lpdwBytesReceived,LPOVERLAPPED lpOverlapped);
BOOL FDAPI_ConnectEx(int rfd,const struct sockaddr *name,int namelen,PVOID lpSendBuffer,DWORD dwSendDataLength,LPDWORD lpdwBytesSent,LPOVERLAPPED lpOverlapped);
void FDAPI_GetAcceptExSockaddrs(int rfd, PVOID lpOutputBuffer,DWORD dwReceiveDataLength,DWORD dwLocalAddressLength,DWORD dwRemoteAddressLength,LPSOCKADDR *LocalSockaddr,LPINT LocalSockaddrLength,LPSOCKADDR *RemoteSockaddr,LPINT RemoteSockaddrLength);
int FDAPI_UpdateAcceptContext(int rfd);
void FDAPI_ClearSocketInfo(int rfd);
void** FDAPI_GetSocketStatePtr(int rfd);
int FDAPI_WSAIoctl(int rfd, DWORD dwIoControlCode, LPVOID lpvInBuffer, DWORD cbInBuffer, LPVOID lpvOutBuffer, DWORD cbOutBuffer, LPDWORD lpcbBytesReturned, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
int FDAPI_WSASend(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD lpNumberOfBytesSent, DWORD dwFlags, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
int FDAPI_WSARecv(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD lpNumberOfBytesRecvd, LPDWORD lpFlags, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine);
BOOL FDAPI_WSAGetOverlappedResult(int rfd, LPWSAOVERLAPPED lpOverlapped, LPDWORD lpcbTransfer, BOOL fWait, LPDWORD lpdwFlags);
int FDAPI_WSADuplicateSocket(int rfd, DWORD dwProcessId, LPWSAPROTOCOL_INFO lpProtocolInfo);
int FDAPI_WSASocket(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags);
int FDAPI_ioctlsocket(int rfd, long cmd, u_long *argp);
void FDAPI_SetCloseSocketState(fnWSIOCP_CloseSocketStateRFD* func);
// Other networking functions
BOOL ParseStorageAddress(const char *ip, int port, SOCKADDR_STORAGE* pSotrageAddr);
// macroize CRT definitions to point to our own
// Macroize CRT definitions to point to our own
#ifndef FDAPI_NOCRTREDEFS
#define close(fd) fdapi_close(fd)
#define setmode(fd,mode) fdapi_setmode(fd,mode)
src/Win32_Interop/Win32_QFork.cpp
+7 -7
View File
@@ -309,12 +309,12 @@ BOOL QForkChildInit(HANDLE QForkConrolMemoryMapHandle, DWORD ParentProcessID) {
LPWSAPROTOCOL_INFO lpProtocolInfo = (LPWSAPROTOCOL_INFO) g_pQForkControl->globalData.protocolInfo;
int pipe_write_fd = fdapi_open_osfhandle((intptr_t)g_pQForkControl->globalData.pipe_write_handle, _O_APPEND);
for (int i = 0; i < g_pQForkControl->globalData.numfds; i++) {
g_pQForkControl->globalData.fds[i] = WSASocket(FROM_PROTOCOL_INFO,
FROM_PROTOCOL_INFO,
FROM_PROTOCOL_INFO,
&lpProtocolInfo[i],
0,
WSA_FLAG_OVERLAPPED);
g_pQForkControl->globalData.fds[i] = FDAPI_WSASocket(FROM_PROTOCOL_INFO,
FROM_PROTOCOL_INFO,
FROM_PROTOCOL_INFO,
&lpProtocolInfo[i],
0,
WSA_FLAG_OVERLAPPED);
}
g_ChildExitCode = do_socketSave(g_pQForkControl->globalData.fds,
@@ -946,7 +946,7 @@ void BeginForkOperation_Socket_PidHook(DWORD dwProcessId) {
WSAPROTOCOL_INFO* protocolInfo = (WSAPROTOCOL_INFO*)dlmalloc(sizeof(WSAPROTOCOL_INFO) * g_pQForkControl->globalData.numfds);
g_pQForkControl->globalData.protocolInfo = protocolInfo;
for(int i = 0; i < g_pQForkControl->globalData.numfds; i++) {
WSADuplicateSocket(g_pQForkControl->globalData.fds[i], dwProcessId, &protocolInfo[i]);
FDAPI_WSADuplicateSocket(g_pQForkControl->globalData.fds[i], dwProcessId, &protocolInfo[i]);
}
}
src/Win32_Interop/win32_rfdmap.cpp
+25 -25
View File
@@ -32,11 +32,11 @@ RFDMap& RFDMap::getInstance() {
RFDMap::RFDMap() {
InitializeCriticalSection(&mutex);
// stdin, assigned rfd = 0
addPosixFD(0);
addCrtFD(0);
// stdout, assigned rfd = 1
addPosixFD(1);
addCrtFD(1);
// stderr, assigned rfd = 2
addPosixFD(2);
addCrtFD(2);
}
RFD RFDMap::getNextRFDAvailable() {
@@ -49,7 +49,7 @@ RFD RFDMap::getNextRFDAvailable() {
if (next_available_rfd < INT_MAX) {
rfd = RFDMap::next_available_rfd++;
} else {
rfd = RFDMap::INVALID_RFD;
rfd = INVALID_FD;
}
}
LeaveCriticalSection(&mutex);
@@ -60,10 +60,10 @@ RFD RFDMap::addSocket(SOCKET s) {
RFD rfd;
EnterCriticalSection(&mutex);
if (SocketToRFDMap.find(s) != SocketToRFDMap.end()) {
rfd = RFDMap::INVALID_RFD;
rfd = INVALID_FD;
} else {
rfd = getNextRFDAvailable();
if (rfd != RFDMap::INVALID_RFD) {
if (rfd != INVALID_FD) {
SocketToRFDMap[s] = rfd;
SocketInfo socket_info;
@@ -90,34 +90,34 @@ void RFDMap::removeRFDToSocketInfo(RFD rfd) {
LeaveCriticalSection(&mutex);
}
RFD RFDMap::addPosixFD(int posixFD) {
RFD RFDMap::addCrtFD(int crt_fd) {
RFD rfd;
EnterCriticalSection(&mutex);
if (PosixFDToRFDMap.find(posixFD) != PosixFDToRFDMap.end()) {
rfd = PosixFDToRFDMap[posixFD];
if (CrtFDToRFDMap.find(crt_fd) != CrtFDToRFDMap.end()) {
rfd = CrtFDToRFDMap[crt_fd];
} else {
rfd = getNextRFDAvailable();
if (rfd != RFDMap::INVALID_RFD) {
PosixFDToRFDMap[posixFD] = rfd;
RFDToPosixFDMap[rfd] = posixFD;
if (rfd != INVALID_FD) {
CrtFDToRFDMap[crt_fd] = rfd;
RFDToCrtFDMap[rfd] = crt_fd;
}
}
LeaveCriticalSection(&mutex);
return rfd;
}
void RFDMap::removePosixFD(int posixFD) {
// posixFD between FIRST_RESERVED_RFD_INDEX and LAST_RESERVED_RFD_INDEX
void RFDMap::removeCrtFD(int crt_fd) {
// crt_fd between FIRST_RESERVED_RFD_INDEX and LAST_RESERVED_RFD_INDEX
// should never be removed.
ASSERT(FIRST_RESERVED_RFD_INDEX == 0);
if (posixFD > RFDMap::LAST_RESERVED_RFD_INDEX) {
if (crt_fd > RFDMap::LAST_RESERVED_RFD_INDEX) {
EnterCriticalSection(&mutex);
map<int, RFD>::iterator mit = PosixFDToRFDMap.find(posixFD);
if (mit != PosixFDToRFDMap.end()) {
map<int, RFD>::iterator mit = CrtFDToRFDMap.find(crt_fd);
if (mit != CrtFDToRFDMap.end()) {
RFD rfd = (*mit).second;
RFDRecyclePool.push(rfd);
RFDToPosixFDMap.erase(rfd);
PosixFDToRFDMap.erase(posixFD);
RFDToCrtFDMap.erase(rfd);
CrtFDToRFDMap.erase(crt_fd);
}
LeaveCriticalSection(&mutex);
}
@@ -143,16 +143,16 @@ SocketInfo* RFDMap::lookupSocketInfo(RFD rfd) {
return socket_info;
}
int RFDMap::lookupPosixFD(RFD rfd) {
int posixFD = RFDMap::INVALID_FD;
int RFDMap::lookupCrtFD(RFD rfd) {
int crt_fd = INVALID_FD;
EnterCriticalSection(&mutex);
if (RFDToPosixFDMap.find(rfd) != RFDToPosixFDMap.end()) {
posixFD = RFDToPosixFDMap[rfd];
if (RFDToCrtFDMap.find(rfd) != RFDToCrtFDMap.end()) {
crt_fd = RFDToCrtFDMap[rfd];
} else if (rfd >= RFDMap::FIRST_RESERVED_RFD_INDEX
&& rfd <= RFDMap::LAST_RESERVED_RFD_INDEX) {
posixFD = rfd;
crt_fd = rfd;
}
LeaveCriticalSection(&mutex);
return posixFD;
return crt_fd;
}
src/Win32_Interop/win32_rfdmap.h
+14 -17
View File
@@ -32,6 +32,7 @@
using namespace std;
typedef int RFD; // Redis File Descriptor
#define INVALID_FD -1
typedef struct {
SOCKET socket;
@@ -61,18 +62,14 @@ private:
private:
map<SOCKET, RFD> SocketToRFDMap;
map<int, RFD> PosixFDToRFDMap;
map<int, RFD> CrtFDToRFDMap;
map<RFD, SocketInfo> RFDToSocketInfoMap;
map<RFD, int> RFDToPosixFDMap;
map<RFD, int> RFDToCrtFDMap;
queue<RFD> RFDRecyclePool;
private:
CRITICAL_SECTION mutex;
public:
const static int INVALID_RFD = -1;
const static int INVALID_FD = -1;
private:
/*
Gets the next available Redis File Descriptor. RFDs are always
@@ -91,10 +88,10 @@ public:
Returns the RFD value for the socket.
Returns INVALID_RFD if the socket is already added to the collection.
*/
RFD addSocket(SOCKET s);
RFD addSocket(SOCKET socket);
/* Removes a socket from SocketToRFDMap. */
void removeSocketToRFD(SOCKET s);
void removeSocketToRFD(SOCKET socket);
/*
Removes a RFD from RFDToSocketInfoMap.
@@ -103,17 +100,17 @@ public:
void removeRFDToSocketInfo(RFD rfd);
/*
Adds a posixFD (used with low-level CRT posix file functions) to RFDToPosixFDMap.
Returns the RFD value for the posixFD.
Returns the existing RFD if the posixFD is already present in the collection.
Adds a CRT fd (used with low-level CRT posix file functions) to RFDToCrtFDMap.
Returns the RFD value for the crt_fd.
Returns the existing RFD if the crt_fd is already present in the collection.
*/
RFD addPosixFD(int posixFD);
RFD addCrtFD(int crt_fd);
/*
Removes a socket from RFDToPosixFDMap.
Removes a socket from RFDToCrtFDMap.
It frees the associated RFD adding it to RFDRecyclePool.
*/
void removePosixFD(int posixFD);
void removeCrtFD(int crt_fd);
/*
Returns the socket associated with a RFD.
@@ -128,8 +125,8 @@ public:
SocketInfo* lookupSocketInfo(RFD rfd);
/*
Returns the posixFD associated with a RFD.
Returns INVALID_FD if the posixFD is not found.
Returns the crt_fd associated with a RFD.
Returns INVALID_FD if the crt_fd is not found.
*/
int lookupPosixFD(RFD rfd);
int lookupCrtFD(RFD rfd);
};
src/Win32_Interop/win32_wsiocp.c
+84 -87
View File
@@ -33,7 +33,7 @@ static HANDLE iocph;
#define SUCCEEDED_WITH_IOCP(result) ((result) || (GetLastError() == ERROR_IO_PENDING))
/* for zero length reads use shared buf */
/* For zero length reads use shared buf */
static DWORD wsarecvflags;
static char zreadchar[1];
@@ -81,7 +81,7 @@ BOOL WSIOCP_CloseSocketStateRFD(int rfd) {
return WSIOCP_CloseSocketState(WSIOCP_GetExistingSocketState(rfd));
}
int aeWinQueueAccept(int listenfd) {
int WSIOCP_QueueAccept(int listenfd) {
aeSockState *sockstate;
aeSockState *accsockstate;
DWORD result, bytes;
@@ -106,7 +106,7 @@ int aeWinQueueAccept(int listenfd) {
}
accsockstate->masks = SOCKET_ATTACHED;
/* keep accept socket in buf len until accepted */
// Keep accept socket in buf len until accepted
areq = (aacceptreq *) CallocMemoryNoCOW(sizeof(aacceptreq));
areq->buf = CallocMemoryNoCOW(sizeof(struct sockaddr_storage) * 2 + 64);
areq->accept = acceptfd;
@@ -132,8 +132,45 @@ int aeWinQueueAccept(int listenfd) {
return 0;
}
/* listen using extension function to get faster accepts */
int aeWinListen(int rfd, int backlog) {
/* For each asynch socket, need to associate completion port */
int WSIOCP_SocketAttach(int fd) {
if (iocph == NULL) {
return -1;
}
aeSockState *sockstate;
if ((sockstate = WSIOCP_GetSocketState(fd)) == NULL) {
errno = WSAEINVAL;
return -1;
}
// Set the socket to nonblocking mode
DWORD yes = 1;
if (FDAPI_ioctlsocket(fd, FIONBIO, &yes) == SOCKET_ERROR) {
errno = WSAGetLastError();
return -1;
}
// Make the socket non-inheritable
if (!FDAPI_SetFDInformation(fd, HANDLE_FLAG_INHERIT, 0)) {
errno = WSAGetLastError();
return -1;
}
// Associate it with the I/O completion port.
// Use fd as completion key.
if (FDAPI_CreateIoCompletionPort(fd, iocph, (ULONG_PTR) fd, 0) == NULL) {
errno = WSAGetLastError();
return -1;
}
sockstate->masks = SOCKET_ATTACHED;
sockstate->wreqs = 0;
return 0;
}
/* Listen using extension function to get faster accepts */
int WSIOCP_Listen(int rfd, int backlog) {
aeSockState *sockstate;
const GUID wsaid_acceptex = WSAID_ACCEPTEX;
const GUID wsaid_acceptexaddrs = WSAID_GETACCEPTEXSOCKADDRS;
@@ -143,11 +180,11 @@ int aeWinListen(int rfd, int backlog) {
return SOCKET_ERROR;
}
aeWinSocketAttach(rfd);
WSIOCP_SocketAttach(rfd);
sockstate->masks |= LISTEN_SOCK;
if (listen(rfd, backlog) == 0) {
if (aeWinQueueAccept(rfd) == -1) {
if (WSIOCP_QueueAccept(rfd) == -1) {
errno = WSAGetLastError();
return SOCKET_ERROR;
}
@@ -158,8 +195,8 @@ int aeWinListen(int rfd, int backlog) {
return 0;
}
/* return the queued accept socket */
int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len) {
/* Return the queued accept socket */
int WSIOCP_Accept(int fd, struct sockaddr *sa, socklen_t *len) {
aeSockState *sockstate;
int acceptfd;
int result;
@@ -212,23 +249,23 @@ int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len) {
}
}
aeWinSocketAttach(acceptfd);
WSIOCP_SocketAttach(acceptfd);
FreeMemoryNoCOW(areq->buf);
FreeMemoryNoCOW(areq);
/* queue another accept */
if (aeWinQueueAccept(fd) == -1) {
// Queue another accept
if (WSIOCP_QueueAccept(fd) == -1) {
return SOCKET_ERROR;
}
return acceptfd;
}
/* after doing read caller needs to call done
* so that we can continue to check for read events.
/* After doing read, caller needs to call done so that we can
* continue to check for read events.
* This is not necessary if caller will delete read events */
int aeWinReceiveDone(int fd) {
int WSIOCP_ReceiveDone(int fd) {
aeSockState *sockstate;
int result;
WSABUF zreadbuf;
@@ -249,13 +286,13 @@ int aeWinReceiveDone(int fd) {
zreadbuf.buf = zreadchar;
zreadbuf.len = 0;
result = WSARecv(fd,
&zreadbuf,
1,
&bytesReceived,
&wsarecvflags,
&sockstate->ov_read,
NULL);
result = FDAPI_WSARecv(fd,
&zreadbuf,
1,
&bytesReceived,
&wsarecvflags,
&sockstate->ov_read,
NULL);
if (SUCCEEDED_WITH_IOCP(result == 0)){
sockstate->masks |= READ_QUEUED;
} else {
@@ -266,11 +303,11 @@ int aeWinReceiveDone(int fd) {
return 0;
}
/* wrapper for send
* enables use of WSA Send to get IOCP notification of completion.
* returns -1 with errno = WSA_IO_PENDING if callback will be invoked later */
int aeWinSocketSend(int fd, char *buf, int len,
void *eventLoop, void *client, void *data, void *proc) {
/* Wrapper for send.
* Enables use of WSASend to get IOCP notification of completion.
* Returns -1 with errno = WSA_IO_PENDING if callback will be invoked later */
int WSIOCP_SocketSend(int fd, char *buf, int len, void *eventLoop,
void *client, void *data, void *proc) {
aeSockState *sockstate;
int result;
asendreq *areq;
@@ -283,7 +320,7 @@ int aeWinSocketSend(int fd, char *buf, int len,
aeWait(fd, AE_WRITABLE, 50);
}
/* if not an async socket, do normal send */
// If not an async socket, do normal send
if (sockstate == NULL ||
(sockstate->masks & SOCKET_ATTACHED) == 0 ||
proc == NULL) {
@@ -294,7 +331,7 @@ int aeWinSocketSend(int fd, char *buf, int len,
return result;
}
/* use overlapped structure to send using IOCP */
// Use overlapped structure to send using IOCP
areq = (asendreq *) CallocMemoryNoCOW(sizeof(asendreq));
areq->wbuf.len = len;
areq->wbuf.buf = buf;
@@ -305,13 +342,13 @@ int aeWinSocketSend(int fd, char *buf, int len,
areq->req.buf = buf;
areq->proc = (aeFileProc *)proc;
result = WSASend(fd,
&areq->wbuf,
1,
&bytesSent,
0,
&areq->ov,
NULL);
result = FDAPI_WSASend(fd,
&areq->wbuf,
1,
&bytesSent,
0,
&areq->ov,
NULL);
if (SUCCEEDED_WITH_IOCP(result == 0)){
errno = WSA_IO_PENDING;
@@ -324,8 +361,8 @@ int aeWinSocketSend(int fd, char *buf, int len,
return SOCKET_ERROR;
}
/* for non-blocking connect with IOCP */
int aeWinSocketConnect(int fd, const SOCKADDR_STORAGE *ss) {
/* For non-blocking connect with IOCP */
int WSIOCP_SocketConnect(int fd, const SOCKADDR_STORAGE *ss) {
const GUID wsaid_connectex = WSAID_CONNECTEX;
DWORD result;
aeSockState *sockstate;
@@ -335,13 +372,13 @@ int aeWinSocketConnect(int fd, const SOCKADDR_STORAGE *ss) {
return SOCKET_ERROR;
}
if (aeWinSocketAttach(fd) != 0) {
if (WSIOCP_SocketAttach(fd) != 0) {
return SOCKET_ERROR;
}
memset(&sockstate->ov_read, 0, sizeof(sockstate->ov_read));
/* need to bind sock before connectex */
// Need to bind sock before connectex
switch (ss->ss_family) {
case AF_INET:
{
@@ -369,7 +406,7 @@ int aeWinSocketConnect(int fd, const SOCKADDR_STORAGE *ss) {
}
default:
{
ASSERT(ss->ss_family == AF_INET || ss->ss_family == AF_INET6)
ASSERT(ss->ss_family == AF_INET || ss->ss_family == AF_INET6);
errno = WSAEINVAL;
return SOCKET_ERROR;
}
@@ -388,7 +425,7 @@ int aeWinSocketConnect(int fd, const SOCKADDR_STORAGE *ss) {
return 0;
}
int aeWinSocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* source_addr) {
int WSIOCP_SocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* source_addr) {
const GUID wsaid_connectex = WSAID_CONNECTEX;
DWORD result;
aeSockState *sockstate;
@@ -398,13 +435,13 @@ int aeWinSocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* sourc
return SOCKET_ERROR;
}
if (aeWinSocketAttach(fd) != 0) {
if (WSIOCP_SocketAttach(fd) != 0) {
return SOCKET_ERROR;
}
memset(&sockstate->ov_read, 0, sizeof(sockstate->ov_read));
/* need to bind sock before connectex */
// Need to bind sock before connectex
int storageSize = 0;
switch (ss->ss_family) {
case AF_INET:
@@ -431,7 +468,7 @@ int aeWinSocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* sourc
}
default:
{
ASSERT(ss->ss_family == AF_INET || ss->ss_family == AF_INET6)
ASSERT(ss->ss_family == AF_INET || ss->ss_family == AF_INET6);
errno = WSAEINVAL;
return SOCKET_ERROR;
}
@@ -451,52 +488,12 @@ int aeWinSocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* sourc
return 0;
}
/* for each asynch socket, need to associate completion port */
int aeWinSocketAttach(int fd) {
if (iocph == NULL) {
return -1;
}
aeSockState *sockstate;
if ((sockstate = WSIOCP_GetSocketState(fd)) == NULL) {
errno = WSAEINVAL;
return -1;
}
/* Set the socket to nonblocking mode */
DWORD yes = 1;
if (ioctlsocket(fd, FIONBIO, &yes) == SOCKET_ERROR) {
errno = WSAGetLastError();
return -1;
}
/* Make the socket non-inheritable */
if (!FDAPI_SetFDInformation(fd, HANDLE_FLAG_INHERIT, 0)) {
errno = WSAGetLastError();
return -1;
}
/* Associate it with the I/O completion port. */
/* Use FD as completion key. */
if (FDAPI_CreateIoCompletionPortOnFD(fd,
iocph,
(ULONG_PTR)fd,
0) == NULL) {
errno = WSAGetLastError();
return -1;
}
sockstate->masks = SOCKET_ATTACHED;
sockstate->wreqs = 0;
return 0;
}
void aeWinInit(HANDLE iocp) {
void WSIOCP_Init(HANDLE iocp) {
iocph = iocp;
FDAPI_SetCloseSocketState(WSIOCP_CloseSocketStateRFD);
}
void aeWinCleanup() {
void WSIOCP_Cleanup() {
iocph = NULL;
}
src/Win32_Interop/win32_wsiocp.h
+2 -2
View File
@@ -66,8 +66,8 @@ typedef struct aeSockState {
#define CONNECT_PENDING 0x002000
#define CLOSE_PENDING 0x004000
void aeWinInit(HANDLE iocp);
void aeWinCleanup();
void WSIOCP_Init(HANDLE iocp);
void WSIOCP_Cleanup();
aeSockState* WSIOCP_GetExistingSocketState(int rfd);
aeSockState* WSIOCP_GetSocketState(int rfd);
src/Win32_Interop/win32fixes.h
+8 -11
View File
@@ -280,9 +280,9 @@ We need to do the check before calling strtod */
double wstrtod(const char *nptr, char **eptr);
/* structs and functions for using IOCP with windows sockets */
/* Structs and functions for using IOCP with windows sockets */
/* need callback on write complete. aeWinSendReq is used to pass parameters */
/* Need a callback on write complete. aeWinSendReq is used to pass parameters */
typedef struct aeWinSendReq {
void *client;
void *data;
@@ -290,15 +290,12 @@ typedef struct aeWinSendReq {
int len;
} aeWinSendReq;
int aeWinSocketAttach(int fd);
int aeWinReceiveDone(int fd);
int aeWinSocketSend(int fd, char *buf, int len,
void *eventLoop, void *client, void *data, void *proc);
int aeWinListen(int rfd, int backlog);
int aeWinAccept(int fd, struct sockaddr *sa, socklen_t *len);
int aeWinSocketConnect(int fd, const SOCKADDR_STORAGE *ss);
int aeWinSocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* source_addr);
int WSIOCP_ReceiveDone(int fd);
int WSIOCP_SocketSend(int fd, char *buf, int len, void *eventLoop, void *client, void *data, void *proc);
int WSIOCP_Listen(int rfd, int backlog);
int WSIOCP_Accept(int fd, struct sockaddr *sa, socklen_t *len);
int WSIOCP_SocketConnect(int fd, const SOCKADDR_STORAGE *ss);
int WSIOCP_SocketConnectBind(int fd, const SOCKADDR_STORAGE *ss, const char* source_addr);
int strerror_r(int err, char* buf, size_t buflen);
char *wsa_strerror(int err);
src/ae_wsiocp.c
+29 -32
View File
@@ -41,27 +41,24 @@ typedef BOOL (WINAPI *sGetQueuedCompletionStatusEx)
BOOL fAlertable);
sGetQueuedCompletionStatusEx pGetQueuedCompletionStatusEx;
/* lookup structure for socket
* socket value is not an index. Convert socket to index
* and then find matching structure in list */
#define MAX_SOCKET_LOOKUP 65535
/* structure that keeps state of sockets and Completion port handle */
/* Structure that keeps IOCP data */
typedef struct aeApiState {
HANDLE iocp;
int setsize;
OVERLAPPED_ENTRY entries[MAX_COMPLETE_PER_POLL];
} aeApiState;
// find matching value in list and remove. If found return 1
int removeMatchFromList(list *socklist, void *value) {
listNode *node;
if (socklist == NULL) return 0;
node = listFirst(socklist);
/* Find matching value in list and remove. If found return 1 */
int removeMatchFromList(list *requestlist, void *value) {
if (requestlist == NULL) {
return 0;
}
listNode* node = listFirst(requestlist);
while (node != NULL) {
if (listNodeValue(node) == value) {
listDelNode(socklist, node);
listDelNode(requestlist, node);
return 1;
}
node = listNextNode(node);
@@ -76,7 +73,7 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
if (!state) return -1;
/* create a single IOCP to be shared by all sockets */
// Create a single IOCP to be shared by all sockets
state->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
NULL,
0,
@@ -96,8 +93,8 @@ static int aeApiCreate(aeEventLoop *eventLoop) {
state->setsize = eventLoop->setsize;
eventLoop->apidata = state;
/* initialize the IOCP socket code with state reference */
aeWinInit(state->iocp);
// Initialize the IOCP socket code with state reference
WSIOCP_Init(state->iocp);
return 0;
}
@@ -107,15 +104,15 @@ static int aeApiResize(aeEventLoop *eventLoop, int setsize) {
}
/* termination */
/* Termination */
static void aeApiFree(aeEventLoop *eventLoop) {
aeApiState *state = (aeApiState *)eventLoop->apidata;
CloseHandle(state->iocp);
FreeMemoryNoCOW(state);
aeWinCleanup();
WSIOCP_Cleanup();
}
/* monitor state changes for a socket */
/* Monitor state changes for a socket */
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeApiState *state = (aeApiState *)eventLoop->apidata;
aeSockState *sockstate = WSIOCP_GetSocketState(fd);
@@ -128,11 +125,11 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
sockstate->masks |= AE_READABLE;
if ((sockstate->masks & CONNECT_PENDING) == 0) {
if (sockstate->masks & LISTEN_SOCK) {
/* actually a listen. Do not treat as read */
// Actually a listen. Do not treat as read
} else {
if ((sockstate->masks & READ_QUEUED) == 0) {
// queue up a 0 byte read
aeWinReceiveDone(fd);
// Queue up a 0 byte read
WSIOCP_ReceiveDone(fd);
}
}
}
@@ -140,13 +137,13 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
if (mask & AE_WRITABLE) {
sockstate->masks |= AE_WRITABLE;
if ((sockstate->masks & CONNECT_PENDING) == 0) {
// if no write active, then need to queue write ready
// If no write active, then need to queue write ready
if (sockstate->wreqs == 0) {
asendreq *areq = (asendreq *) CallocMemoryNoCOW(sizeof(asendreq));
if (PostQueuedCompletionStatus(state->iocp,
0,
fd,
&areq->ov) == 0) {
0,
fd,
&areq->ov) == 0) {
errno = GetLastError();
FreeMemoryNoCOW(areq);
return -1;
@@ -159,7 +156,7 @@ static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
return 0;
}
/* stop monitoring state changes for a socket */
/* Stop monitoring state changes for a socket */
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
aeSockState *sockstate = WSIOCP_GetExistingSocketState(fd);
if (sockstate == NULL) {
@@ -171,7 +168,7 @@ static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
if (mask & AE_WRITABLE) sockstate->masks &= ~AE_WRITABLE;
}
/* return array of sockets that are ready for read or write
/* Return array of sockets that are ready for read or write
depending on the mask for each socket */
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
aeApiState *state = (aeApiState *)eventLoop->apidata;
@@ -185,11 +182,11 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
if (pGetQueuedCompletionStatusEx != NULL) {
/* first get an array of completion notifications */
rc = pGetQueuedCompletionStatusEx(state->iocp,
state->entries,
MAX_COMPLETE_PER_POLL,
&numComplete,
mswait,
FALSE);
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,
@@ -270,7 +267,7 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
if (areq->proc != NULL) {
DWORD written = 0;
DWORD flags;
WSAGetOverlappedResult(rfd, &areq->ov, &written, FALSE, &flags);
FDAPI_WSAGetOverlappedResult(rfd, &areq->ov, &written, FALSE, &flags);
areq->proc(areq->eventLoop, rfd, &areq->req, (int)written);
}
sockstate->wreqs--;
src/anet.c
+23 -14
View File
@@ -118,19 +118,28 @@ int anetKeepAlive(char *err, int fd, int interval)
DWORD dwBytesRet = 0;
alive.onoff = TRUE;
alive.keepalivetime = interval * 1000;
/* According to http://msdn.microsoft.com/en-us/library/windows/desktop/ee470551(v=vs.85).aspx
On Windows Vista and later, the number of keep-alive probes (data retransmissions) is set to 10 and cannot be changed.
So we set the keep alive interval as interval/10, as 10 probes will be send before
detecting an error
*/
/* According to http://msdn.microsoft.com/en-us/library/windows/desktop/ee470551(v=vs.85).aspx
* On Windows Vista and later, the number of keep-alive probes (data retransmissions)
* is set to 10 and cannot be changed.
* So we set the keep alive interval as interval/10, as 10 probes will be send before
* detecting an error. */
val = interval/10;
if (val == 0) val = 1;
if (val == 0) {
val = 1;
}
alive.keepaliveinterval = val*1000;
if(WSAIoctl(fd, SIO_KEEPALIVE_VALS, &alive, sizeof(alive),
NULL, 0, &dwBytesRet, NULL, NULL) == SOCKET_ERROR) {
anetSetError(err, "WSAIotcl(SIO_KEEPALIVE_VALS) failed with error code %d\n", WSAGetLastError());
return ANET_ERR;
}
if (SOCKET_ERROR == FDAPI_WSAIoctl(fd,
SIO_KEEPALIVE_VALS,
&alive,
sizeof(alive),
NULL,
0,
&dwBytesRet,
NULL,
NULL)) {
anetSetError(err, "WSAIotcl(SIO_KEEPALIVE_VALS) failed with error code %d\n", WSAGetLastError());
return ANET_ERR;
}
#else
/* Default settings are more or less garbage, with the keepalive time
* set to 7200 by default on Linux. Modify settings to make the feature
@@ -318,7 +327,7 @@ static int anetTcpGenericConnect(char *err, char *addr, int port, int flags) {
if ((rfd = anetCreateSocket(err,ss.ss_family)) == ANET_ERR) {
return ANET_ERR;
}
if (aeWinSocketConnect(rfd, &ss ) == SOCKET_ERROR) {
if (WSIOCP_SocketConnect(rfd, &ss ) == SOCKET_ERROR) {
if ((errno == WSAEWOULDBLOCK || errno == WSA_IO_PENDING)) errno = EINPROGRESS;
if (errno == EINPROGRESS && flags & ANET_CONNECT_NONBLOCK) {
return rfd;
@@ -494,7 +503,7 @@ static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int
}
#ifdef WIN32_IOCP
if (aeWinListen(s, backlog) == SOCKET_ERROR) {
if (WSIOCP_Listen(s, backlog) == SOCKET_ERROR) {
#else
if (listen(s, backlog) == -1) {
#endif
@@ -591,7 +600,7 @@ int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
int fd;
while(1) {
#ifdef WIN32_IOCP
fd = aeWinAccept(s,sa,len);
fd = WSIOCP_Accept(s,sa,len);
#else
fd = accept(s,sa,len);
#endif
src/networking.c
+7 -7
View File
@@ -36,7 +36,7 @@
#endif
#include <math.h>
WIN32_ONLY(extern int aeWinQueueAccept(int listenfd);)
WIN32_ONLY(extern int WSIOCP_QueueAccept(int listenfd);)
static void setProtocolError(redisClient *c, int pos);
@@ -589,7 +589,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
redisLog(REDIS_WARNING,
"Accepting client connection: %s", server.neterr);
#ifdef _WIN32
if (aeWinQueueAccept(fd) == -1) {
if (WSIOCP_QueueAccept(fd) == -1) {
redisLog(REDIS_WARNING,
"acceptTcpHandler: failed to queue another accept.");
}
@@ -867,8 +867,8 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
while(c->bufpos > c->sentlen || ln != NULL) {
if (c->bufpos > c->sentlen) {
nwritten = c->bufpos - c->sentlen;
result = aeWinSocketSend(fd,c->buf+c->sentlen, nwritten,
el, c, c->buf, sendReplyBufferDone);
result = WSIOCP_SocketSend(fd, c->buf + c->sentlen, nwritten,
el, c, c->buf, sendReplyBufferDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
redisLog(REDIS_VERBOSE, "Error writing to client: %s", wsa_strerror(errno));
freeClient(c);
@@ -890,8 +890,8 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
/* object ref placed in request, release in sendReplyListDone */
incrRefCount(o);
result = aeWinSocketSend(fd, ((char*)o->ptr), objlen,
el, c, o, sendReplyListDone);
result = WSIOCP_SocketSend(fd, ((char*) o->ptr), objlen,
el, c, o, sendReplyListDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
redisLog(REDIS_VERBOSE,
"Error writing to client: %s", wsa_strerror(errno));
@@ -1318,7 +1318,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
return;
}
#ifdef WIN32_IOCP
aeWinReceiveDone(fd);
WSIOCP_ReceiveDone(fd);
#endif
if (nread) {
sdsIncrLen(c->querybuf,nread);
src/redis-benchmark.c
+6 -5
View File
@@ -227,7 +227,7 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
errno = WSAGetLastError();
if ((errno == ENOENT) || (errno == WSAEWOULDBLOCK)) {
errno = EAGAIN;
aeWinReceiveDone((int)c->context->fd);
WSIOCP_ReceiveDone((int) c->context->fd);
return;
} else {
fprintf(stderr,"Error: %s\n",c->context->errstr);
@@ -241,7 +241,7 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
exit(1);
} else {
#ifdef WIN32_IOCP
aeWinReceiveDone((int)c->context->fd);
WSIOCP_ReceiveDone((int) c->context->fd);
#endif
while(c->pending) {
if (redisGetReply(c->context,&reply) != REDIS_OK) {
@@ -322,8 +322,9 @@ static void writeHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
if (sdslen(c->obuf) > c->written) {
void *ptr = c->obuf+c->written;
#ifdef WIN32_IOCP
int result = aeWinSocketSend(c->context->fd,(char*)ptr,(int)(sdslen(c->obuf)-c->written),
el, c, NULL, writeHandlerDone);
int result = WSIOCP_SocketSend(c->context->fd, (char*) ptr,
(int) (sdslen(c->obuf) - c->written),
el, c, NULL, writeHandlerDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
if (errno != EPIPE)
fprintf(stderr, "Writing to socket %s\n", wsa_strerror(errno));
@@ -376,7 +377,7 @@ static client createClient(char *cmd, size_t len, client from) {
#ifdef WIN32_IOCP
SOCKADDR_STORAGE ss;
c->context = redisPreConnectNonBlock(config.hostip,config.hostport, &ss);
if (aeWinSocketConnect(c->context->fd, &ss) != 0) {
if (WSIOCP_SocketConnect(c->context->fd, &ss) != 0) {
c->context->err = errno;
strerror_r(errno,c->context->errstr,sizeof(c->context->errstr));
}
src/replication.c
+9 -9
View File
@@ -724,8 +724,8 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(PORT_ULONGLONG)
slave->repldbsize);
result = aeWinSocketSend(fd,bulkcount,(int)sdslen(bulkcount),
el, slave, bulkcount, sendBulkToSlaveLenDone);
result = WSIOCP_SocketSend(fd, bulkcount, (int) sdslen(bulkcount), el,
slave, bulkcount, sendBulkToSlaveLenDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
sdsfree(bulkcount);
freeClient(slave);
@@ -742,8 +742,8 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
return;
}
result = aeWinSocketSend(fd,buf,(int)buflen,
el, slave, buf, sendBulkToSlaveDataDone);
result = WSIOCP_SocketSend(fd, buf, (int) buflen, el, slave, buf,
sendBulkToSlaveDataDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
strerror(errno));
@@ -991,7 +991,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
}
#ifdef WIN32_IOCP
aeWinReceiveDone(fd);
WSIOCP_ReceiveDone(fd);
#endif
if (buf[0] == '-') {
redisLog(REDIS_WARNING,
@@ -1065,7 +1065,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
replicationAbortSyncTransfer();
return;
}
aeWinReceiveDone(fd);
WSIOCP_ReceiveDone(fd);
#else
nread = read(fd,buf,readlen);
if (nread <= 0) {
@@ -2019,10 +2019,10 @@ void replicationCron(void) {
server.rdb_child_type != REDIS_RDB_CHILD_TYPE_SOCKET))
{
#ifdef WIN32_IOCP
if (aeWinSocketSend(slave->fd, "\n", 1,
server.el, NULL, NULL, NULL) == -1) {
if (WSIOCP_SocketSend(slave->fd, "\n", 1, server.el,
NULL, NULL, NULL) == -1) {
#else
if (write(slave->fd, "\n", 1) == -1) {
if (write(slave->fd, "\n", 1) == -1) {
#endif
/* Don't worry, it's just a ping. */
}
src/sentinel.c
+4 -3
View File
@@ -253,7 +253,7 @@ static void redisAeReadEvent(aeEventLoop *el, int fd, void *privdata, int mask)
redisAsyncHandleRead(e->context);
#ifdef WIN32_IOCP
aeWinReceiveDone(fd);
WSIOCP_ReceiveDone(fd);
#endif
}
@@ -272,8 +272,9 @@ static void redisAeWriteEvent(aeEventLoop *el, int fd, void *privdata, int mask)
((void)el); ((void)fd); ((void)mask);
if (redisAsyncHandleWritePrep(e->context) == REDIS_OK) {
result = aeWinSocketSend((int)c->fd,(char*)c->obuf,(int)(sdslen(c->obuf)),
el, e, NULL, writeHandlerDone);
result = WSIOCP_SocketSend((int) c->fd, (char*) c->obuf,
(int) (sdslen(c->obuf)),
el, e, NULL, writeHandlerDone);
if (result == SOCKET_ERROR && errno != WSA_IO_PENDING) {
if (errno != EPIPE) {
redisLog(REDIS_VERBOSE, "Writing to socket %s (%d)\n", wsa_strerror(errno), errno);