Merge 'Internalize reactor::posix_... API methods' from Pavel Emelyanov

There are two of those -- posix_connect() and posix_listen(). Both have nothing to do with reactor and are only called by posix_stack.cc. Despite both just deserve being in posix_stack.cc, they are public methods, and need to be deprecated first, for safety.

Closes #3031

* github.com:scylladb/seastar:
  reactor: Deprecate and internalize posix_connect()
  reactor: Deprecate and internalize posix_listen()

Author: avikivity

include/seastar/core/reactor.hh

@@ -135,6 +135,9 @@ void at_destroy(Func&& func);
 void at_exit(noncopyable_function<future<> ()> func);
 void set_current_task(task* t);
 
+pollable_fd posix_listen(socket_address sa, listen_options opts = {});
+future<> posix_connect(pollable_fd pfd, socket_address sa, socket_address local);
+
 }
 
 class io_queue;
@@ -506,15 +509,21 @@ public:
     future<connected_socket> connect(socket_address sa);
     future<connected_socket> connect(socket_address, socket_address, transport proto = transport::TCP);
 
-    pollable_fd posix_listen(socket_address sa, listen_options opts = {});
+    [[deprecated("Internal reactor function, consider using seastar::listen()")]]
+    pollable_fd posix_listen(socket_address sa, listen_options opts = {}) {
+        return internal::posix_listen(sa, opts);
+    }
 
     // FIXME: reuseport currently leads to heavy load imbalance.
     // Until we fix that, just disable it unconditionally.
     bool posix_reuseport_available() const { return false; }
 
     pollable_fd make_pollable_fd(socket_address sa, int proto);
 
-    future<> posix_connect(pollable_fd pfd, socket_address sa, socket_address local);
+    [[deprecated("Internal reactor function, consider using seastar::connect)")]]
+    future<> posix_connect(pollable_fd pfd, socket_address sa, socket_address local) {
+        return internal::posix_connect(std::move(pfd), sa, local);
+    }
 
     future<> send_all(pollable_fd_state& fd, const void* buffer, size_t size);
 

src/core/reactor.cc

@@ -1520,8 +1520,9 @@ void reactor::configure(const reactor_options& opts) {
     }
 }
 
-pollable_fd
-reactor::posix_listen(socket_address sa, listen_options opts) {
+namespace internal {
+
+pollable_fd posix_listen(socket_address sa, listen_options opts) {
     auto specific_protocol = (int)(opts.proto);
     if (sa.is_af_unix()) {
         // no type-safe way to create listen_opts with proto=0
@@ -1579,6 +1580,8 @@ reactor::posix_listen(socket_address sa, listen_options opts) {
     return pollable_fd(std::move(fd));
 }
 
+} // internel namespace
+
 void pollable_fd_state::maybe_no_more_recv() {
     if (shutdown_mask & posix::rcv_shutdown) {
         throw std::system_error(std::error_code(ECONNABORTED, std::system_category()));
@@ -1626,8 +1629,9 @@ reactor::make_pollable_fd(socket_address sa, int proto) {
     return pollable_fd(std::move(fd));
 }
 
-future<>
-reactor::posix_connect(pollable_fd pfd, socket_address sa, socket_address local) {
+namespace internal {
+
+future<> posix_connect(pollable_fd pfd, socket_address sa, socket_address local) {
 #ifdef IP_BIND_ADDRESS_NO_PORT
     if (!sa.is_af_unix()) {
         try {
@@ -1650,6 +1654,8 @@ reactor::posix_connect(pollable_fd pfd, socket_address sa, socket_address local)
     return pfd.connect(sa).finally([pfd] {});
 }
 
+} // internal namespace
+
 server_socket
 reactor::listen(socket_address sa, listen_options opt) {
     return server_socket(_network_stack->listen(sa, opt));

src/net/posix-stack.cc

@@ -429,7 +429,7 @@ class posix_socket_impl final : public socket_impl {
             _fd.get_file_desc().setsockopt(SOL_SOCKET, SO_REUSEADDR, int(_reuseaddr));
             uint16_t port = attempts++ < 5 && requested_port == 0 && proto == transport::TCP ? u(random_engine) * smp::count + this_shard_id() : requested_port;
             local.as_posix_sockaddr_in().sin_port = hton(port);
-            return futurize_invoke([this, sa, local] { return engine().posix_connect(_fd, sa, local); }).then_wrapped([port, requested_port] (future<> f) {
+            return futurize_invoke([this, sa, local] { return internal::posix_connect(_fd, sa, local); }).then_wrapped([port, requested_port] (future<> f) {
                 try {
                     f.get();
                     return stop_iteration::yes;
@@ -451,7 +451,7 @@ class posix_socket_impl final : public socket_impl {
         }
 
         _fd = engine().make_pollable_fd(sa, 0);
-        return engine().posix_connect(_fd, sa, local).then(
+        return internal::posix_connect(_fd, sa, local).then(
             [fd = _fd, allocator = _allocator](){
                 // a problem with 'private' interaction with 'unique_ptr'
                 std::unique_ptr<connected_socket_impl> csi(new posix_connected_socket_impl{AF_UNIX, 0, std::move(fd), allocator});
@@ -705,13 +705,13 @@ posix_network_stack::listen(socket_address sa, listen_options opt) {
         sa = inet_address(inet_address::family::INET);
     }
     if (sa.is_af_unix()) {
-        return server_socket(std::make_unique<posix_server_socket_impl>(0, sa, engine().posix_listen(sa, opt), opt.lba, opt.fixed_cpu, _allocator));
+        return server_socket(std::make_unique<posix_server_socket_impl>(0, sa, internal::posix_listen(sa, opt), opt.lba, opt.fixed_cpu, _allocator));
     }
     auto protocol = static_cast<int>(opt.proto);
     return _reuseport ?
-        server_socket(std::make_unique<posix_reuseport_server_socket_impl>(protocol, sa, engine().posix_listen(sa, opt), _allocator))
+        server_socket(std::make_unique<posix_reuseport_server_socket_impl>(protocol, sa, internal::posix_listen(sa, opt), _allocator))
         :
-        server_socket(std::make_unique<posix_server_socket_impl>(protocol, sa, engine().posix_listen(sa, opt), opt.lba, opt.fixed_cpu, _allocator));
+        server_socket(std::make_unique<posix_server_socket_impl>(protocol, sa, internal::posix_listen(sa, opt), opt.lba, opt.fixed_cpu, _allocator));
 }
 
 ::seastar::socket posix_network_stack::socket() {
@@ -734,7 +734,7 @@ posix_ap_network_stack::listen(socket_address sa, listen_options opt) {
     }
     auto protocol = static_cast<int>(opt.proto);
     return _reuseport ?
-        server_socket(std::make_unique<posix_reuseport_server_socket_impl>(protocol, sa, engine().posix_listen(sa, opt), _allocator))
+        server_socket(std::make_unique<posix_reuseport_server_socket_impl>(protocol, sa, internal::posix_listen(sa, opt), _allocator))
         :
         server_socket(std::make_unique<posix_ap_server_socket_impl>(protocol, sa, _allocator));
 }