diff --git a/deps/hiredis/async.c b/deps/hiredis/async.c index 226939f3..00e81d75 100644 --- a/deps/hiredis/async.c +++ b/deps/hiredis/async.c @@ -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)); } diff --git a/deps/hiredis/net.c b/deps/hiredis/net.c index 2752499e..e580e3c4 100644 --- a/deps/hiredis/net.c +++ b/deps/hiredis/net.c @@ -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; diff --git a/src/Win32_Interop/Win32_Assert.h b/src/Win32_Interop/Win32_Assert.h index 9c808798..c58b384f 100644 --- a/src/Win32_Interop/Win32_Assert.h +++ b/src/Win32_Interop/Win32_Assert.h @@ -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 diff --git a/src/Win32_Interop/Win32_FDAPI.cpp b/src/Win32_Interop/Win32_FDAPI.cpp index 01806e6c..87c39a33 100644 --- a/src/Win32_Interop/Win32_FDAPI.cpp +++ b/src/Win32_Interop/Win32_FDAPI.cpp @@ -45,15 +45,7 @@ void FDAPI_SetCloseSocketState(fnWSIOCP_CloseSocketStateRFD* func) { extern "C" { // FD lookup Winsock equivalents for Win32_wsiocp.c -redis_WSASetLastError WSASetLastError = NULL; redis_WSAGetLastError WSAGetLastError = NULL; -redis_WSAIoctl WSAIoctl = NULL; -redis_WSASend WSASend = NULL; -redis_WSARecv WSARecv = NULL; -redis_WSACleanup WSACleanup = NULL; -redis_WSAGetOverlappedResult WSAGetOverlappedResult = NULL; -redis_WSADuplicateSocket WSADuplicateSocket = NULL; -redis_WSASocket WSASocket = NULL; // other API forwards redis_fwrite fdapi_fwrite = NULL; @@ -67,15 +59,13 @@ redis_access access = NULL; redis_lseek64 lseek64 = NULL; redis_get_osfhandle fdapi_get_osfhandle = NULL; redis_open_osfhandle fdapi_open_osfhandle = NULL; +_redis_fstat fdapi_fstat64 = NULL; // Unix compatible FD based routines redis_pipe pipe = NULL; redis_socket socket = NULL; redis_close fdapi_close = NULL; redis_open open = NULL; -redis_ioctlsocket ioctlsocket = NULL; -redis_inet_addr inet_addr = NULL; -redis_inet_ntoa inet_ntoa = NULL; redis_accept accept = NULL; redis_setsockopt setsockopt = NULL; redis_fcntl fcntl = NULL; @@ -85,17 +75,14 @@ redis_connect connect = NULL; redis_read read = NULL; redis_write write = NULL; redis_fsync fsync = NULL; -_redis_fstat fdapi_fstat64 = NULL; redis_listen listen = NULL; redis_ftruncate ftruncate = NULL; redis_bind bind = NULL; -redis_shutdown shutdown = NULL; -redis_gethostbyname gethostbyname = NULL; redis_htons htons = NULL; redis_htonl htonl = NULL; +redis_ntohs ntohs = NULL; redis_getpeername getpeername = NULL; redis_getsockname getsockname = NULL; -redis_ntohs ntohs = NULL; redis_freeaddrinfo freeaddrinfo = NULL; redis_getaddrinfo getaddrinfo = NULL; redis_inet_ntop inet_ntop = NULL; @@ -112,10 +99,6 @@ auto f_recv = dllfunctor_stdcall("ws2_32.dll", "re auto f_send = dllfunctor_stdcall("ws2_32.dll", "send"); auto f_listen = dllfunctor_stdcall("ws2_32.dll", "listen"); auto f_bind = dllfunctor_stdcall("ws2_32.dll", "bind"); -auto f_shutdown = dllfunctor_stdcall("ws2_32.dll", "shutdown"); -auto f_inet_addr = dllfunctor_stdcall("ws2_32.dll", "inet_addr"); -auto f_gethostbyname = dllfunctor_stdcall("ws2_32.dll", "gethostbyname"); -auto f_inet_ntoa = dllfunctor_stdcall("ws2_32.dll", "inet_ntoa"); auto f_htons = dllfunctor_stdcall("ws2_32.dll", "htons"); auto f_htonl = dllfunctor_stdcall("ws2_32.dll", "htonl"); auto f_getpeername = dllfunctor_stdcall("ws2_32.dll", "getpeername"); @@ -141,54 +124,67 @@ auto f_WSARecv = dllfunctor_stdcallsocket == INVALID_SOCKET) + ASSERT(socketInfo != NULL); + ASSERT(socketInfo->socket == INVALID_SOCKET); - if (socketInfo != NULL) { - if (socketInfo->socket == INVALID_SOCKET) { - RFDMap::getInstance().removeRFDToSocketInfo(rfd); - return; - } else { - redisLog(REDIS_WARNING, "FDAPI_ClearSocketInfo called on non closed socket."); - } + if (socketInfo != NULL) { + if (socketInfo->socket == INVALID_SOCKET) { + RFDMap::getInstance().removeRFDToSocketInfo(rfd); + return; } else { - redisLog(REDIS_WARNING, "FDAPI_ClearSocketInfo called on non attached socket."); + redisLog(REDIS_WARNING, "FDAPI_ClearSocketInfo called on non closed socket."); } + } else { + redisLog(REDIS_WARNING, "FDAPI_ClearSocketInfo called on non attached socket."); + } } -int redis_socket_impl(int af,int type,int protocol) { - RFD rfd = RFDMap::INVALID_RFD; +int redis_socket_impl(int domain, int type, int protocol) { + RFD rfd = INVALID_FD; try { - SOCKET s = f_socket( af, type, protocol ); - if( s != INVALID_SOCKET ) { - rfd = RFDMap::getInstance().addSocket( s ); + SOCKET socket = f_socket(domain, type, protocol); + if (socket != INVALID_SOCKET) { + rfd = RFDMap::getInstance().addSocket(socket); } - return rfd; - } CATCH_AND_REPORT() + return rfd; + } CATCH_AND_REPORT(); return rfd; } int redis_pipe_impl(int *pfds) { - int err = -1; + int result = -1; try { - // Not passing _O_NOINHERIT, the underlying handles are inheritable by default - err = crt_pipe(pfds, 8192, _O_BINARY); - if(err == 0) { - pfds[0] = RFDMap::getInstance().addPosixFD(pfds[0]); - pfds[1] = RFDMap::getInstance().addPosixFD(pfds[1]); + // Not passing _O_NOINHERIT since the underlying handles are + // inheritable by default + result = crt_pipe(pfds, 8192, _O_BINARY); + if (result == 0) { + pfds[0] = RFDMap::getInstance().addCrtFD(pfds[0]); + pfds[1] = RFDMap::getInstance().addCrtFD(pfds[1]); } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); - return err; + return result; } // In unix a fd is a fd. All are closed with close(). @@ -346,7 +367,7 @@ int redis_close_impl(RFD rfd) { SocketInfo* socketInfo = RFDMap::getInstance().lookupSocketInfo(rfd); if (socketInfo != NULL) { - ASSERT(socketInfo->socket != INVALID_SOCKET) + ASSERT(socketInfo->socket != INVALID_SOCKET); if (socketInfo->socket != INVALID_SOCKET) { SOCKET socket = socketInfo->socket; @@ -365,66 +386,60 @@ int redis_close_impl(RFD rfd) { return f_closesocket(socket); } } else { - int posixFD = RFDMap::getInstance().lookupPosixFD(rfd); - if (posixFD != RFDMap::INVALID_FD) { - RFDMap::getInstance().removePosixFD(posixFD); - int retval = crt_close(posixFD); - if (retval == -1) { - errno = GetLastError(); - } - return retval; + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + RFDMap::getInstance().removeCrtFD(crt_fd); + return crt_close(crt_fd); } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int __cdecl redis_open_impl(const char * _Filename, int _OpenFlag, int flags = 0) { - RFD rfd = RFDMap::INVALID_RFD; +int __cdecl redis_open_impl(const char * filename, int openFlag, int flags = 0) { + RFD rfd = INVALID_FD; try { - int posixFD = crt_open(_Filename,_OpenFlag,flags); - if(posixFD != -1) { - rfd = RFDMap::getInstance().addPosixFD(posixFD); - return rfd; - } else { - errno = GetLastError(); - return -1; + int crt_fd = crt_open(filename, openFlag, flags); + if (crt_fd != -1) { + rfd = RFDMap::getInstance().addCrtFD(crt_fd); } - } CATCH_AND_REPORT() + return rfd; + } CATCH_AND_REPORT(); return rfd; } -int redis_accept_impl(int sockfd, struct sockaddr *addr, socklen_t *addrlen) { +int redis_accept_impl(int rfd, struct sockaddr *addr, socklen_t *addrlen) { try { - SOCKET s = RFDMap::getInstance().lookupSocket(sockfd); - if( s != INVALID_SOCKET ) { - SOCKET sAccept = f_accept(s, addr, addrlen); - if( sAccept != INVALID_SOCKET ) { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + SOCKET sAccept = f_accept(socket, addr, addrlen); + if (sAccept != INVALID_SOCKET) { return RFDMap::getInstance().addSocket(sAccept); } else { errno = WSAGetLastError(); - if((errno==ENOENT)||(errno==WSAEWOULDBLOCK)) { + if ((errno == ENOENT) || (errno == WSAEWOULDBLOCK)) { errno = EAGAIN; - return RFDMap::INVALID_RFD; + return INVALID_FD; } } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; - return RFDMap::INVALID_RFD; + return INVALID_FD; } -int redis_setsockopt_impl(int sockfd, int level, int optname, const void *optval, socklen_t optlen) { +int redis_setsockopt_impl(int rfd, int level, int optname, const void *optval, + socklen_t optlen) { try { - SOCKET s = RFDMap::getInstance().lookupSocket(sockfd); - if( s != INVALID_SOCKET ) { - return f_setsockopt(s, level, optname,(const char*)optval, optlen); + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_setsockopt(socket, level, optname, (const char*) optval, optlen); } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; @@ -444,7 +459,9 @@ int redis_fcntl_impl(int rfd, int cmd, int flags = 0 ) { case F_SETFL: { u_long fionbio_flags = (flags & O_NONBLOCK); - if (f_ioctlsocket(socket_info->socket, FIONBIO, &fionbio_flags) == SOCKET_ERROR) { + if (SOCKET_ERROR == f_ioctlsocket(socket_info->socket, + FIONBIO, + &fionbio_flags)) { errno = WSAGetLastError(); return -1; } else { @@ -455,12 +472,12 @@ int redis_fcntl_impl(int rfd, int cmd, int flags = 0 ) { } default: { - ASSERT(cmd == F_GETFL || cmd == F_SETFL) + ASSERT(cmd == F_GETFL || cmd == F_SETFL); return -1; } } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; @@ -474,7 +491,7 @@ int redis_poll_impl(struct pollfd *fds, nfds_t nfds, int timeout) { return -1; } - // NOTE: Treating the fds.fd as a Redis file descriptor and converting to a SOCKET for WSAPoll. + // NOTE: Treating the fds.fd as a RFD and converting to a SOCKET for WSAPoll. for (nfds_t n = 0; n < nfds; n ++) { pollCopy[n].fd = RFDMap::getInstance().lookupSocket((RFD)(fds[n].fd)); pollCopy[n].events = fds[n].events; @@ -484,8 +501,7 @@ int redis_poll_impl(struct pollfd *fds, nfds_t nfds, int timeout) { if (IsWindowsVersionAtLeast(HIBYTE(_WIN32_WINNT_WIN6), LOBYTE(_WIN32_WINNT_WIN6), 0)) { static auto f_WSAPoll = dllfunctor_stdcall("ws2_32.dll", "WSAPoll"); - // See the community addition comments at http://msdn.microsoft.com/en-us/library/windows/desktop/ms741669%28v=vs.85%29.aspx for this API. - // BugCheck seems to indicate that problems with this API in Win8 have been addressed, but this needs to be verified. + // WSAPoll will wait forever if timeout = -1 and the endpoint is not reachable int ret = f_WSAPoll(pollCopy, nfds, timeout); for (nfds_t n = 0; n < nfds; n++) { @@ -554,52 +570,47 @@ int redis_poll_impl(struct pollfd *fds, nfds_t nfds, int timeout) { return ret; } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int redis_getsockopt_impl(int sockfd, int level, int optname, void *optval, socklen_t *optlen) { +int redis_getsockopt_impl(int rfd, int level, int optname, void *optval, socklen_t *optlen) { try { - SOCKET s = RFDMap::getInstance().lookupSocket(sockfd); - if( s == INVALID_SOCKET ) { - errno = EBADF; - return -1; + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_getsockopt(socket, level, optname, (char*) optval, optlen); } - - return f_getsockopt(s,level,optname,(char*)optval,optlen); - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int redis_connect_impl(int sockfd, const struct sockaddr *addr, size_t addrlen) { +int redis_connect_impl(int rfd, const struct sockaddr *addr, size_t addrlen) { try { - SOCKET s = RFDMap::getInstance().lookupSocket(sockfd); - if( s == INVALID_SOCKET ) { - errno = EBADF; - return -1; + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + EnableFastLoopback(socket); + int result = f_connect(socket, addr, (int) addrlen); + errno = WSAGetLastError(); + if ((errno == WSAEINVAL) || (errno == WSAEWOULDBLOCK) || (errno == WSA_IO_PENDING)) { + errno = EINPROGRESS; + } + return result; } - EnableFastLoopback(s); - int r = f_connect(s, addr, (int)addrlen); - errno = WSAGetLastError(); - if ((errno == WSAEINVAL) || (errno == WSAEWOULDBLOCK) || (errno == WSA_IO_PENDING)) { - errno = EINPROGRESS; - } - return r; - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -ssize_t redis_read_impl(int fd, void *buf, size_t count) { +ssize_t redis_read_impl(int rfd, void *buf, size_t count) { try { - SOCKET s = RFDMap::getInstance().lookupSocket( fd ); - if( s != INVALID_SOCKET ) { - int retval = f_recv( s, (char*)buf, (unsigned int)count, 0); + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + int retval = f_recv(socket, (char*) buf, (unsigned int) count, 0); if (retval == -1) { errno = GetLastError(); if (errno == WSAEWOULDBLOCK) { @@ -608,77 +619,81 @@ ssize_t redis_read_impl(int fd, void *buf, size_t count) { } return retval; } else { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD != -1 ) { - int retval = crt_read(posixFD, buf,(unsigned int)count); - if(retval == -1) { + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != -1) { + int retval = crt_read(crt_fd, buf, (unsigned int) count); + if (retval == -1) { errno = GetLastError(); } return retval; - } - else { + } else { errno = EBADF; return 0; } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -ssize_t redis_write_impl(int fd, const void *buf, size_t count) { +ssize_t redis_write_impl(int rfd, const void *buf, size_t count) { try { - SOCKET s = RFDMap::getInstance().lookupSocket( fd ); - if( s != INVALID_SOCKET ) { - int ret = f_send( s, (char*)buf, (unsigned int)count, 0); + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + int ret = f_send(socket, (char*) buf, (unsigned int) count, 0); if (ret == SOCKET_ERROR) { set_errno_from_last_error(); } return ret; } else { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD != -1 ) { - if (posixFD == _fileno(stdout)) { + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != -1) { + if (crt_fd == _fileno(stdout)) { DWORD bytesWritten = 0; - if (FALSE != ParseAndPrintANSIString(GetStdHandle(STD_OUTPUT_HANDLE), buf, (DWORD)count, &bytesWritten)) { - return (int)bytesWritten; + if (FALSE != ParseAndPrintANSIString(GetStdHandle(STD_OUTPUT_HANDLE), + buf, + (DWORD) count, + &bytesWritten)) { + return (int) bytesWritten; } else { errno = GetLastError(); return 0; } - } else if (posixFD == _fileno(stderr)) { + } else if (crt_fd == _fileno(stderr)) { DWORD bytesWritten = 0; - if (FALSE != ParseAndPrintANSIString(GetStdHandle(STD_ERROR_HANDLE), buf, (DWORD)count, &bytesWritten)) { - return (int)bytesWritten; + if (FALSE != ParseAndPrintANSIString(GetStdHandle(STD_ERROR_HANDLE), + buf, + (DWORD) count, + &bytesWritten)) { + return (int) bytesWritten; } else { errno = GetLastError(); return 0; } } else { - int retval = crt_write(posixFD, buf, (unsigned int)count); + int retval = crt_write(crt_fd, buf, (unsigned int) count); if (retval == -1) { errno = GetLastError(); } return retval; } - } - else { + } else { errno = EBADF; return 0; } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int redis_fsync_impl(int fd) { +int redis_fsync_impl(int rfd) { try { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if (posixFD != -1) { - HANDLE h = (HANDLE) crtget_osfhandle(posixFD); + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + HANDLE h = (HANDLE) crtget_osfhandle(crt_fd); if (h == INVALID_HANDLE_VALUE) { errno = EBADF; return -1; @@ -698,30 +713,30 @@ int redis_fsync_impl(int fd) { } } return 0; - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int redis_fstat_impl(int fd, struct __stat64 *buffer) { +int redis_fstat_impl(int rfd, struct __stat64 *buffer) { try { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if (posixFD != -1) { - return _fstat64(posixFD, buffer); + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + return _fstat64(crt_fd, buffer); } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } -int redis_listen_impl(int sockfd, int backlog) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( sockfd ); - if( s != INVALID_SOCKET ) { - EnableFastLoopback(s); - return f_listen( s, backlog ); +int redis_listen_impl(int rfd, int backlog) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + EnableFastLoopback(socket); + return f_listen(socket, backlog); } else { errno = EBADF; return 0; @@ -732,12 +747,12 @@ int redis_listen_impl(int sockfd, int backlog) { return -1; } -int redis_ftruncate_impl(int fd, PORT_LONGLONG length) { +int redis_ftruncate_impl(int rfd, PORT_LONGLONG length) { try { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if (posixFD != -1) { - HANDLE h = (HANDLE) crtget_osfhandle(posixFD); + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + HANDLE h = (HANDLE) crtget_osfhandle(crt_fd); if (h == INVALID_HANDLE_VALUE) { errno = EBADF; return -1; @@ -756,26 +771,11 @@ int redis_ftruncate_impl(int fd, PORT_LONGLONG length) { return -1; } -int redis_bind_impl(int sockfd, const struct sockaddr *addr, socklen_t addrlen) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( sockfd ); - if( s != INVALID_SOCKET ) { - return f_bind(s, addr, addrlen); - } else { - errno = EBADF; - return 0; - } - } CATCH_AND_REPORT(); - - errno = EBADF; - return -1; -} - -int redis_shutdown_impl(int sockfd, int how) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( sockfd ); - if( s != INVALID_SOCKET ) { - return f_shutdown(s, how); +int redis_bind_impl(int rfd, const struct sockaddr *addr, socklen_t addrlen) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_bind(socket, addr, addrlen); } else { errno = EBADF; return 0; @@ -794,11 +794,12 @@ int redis_WSAGetLastError_impl(void) { return f_WSAGetLastError(); } -BOOL redis_WSAGetOverlappedResult_impl(int rfd, LPWSAOVERLAPPED lpOverlapped, LPDWORD lpcbTransfer, BOOL fWait, LPDWORD lpdwFlags) { +BOOL FDAPI_WSAGetOverlappedResult(int rfd, LPWSAOVERLAPPED lpOverlapped, + LPDWORD lpcbTransfer, BOOL fWait, LPDWORD lpdwFlags) { try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_WSAGetOverlappedResult(s,lpOverlapped, lpcbTransfer, fWait, lpdwFlags); + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_WSAGetOverlappedResult(socket, lpOverlapped, lpcbTransfer, fWait, lpdwFlags); } else { errno = EBADF; return SOCKET_ERROR; @@ -808,11 +809,12 @@ BOOL redis_WSAGetOverlappedResult_impl(int rfd, LPWSAOVERLAPPED lpOverlapped, LP return SOCKET_ERROR; } -int redis_WSADuplicateSocket_impl(int rfd, DWORD dwProcessId, LPWSAPROTOCOL_INFO lpProtocolInfo) { +int FDAPI_WSADuplicateSocket(int rfd, DWORD dwProcessId, + LPWSAPROTOCOL_INFO lpProtocolInfo) { try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_WSADuplicateSocket(s, dwProcessId, lpProtocolInfo); + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_WSADuplicateSocket(socket, dwProcessId, lpProtocolInfo); } else { errno = EBADF; return SOCKET_ERROR; @@ -822,8 +824,9 @@ int redis_WSADuplicateSocket_impl(int rfd, DWORD dwProcessId, LPWSAPROTOCOL_INFO return SOCKET_ERROR; } -int redis_WSASocket_impl(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags) { - RFD rfd = RFDMap::INVALID_RFD; +int FDAPI_WSASocket(int af, int type, int protocol, + LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags) { + RFD rfd = INVALID_FD; try { SOCKET socket = f_WSASocket(af, type, @@ -832,19 +835,30 @@ int redis_WSASocket_impl(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpPr g, dwFlags); - if(socket != INVALID_SOCKET) { + if (socket != INVALID_SOCKET) { rfd = RFDMap::getInstance().addSocket(socket); } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); return rfd; } -int redis_WSAIoctl_impl(RFD rfd,DWORD dwIoControlCode,LPVOID lpvInBuffer,DWORD cbInBuffer,LPVOID lpvOutBuffer,DWORD cbOutBuffer,LPDWORD lpcbBytesReturned,LPWSAOVERLAPPED lpOverlapped,LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_WSAIoctl(s,dwIoControlCode,lpvInBuffer,cbInBuffer,lpvOutBuffer,cbOutBuffer,lpcbBytesReturned,lpOverlapped,lpCompletionRoutine); +int FDAPI_WSAIoctl(RFD rfd, DWORD dwIoControlCode, LPVOID lpvInBuffer, + DWORD cbInBuffer, LPVOID lpvOutBuffer, DWORD cbOutBuffer, + LPDWORD lpcbBytesReturned, LPWSAOVERLAPPED lpOverlapped, + LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_WSAIoctl(socket, + dwIoControlCode, + lpvInBuffer, + cbInBuffer, + lpvOutBuffer, + cbOutBuffer, + lpcbBytesReturned, + lpOverlapped, + lpCompletionRoutine); } else { errno = EBADF; return SOCKET_ERROR; @@ -854,11 +868,19 @@ int redis_WSAIoctl_impl(RFD rfd,DWORD dwIoControlCode,LPVOID lpvInBuffer,DWORD c return SOCKET_ERROR; } -int redis_WSASend_impl(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD lpNumberOfBytesSent, DWORD dwFlags, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_WSASend( s, lpBuffers, dwBufferCount, lpNumberOfBytesSent, dwFlags, lpOverlapped, lpCompletionRoutine ); +int FDAPI_WSASend(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, + LPDWORD lpNumberOfBytesSent, DWORD dwFlags, LPWSAOVERLAPPED lpOverlapped, + LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_WSASend(socket, + lpBuffers, + dwBufferCount, + lpNumberOfBytesSent, + dwFlags, + lpOverlapped, + lpCompletionRoutine); } else { errno = EBADF; return SOCKET_ERROR; @@ -868,11 +890,19 @@ int redis_WSASend_impl(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD return SOCKET_ERROR; } -int redis_WSARecv_impl(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD lpNumberOfBytesRecvd, LPDWORD lpFlags, LPWSAOVERLAPPED lpOverlapped, LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_WSARecv( s, lpBuffers, dwBufferCount, lpNumberOfBytesRecvd, lpFlags, lpOverlapped, lpCompletionRoutine ); +int FDAPI_WSARecv(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, + LPDWORD lpNumberOfBytesRecvd, LPDWORD lpFlags, LPWSAOVERLAPPED lpOverlapped, + LPWSAOVERLAPPED_COMPLETION_ROUTINE lpCompletionRoutine) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_WSARecv(socket, + lpBuffers, + dwBufferCount, + lpNumberOfBytesRecvd, + lpFlags, + lpOverlapped, + lpCompletionRoutine); } else { errno = EBADF; return SOCKET_ERROR; @@ -882,15 +912,11 @@ int redis_WSARecv_impl(int rfd, LPWSABUF lpBuffers, DWORD dwBufferCount, LPDWORD return SOCKET_ERROR; } -int redis_WSACleanup_impl(void) { - return f_WSACleanup(); -} - -int redis_ioctlsocket_impl(int rfd, long cmd, u_long *argp) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( rfd ); - if( s != INVALID_SOCKET ) { - return f_ioctlsocket(s,cmd,argp); +int FDAPI_ioctlsocket(int rfd, long cmd, u_long *argp) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_ioctlsocket(socket, cmd, argp); } else { errno = EBADF; return SOCKET_ERROR; @@ -900,18 +926,6 @@ int redis_ioctlsocket_impl(int rfd, long cmd, u_long *argp) { return SOCKET_ERROR; } -unsigned long redis_inet_addr_impl(const char *cp) { - return f_inet_addr(cp); -} - -struct hostent* redis_gethostbyname_impl(const char *name) { - return f_gethostbyname(name); -} - -char* redis_inet_ntoa_impl(struct in_addr in) { - return f_inet_ntoa(in); -} - u_short redis_htons_impl(u_short hostshort) { return f_htons(hostshort); } @@ -920,11 +934,11 @@ u_long redis_htonl_impl(u_long hostlong) { return f_htonl(hostlong); } -int redis_getpeername_impl(int sockfd, struct sockaddr *addr, socklen_t * addrlen) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( sockfd ); - if( s != INVALID_SOCKET ) { - return f_getpeername(s,addr, addrlen); +int redis_getpeername_impl(int rfd, struct sockaddr *addr, socklen_t * addrlen) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_getpeername(socket, addr, addrlen); } else { errno = EBADF; return SOCKET_ERROR; @@ -935,11 +949,11 @@ int redis_getpeername_impl(int sockfd, struct sockaddr *addr, socklen_t * addrle return -1; } -int redis_getsockname_impl(int sockfd, struct sockaddr* addrsock, int* addrlen) { - try { - SOCKET s = RFDMap::getInstance().lookupSocket( sockfd ); - if( s != INVALID_SOCKET ) { - return f_getsockname(s,addrsock, addrlen); +int redis_getsockname_impl(int rfd, struct sockaddr* addrsock, int* addrlen) { + try { + SOCKET socket = RFDMap::getInstance().lookupSocket(rfd); + if (socket != INVALID_SOCKET) { + return f_getsockname(socket, addrsock, addrlen); } else { errno = EBADF; return 0; @@ -951,11 +965,11 @@ int redis_getsockname_impl(int sockfd, struct sockaddr* addrsock, int* addrlen) } u_short redis_ntohs_impl(u_short netshort) { - return f_ntohs( netshort ); + return f_ntohs(netshort); } -int redis_setmode_impl(int fd,int mode) { - return crt_setmode(fd,mode); +int redis_setmode_impl(int rfd, int mode) { + return crt_setmode(rfd, mode); } size_t redis_fwrite_impl(const void * _Str, size_t _Size, size_t _Count, FILE * _File) { @@ -963,43 +977,44 @@ size_t redis_fwrite_impl(const void * _Str, size_t _Size, size_t _Count, FILE * } int redis_fclose_impl(FILE * file) { - int posixFD = crt_fileno(file); - if (posixFD != -1) { - RFDMap::getInstance().removePosixFD(posixFD); + int crt_fd = crt_fileno(file); + if (crt_fd != INVALID_FD) { + RFDMap::getInstance().removeCrtFD(crt_fd); } return crt_fclose(file); } int redis_fileno_impl(FILE* file) { - int rfd = -1; - int posixFD = crt_fileno(file); - if (posixFD != -1) { - // if posixFD is already mapped, addPosixFD() will return the existing rfd. - rfd = RFDMap::getInstance().addPosixFD(posixFD); + int crt_fd = crt_fileno(file); + if (crt_fd != INVALID_FD) { + // If crt_fd is already mapped, addCrtFD() will return the existing rfd. + return RFDMap::getInstance().addCrtFD(crt_fd); + } else { + return INVALID_FD; } - return rfd; } -int redis_select_impl(int nfds, fd_set *readfds, fd_set *writefds,fd_set *exceptfds, struct timeval *timeout) { - try { - if (readfds != NULL) { - for(u_int r = 0; r < readfds->fd_count; r++) { - readfds->fd_array[r] = RFDMap::getInstance().lookupSocket( (RFD)readfds->fd_array[r] ); - } - } - if (writefds != NULL) { - for(u_int r = 0; r < writefds->fd_count; r++) { - writefds->fd_array[r] = RFDMap::getInstance().lookupSocket( (RFD)writefds->fd_array[r] ); +int redis_select_impl(int nfds, fd_set *readfds, fd_set *writefds, + fd_set *exceptfds, struct timeval *timeout) { + try { + if (readfds != NULL) { + for (u_int r = 0; r < readfds->fd_count; r++) { + readfds->fd_array[r] = RFDMap::getInstance().lookupSocket((RFD) readfds->fd_array[r]); } - } - if (exceptfds != NULL ) { - for(u_int r = 0; r < exceptfds->fd_count; r++) { - exceptfds->fd_array[r] = RFDMap::getInstance().lookupSocket( (RFD)exceptfds->fd_array[r] ); - } - } + } + if (writefds != NULL) { + for (u_int r = 0; r < writefds->fd_count; r++) { + writefds->fd_array[r] = RFDMap::getInstance().lookupSocket((RFD) writefds->fd_array[r]); + } + } + if (exceptfds != NULL) { + for (u_int r = 0; r < exceptfds->fd_count; r++) { + exceptfds->fd_array[r] = RFDMap::getInstance().lookupSocket((RFD) exceptfds->fd_array[r]); + } + } - return f_select(nfds,readfds,writefds,exceptfds,timeout); - } CATCH_AND_REPORT(); + return f_select(nfds, readfds, writefds, exceptfds, timeout); + } CATCH_AND_REPORT(); errno = EBADF; return -1; @@ -1009,17 +1024,17 @@ u_int redis_ntohl_impl(u_int netlong){ return f_ntohl(netlong); } -int redis_isatty_impl(int fd) { - try { - int posixFD = RFDMap::getInstance().lookupPosixFD(fd); - if( posixFD != -1) { - return crt_isatty(posixFD); - } else if (fd >= 0 && fd <= 2) { - return crt_isatty(fd); - } else { - errno = EBADF; - return 0; - } +int redis_isatty_impl(int rfd) { + try { + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + return crt_isatty(crt_fd); + } else if (rfd >= 0 && rfd <= 2) { + return crt_isatty(rfd); + } else { + errno = EBADF; + return 0; + } } CATCH_AND_REPORT(); errno = EBADF; @@ -1030,11 +1045,11 @@ int redis_access_impl(const char *pathname, int mode) { return crt_access(pathname, mode); } -u_int64 redis_lseek64_impl(int fd, u_int64 offset, int whence) { - try { - int posixFD = RFDMap::getInstance().lookupPosixFD(fd); - if (posixFD != -1) { - return crt_lseek64(posixFD, offset, whence); +u_int64 redis_lseek64_impl(int rfd, u_int64 offset, int whence) { + try { + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + return crt_lseek64(crt_fd, offset, whence); } } CATCH_AND_REPORT(); @@ -1042,11 +1057,11 @@ u_int64 redis_lseek64_impl(int fd, u_int64 offset, int whence) { return -1; } -intptr_t redis_get_osfhandle_impl(int fd) { - try { - int posixFD = RFDMap::getInstance().lookupPosixFD(fd); - if (posixFD != -1) { - return crtget_osfhandle(posixFD); +intptr_t redis_get_osfhandle_impl(int rfd) { + try { + int crt_fd = RFDMap::getInstance().lookupCrtFD(rfd); + if (crt_fd != INVALID_FD) { + return crtget_osfhandle(crt_fd); } } CATCH_AND_REPORT(); @@ -1055,13 +1070,13 @@ intptr_t redis_get_osfhandle_impl(int fd) { } int redis_open_osfhandle_impl(intptr_t osfhandle, int flags) { - RFD rfd = RFDMap::INVALID_RFD; + RFD rfd = INVALID_FD; try { - int posixFD = crt_open_osfhandle(osfhandle, flags); - if(posixFD != -1) { - rfd = RFDMap::getInstance().addPosixFD(posixFD); + int crt_fd = crt_open_osfhandle(osfhandle, flags); + if (crt_fd != INVALID_FD) { + rfd = RFDMap::getInstance().addCrtFD(crt_fd); } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); return rfd; } @@ -1086,7 +1101,7 @@ const char* redis_inet_ntop_impl(int af, const void *src, char *dst, size_t size memcpy(&(srcaddr.sin_addr), src, sizeof(srcaddr.sin_addr)); srcaddr.sin_family = af; - if (f_WSAAddressToStringA((struct sockaddr*) &srcaddr, sizeof(struct sockaddr_in), 0, dst, (LPDWORD)&size) != 0) { + if (f_WSAAddressToStringA((struct sockaddr*) &srcaddr, sizeof(struct sockaddr_in), 0, dst, (LPDWORD) &size) != 0) { return NULL; } return dst; @@ -1107,9 +1122,6 @@ BOOL ParseStorageAddress(const char *ip, int port, SOCKADDR_STORAGE* pSotrageAdd hints.ai_flags = AI_NUMERICSERV | AI_PASSIVE; if ((status = getaddrinfo(ip, port_buffer, &hints, &res) != 0)) { -#ifdef _DEBUG - fprintf(stderr, "getaddrinfo: %S\n", gai_strerror(status)); -#endif return FALSE; } @@ -1162,29 +1174,16 @@ private: listen = redis_listen_impl; ftruncate = redis_ftruncate_impl; bind = redis_bind_impl; - shutdown = redis_shutdown_impl; htons = redis_htons_impl; htonl = redis_htonl_impl; getpeername = redis_getpeername_impl; getsockname = redis_getsockname_impl; ntohs = redis_ntohs_impl; - ioctlsocket = redis_ioctlsocket_impl; - inet_addr = redis_inet_addr_impl; - gethostbyname = redis_gethostbyname_impl; - inet_ntoa = redis_inet_ntoa_impl; fdapi_fwrite = redis_fwrite_impl; fdapi_fclose = redis_fclose_impl; fdapi_fileno = redis_fileno_impl; fdapi_setmode = redis_setmode_impl; - WSASetLastError = redis_WSASetLastError_impl; WSAGetLastError = redis_WSAGetLastError_impl; - WSAIoctl = redis_WSAIoctl_impl; - WSASend = redis_WSASend_impl; - WSARecv = redis_WSARecv_impl; - WSACleanup = redis_WSACleanup_impl; - WSAGetOverlappedResult = redis_WSAGetOverlappedResult_impl; - WSADuplicateSocket = redis_WSADuplicateSocket_impl; - WSASocket = redis_WSASocket_impl; select = redis_select_impl; ntohl = redis_ntohl_impl; isatty = redis_isatty_impl; diff --git a/src/Win32_Interop/Win32_FDAPI.h b/src/Win32_Interop/Win32_FDAPI.h index 840f2d5a..3ee2a2a9 100644 --- a/src/Win32_Interop/Win32_FDAPI.h +++ b/src/Win32_Interop/Win32_FDAPI.h @@ -40,18 +40,14 @@ typedef unsigned long nfds_t; #include // 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) diff --git a/src/Win32_Interop/Win32_QFork.cpp b/src/Win32_Interop/Win32_QFork.cpp index 6f3d380f..f285e5ab 100644 --- a/src/Win32_Interop/Win32_QFork.cpp +++ b/src/Win32_Interop/Win32_QFork.cpp @@ -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]); } } diff --git a/src/Win32_Interop/win32_rfdmap.cpp b/src/Win32_Interop/win32_rfdmap.cpp index 41caecf5..31ab48f8 100644 --- a/src/Win32_Interop/win32_rfdmap.cpp +++ b/src/Win32_Interop/win32_rfdmap.cpp @@ -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::iterator mit = PosixFDToRFDMap.find(posixFD); - if (mit != PosixFDToRFDMap.end()) { + map::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; } diff --git a/src/Win32_Interop/win32_rfdmap.h b/src/Win32_Interop/win32_rfdmap.h index 73acddd4..2499ad85 100644 --- a/src/Win32_Interop/win32_rfdmap.h +++ b/src/Win32_Interop/win32_rfdmap.h @@ -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 SocketToRFDMap; - map PosixFDToRFDMap; + map CrtFDToRFDMap; map RFDToSocketInfoMap; - map RFDToPosixFDMap; + map RFDToCrtFDMap; queue 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); }; diff --git a/src/Win32_Interop/win32_wsiocp.c b/src/Win32_Interop/win32_wsiocp.c index bd8af3e9..0da75058 100644 --- a/src/Win32_Interop/win32_wsiocp.c +++ b/src/Win32_Interop/win32_wsiocp.c @@ -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; } diff --git a/src/Win32_Interop/win32_wsiocp.h b/src/Win32_Interop/win32_wsiocp.h index f59e07ef..ba70cdd0 100644 --- a/src/Win32_Interop/win32_wsiocp.h +++ b/src/Win32_Interop/win32_wsiocp.h @@ -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); diff --git a/src/Win32_Interop/win32fixes.h b/src/Win32_Interop/win32fixes.h index c3e398b0..93e6dd03 100644 --- a/src/Win32_Interop/win32fixes.h +++ b/src/Win32_Interop/win32fixes.h @@ -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); diff --git a/src/ae_wsiocp.c b/src/ae_wsiocp.c index 5ad533a7..a9770827 100644 --- a/src/ae_wsiocp.c +++ b/src/ae_wsiocp.c @@ -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--; diff --git a/src/anet.c b/src/anet.c index 57b73be5..2ba2f8b6 100644 --- a/src/anet.c +++ b/src/anet.c @@ -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 diff --git a/src/networking.c b/src/networking.c index 73fd4473..35faedd8 100644 --- a/src/networking.c +++ b/src/networking.c @@ -36,7 +36,7 @@ #endif #include -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); diff --git a/src/redis-benchmark.c b/src/redis-benchmark.c index 53d4c113..e285bf75 100644 --- a/src/redis-benchmark.c +++ b/src/redis-benchmark.c @@ -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)); } diff --git a/src/replication.c b/src/replication.c index c579fa30..11f3a1ab 100644 --- a/src/replication.c +++ b/src/replication.c @@ -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. */ } diff --git a/src/sentinel.c b/src/sentinel.c index 6b1365f6..b132558f 100644 --- a/src/sentinel.c +++ b/src/sentinel.c @@ -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);