wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-08 21:34:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Redis on Windows (#7509)

Browse Source 
* Switch hiredis on Windows to that of the Windows port of Redis

* Use boost::asio::ip::tcp::socket::native_handle_type

* Use normal hiredis instead of Windows-specific one

* Finish up using normal hiredis

Co-authored-by: Mehrdad <noreply@github.com>
This commit is contained in:
mehrdadn
2020-03-09 18:49:54 -07:00
committed by GitHub
parent 90e23a5c43
commit fc76586518
8 changed files with 188 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BUILD.bazel
+18 -9
View File
@@ -890,6 +890,11 @@ cc_library(
includes = ["src/ray/thirdparty"],
)
alias(
name = "hiredis",
actual = "@com_github_antirez_redis//:hiredis",
)
cc_library(
name = "gcs",
srcs = glob(
@@ -907,13 +912,13 @@ cc_library(
copts = COPTS,
deps = [
":gcs_cc_proto",
":hiredis",
":node_manager_fbs",
":node_manager_rpc",
":ray_common",
":ray_util",
":stats_lib",
"@boost//:asio",
"@redis//:hiredis",
],
)
@@ -1206,10 +1211,11 @@ genrule(
] + select({
# Windows is not supported yet, so make stubs for it
"@bazel_tools//src/conditions:windows": [
"@win-redis-bin//file",
"@com_github_tporadowski_redis_bin//file",
],
"//conditions:default": [
"@redis-src//file",
"@com_github_antirez_redis//:file", # This is necessary so we can access the root of the directory
"@com_github_antirez_redis//:files", # This is necessary to ensure entire directory tree is accessible
],
}),
outs = [
@@ -1219,18 +1225,21 @@ genrule(
cmd = select({
# Windows is not supported yet, so make stubs for it
"@bazel_tools//src/conditions:windows": """
unzip -q -o -- $(location @win-redis-bin//file) redis-server.exe redis-cli.exe &&
unzip -q -o -- $(location @com_github_tporadowski_redis_bin//file) redis-server.exe redis-cli.exe &&
mv -f -- redis-server.exe $(location redis-server) &&
mv -f -- redis-cli.exe $(location redis-cli)
""",
"//conditions:default": """
set -x &&
tar -xzf $(location @redis-src//file) --strip-components=1 -C . &&
make &&
mv ./src/redis-server $(location redis-server) &&
tmpdir="redis.tmp" &&
path=$(location @com_github_antirez_redis//:file) &&
cp -p -L -R -- "$${path%/*}" "$${tmpdir}" &&
make -s -C "$${tmpdir}" -j"$$(getconf _NPROCESSORS_ONLN || echo 1)" &&
mv "$${tmpdir}"/src/redis-server $(location redis-server) &&
chmod +x $(location redis-server) &&
mv ./src/redis-cli $(location redis-cli) &&
chmod +x $(location redis-cli)
mv "$${tmpdir}"/src/redis-cli $(location redis-cli) &&
chmod +x $(location redis-cli) &&
rm -r -f -- "$${tmpdir}"
""",
}),
visibility = ["//java:__subpackages__"],
bazel/BUILD.redis
+27 -6
View File
@@ -1,13 +1,34 @@
COPTS = [] + select({
"@bazel_tools//src/conditions:windows": [
"-Dssize_t=ptrdiff_t", # no ssize_t on Windows
"-fcolor-diagnostics",
"-fansi-escape-codes",
"-D_CRT_DECLARE_NONSTDC_NAMES=0", # don't define off_t, to avoid conflicts
],
"//conditions:default": [
],
})
filegroup(
name = "files",
srcs = glob(["**"]),
visibility = ["//visibility:public"],
)
filegroup(
name = "file",
srcs = ["Makefile"],
visibility = ["//visibility:public"],
)
# This library is for internal hiredis use, because hiredis assumes a
# different include prefix for itself than external libraries do.
cc_library(
name = "_hiredis",
hdrs = [
"deps/hiredis/dict.c",
],
copts = COPTS,
strip_include_prefix = "deps/hiredis",
)
cc_library(
name = "hiredis",
srcs = glob(
@@ -23,13 +44,13 @@ cc_library(
hdrs = glob([
"deps/hiredis/*.h",
"deps/hiredis/adapters/*.h",
"deps/hiredis/dict.c",
]),
copts = COPTS,
includes = ["deps"],
includes = ["deps/hiredis"],
strip_include_prefix = "deps",
deps = [
"@com_github_ray_project_ray//:platform_shims",
":_hiredis",
"@//:platform_shims",
],
visibility = ["//visibility:public"],
)
bazel/ray_deps_setup.bzl
+4 -10
View File
@@ -73,34 +73,28 @@ def auto_http_archive(*, name=None, url=None, urls=True,
def ray_deps_setup():
auto_http_archive(
name = "redis",
build_file = True,
name = "com_github_antirez_redis",
build_file = "//bazel:BUILD.redis",
url = "https://github.com/antirez/redis/archive/5.0.3.tar.gz",
sha256 = "7084e8bd9e5dedf2dbb2a1e1d862d0c46e66cc0872654bdc677f4470d28d84c5",
patches = [
"//thirdparty/patches:hiredis-async-include-dict.patch",
"//thirdparty/patches:hiredis-casts.patch",
"//thirdparty/patches:hiredis-connect-rename.patch",
"//thirdparty/patches:hiredis-windows-sigpipe.patch",
"//thirdparty/patches:hiredis-windows-sockets.patch",
"//thirdparty/patches:hiredis-windows-strerror.patch",
"//thirdparty/patches:hiredis-windows-poll.patch",
"//thirdparty/patches:hiredis-windows-translations.patch",
"//thirdparty/patches:redis-windows-poll.patch",
],
)
http_file(
name = "win-redis-bin",
name = "com_github_tporadowski_redis_bin",
sha256 = "6fac443543244c803311de5883b714a7ae3c4fa0594cad51d75b24c4ef45b353",
urls = ["https://github.com/tporadowski/redis/releases/download/v4.0.14.2/Redis-x64-4.0.14.2.zip"],
)
http_file(
name = "redis-src",
sha256 = "7084e8bd9e5dedf2dbb2a1e1d862d0c46e66cc0872654bdc677f4470d28d84c5",
urls = ["https://github.com/antirez/redis/archive/5.0.3.tar.gz"],
)
auto_http_archive(
name = "rules_jvm_external",
url = "https://github.com/bazelbuild/rules_jvm_external/archive/2.10.tar.gz",
src/ray/gcs/asio.cc
+17 -1
View File
@@ -2,6 +2,10 @@
#include "ray/util/logging.h"
#ifdef _WIN32
#include <io.h>
#endif
RedisAsioClient::RedisAsioClient(boost::asio::io_service &io_service,
ray::gcs::RedisAsyncContext &redis_async_context)
: redis_async_context_(redis_async_context),
@@ -16,9 +20,21 @@ RedisAsioClient::RedisAsioClient(boost::asio::io_service &io_service,
// gives access to c->fd
redisContext *c = &(async_context->c);
#ifdef _WIN32
SOCKET sock = SOCKET_ERROR;
WSAPROTOCOL_INFO pi;
if (WSADuplicateSocket(_get_osfhandle(c->fd), GetCurrentProcessId(), &pi) == 0) {
DWORD flag = WSA_FLAG_OVERLAPPED;
sock = WSASocket(pi.iAddressFamily, pi.iSocketType, pi.iProtocol, &pi, 0, flag);
}
boost::asio::ip::tcp::socket::native_handle_type handle(sock);
#else
boost::asio::ip::tcp::socket::native_handle_type handle(dup(c->fd));
#endif
// hiredis is already connected
// use the existing native socket
socket_.assign(boost::asio::ip::tcp::v4(), c->fd);
socket_.assign(boost::asio::ip::tcp::v4(), handle);
// register hooks with the hiredis async context
async_context->ev.addRead = call_C_addRead;
thirdparty/patches/hiredis-async-include-dict.patch
Vendored
-7
View File
@@ -1,7 +0,0 @@
diff --git deps/hiredis/async.c deps/hiredis/async.c
--- deps/hiredis/async.c
+++ deps/hiredis/async.c
@@ -41,1 +41,1 @@
-#include "dict.c"
+#include "hiredis/dict.c"
--
thirdparty/patches/hiredis-windows-poll.patch
Vendored
+2 -2
View File
@@ -7,7 +7,7 @@ diff --git deps/hiredis/net.c deps/hiredis/net.c
+ fd_set wset;
+
+ FD_ZERO(&wset);
+ FD_SET(c->fd, &wset);
+ FD_SET(_get_osfhandle(c->fd), &wset);
+#else
struct pollfd wfd[1];
@@ -20,7 +20,7 @@ diff --git deps/hiredis/net.c deps/hiredis/net.c
+#ifdef _WINSOCKAPI_
+ struct timeval tv = { msec / 1000, (msec % 1000) * 1000 };
+ res = select(c->fd + 1, NULL, &wset, NULL, msec >= 0 ? &tv : NULL);
+ res = select(_get_osfhandle(c->fd) + 1, NULL, &wset, NULL, msec >= 0 ? &tv : NULL);
+#else
+ res = poll(wfd, 1, msec);
+#endif
thirdparty/patches/hiredis-windows-sockets.patch
Vendored
+18 -2
View File
@@ -1,11 +1,27 @@
diff --git deps/hiredis/hiredis.c deps/hiredis/hiredis.c
--- deps/hiredis/hiredis.c
+++ deps/hiredis/hiredis.c
@@ -620,1 +620,1 @@
- close(c->fd);
+ redisContextCloseFd(c);
@@ -648,1 +648,1 @@
- close(c->fd);
+ redisContextCloseFd(c);
diff --git deps/hiredis/net.h deps/hiredis/net.h
--- deps/hiredis/net.h
+++ deps/hiredis/net.h
@@ -44,1 +44,2 @@
+void redisContextCloseFd(redisContext *c);
int redisContextConnectTcp(redisContext *c, const char *addr, int port, const struct timeval *timeout);
diff --git deps/hiredis/net.c deps/hiredis/net.c
--- deps/hiredis/net.c
+++ deps/hiredis/net.c
@@ -60,6 +60,10 @@
static void redisContextCloseFd(redisContext *c) {
-static void redisContextCloseFd(redisContext *c) {
+void redisContextCloseFd(redisContext *c) {
if (c && c->fd >= 0) {
+#ifdef _WINSOCKAPI_
+ closesocket(c->fd);
+ closesocket(_get_osfhandle(c->fd)) == 0 || (WSAGetLastError() == WSAENOTSOCK && _close(c->fd) == 0);
+#else
close(c->fd);
+#endif
thirdparty/patches/hiredis-windows-translations.patch
Vendored
+102
View File
@@ -0,0 +1,102 @@
diff --git deps/hiredis/hiredis.c deps/hiredis/hiredis.c
--- deps/hiredis/hiredis.c
+++ deps/hiredis/hiredis.c
@@ -46,0 +46,37 @@
+#ifdef _WINSOCKAPI_
+static ptrdiff_t read_redis(int fildes, void *buf, size_t nbyte) {
+ WSABUF wsabuf = { (u_long)nbyte, (char *)buf };
+ WSAOVERLAPPED overlapped = { 0 };
+ SOCKET sock = _get_osfhandle(fildes);
+ int flags = 0;
+ ptrdiff_t r = WSARecv(sock, &wsabuf, 1, NULL, &flags, &overlapped, NULL);
+ if (r == SOCKET_ERROR && WSAGetLastError() == WSA_IO_PENDING) {
+ r = WSAWaitForMultipleEvents(1, &sock, FALSE, WSA_INFINITE, FALSE) == WSA_WAIT_EVENT_0 ? overlapped.InternalHigh : -1;
+ } else if (r != SOCKET_ERROR) {
+ r = overlapped.InternalHigh;
+ }
+ if (r == -1) {
+ switch (errno) {
+ case WSAEWOULDBLOCK:
+ WSASetLastError(EAGAIN);
+ break;
+ }
+ }
+ return r;
+}
+#define read read_redis
+static ptrdiff_t write_redis(int fildes, const void *buf, size_t nbyte) {
+ WSABUF wsabuf = { (u_long)nbyte, (char *)buf };
+ WSAOVERLAPPED overlapped = { 0 };
+ SOCKET sock = _get_osfhandle(fildes);
+ int flags = 0;
+ ptrdiff_t r = WSASend(sock, &wsabuf, 1, NULL, flags, &overlapped, NULL);
+ if (r == SOCKET_ERROR && WSAGetLastError() == WSA_IO_PENDING) {
+ r = WSAWaitForMultipleEvents(1, &sock, FALSE, WSA_INFINITE, FALSE) == WSA_WAIT_EVENT_0 ? overlapped.InternalHigh : -1;
+ } else if (r != SOCKET_ERROR) {
+ r = overlapped.InternalHigh;
+ }
+ return r;
+}
+#define write write_redis
+#endif
diff --git deps/hiredis/net.c deps/hiredis/net.c
--- deps/hiredis/net.c
+++ deps/hiredis/net.c
@@ -47,1 +47,45 @@
+#ifdef _WINSOCKAPI_
+#define errno WSAGetLastError()
+static int socket_redis(int af, int type, int protocol) {
+ int r;
+ SOCKET sock = WSASocket(af, type, protocol, NULL, 0, WSA_FLAG_OVERLAPPED);
+ if (sock != -1) {
+ r = _open_osfhandle(sock, _O_BINARY | _O_RDWR);
+ if (r == -1) { closesocket(sock); }
+ } else {
+ r = -1;
+ }
+ return r;
+}
+#define socket socket_redis
+static int connect_redis(int socketfd, const struct sockaddr *address, socklen_t address_len) {
+ int r = connect(_get_osfhandle(socketfd), address, address_len);
+ if (r == -1) {
+ switch (errno) {
+ case WSAEINVAL:
+ case WSAEWOULDBLOCK:
+ case WSA_IO_PENDING:
+ WSASetLastError(EINPROGRESS);
+ break;
+ }
+ }
+ return r;
+}
+#define connect connect_redis
+static int ioctlsocket_redis(int socketfd, long cmd, u_long* argp) {
+ int r = ioctlsocket(_get_osfhandle(socketfd), cmd, argp);
+ return r;
+}
+#define ioctlsocket ioctlsocket_redis
+static int getsockopt_redis(int socketfd, int level, int optname, char* optval, int* optlen) {
+ return getsockopt(_get_osfhandle(socketfd), level, optname, optval, optlen);
+}
+#define getsockopt getsockopt_redis
+static int setsockopt_redis(int socketfd, int level, int optname, const char* optval, int optlen) {
+ return setsockopt(_get_osfhandle(socketfd), level, optname, optval, optlen);
+}
+#define setsockopt setsockopt_redis
+#else
#include <errno.h>
+#endif
@@ -218,1 +258,5 @@
+#ifdef _WINSOCKAPI_
+ WSASetLastError(ETIMEDOUT);
+#else
errno = ETIMEDOUT;
+#endif
@@ -245,1 +285,5 @@
+#ifdef _WINSOCKAPI_
+ WSASetLastError(err);
+#else
errno = err;
+#endif
--