diff --git a/BUILD.bazel b/BUILD.bazel index 7d0955bb8..f59963167 100644 --- a/BUILD.bazel +++ b/BUILD.bazel @@ -638,6 +638,17 @@ cc_test( ], ) +cc_test( + name = "util_test", + srcs = ["src/ray/util/util_test.cc"], + copts = COPTS, + deps = [ + ":ray_util", + "@boost//:asio", + "@com_google_googletest//:gtest_main", + ], +) + cc_test( name = "sample_test", srcs = ["src/ray/util/sample_test.cc"], @@ -678,17 +689,6 @@ cc_test( ], ) -cc_test( - name = "url_test", - srcs = ["src/ray/util/url_test.cc"], - copts = COPTS, - deps = [ - ":raylet_lib", - "@boost//:asio", - "@com_google_googletest//:gtest_main", - ], -) - cc_test( name = "sequencer_test", srcs = ["src/ray/util/sequencer_test.cc"], diff --git a/src/ray/common/client_connection.cc b/src/ray/common/client_connection.cc index 7fbc6b672..3abc1908e 100644 --- a/src/ray/common/client_connection.cc +++ b/src/ray/common/client_connection.cc @@ -25,7 +25,6 @@ #include #include "ray/common/ray_config.h" -#include "ray/util/url.h" #include "ray/util/util.h" namespace ray { @@ -299,7 +298,7 @@ bool ClientConnection::CheckRayCookie() { } std::string ClientConnection::RemoteEndpointInfo() { - return endpoint_to_url(ServerConnection::socket_.remote_endpoint(), false); + return EndpointToUrl(ServerConnection::socket_.remote_endpoint(), false); } void ClientConnection::ProcessMessage(const boost::system::error_code &error) { diff --git a/src/ray/raylet/main.cc b/src/ray/raylet/main.cc index 42db174b2..e282c9527 100644 --- a/src/ray/raylet/main.cc +++ b/src/ray/raylet/main.cc @@ -124,11 +124,11 @@ int main(int argc, char *argv[]) { if (!python_worker_command.empty()) { node_manager_config.worker_commands.emplace( - make_pair(ray::Language::PYTHON, SplitStrByWhitespaces(python_worker_command))); + make_pair(ray::Language::PYTHON, ParseCommandLine(python_worker_command))); } if (!java_worker_command.empty()) { node_manager_config.worker_commands.emplace( - make_pair(ray::Language::JAVA, SplitStrByWhitespaces(java_worker_command))); + make_pair(ray::Language::JAVA, ParseCommandLine(java_worker_command))); } if (python_worker_command.empty() && java_worker_command.empty()) { RAY_CHECK(0) diff --git a/src/ray/raylet/raylet.cc b/src/ray/raylet/raylet.cc index 9d33a10fb..5fcbdcc34 100644 --- a/src/ray/raylet/raylet.cc +++ b/src/ray/raylet/raylet.cc @@ -20,7 +20,7 @@ #include #include "ray/common/status.h" -#include "ray/util/url.h" +#include "ray/util/util.h" namespace { @@ -68,7 +68,7 @@ Raylet::Raylet(boost::asio::io_service &main_service, const std::string &socket_ node_manager_(main_service, self_node_id_, node_manager_config, object_manager_, gcs_client_, object_directory_), socket_name_(socket_name), - acceptor_(main_service, parse_url_endpoint(socket_name)), + acceptor_(main_service, ParseUrlEndpoint(socket_name)), socket_(main_service) { self_node_info_.set_node_id(self_node_id_.Binary()); self_node_info_.set_state(GcsNodeInfo::ALIVE); diff --git a/src/ray/raylet/raylet_client.cc b/src/ray/raylet/raylet_client.cc index a6aa423d4..a68b9c35e 100644 --- a/src/ray/raylet/raylet_client.cc +++ b/src/ray/raylet/raylet_client.cc @@ -30,7 +30,7 @@ #include "ray/common/task/task_spec.h" #include "ray/raylet/format/node_manager_generated.h" #include "ray/util/logging.h" -#include "ray/util/url.h" +#include "ray/util/util.h" using MessageType = ray::protocol::MessageType; @@ -62,7 +62,7 @@ raylet::RayletConnection::RayletConnection(boost::asio::io_service &io_service, RAY_CHECK(!raylet_socket.empty()); boost::system::error_code ec; for (int num_attempts = 0; num_attempts < num_retries; ++num_attempts) { - if (!conn_.connect(parse_url_endpoint(raylet_socket), ec)) { + if (!conn_.connect(ParseUrlEndpoint(raylet_socket), ec)) { break; } if (num_attempts > 0) { diff --git a/src/ray/raylet/worker_pool.cc b/src/ray/raylet/worker_pool.cc index 88a52678b..c88a52fbb 100644 --- a/src/ray/raylet/worker_pool.cc +++ b/src/ray/raylet/worker_pool.cc @@ -186,7 +186,7 @@ Process WorkerPool::StartWorkerProcess(const Language &language, if (token == option_placeholder) { if (!dynamic_options.empty()) { RAY_CHECK(dynamic_option_index < dynamic_options.size()); - auto options = SplitStrByWhitespaces(dynamic_options[dynamic_option_index]); + auto options = ParseCommandLine(dynamic_options[dynamic_option_index]); worker_command_args.insert(worker_command_args.end(), options.begin(), options.end()); ++dynamic_option_index; diff --git a/src/ray/util/process.cc b/src/ray/util/process.cc index 2834ece33..6a84e885e 100644 --- a/src/ray/util/process.cc +++ b/src/ray/util/process.cc @@ -28,6 +28,7 @@ #include #include "ray/util/logging.h" +#include "ray/util/util.h" namespace ray { @@ -54,7 +55,15 @@ class ProcessFD { intptr_t fd; pid_t pid; #ifdef _WIN32 - fd = _spawnvp(P_NOWAIT, argv[0], argv); + std::vector args; + for (size_t i = 0; argv[i]; ++i) { + args.push_back(argv[i]); + } + // Calling CreateCommandLine() here wouldn't make sense here if the + // Microsoft C runtime properly quoted each command-argument argument. + // However, it doesn't quote at all. It just joins arguments with a space. + // So we have to do the quoting manually and pass everything as a single argument. + fd = _spawnlp(P_NOWAIT, args[0].c_str(), CreateCommandLine(args).c_str(), NULL); if (fd != -1) { pid = static_cast(GetProcessId(reinterpret_cast(fd))); if (pid == 0) { diff --git a/src/ray/util/url.cc b/src/ray/util/url.cc deleted file mode 100644 index 6c5f063fe..000000000 --- a/src/ray/util/url.cc +++ /dev/null @@ -1,140 +0,0 @@ -// Copyright 2017 The Ray Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "ray/util/url.h" - -#include - -#ifdef _WIN32 -#include -#else -#include -#endif - -#include -#include -#include - -#include -#ifndef _WIN32 -#include -#endif -#include - -#include "ray/util/filesystem.h" -#include "ray/util/logging.h" -#include "ray/util/util.h" - -namespace ray { - -/// Uses sscanf() to read a token matching from the string, advancing the iterator. -/// \param c_str A string iterator that is dereferenceable. (i.e.: c_str < string::end()) -/// \param format The pattern. It must not produce any output. (e.g., use %*d, not %d.) -/// \return The scanned prefix of the string, if any. -static std::string ScanToken(std::string::const_iterator &c_str, std::string format) { - int i = 0; - std::string result; - format += "%n"; - if (static_cast(sscanf(&*c_str, format.c_str(), &i)) <= 1) { - result.insert(result.end(), c_str, c_str + i); - c_str += i; - } - return result; -} - -std::string endpoint_to_url( - const boost::asio::generic::basic_endpoint &ep, - bool include_scheme) { - std::string result, scheme; - switch (ep.protocol().family()) { - case AF_INET: { - scheme = "tcp://"; - boost::asio::ip::tcp::endpoint e(boost::asio::ip::tcp::v4(), 0); - RAY_CHECK(e.size() == ep.size()); - const sockaddr *src = ep.data(); - sockaddr *dst = e.data(); - *reinterpret_cast(dst) = *reinterpret_cast(src); - std::ostringstream ss; - ss << e; - result = ss.str(); - break; - } - case AF_INET6: { - scheme = "tcp://"; - boost::asio::ip::tcp::endpoint e(boost::asio::ip::tcp::v6(), 0); - RAY_CHECK(e.size() == ep.size()); - const sockaddr *src = ep.data(); - sockaddr *dst = e.data(); - *reinterpret_cast(dst) = *reinterpret_cast(src); - std::ostringstream ss; - ss << e; - result = ss.str(); - break; - } - case AF_UNIX: - scheme = "unix://"; -#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS - result.append(reinterpret_cast(ep.data())->sun_path, - ep.size() - offsetof(sockaddr_un, sun_path)); -#else - RAY_LOG(FATAL) << "UNIX-domain socket endpoints are not supported"; -#endif - break; - default: - RAY_LOG(FATAL) << "unsupported protocol family: " << ep.protocol().family(); - break; - } - if (include_scheme) { - result.insert(0, scheme); - } - return result; -} - -boost::asio::generic::basic_endpoint -parse_url_endpoint(const std::string &endpoint, int default_port) { - // Syntax reference: https://en.wikipedia.org/wiki/URL#Syntax - // Note that we're a bit more flexible, to allow parsing "127.0.0.1" as a URL. - boost::asio::generic::stream_protocol::endpoint result; - std::string address = endpoint, scheme; - if (address.find("unix://") == 0) { - scheme = "unix://"; - address.erase(0, scheme.size()); - } else if (address.size() > 0 && ray::IsDirSep(address[0])) { - scheme = "unix://"; - } else if (address.find("tcp://") == 0) { - scheme = "tcp://"; - address.erase(0, scheme.size()); - } else { - scheme = "tcp://"; - } - if (scheme == "unix://") { -#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS - result = boost::asio::local::stream_protocol::endpoint(address); -#else - RAY_LOG(FATAL) << "UNIX-domain socket endpoints are not supported: " << endpoint; -#endif - } else if (scheme == "tcp://") { - std::string::const_iterator i = address.begin(); - std::string host = ScanToken(i, "[%*[^][/]]"); - host = host.empty() ? ScanToken(i, "%*[^/:]") : host.substr(1, host.size() - 2); - std::string port_str = ScanToken(i, ":%*d"); - int port = port_str.empty() ? default_port : std::stoi(port_str.substr(1)); - result = boost::asio::ip::tcp::endpoint(boost::asio::ip::make_address(host), port); - } else { - RAY_LOG(FATAL) << "Unable to parse socket endpoint: " << endpoint; - } - return result; -} - -} // namespace ray diff --git a/src/ray/util/url.h b/src/ray/util/url.h deleted file mode 100644 index b4e5c3845..000000000 --- a/src/ray/util/url.h +++ /dev/null @@ -1,39 +0,0 @@ -// Copyright 2017 The Ray Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#ifndef RAY_UTIL_URL_H -#define RAY_UTIL_URL_H - -#include -#include - -namespace ray { - -/// Converts the given endpoint (such as TCP or UNIX domain socket address) to a string. -/// \param include_scheme Whether to include the scheme prefix (such as tcp://). -/// This is recommended to avoid later ambiguity when parsing. -std::string endpoint_to_url( - const boost::asio::generic::basic_endpoint &ep, - bool include_scheme = true); - -/// Parses the endpoint socket address of a URL. -/// If a scheme:// prefix is absent, the address family is guessed automatically. -/// For TCP/IP, the endpoint comprises the IP address and port number in the URL. -/// For UNIX domain sockets, the endpoint comprises the socket path. -boost::asio::generic::basic_endpoint -parse_url_endpoint(const std::string &endpoint, int default_port = 0); - -} // namespace ray - -#endif diff --git a/src/ray/util/url_test.cc b/src/ray/util/url_test.cc deleted file mode 100644 index fbadaad98..000000000 --- a/src/ray/util/url_test.cc +++ /dev/null @@ -1,54 +0,0 @@ -// Copyright 2017 The Ray Authors. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -#include "ray/util/url.h" - -#include "gtest/gtest.h" - -namespace ray { - -template -static std::string to_str(const T &obj, bool include_scheme) { - return endpoint_to_url(obj, include_scheme); -} - -TEST(UrlTest, UrlIpTcpParseTest) { - ASSERT_EQ(to_str(parse_url_endpoint("tcp://[::1]:1/", 0), false), "[::1]:1"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://[::1]/", 0), false), "[::1]:0"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://[::1]:1", 0), false), "[::1]:1"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://[::1]", 0), false), "[::1]:0"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://127.0.0.1:1/", 0), false), "127.0.0.1:1"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://127.0.0.1/", 0), false), "127.0.0.1:0"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://127.0.0.1:1", 0), false), "127.0.0.1:1"); - ASSERT_EQ(to_str(parse_url_endpoint("tcp://127.0.0.1", 0), false), "127.0.0.1:0"); - ASSERT_EQ(to_str(parse_url_endpoint("[::1]:1/", 0), false), "[::1]:1"); - ASSERT_EQ(to_str(parse_url_endpoint("[::1]/", 0), false), "[::1]:0"); - ASSERT_EQ(to_str(parse_url_endpoint("[::1]:1", 0), false), "[::1]:1"); - ASSERT_EQ(to_str(parse_url_endpoint("[::1]", 0), false), "[::1]:0"); - ASSERT_EQ(to_str(parse_url_endpoint("127.0.0.1:1/", 0), false), "127.0.0.1:1"); - ASSERT_EQ(to_str(parse_url_endpoint("127.0.0.1/", 0), false), "127.0.0.1:0"); - ASSERT_EQ(to_str(parse_url_endpoint("127.0.0.1:1", 0), false), "127.0.0.1:1"); - ASSERT_EQ(to_str(parse_url_endpoint("127.0.0.1", 0), false), "127.0.0.1:0"); -#ifndef _WIN32 - ASSERT_EQ(to_str(parse_url_endpoint("unix:///tmp/sock"), false), "/tmp/sock"); - ASSERT_EQ(to_str(parse_url_endpoint("/tmp/sock"), false), "/tmp/sock"); -#endif -} - -} // namespace ray - -int main(int argc, char **argv) { - ::testing::InitGoogleTest(&argc, argv); - return RUN_ALL_TESTS(); -} diff --git a/src/ray/util/util.cc b/src/ray/util/util.cc new file mode 100644 index 000000000..f94a24dc2 --- /dev/null +++ b/src/ray/util/util.cc @@ -0,0 +1,322 @@ +#include "ray/util/util.h" + +#include +#include +#ifndef _WIN32 +#include +#endif + +#include +#include +#include +#include + +#include +#ifndef _WIN32 +#include +#endif +#include + +#include "ray/util/filesystem.h" +#include "ray/util/logging.h" + +/// Uses sscanf() to read a token matching from the string, advancing the iterator. +/// \param c_str A string iterator that is dereferenceable. (i.e.: c_str < string::end()) +/// \param format The pattern. It must not produce any output. (e.g., use %*d, not %d.) +/// \return The scanned prefix of the string, if any. +static std::string ScanToken(std::string::const_iterator &c_str, std::string format) { + int i = 0; + std::string result; + format += "%n"; + if (static_cast(sscanf(&*c_str, format.c_str(), &i)) <= 1) { + result.insert(result.end(), c_str, c_str + i); + c_str += i; + } + return result; +} + +std::string EndpointToUrl( + const boost::asio::generic::basic_endpoint &ep, + bool include_scheme) { + std::string result, scheme; + switch (ep.protocol().family()) { + case AF_INET: { + scheme = "tcp://"; + boost::asio::ip::tcp::endpoint e(boost::asio::ip::tcp::v4(), 0); + RAY_CHECK(e.size() == ep.size()); + const sockaddr *src = ep.data(); + sockaddr *dst = e.data(); + *reinterpret_cast(dst) = *reinterpret_cast(src); + std::ostringstream ss; + ss << e; + result = ss.str(); + break; + } + case AF_INET6: { + scheme = "tcp://"; + boost::asio::ip::tcp::endpoint e(boost::asio::ip::tcp::v6(), 0); + RAY_CHECK(e.size() == ep.size()); + const sockaddr *src = ep.data(); + sockaddr *dst = e.data(); + *reinterpret_cast(dst) = *reinterpret_cast(src); + std::ostringstream ss; + ss << e; + result = ss.str(); + break; + } +#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS + case AF_UNIX: + scheme = "unix://"; + result.append(reinterpret_cast(ep.data())->sun_path, + ep.size() - offsetof(sockaddr_un, sun_path)); + break; +#endif + default: + RAY_LOG(FATAL) << "unsupported protocol family: " << ep.protocol().family(); + break; + } + if (include_scheme) { + result.insert(0, scheme); + } + return result; +} + +boost::asio::generic::basic_endpoint +ParseUrlEndpoint(const std::string &endpoint, int default_port) { + // Syntax reference: https://en.wikipedia.org/wiki/URL#Syntax + // Note that we're a bit more flexible, to allow parsing "127.0.0.1" as a URL. + boost::asio::generic::stream_protocol::endpoint result; + std::string address = endpoint, scheme; + if (address.find("unix://") == 0) { + scheme = "unix://"; + address.erase(0, scheme.size()); + } else if (address.size() > 0 && ray::IsDirSep(address[0])) { + scheme = "unix://"; + } else if (address.find("tcp://") == 0) { + scheme = "tcp://"; + address.erase(0, scheme.size()); + } else { + scheme = "tcp://"; + } + if (scheme == "unix://") { +#ifdef BOOST_ASIO_HAS_LOCAL_SOCKETS + result = boost::asio::local::stream_protocol::endpoint(address); +#else + RAY_LOG(FATAL) << "UNIX-domain socket endpoints are not supported: " << endpoint; +#endif + } else if (scheme == "tcp://") { + std::string::const_iterator i = address.begin(); + std::string host = ScanToken(i, "[%*[^][/]]"); + host = host.empty() ? ScanToken(i, "%*[^/:]") : host.substr(1, host.size() - 2); + std::string port_str = ScanToken(i, ":%*d"); + int port = port_str.empty() ? default_port : std::stoi(port_str.substr(1)); + result = boost::asio::ip::tcp::endpoint(boost::asio::ip::make_address(host), port); + } else { + RAY_LOG(FATAL) << "Unable to parse socket endpoint: " << endpoint; + } + return result; +} + +/// Rules: +/// 1. Adjacent tokens are concatenated, so "a"b'c' is just abc +/// 2. Outside quotes: backslashes make the next character literal; space & tab delimit +/// 3. Inside "...": backslashes escape a following " and \ and otherwise stay literal +/// 4. Inside '...': no escaping occurs +/// 5. [&|;<>`#()!] etc. are literal, but should be in '...' to avoid confusion. +/// Note: POSIX shells can perform additional processing (like piping) not reflected here. +/// Refer to the unit tests for examples. +/// To compare against the platform's behavior, try a command like the following: +/// $ python3 -c "import sys; [print(a) for a in sys.argv[1:]]" \x\\y "\x\\y" '\x\\y' +/// Python analog: shlex.split(s) +static std::vector ParsePosixCommandLine(const std::string &s) { + RAY_CHECK(s.find('\0') >= s.size()) << "Invalid null character in command line"; + const char space = ' ', tab = '\t', backslash = '\\', squote = '\'', dquote = '\"'; + char surroundings = space; + bool escaping = false, arg_started = false; + std::vector result; + std::string arg; + for (char ch : s) { + bool is_delimeter = false; + if (escaping) { + if (surroundings == dquote && (ch == backslash || ch == dquote)) { + arg.pop_back(); // remove backslash because it precedes \ or " in double-quotes + } + arg += ch; + escaping = false; + } else if (surroundings == dquote || surroundings == squote) { // inside quotes + if (ch == surroundings) { + surroundings = space; // leaving quotes + } else { + arg += ch; + escaping = surroundings == dquote && ch == backslash; // backslash in "..." + } + } else { // outside quotes + if (ch == space || ch == tab) { + is_delimeter = true; + if (arg_started) { // we just finished an argument + result.push_back(arg); + } + arg.clear(); + } else if (ch == dquote || ch == squote) { + surroundings = ch; // entering quotes + } else if (ch == backslash) { + escaping = true; + } else { + arg += ch; + } + } + arg_started = !is_delimeter; + } + if (arg_started) { + result.push_back(arg); + } + return result; +} + +/// Rules: +/// 1. Adjacent tokens are concatenated, so "a"b"c" is just abc +/// 2. Backslashes escape when eventually followed by " but stay literal otherwise +/// 3. Outside "...": space and tab are delimiters +/// 4. [&|:<>^#()] etc. are literal, but should be in "..." to avoid confusion. +/// Note: Windows tools have additional processing & quirks not reflected here. +/// Refer to the unit tests for examples. +/// To compare against the platform's behavior, try a command like the following: +/// > python3 -c "import sys; [print(a) for a in sys.argv[1:]]" \x\\y "\x\\y" +/// Python analog: None (would be shlex.split(s, posix=False), but it doesn't unquote) +static std::vector ParseWindowsCommandLine(const std::string &s) { + RAY_CHECK(s.find('\0') >= s.size()) << "Invalid null character in command line"; + std::vector result; + std::string arg, c_str = s + '\0'; + std::string::const_iterator i = c_str.begin(), j = c_str.end() - 1; + for (bool stop = false, in_dquotes = false; !stop;) { + if (!in_dquotes && (i >= j || ScanToken(i, "%*[ \t]").size())) { + result.push_back(arg); + arg.clear(); + } + stop |= i >= j && !in_dquotes; + arg += ScanToken(i, in_dquotes ? "%*[^\\\"]" : "%*[^\\\" \t]"); + std::string possible_escape = ScanToken(i, "%*[\\]"); + bool escaping = possible_escape.size() % 2 != 0; + if (*i == '\"') { + possible_escape.erase(possible_escape.size() / 2); + possible_escape.append(escaping ? 1 : 0, *i); + in_dquotes ^= !escaping; + ++i; + } + arg += possible_escape; + } + return result; +} + +std::vector ParseCommandLine(const std::string &s, CommandLineSyntax kind) { + if (kind == CommandLineSyntax::System) { +#ifdef _WIN32 + kind = CommandLineSyntax::Windows; +#else + kind = CommandLineSyntax::POSIX; +#endif + } + std::vector result; + switch (kind) { + case CommandLineSyntax::POSIX: + result = ParsePosixCommandLine(s); + break; + case CommandLineSyntax::Windows: + result = ParseWindowsCommandLine(s); + break; + default: + RAY_LOG(FATAL) << "invalid command line syntax"; + break; + } + return result; +} + +/// Python analog: shlex.join(args) +std::string CreatePosixCommandLine(const std::vector &args) { + std::string result; + const std::string safe_chars("%*[-A-Za-z0-9%_=+]"); + const char single_quote = '\''; + for (size_t a = 0; a != args.size(); ++a) { + std::string arg = args[a], arg_with_null = arg + '\0'; + std::string::const_iterator i = arg_with_null.begin(); + if (ScanToken(i, safe_chars) != arg) { + // Prefer single-quotes. Double-quotes have unpredictable behavior, e.g. for "\!". + std::string quoted; + quoted += single_quote; + for (char ch : arg) { + if (ch == single_quote) { + quoted += single_quote; + quoted += '\\'; + } + quoted += ch; + if (ch == single_quote) { + quoted += single_quote; + } + } + quoted += single_quote; + arg = quoted; + } + if (a > 0) { + result += ' '; + } + result += arg; + } + return result; +} + +// Python analog: subprocess.list2cmdline(args) +static std::string CreateWindowsCommandLine(const std::vector &args) { + std::string result; + const std::string safe_chars("%*[-A-Za-z0-9%_=+]"); + const char double_quote = '\"'; + for (size_t a = 0; a != args.size(); ++a) { + std::string arg = args[a], arg_with_null = arg + '\0'; + std::string::const_iterator i = arg_with_null.begin(); + if (ScanToken(i, safe_chars) != arg) { + // Escape only backslashes that precede double-quotes + std::string quoted; + quoted += double_quote; + size_t backslashes = 0; + for (char ch : arg) { + if (ch == double_quote) { + quoted.append(backslashes, '\\'); + quoted += '\\'; + } + quoted += ch; + backslashes = ch == '\\' ? backslashes + 1 : 0; + } + quoted.append(backslashes, '\\'); + quoted += double_quote; + arg = quoted; + } + if (a > 0) { + result += ' '; + } + result += arg; + } + return result; +} + +std::string CreateCommandLine(const std::vector &args, + CommandLineSyntax kind) { + if (kind == CommandLineSyntax::System) { +#ifdef _WIN32 + kind = CommandLineSyntax::Windows; +#else + kind = CommandLineSyntax::POSIX; +#endif + } + std::string result; + switch (kind) { + case CommandLineSyntax::POSIX: + result = CreatePosixCommandLine(args); + break; + case CommandLineSyntax::Windows: + result = CreateWindowsCommandLine(args); + break; + default: + RAY_LOG(FATAL) << "invalid command line syntax"; + break; + } + return result; +} diff --git a/src/ray/util/util.h b/src/ray/util/util.h index 17c438dbd..782f211b9 100644 --- a/src/ray/util/util.h +++ b/src/ray/util/util.h @@ -24,6 +24,26 @@ #include #include +// Boost forward-declarations (to avoid forcing slow header inclusions) +namespace boost { + +namespace asio { + +namespace generic { + +template +class basic_endpoint; + +class stream_protocol; + +} // namespace generic + +} // namespace asio + +} // namespace boost + +enum class CommandLineSyntax { System, POSIX, Windows }; + /// Return the number of milliseconds since the steady clock epoch. NOTE: The /// returned timestamp may be used for accurately measuring intervals but has /// no relation to wall clock time. It must not be used for synchronization @@ -47,17 +67,36 @@ inline int64_t current_sys_time_ms() { return ms_since_epoch.count(); } -/// A helper function to split a string by whitespaces. +/// A helper function to parse command-line arguments in a platform-compatible manner. /// -/// \param str The string with whitespaces. +/// \param cmdline The command-line to split. /// -/// \return A vector that contains strings split by whitespaces. -inline std::vector SplitStrByWhitespaces(const std::string &str) { - std::istringstream iss(str); - std::vector result(std::istream_iterator{iss}, - std::istream_iterator()); - return result; -} +/// \return The command-line arguments, after processing any escape sequences. +std::vector ParseCommandLine( + const std::string &cmdline, CommandLineSyntax syntax = CommandLineSyntax::System); + +/// A helper function to combine command-line arguments in a platform-compatible manner. +/// The result of this function is intended to be suitable for the shell used by popen(). +/// +/// \param cmdline The command-line arguments to combine. +/// +/// \return The command-line string, including any necessary escape sequences. +std::string CreateCommandLine(const std::vector &args, + CommandLineSyntax syntax = CommandLineSyntax::System); + +/// Converts the given endpoint (such as TCP or UNIX domain socket address) to a string. +/// \param include_scheme Whether to include the scheme prefix (such as tcp://). +/// This is recommended to avoid later ambiguity when parsing. +std::string EndpointToUrl( + const boost::asio::generic::basic_endpoint &ep, + bool include_scheme = true); + +/// Parses the endpoint socket address of a URL. +/// If a scheme:// prefix is absent, the address family is guessed automatically. +/// For TCP/IP, the endpoint comprises the IP address and port number in the URL. +/// For UNIX domain sockets, the endpoint comprises the socket path. +boost::asio::generic::basic_endpoint +ParseUrlEndpoint(const std::string &endpoint, int default_port = 0); class InitShutdownRAII { public: diff --git a/src/ray/util/util_test.cc b/src/ray/util/util_test.cc new file mode 100644 index 000000000..631cb8a8c --- /dev/null +++ b/src/ray/util/util_test.cc @@ -0,0 +1,161 @@ +#include "ray/util/util.h" + +#include + +#include + +#include "gtest/gtest.h" + +static const char *argv0 = NULL; + +namespace ray { + +template +static std::string to_str(const T &obj, bool include_scheme) { + return EndpointToUrl(obj, include_scheme); +} + +TEST(UtilTest, UrlIpTcpParseTest) { + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://[::1]:1/", 0), false), "[::1]:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://[::1]/", 0), false), "[::1]:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://[::1]:1", 0), false), "[::1]:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://[::1]", 0), false), "[::1]:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://127.0.0.1:1/", 0), false), "127.0.0.1:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://127.0.0.1/", 0), false), "127.0.0.1:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://127.0.0.1:1", 0), false), "127.0.0.1:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("tcp://127.0.0.1", 0), false), "127.0.0.1:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("[::1]:1/", 0), false), "[::1]:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("[::1]/", 0), false), "[::1]:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("[::1]:1", 0), false), "[::1]:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("[::1]", 0), false), "[::1]:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("127.0.0.1:1/", 0), false), "127.0.0.1:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("127.0.0.1/", 0), false), "127.0.0.1:0"); + ASSERT_EQ(to_str(ParseUrlEndpoint("127.0.0.1:1", 0), false), "127.0.0.1:1"); + ASSERT_EQ(to_str(ParseUrlEndpoint("127.0.0.1", 0), false), "127.0.0.1:0"); +#ifndef _WIN32 + ASSERT_EQ(to_str(ParseUrlEndpoint("unix:///tmp/sock"), false), "/tmp/sock"); + ASSERT_EQ(to_str(ParseUrlEndpoint("/tmp/sock"), false), "/tmp/sock"); +#endif +} + +TEST(UtilTest, ParseCommandLineTest) { + typedef std::vector ArgList; + CommandLineSyntax posix = CommandLineSyntax::POSIX, win32 = CommandLineSyntax::Windows, + all[] = {posix, win32}; + for (CommandLineSyntax syn : all) { + ASSERT_EQ(ParseCommandLine(R"(aa)", syn), ArgList({R"(aa)"})); + ASSERT_EQ(ParseCommandLine(R"(a )", syn), ArgList({R"(a)"})); + ASSERT_EQ(ParseCommandLine(R"(\" )", syn), ArgList({R"(")"})); + ASSERT_EQ(ParseCommandLine(R"(" a")", syn), ArgList({R"( a)"})); + ASSERT_EQ(ParseCommandLine(R"("\\")", syn), ArgList({R"(\)"})); + ASSERT_EQ(ParseCommandLine(R"("\"")", syn), ArgList({R"(")"})); + ASSERT_EQ(ParseCommandLine(R"(a" b c"d )", syn), ArgList({R"(a b cd)"})); + ASSERT_EQ(ParseCommandLine(R"(\"a b)", syn), ArgList({R"("a)", R"(b)"})); + ASSERT_EQ(ParseCommandLine(R"(| ! ^ # [)", syn), ArgList({"|", "!", "^", "#", "["})); + ASSERT_EQ(ParseCommandLine(R"(; ? * $ &)", syn), ArgList({";", "?", "*", "$", "&"})); + ASSERT_EQ(ParseCommandLine(R"(: ` < > ~)", syn), ArgList({":", "`", "<", ">", "~"})); + } + ASSERT_EQ(ParseCommandLine(R"( a)", posix), ArgList({R"(a)"})); + ASSERT_EQ(ParseCommandLine(R"( a)", win32), ArgList({R"()", R"(a)"})); + ASSERT_EQ(ParseCommandLine(R"(\ a)", posix), ArgList({R"( a)"})); + ASSERT_EQ(ParseCommandLine(R"(\ a)", win32), ArgList({R"(\)", R"(a)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\ D)", posix), ArgList({R"(C: D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\ D)", win32), ArgList({R"(C:\)", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\\ D)", posix), ArgList({R"(C:\)", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\\ D)", win32), ArgList({R"(C:\\)", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\ D)", posix), ArgList({R"(C: )", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\ D)", win32), ArgList({R"(C:\)", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\\\ D)", posix), ArgList({R"(C:\ )", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(C:\\\ D)", win32), ArgList({R"(C:\\\)", R"(D)"})); + ASSERT_EQ(ParseCommandLine(R"(\)", posix), ArgList({R"()"})); + ASSERT_EQ(ParseCommandLine(R"(\)", win32), ArgList({R"(\)"})); + ASSERT_EQ(ParseCommandLine(R"(\\a)", posix), ArgList({R"(\a)"})); + ASSERT_EQ(ParseCommandLine(R"(\\a)", win32), ArgList({R"(\\a)"})); + ASSERT_EQ(ParseCommandLine(R"(\\\a)", posix), ArgList({R"(\a)"})); + ASSERT_EQ(ParseCommandLine(R"(\\\a)", win32), ArgList({R"(\\\a)"})); + ASSERT_EQ(ParseCommandLine(R"(\\)", posix), ArgList({R"(\)"})); + ASSERT_EQ(ParseCommandLine(R"(\\)", win32), ArgList({R"(\\)"})); + ASSERT_EQ(ParseCommandLine(R"("\\a")", posix), ArgList({R"(\a)"})); + ASSERT_EQ(ParseCommandLine(R"("\\a")", win32), ArgList({R"(\\a)"})); + ASSERT_EQ(ParseCommandLine(R"("\\\a")", posix), ArgList({R"(\\a)"})); + ASSERT_EQ(ParseCommandLine(R"("\\\a")", win32), ArgList({R"(\\\a)"})); + ASSERT_EQ(ParseCommandLine(R"('a'' b')", posix), ArgList({R"(a b)"})); + ASSERT_EQ(ParseCommandLine(R"('a'' b')", win32), ArgList({R"('a'')", R"(b')"})); + ASSERT_EQ(ParseCommandLine(R"('a')", posix), ArgList({R"(a)"})); + ASSERT_EQ(ParseCommandLine(R"('a')", win32), ArgList({R"('a')"})); + ASSERT_EQ(ParseCommandLine(R"(x' a \b')", posix), ArgList({R"(x a \b)"})); + ASSERT_EQ(ParseCommandLine(R"(x' a \b')", win32), ArgList({R"(x')", R"(a)", R"(\b')"})); +} + +TEST(UtilTest, CreateCommandLineTest) { + typedef std::vector ArgList; + CommandLineSyntax posix = CommandLineSyntax::POSIX, win32 = CommandLineSyntax::Windows, + all[] = {posix, win32}; + std::vector test_cases({ + ArgList({R"(a)"}), + ArgList({R"(a b)"}), + ArgList({R"(")"}), + ArgList({R"(')"}), + ArgList({R"(\)"}), + ArgList({R"(/)"}), + ArgList({R"(#)"}), + ArgList({R"($)"}), + ArgList({R"(!)"}), + ArgList({R"(@)"}), + ArgList({R"(`)"}), + ArgList({R"(&)"}), + ArgList({R"(|)"}), + ArgList({R"(a")", R"('x)", R"(?'"{)", R"(]))", R"(!)", R"(~`\)"}), + }); + for (CommandLineSyntax syn : all) { + for (const ArgList &arglist : test_cases) { + ASSERT_EQ(ParseCommandLine(CreateCommandLine(arglist, syn), syn), arglist); + std::string cmdline = CreateCommandLine(arglist, syn); + std::string buf((2 + cmdline.size()) * 6, '\0'); + std::string test_command = std::string(argv0); + test_command = "\"" + test_command + "\""; +#ifdef _WIN32 + test_command = "\"" + test_command; +#endif + test_command += " --println " + cmdline; +#ifdef _WIN32 + test_command = test_command + "\""; +#endif + FILE *proc; +#ifdef _WIN32 + proc = syn == win32 ? _popen(test_command.c_str(), "r") : NULL; +#else + proc = syn == posix ? popen(test_command.c_str(), "r") : NULL; +#endif + if (proc) { + std::vector lines; + while (fgets(&*buf.begin(), static_cast(buf.size()), proc)) { + lines.push_back(buf.substr(0, buf.find_first_of(std::string({'\0', '\n'})))); + } + ASSERT_EQ(lines, arglist); +#ifdef _WIN32 + _pclose(proc); +#else + pclose(proc); +#endif + } + } + } +} + +} // namespace ray + +int main(int argc, char **argv) { + argv0 = argv[0]; + int result = 0; + if (argc > 1 && strcmp(argv[1], "--println") == 0) { + // If we're given this special command, emit each argument on a new line + for (int i = 2; i < argc; ++i) { + fprintf(stdout, "%s\n", argv[i]); + } + } else { + ::testing::InitGoogleTest(&argc, argv); + result = RUN_ALL_TESTS(); + } + return result; +}