Method dispatch_sockets

src/test/crimson/test_socket.cc:173–259

 public:
  template <typename FuncC, typename FuncS>
  static future<> dispatch_sockets(
      bool is_fixed_cpu,
      FuncC&& cb_client,
      FuncS&& cb_server) {
    ceph_assert_always(seastar::this_shard_id() == CLIENT_CPU);
    auto owner = std::make_unique<SocketFactory>();
    auto psf = owner.get();
    auto saddr = get_server_addr();
    return seastar::smp::submit_to(SERVER_CPU, [psf, saddr, is_fixed_cpu] {
      return ShardedServerSocket::create(is_fixed_cpu
      ).then([psf, saddr](auto pss) {
        psf->pss = pss;
        return pss->listen(saddr
        ).safe_then([] {
        }, listen_ertr::assert_all_func(
	  [saddr](const std::error_code& e) {
          logger().error("dispatch_sockets(): there is another instance running at {}",
                         saddr);
        }));
      });
    }).then([psf, saddr] {
      return seastar::when_all_succeed(
        seastar::smp::submit_to(CLIENT_CPU, [psf, saddr] {
          return socket_connect(saddr).then([psf](auto socket) {
            ceph_assert_always(seastar::this_shard_id() == CLIENT_CPU);
            psf->client_socket = std::move(socket);
          });
        }),
        seastar::smp::submit_to(SERVER_CPU, [psf] {
          return psf->pss->accept([psf](auto _socket, auto paddr) {
            logger().info("dispatch_sockets(): accepted at shard {}",
                          seastar::this_shard_id());
            psf->server_socket_CPU = seastar::this_shard_id();
            if (psf->pss->is_fixed_shard_dispatching()) {
              ceph_assert_always(SERVER_CPU == seastar::this_shard_id());
            }
            SocketFRef socket = seastar::make_foreign(std::move(_socket));
            psf->server_socket = std::move(socket);
            return seastar::smp::submit_to(CLIENT_CPU, [psf] {
              psf->server_connected.set_value();
            });
          });
        })
      );
    }).then_unpack([] {
      return seastar::now();
    }).then([psf] {
      return psf->server_connected.get_future();
    }).then([psf] {
      if (psf->pss) {
        return seastar::smp::submit_to(SERVER_CPU, [psf] {
          return psf->pss->shutdown_destroy();
        });
      }
      return seastar::now();
    }).then([psf,
             cb_client = std::move(cb_client),
             cb_server = std::move(cb_server)]() mutable {