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

[Change/Fix] Added API mapping for fclose/fileno.

Browse Source 
[Change] RFDMap code refactoring.
This commit is contained in:
Enrico Giordani
2015-07-17 14:43:23 +02:00
parent c2432fcbc1
commit a078f2080b
6 changed files with 360 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Win32_Interop/Win32_FDAPI.cpp
+162 -148
View File
@@ -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<SOCKET, int, int, int>("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<int, SOCKET>("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<int, SOCKET, int, int, const char*, int>("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<int, SOCKET, long, u_long*>("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<int, SOCKET, char*, int, int>("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<int, SOCKET, const char*, int, int>("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<SOCKET, int, int, int, LPWSAPROTOCOL_INFO, GROUP, DWORD>("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<int, int, fd_set*, fd_set*, fd_set*, const struct timeval*>("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;
src/Win32_Interop/Win32_FDAPI.h
+6
View File
@@ -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
src/Win32_Interop/Win32_fdapi_crt.cpp
+8 -1
View File
@@ -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);
src/Win32_Interop/Win32_fdapi_crt.h
+4 -1
View File
@@ -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);
src/Win32_Interop/win32_rfdmap.cpp
+136 -78
View File
@@ -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<SOCKET, RFD>::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<int, RFD>::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<RFD, RFD_INFO>::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<RFD, RFD_INFO>::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;
}
src/Win32_Interop/win32_rfdmap.h
+44 -51
View File
@@ -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<SOCKET,RFD> SocketToRFDMapType;
typedef map<SOCKET,RedisSocketState> SocketToStateMapType;
typedef map<int,RFD> PosixFDToRFDMapType;
typedef map<RFD,SOCKET> RFDToSocketMapType;
typedef map<RFD,int> RFDToPosixFDMapType;
typedef queue<RFD> 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<SOCKET, RFD> SocketToRFDMap;
map<RFD, RFD_INFO> RFDToSocketInfoMap;
map<int, RFD> CrtFDToRFDMap;
map<RFD, RFD_INFO> RFDToCrtFDInfoMap;
queue<int> 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);
};