diff --git a/src/Win32_Interop/Win32_FDAPI.cpp b/src/Win32_Interop/Win32_FDAPI.cpp index 5d164d4c..cc6a45cf 100644 --- a/src/Win32_Interop/Win32_FDAPI.cpp +++ b/src/Win32_Interop/Win32_FDAPI.cpp @@ -51,6 +51,8 @@ redis_WSASocket WSASocket = NULL; // other API forwards redis_fwrite fdapi_fwrite = NULL; +redis_fclose fdapi_fclose = NULL; +redis_fileno fdapi_fileno = NULL; redis_setmode fdapi_setmode = NULL; redis_select select = NULL; redis_ntohl ntohl = NULL; @@ -280,12 +282,12 @@ int FDAPI_UpdateAcceptContext(int fd) } } CATCH_AND_REPORT() errno = EBADF; - return RFDMap::invalidRFD; + return -1; } auto f_socket = dllfunctor_stdcall("ws2_32.dll", "socket"); int redis_socket_impl(int af,int type,int protocol) { - RFD rfd = RFDMap::invalidRFD; + RFD rfd = -1; try { SOCKET s = f_socket( af, type, protocol ); if( s != INVALID_SOCKET ) { @@ -303,8 +305,8 @@ int redis_pipe_impl(int *pfds) { // 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]); + pfds[0] = RFDMap::getInstance().addCrtFD(pfds[0]); + pfds[1] = RFDMap::getInstance().addCrtFD(pfds[1]); } } CATCH_AND_REPORT() @@ -315,37 +317,33 @@ int redis_pipe_impl(int *pfds) { // In unix a fd is a fd. All are closed with close(). auto f_closesocket = dllfunctor_stdcall("ws2_32.dll", "closesocket"); int redis_close_impl(RFD rfd) { + int retval = -1; try { - SOCKET s = RFDMap::getInstance().lookupSocket(rfd); - if( s != INVALID_SOCKET ) { - RFDMap::getInstance().removeSocket(s); - return f_closesocket(s); - } else { - int posixFD = RFDMap::getInstance().lookupPosixFD(rfd); - if(posixFD != -1) { - RFDMap::getInstance().removePosixFD(posixFD); - int retval = crt_close(posixFD); - if( retval == -1 ) { + RFD_INFO rfd_info = RFDMap::getInstance().GetRFDInfo(rfd); + RFDMap::getInstance().removeRFD(rfd); + switch (rfd_info.type) { + case RFD_TYPE::SOCKET: + retval = f_closesocket(rfd_info.value.socket); + break; + case RFD_TYPE::CRTFD: + retval = crt_close(rfd_info.value.crtFD); + if (retval == -1) { errno = GetLastError(); } - return retval; - } - else { + break; + default: errno = EBADF; - return -1; - } } - } CATCH_AND_REPORT() - - return -1; + } CATCH_AND_REPORT(); + return retval; } int __cdecl redis_open_impl(const char * _Filename, int _OpenFlag, int flags = 0) { - RFD rfd = RFDMap::invalidRFD; + RFD rfd = -1; try { int posixFD = crt_open(_Filename,_OpenFlag,flags); if(posixFD != -1) { - rfd = RFDMap::getInstance().addPosixFD(posixFD); + rfd = RFDMap::getInstance().addCrtFD(posixFD); return rfd; } else { errno = GetLastError(); @@ -368,14 +366,14 @@ int redis_accept_impl(int sockfd, struct sockaddr *addr, socklen_t *addrlen) { errno = WSAGetLastError(); if((errno==ENOENT)||(errno==WSAEWOULDBLOCK)) { errno = EAGAIN; - return RFDMap::invalidRFD; + return -1; } } } } CATCH_AND_REPORT() errno = EBADF; - return RFDMap::invalidRFD; + return -1; } auto f_setsockopt = dllfunctor_stdcall("ws2_32.dll", "setsockopt"); @@ -392,40 +390,49 @@ int redis_setsockopt_impl(int sockfd, int level, int optname, const void *optval } auto f_ioctlsocket = dllfunctor_stdcall("ws2_32.dll", "ioctlsocket"); -int redis_fcntl_impl(int fd, int cmd, int flags = 0 ) { +int redis_fcntl_impl(int rfd, int cmd, int flags = 0 ) { try { - SOCKET s = RFDMap::getInstance().lookupSocket(fd); - if( s != INVALID_SOCKET ) { - switch(cmd) { - case F_GETFL: - { - // Since there is no way to determine if a socket is blocking in winsock, we keep track of this separately. - RedisSocketState state; - RFDMap::getInstance().GetSocketState(s, state); - return state.IsBlockingSocket ? O_NONBLOCK : 0; + switch (cmd) { + case F_GETFL: { + // Since there is no way to determine if a socket/pipe is blocking in winsock, + // we keep track of this separately. + RFD_INFO rfd_info = RFDMap::getInstance().GetRFDInfo(rfd); + if (rfd_info.type != RFD_TYPE::INVALID) { + return rfd_info.flags; + } else { + return 0; } - case F_SETFL: - { - RedisSocketState state; - state.IsBlockingSocket = ((flags & O_NONBLOCK) != 0); - u_long fionbio_flags = state.IsBlockingSocket; - if( f_ioctlsocket(s, FIONBIO, &fionbio_flags) == SOCKET_ERROR ) { - errno = WSAGetLastError(); - return -1; - } else { - RFDMap::getInstance().SetSocketState( s, state ); + } + break; + case F_SETFL: { + RFD_INFO rfd_info = RFDMap::getInstance().GetRFDInfo(rfd); + switch (rfd_info.type) { + case RFD_TYPE::SOCKET: { + u_long fionbio_flags = (flags & O_NONBLOCK); + if (f_ioctlsocket(rfd_info.value.socket, FIONBIO, &fionbio_flags) == SOCKET_ERROR) { + errno = WSAGetLastError(); + return -1; + } else { + rfd_info.flags = flags; + RFDMap::getInstance().SetRFDInfo(rfd, rfd_info); + return 0; + } + } + break; + case RFD_TYPE::CRTFD: { + // TODO return 0; } break; - } - default: - { - DebugBreak(); - return -1; + default: + return -1; } } - } - } CATCH_AND_REPORT() + break; + default: + return -1; + } + } CATCH_AND_REPORT(); errno = EBADF; return -1; @@ -585,81 +592,84 @@ int redis_connect_impl(int sockfd, const struct sockaddr *addr, size_t addrlen) auto f_recv = dllfunctor_stdcall("ws2_32.dll", "recv"); -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); - if (retval == -1) { - errno = GetLastError(); - if (errno == WSAEWOULDBLOCK) { - errno = EAGAIN; + RFD_INFO rfd_info = RFDMap::getInstance().GetRFDInfo(rfd); + switch (rfd_info.type) { + case RFD_TYPE::SOCKET: { + int retval = f_recv(rfd_info.value.socket, (char*) buf, (unsigned int) count, 0); + if (retval == -1) { + errno = GetLastError(); + if (errno == WSAEWOULDBLOCK) { + errno = EAGAIN; + } } + return retval; } - return retval; - } else { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD != -1 ) { - int retval = crt_read(posixFD, buf,(unsigned int)count); - if(retval == -1) { + break; + case RFD_TYPE::CRTFD: { + int retval = crt_read(rfd_info.value.crtFD, buf, (unsigned int) count); + if (retval == -1) { errno = GetLastError(); } return retval; } - else { + break; + default: errno = EBADF; return 0; - } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; } auto f_send = dllfunctor_stdcall("ws2_32.dll", "send"); -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) { + ssize_t retval; try { - SOCKET s = RFDMap::getInstance().lookupSocket( fd ); - if( s != INVALID_SOCKET ) { - int ret = f_send( s, (char*)buf, (unsigned int)count, 0); - if (ret == SOCKET_ERROR) { - set_errno_from_last_error(); + RFD_INFO rfd_info = RFDMap::getInstance().GetRFDInfo(rfd); + switch (rfd_info.type) { + case RFD_TYPE::SOCKET: { + retval = f_send(rfd_info.value.socket, (char*) buf, (unsigned int) count, 0); + if (retval == SOCKET_ERROR) { + set_errno_from_last_error(); + } + return retval; } - return ret; - } else { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD != -1 ) { - if (posixFD == _fileno(stdout)) { + break; + case RFD_TYPE::CRTFD: { + if (rfd_info.value.crtFD == _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 (rfd_info.value.crtFD == _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); + retval = crt_write(rfd_info.value.crtFD, buf, (unsigned int) count); if (retval == -1) { errno = GetLastError(); } return retval; } } - else { + break; + default: errno = EBADF; return 0; - } } - } CATCH_AND_REPORT() + } CATCH_AND_REPORT(); errno = EBADF; return -1; @@ -667,14 +677,9 @@ ssize_t redis_write_impl(int fd, const void *buf, size_t count) { int redis_fsync_impl(int fd) { try { - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD == -1 ) { - // There is one place in Redis where we are not tracking posix FDs because it involves - // direct ocnversion of a FILE* to an FD. - posixFD = fd; - } + int crtFD = RFDMap::getInstance().lookupCrtFD(fd); - HANDLE h = (HANDLE) crtget_osfhandle(posixFD); + HANDLE h = (HANDLE) crtget_osfhandle(crtFD); DWORD err; if (h == INVALID_HANDLE_VALUE) { @@ -686,15 +691,13 @@ int redis_fsync_impl(int fd) { err = GetLastError(); switch (err) { case ERROR_INVALID_HANDLE: - errno = EINVAL; + errno = EINVAL; break; - default: errno = EIO; } return -1; } - return 0; } CATCH_AND_REPORT() @@ -702,14 +705,12 @@ int redis_fsync_impl(int fd) { 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 ) { - posixFD = fd; + int crtFD = RFDMap::getInstance().lookupCrtFD(rfd); + if (crtFD != -1) { + return _fstat64(crtFD, buffer); } - - return _fstat64(posixFD, buffer); } CATCH_AND_REPORT() errno = EBADF; @@ -734,30 +735,25 @@ int redis_listen_impl(int sockfd, int backlog) { return -1; } -int redis_ftruncate_impl(int fd, PORT_LONGLONG length) { - try - { - LARGE_INTEGER l, o; - HANDLE h = INVALID_HANDLE_VALUE; +int redis_ftruncate_impl(int rfd, PORT_LONGLONG length) { + try { + int crtFD = RFDMap::getInstance().lookupCrtFD(rfd); + if (crtFD != -1) { + HANDLE h = (HANDLE) crtget_osfhandle(crtFD); - int posixFD = RFDMap::getInstance().lookupPosixFD( fd ); - if( posixFD == -1 ) { - h = (HANDLE) crtget_osfhandle (fd); - } else { - h = (HANDLE) crtget_osfhandle (posixFD); + if (h == INVALID_HANDLE_VALUE) { + errno = EBADF; + return -1; + } + + LARGE_INTEGER l, o; + l.QuadPart = length; + + if (!SetFilePointerEx(h, l, &o, FILE_BEGIN)) return -1; + if (!SetEndOfFile(h)) return -1; + + return 0; } - - if( h == INVALID_HANDLE_VALUE) { - errno = EBADF; - return -1; - } - - l.QuadPart = length; - - if (!SetFilePointerEx(h, l, &o, FILE_BEGIN)) return -1; - if (!SetEndOfFile(h)) return -1; - - return 0; } CATCH_AND_REPORT(); errno = EBADF; @@ -838,7 +834,7 @@ int redis_WSADuplicateSocket_impl(int rfd, DWORD dwProcessId, LPWSAPROTOCOL_INFO auto f_WSASocket = dllfunctor_stdcall("ws2_32.dll", "WSASocketW"); int redis_WSASocket_impl(int af, int type, int protocol, LPWSAPROTOCOL_INFO lpProtocolInfo, GROUP g, DWORD dwFlags) { - RFD rfd = RFDMap::invalidRFD; + RFD rfd = -1; try { SOCKET socket = f_WSASocket(af, type, @@ -987,7 +983,25 @@ int redis_setmode_impl(int fd,int mode) { } size_t redis_fwrite_impl(const void * _Str, size_t _Size, size_t _Count, FILE * _File) { - return crtfwrite(_Str, _Size, _Count, _File); + return crt_fwrite(_Str, _Size, _Count, _File); +} + +int redis_fclose_impl(FILE * file) { + int crtFD = crt_fileno(file); + if (crtFD != -1) { + RFDMap::getInstance().removeCrtRFD(crtFD); + } + return crt_fclose(file); +} + +int redis_fileno_impl(FILE* file) { + int rfd = -1; + int crtFD = crt_fileno(file); + if (crtFD != -1) { + // if crtFD is already mapped, addCrtFD() will return the existing rfd. + rfd = RFDMap::getInstance().addCrtFD(crtFD); + } + return rfd; } auto f_select = dllfunctor_stdcall("ws2_32.dll", "select"); @@ -1021,13 +1035,11 @@ u_int redis_ntohl_impl(u_int netlong){ return f_ntohl(netlong); } -int redis_isatty_impl(int fd) { +int redis_isatty_impl(int rfd) { try { - int posixFD = RFDMap::getInstance().lookupPosixFD(fd); - if( posixFD != -1) { - return crt_isatty(posixFD); - } else if (fd >= 0 && fd <= 2) { - return crt_isatty(fd); + int crtFD = RFDMap::getInstance().lookupCrtFD(rfd); + if (crtFD != -1) { + return crt_isatty(crtFD); } else { errno = EBADF; return 0; @@ -1042,11 +1054,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 crtFD = RFDMap::getInstance().lookupCrtFD(rfd); + if (crtFD != -1) { + return crt_lseek64(crtFD, offset, whence); } else { errno = EBADF; return 0; @@ -1057,11 +1069,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(RFD rfd) { + try { + int crtFD = RFDMap::getInstance().lookupCrtFD(rfd); + if (crtFD != -1) { + return crtget_osfhandle(crtFD); } else { errno = EBADF; return 0; @@ -1073,11 +1085,11 @@ intptr_t redis_get_osfhandle_impl(int fd) { } int redis_open_osfhandle_impl(intptr_t osfhandle, int flags) { - RFD rfd = RFDMap::invalidRFD; + RFD rfd = -1; try { int posixFD = crt_open_osfhandle(osfhandle, flags); if(posixFD != -1) { - rfd = RFDMap::getInstance().addPosixFD(posixFD); + rfd = RFDMap::getInstance().addCrtFD(posixFD); } } CATCH_AND_REPORT() @@ -1188,6 +1200,8 @@ private: 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; diff --git a/src/Win32_Interop/Win32_FDAPI.h b/src/Win32_Interop/Win32_FDAPI.h index 35024ebd..1afc03ef 100644 --- a/src/Win32_Interop/Win32_FDAPI.h +++ b/src/Win32_Interop/Win32_FDAPI.h @@ -152,6 +152,8 @@ typedef int (*redis_WSASocket)(int af, int type, int protocol, LPWSAPROTOCOL_INF // 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); +typedef int (*redis_fclose)(FILE* file); +typedef int (*redis_fileno)(FILE* file); // API prototypes must match the unix implementation typedef int (*redis_pipe)(int pipefd[2]); @@ -237,6 +239,8 @@ 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; @@ -267,6 +271,8 @@ int StorageSize(const SOCKADDR_STORAGE *ss); #define close(fd) fdapi_close(fd) #define setmode(fd,mode) fdapi_setmode(fd,mode) #define fwrite(Str, Size, Count, File) fdapi_fwrite(Str,Size,Count,File) +#define fclose(File) fdapi_fclose(File) +#define fileno(File) fdapi_fileno(File) #define _get_osfhandle(fd) fdapi_get_osfhandle(fd) #define _INC_STAT_INL diff --git a/src/Win32_Interop/Win32_fdapi_crt.cpp b/src/Win32_Interop/Win32_fdapi_crt.cpp index 5da484b0..f112ca1c 100644 --- a/src/Win32_Interop/Win32_fdapi_crt.cpp +++ b/src/Win32_Interop/Win32_fdapi_crt.cpp @@ -56,7 +56,7 @@ int crtsetmode(int fd, int mode) { return ::_setmode(fd, mode); } -size_t crtfwrite(const void * _Str, size_t _Size, size_t _Count, FILE * _File) { +size_t crt_fwrite(const void * _Str, size_t _Size, size_t _Count, FILE * _File) { // fwrite() somehow locks its view of the buffer. If during a fork operation the buffer has not been loaded into the forkee's process space, // the VEH will be called to load the missing pages. Although the page gets loaded, fwrite() will not see the loaded page. The result is // that fwrite will fail with errno set to ERROR_INVALID_USER_BUFFER. The fix is to force the buffer into memory before fwrite(). This only @@ -74,6 +74,13 @@ size_t crtfwrite(const void * _Str, size_t _Size, size_t _Count, FILE * _File) { return ::fwrite(_Str, _Size, _Count, _File); } +int crt_fclose(FILE* file) { + return ::fclose(file); +} + +int crt_fileno(FILE* file) { + return ::_fileno(file); +} int crt_isatty(int fd) { return _isatty(fd); diff --git a/src/Win32_Interop/Win32_fdapi_crt.h b/src/Win32_Interop/Win32_fdapi_crt.h index 9379a163..6b59515e 100644 --- a/src/Win32_Interop/Win32_fdapi_crt.h +++ b/src/Win32_Interop/Win32_fdapi_crt.h @@ -32,7 +32,10 @@ int crt_open(const char *filename, int oflag, int pmode); int crt_open_osfhandle(intptr_t osfhandle, int flags); intptr_t crtget_osfhandle(int fd); int crtsetmode(int fd, int mode); -size_t crtfwrite(const void * _Str, size_t _Size, size_t _Count, FILE * _File); +size_t crt_fwrite(const void * _Str, size_t _Size, size_t _Count, FILE * _File); +int crt_fclose(FILE* file); +int crt_fileno(FILE* file); + int crt_isatty(int fd); int crt_access(const char *pathname, int mode); __int64 crt_lseek64(int fd, __int64 offset, int origin); diff --git a/src/Win32_Interop/win32_rfdmap.cpp b/src/Win32_Interop/win32_rfdmap.cpp index 2375a6a6..d619436e 100644 --- a/src/Win32_Interop/win32_rfdmap.cpp +++ b/src/Win32_Interop/win32_rfdmap.cpp @@ -20,7 +20,6 @@ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ - #include "win32_types.h" #include "win32_rfdmap.h" @@ -29,9 +28,41 @@ RFDMap& RFDMap::getInstance() { return instance; } +#ifndef STDIN_FILENO +#define STDIN_FILENO (_fileno(stdin)) +#endif +#ifndef STDOUT_FILENO +#define STDOUT_FILENO (_fileno(stdout)) +#endif +#ifndef STDERR_FILENO +#define STDERR_FILENO (_fileno(stderr)) +#endif + RFDMap::RFDMap() { InitializeCriticalSection(&mutex); - maxRFD = minRFD; + // stdin, assign rfd = 0 (STDIN_FILENO) + RFD_INFO stdInRFDInfo; + stdInRFDInfo.flags = 0; + stdInRFDInfo.type = RFD_TYPE::CRTFD; + stdInRFDInfo.value.crtFD = STDIN_FILENO; + RFDToCrtFDInfoMap[STDIN_FILENO] = stdInRFDInfo; + CrtFDToRFDMap[STDIN_FILENO] = STDIN_FILENO; + + // stdout, assign rfd = 1 (STDOUT_FILENO) + RFD_INFO stdOutRFDInfo; + stdOutRFDInfo.flags = 0; + stdOutRFDInfo.type = RFD_TYPE::CRTFD; + stdOutRFDInfo.value.crtFD = STDOUT_FILENO; + RFDToCrtFDInfoMap[STDOUT_FILENO] = stdOutRFDInfo; + CrtFDToRFDMap[STDOUT_FILENO] = STDOUT_FILENO; + + // stderr, assign rfd = 2 (STDERR_FILENO) + RFD_INFO stdErrRFDInfo; + stdErrRFDInfo.flags = 0; + stdErrRFDInfo.type = RFD_TYPE::CRTFD; + stdErrRFDInfo.value.crtFD = STDERR_FILENO; + RFDToCrtFDInfoMap[STDERR_FILENO] = stdErrRFDInfo; + CrtFDToRFDMap[STDERR_FILENO] = STDERR_FILENO; } RFD RFDMap::getNextRFDAvailable() { @@ -41,107 +72,134 @@ RFD RFDMap::getNextRFDAvailable() { rfd = RFDRecyclePool.front(); RFDRecyclePool.pop(); } else { - maxRFD = minRFD + (int) SocketToRFDMap.size() + (int) PosixFDToRFDMap.size(); - rfd = maxRFD; + // We need to make sure a rfd is a unique value ragardless it's a socket or a crt fd + rfd = (int) RFDToSocketInfoMap.size() + (int) RFDToCrtFDInfoMap.size(); } LeaveCriticalSection(&mutex); + return rfd; } RFD RFDMap::addSocket(SOCKET s) { - RFD rfd; + RFD rfd = -1; EnterCriticalSection(&mutex); - if (SocketToRFDMap.find(s) != SocketToRFDMap.end()) { - rfd = invalidRFD; - } else { + map::iterator iter = SocketToRFDMap.find(s); + if (iter == SocketToRFDMap.end()) { + RFD_INFO rfd_info; + rfd_info.flags = 0; + rfd_info.type = RFD_TYPE::SOCKET; + rfd_info.value.socket = s; rfd = getNextRFDAvailable(); + RFDToSocketInfoMap[rfd] = rfd_info; SocketToRFDMap[s] = rfd; - RFDToSocketMap[rfd] = s; - } - LeaveCriticalSection(&mutex); - return rfd; -} - -void RFDMap::removeSocket(SOCKET s) { - EnterCriticalSection(&mutex); - S2RFDIterator mit = SocketToRFDMap.find(s); - if (mit != SocketToRFDMap.end()) { - RFD rfd = (*mit).second; - RFDRecyclePool.push(rfd); - RFDToSocketMap.erase(rfd); - SocketToRFDMap.erase(s); - } - LeaveCriticalSection(&mutex); -} - -RFD RFDMap::addPosixFD(int posixFD) { - RFD rfd; - EnterCriticalSection(&mutex); - if (PosixFDToRFDMap.find(posixFD) != PosixFDToRFDMap.end()) { - rfd = invalidRFD; } else { - rfd = getNextRFDAvailable(); - PosixFDToRFDMap[posixFD] = rfd; - RFDToPosixFDMap[rfd] = posixFD; + rfd = iter->second; } LeaveCriticalSection(&mutex); + return rfd; } -void RFDMap::removePosixFD(int posixFD) { +RFD RFDMap::addCrtFD(int crtFD) { + RFD rfd = -1; EnterCriticalSection(&mutex); - PosixFD2RFDIterator mit = PosixFDToRFDMap.find(posixFD); - if (mit != PosixFDToRFDMap.end()) { - RFD rfd = (*mit).second; - RFDRecyclePool.push(rfd); - RFDToPosixFDMap.erase(rfd); - PosixFDToRFDMap.erase(posixFD); + map::iterator iter = CrtFDToRFDMap.find(crtFD); + if (iter == CrtFDToRFDMap.end()) { + RFD_INFO rfd_info; + rfd_info.flags = 0; + rfd_info.type = RFD_TYPE::CRTFD; + rfd_info.value.crtFD = crtFD; + rfd = getNextRFDAvailable(); + RFDToCrtFDInfoMap[rfd] = rfd_info; + CrtFDToRFDMap[crtFD] = rfd; + } else { + rfd = iter->second; } LeaveCriticalSection(&mutex); + + return rfd; +} + +void RFDMap::removeRFD(RFD rfd) { + // stdin, stderr and stdout should never be removed + if (rfd > 2) { + EnterCriticalSection(&mutex); + map::iterator iter = RFDToSocketInfoMap.find(rfd); + if (iter != RFDToSocketInfoMap.end()) { + SocketToRFDMap.erase(iter->second.value.socket); + RFDToSocketInfoMap.erase(rfd); + RFDRecyclePool.push(rfd); + } else { + iter = RFDToCrtFDInfoMap.find(rfd); + if (iter != RFDToCrtFDInfoMap.end()) { + CrtFDToRFDMap.erase(iter->second.value.crtFD); + RFDToCrtFDInfoMap.erase(rfd); + RFDRecyclePool.push(rfd); + } + } + LeaveCriticalSection(&mutex); + } +} + +void RFDMap::removeCrtRFD(RFD rfd) { + // stdin, stderr and stdout should never be removed + if (rfd > 2) { + EnterCriticalSection(&mutex); + map::iterator iter = RFDToCrtFDInfoMap.find(rfd); + if (iter != RFDToCrtFDInfoMap.end()) { + CrtFDToRFDMap.erase(iter->second.value.crtFD); + RFDToCrtFDInfoMap.erase(rfd); + RFDRecyclePool.push(rfd); + } + LeaveCriticalSection(&mutex); + } +} + +RFD_INFO RFDMap::GetRFDInfo(RFD rfd) { + RFD_INFO rfd_info; + rfd_info.type = RFD_TYPE::INVALID; + EnterCriticalSection(&mutex); + if (RFDToSocketInfoMap.find(rfd) != RFDToSocketInfoMap.end()) { + rfd_info = RFDToSocketInfoMap[rfd]; + } else if (RFDToCrtFDInfoMap.find(rfd) != RFDToCrtFDInfoMap.end()) { + rfd_info = RFDToCrtFDInfoMap[rfd]; + } + LeaveCriticalSection(&mutex); + return rfd_info; +} + +bool RFDMap::SetRFDInfo(RFD rfd, RFD_INFO rfd_info) { + bool retVal = false; + EnterCriticalSection(&mutex); + if (rfd_info.type == RFD_TYPE::SOCKET) { + if (RFDToSocketInfoMap.find(rfd) != RFDToSocketInfoMap.end()) { + RFDToSocketInfoMap[rfd] = rfd_info; + retVal = true; + } + } else if (rfd_info.type == RFD_TYPE::CRTFD) { + if (RFDToCrtFDInfoMap.find(rfd) != RFDToCrtFDInfoMap.end()) { + RFDToCrtFDInfoMap[rfd] = rfd_info; + retVal = true; + } + } + LeaveCriticalSection(&mutex); + return retVal; } SOCKET RFDMap::lookupSocket(RFD rfd) { SOCKET socket = INVALID_SOCKET; - EnterCriticalSection(&mutex); - if (RFDToSocketMap.find(rfd) != RFDToSocketMap.end()) { - socket = RFDToSocketMap[rfd]; + RFD_INFO rfd_info = GetRFDInfo(rfd); + if (rfd_info.type == RFD_TYPE::SOCKET) { + socket = rfd_info.value.socket; } - LeaveCriticalSection(&mutex); return socket; } -int RFDMap::lookupPosixFD(RFD rfd) { - int posixFD = -1; - EnterCriticalSection(&mutex); - if (RFDToPosixFDMap.find(rfd) != RFDToPosixFDMap.end()) { - posixFD = RFDToPosixFDMap[rfd]; - } else if (rfd >= 0 && rfd <= 2) { - posixFD = rfd; +int RFDMap::lookupCrtFD(RFD rfd) { + int crtFD = -1; + RFD_INFO rfd_info = GetRFDInfo(rfd); + if (rfd_info.type == RFD_TYPE::CRTFD) { + crtFD = rfd_info.value.crtFD; } - LeaveCriticalSection(&mutex); - return posixFD; -} - -bool RFDMap::SetSocketState(SOCKET s, RedisSocketState state) { - bool result = false; - EnterCriticalSection(&mutex); - S2StateIterator sit = SocketToStateMap.find(s); - if (sit != SocketToStateMap.end()) { - SocketToStateMap[s] = state; - result = true; - } - LeaveCriticalSection(&mutex); - return result; -} - -bool RFDMap::GetSocketState(SOCKET s, RedisSocketState& state) { - bool result = false; - EnterCriticalSection(&mutex); - S2StateIterator sit = SocketToStateMap.find(s); - if (sit != SocketToStateMap.end()) { - state = SocketToStateMap[s]; - result = true; - } - LeaveCriticalSection(&mutex); - return result; + return crtFD; } diff --git a/src/Win32_Interop/win32_rfdmap.h b/src/Win32_Interop/win32_rfdmap.h index 227a0274..e2b9bca3 100644 --- a/src/Win32_Interop/win32_rfdmap.h +++ b/src/Win32_Interop/win32_rfdmap.h @@ -31,23 +31,20 @@ using namespace std; +typedef int RFD; // Redis File Descriptor, just an index in the SocketOrCrtFD_To_RFD map + +enum class RFD_TYPE { SOCKET, CRTFD, INVALID }; + +union RFD_VALUE { + SOCKET socket; + int crtFD; +}; + typedef struct { - bool IsBlockingSocket; -} RedisSocketState; - -typedef int RFD; // Redis File Descriptor -typedef map SocketToRFDMapType; -typedef map SocketToStateMapType; -typedef map PosixFDToRFDMapType; -typedef map RFDToSocketMapType; -typedef map RFDToPosixFDMapType; -typedef queue RFDRecyclePoolType; -typedef SocketToRFDMapType::iterator S2RFDIterator; -typedef SocketToStateMapType::iterator S2StateIterator; -typedef PosixFDToRFDMapType::iterator PosixFD2RFDIterator; -typedef RFDToSocketMapType::iterator RFD2SIterator; -typedef RFDToPosixFDMapType::iterator RFD2PosixFDIterator; - + int flags; + RFD_TYPE type; + RFD_VALUE value; +} RFD_INFO; /* In UNIX File Descriptors increment by one for each new one. Windows handles * do not follow the same rule. Additionally UNIX uses a 32-bit int to @@ -66,56 +63,52 @@ public: private: RFDMap(); - RFDMap(RFDMap const&); // Don't implement to guarantee singleton semantics - void operator=(RFDMap const&); // Don't implement to guarantee singleton semantics + RFDMap(RFDMap const&); // Don't implement to guarantee singleton semantics + void operator=(RFDMap const&); // Don't implement to guarantee singleton semantics private: - SocketToRFDMapType SocketToRFDMap; - SocketToStateMapType SocketToStateMap; - PosixFDToRFDMapType PosixFDToRFDMap; - RFDToSocketMapType RFDToSocketMap; - RFDToPosixFDMapType RFDToPosixFDMap; - RFDRecyclePoolType RFDRecyclePool; + map SocketToRFDMap; + map RFDToSocketInfoMap; + + map CrtFDToRFDMap; + map RFDToCrtFDInfoMap; + + queue RFDRecyclePool; private: - const static int minRFD = 3; // 0, 1 and 2 are reserved for stdin, stdout and stderr - RFD maxRFD; CRITICAL_SECTION mutex; -public: - const static int invalidRFD = -1; - private: - /* Gets the next available Redis File Descriptor. Redis File Descriptors are always - non-negative integers, with the first three being reserved for stdin(0), - stdout(1) and stderr(2). */ + /* Gets the next available Redis File Descriptor. Redis File Descriptors + are always non-negative integers, with the first three being reserved + for stdin(0), stdout(1) and stderr(2). */ RFD getNextRFDAvailable(); public: - /* Adds a socket to the socket map. Returns the redis file descriptor value for - the socket. Returns invalidRFD if the socket is already added to the - collection. */ + /* Adds a socket to the socket map and returns the RFD value for the socket. + If the socket already exists, returns the RFD. + */ RFD addSocket(SOCKET s); - /* Removes a socket from the list of sockets. Also removes the associated - file descriptor. */ - void removeSocket(SOCKET s); + /* Adds a fd obtained with low-level CRT file/pipe functions and returns the + RFD value for the CrtFD. If the crtFD already exists returns the RFD. */ + RFD addCrtFD(int crtFD); - /* Adds a posixFD (used with low-level CRT posix file functions) to the posixFD map. Returns - the redis file descriptor value for the posixFD. Returns invalidRFD if the posicFD is already - added to the collection. */ - RFD addPosixFD(int posixFD); + /* Removes a generic RFD from the list of RFDs (SOCKET or CRTFD). */ + void removeRFD(RFD rfd); + + /* Removes a RFD from the list of the CRTFDs */ + void removeCrtRFD(RFD rfd); - /* Removes a socket from the list of sockets. Also removes the associated - file descriptor. */ - void removePosixFD(int posixFD); + /* Gets the RFD_INFO data associated with a RFD */ + RFD_INFO GetRFDInfo(RFD rfd); - /* Returns the socket associated with a file descriptor. */ + /* Sets the RFD_INFO data associated with a RFD */ + bool SetRFDInfo(RFD rfd, RFD_INFO rfd_info); + + /* Returns the SOCKET associated with a RFD. */ SOCKET lookupSocket(RFD rfd); - /* Returns the socket associated with a file descriptor. */ - int lookupPosixFD(RFD rfd); - - bool SetSocketState(SOCKET s, RedisSocketState state); - bool GetSocketState(SOCKET s, RedisSocketState& state); + /* Returns the CRTFD associated with a RFD. */ + int lookupCrtFD(RFD rfd); };