network: configure io_uring and add io_uring test CI job

.azure-pipelines/stage/checks.yml

@@ -89,6 +89,48 @@ jobs:
  timeoutInMinutes: 10
  condition: and(failed(), eq(variables['CI_TARGET'], 'bazel.clang_tidy'))
 
+# Exempt job will continue on error and will be regarded as a warning instead
+# of an error on failed.
+- job: exempt
+ displayName: "Linux x64"
+ dependsOn: []
+ condition: |
+ and(not(canceled()),
+ eq(${{ parameters.runChecks }}, 'true'))
+ continueOnError: true
+ variables:
+ REPO_FETCH_DEPTH: 1
+ REPO_FETCH_TAGS: false
+ PUBLISH_TEST_RESULTS: true
+ PUBLISH_ENVOY: true
+ strategy:
+ maxParallel: ${{ parameters.concurrencyChecks }}
+ matrix:
+ io_uring:
+ CI_TARGET: "bazel.io_uring"
+ timeoutInMinutes: 180
+ pool:
+ vmImage: $(agentUbuntu)
+ steps:
+ - template: ../bazel.yml
+ parameters:
+ ciTarget: $(CI_TARGET)
+ envoyBuildFilterExample: $(ENVOY_FILTER_EXAMPLE)
+ cacheTestResults: ${{ parameters.cacheTestResults }}
+ repoFetchDepth: $(REPO_FETCH_DEPTH)
+ repoFetchTags: $(REPO_FETCH_TAGS)
+ publishTestResults: variables.PUBLISH_TEST_RESULTS
+ publishEnvoy: variables.PUBLISH_ENVOY
+ stepsPost:
+
+ # TODO(phlax): consolidate "fix" paths/jobs
+ - task: PublishBuildArtifacts@1
+ inputs:
+ pathtoPublish: "$(Build.StagingDirectory)/tmp/lint-fixes"
+ artifactName: "$(CI_TARGET).fixes"
+ timeoutInMinutes: 10
+ condition: and(failed(), eq(variables['CI_TARGET'], 'bazel.clang_tidy'))
+
 - job: coverage
  displayName: "Linux x64"
  dependsOn: []

api/envoy/extensions/network/socket_interface/v3/default_socket_interface.proto

@@ -2,7 +2,10 @@ syntax = "proto3";
 
 package envoy.extensions.network.socket_interface.v3;
 
+import "google/protobuf/wrappers.proto";
+
 import "udpa/annotations/status.proto";
+import "validate/validate.proto";
 
 option java_package = "io.envoyproxy.envoy.extensions.network.socket_interface.v3";
 option java_outer_classname = "DefaultSocketInterfaceProto";
@@ -15,4 +18,38 @@ option (udpa.annotations.file_status).package_version_status = ACTIVE;
 // Configuration for default socket interface that relies on OS dependent syscall to create
 // sockets.
 message DefaultSocketInterface {
+ // Enable io_uring for socket operations if the kernel supports. io_uring is only valid in
+ // Linux with minimum kernel version 5.6. Otherwise, Envoy will fallback to use the default
+ // socket operations. Default to false.
+ bool enable_io_uring = 1;
+
+ // The size for io_uring submission queues (SQ) and completion queues (CQ). io_uring is built
+ // up during configuration with a fixed size in each threads, and each io_uring operation will
+ // create a submission queue entry (SQE). Once the SQ is used up, more operations will not be
+ // added to the io_uring. Default to 1000.
+ google.protobuf.UInt32Value io_uring_size = 2;
+
+ // Enable io_uring submission queue polling (SQPOLL). io_uring SQPOLL mode polls all SQEs in SQ
+ // in the kernel thread. io_uring SQPOLL mode may reduce latency and increase CPU usage as a
+ // cost. Default to false.
+ bool enable_io_uring_submission_queue_polling = 3;
+
+ // The size of a io_uring TCP accept socket's pending connections queue can grow to. The value is
+ // different from :ref:`tcp_backlog_size <envoy_v3_api_field_config.listener.v3.Listener.tcp_backlog_size>`.
+ // Connections in TCP listener's queue are not being accepted, while connections in io_uring TCP
+ // accept socket's queue are accepted but not handled. io_uring accepts sockets asynchronously,
+ // and a large backlog value will have a better performance in situation with large number of no
+ // keep-alive connections. Default to 5.
+ google.protobuf.UInt32Value io_uring_accept_backlog = 4;
+
+ // The size of a io_uring socket's read buffer. Each io_uring read operation will allocate buffer
+ // with the given size, and if the buffer provided is to small, the socket may read multiple
+ // times to read all the data. Default to 8192.
+ google.protobuf.UInt32Value io_uring_read_buffer_size = 5;
+
+ // The timeout of a io_uring socket's write on closing in ms. io_uring writes and closes
+ // asynchronously. If the remote stops reading, the io_uring write operation may never complete.
+ // Connections may have timeout like per_try_timeout before closing, and io_uring adds another
+ // timeout period on top of it. Default to 1000.
+ google.protobuf.UInt32Value io_uring_write_timeout_ms = 6;
 }

bazel/BUILD

@@ -524,6 +524,14 @@ config_setting(
  values = {"define": "uhv=enabled"},
 )
 
+# This config setting enables using io_uring socket handle in tests and works
+# in bazel.io_uring CI target. io_uring socket handle is in its early stage,
+# and a separate warning-only CI target is responsilbe for check its status.
+config_setting(
+ name = "enable_test_io_uring",
+ values = {"define": "test_io_uring=enabled"},
+)
+
 # Alias pointing to the selected version of BoringSSL:
 # - BoringSSL FIPS from @boringssl_fips//:ssl,
 # - non-FIPS BoringSSL from @boringssl//:ssl.

bazel/envoy_internal.bzl

@@ -118,6 +118,9 @@ def envoy_copts(repository, test = False):
  }) + select({
  repository + "//bazel:uhv_enabled": ["-DENVOY_ENABLE_UHV"],
  "//conditions:default": [],
+ }) + select({
+ repository + "//bazel:enable_test_io_uring": ["-DENVOY_TEST_IO_URING"],
+ "//conditions:default": [],
  }) + envoy_select_hot_restart(["-DENVOY_HOT_RESTART"], repository) + \
  envoy_select_disable_exceptions(["-fno-unwind-tables", "-fno-exceptions"], repository) + \
  envoy_select_admin_html(["-DENVOY_ADMIN_HTML"], repository) + \

ci/README.md

@@ -145,6 +145,8 @@ The `./ci/run_envoy_docker.sh './ci/do_ci.sh <TARGET>'` targets are:
 * `bazel.fuzz <test>` &mdash; build and run a specified fuzz test or test dir under `-c dbg --config=asan-fuzzer` with clang. If specifying a single fuzz test, must use the full target name with "_with_libfuzzer" for `<test>`.
 * `bazel.compile_time_options` &mdash; build Envoy and run tests with various compile-time options toggled to their non-default state, to ensure they still build.
 * `bazel.compile_time_options <test>` &mdash; build Envoy and run a specified test or test dir with various compile-time options toggled to their non-default state, to ensure they still build.
+* `bazel.io_uring` &mdash; build and run tests with io_uring enabled.
+* `bazel.io_uring <test>` &mdash; build and run a specified test or test dir with io_uring enabled.
 * `bazel.clang_tidy <files>` &mdash; build and run clang-tidy specified source files, if no files specified, runs against the diff with the last GitHub commit.
 * `check_proto_format`&mdash; check configuration, formatting and build issues in API proto files.
 * `fix_proto_format`&mdash; fix configuration, formatting and build issues in API proto files.

ci/do_ci.sh

@@ -587,6 +587,17 @@ case $CI_TARGET in
  --test_arg="-runs=10"
  ;;
 
+ io_uring)
+ setup_clang_toolchain
+ echo "bazel build with tests and io_uring enabled"
+ echo "Building and testing envoy tests ${TEST_TARGETS[*]}"
+ bazel_with_collection \
+ test "${BAZEL_BUILD_OPTIONS[@]}" \
+ --define test_io_uring=enabled \
+ --remote_download_minimal \
+ "${TEST_TARGETS[@]}"
+ ;;
+
  gcc)
  BAZEL_BUILD_OPTIONS+=("--test_env=HEAPCHECK=")
  setup_gcc_toolchain

source/common/network/socket_interface_impl.cc

@@ -2,6 +2,7 @@
 
 #include "envoy/common/exception.h"
 #include "envoy/extensions/network/socket_interface/v3/default_socket_interface.pb.h"
+#include "envoy/extensions/network/socket_interface/v3/default_socket_interface.pb.validate.h"
 
 #include "source/common/api/os_sys_calls_impl.h"
 #include "source/common/common/assert.h"
@@ -162,15 +163,21 @@ bool SocketInterfaceImpl::ipFamilySupported(int domain) {
 }
 
 Server::BootstrapExtensionPtr SocketInterfaceImpl::createBootstrapExtension(
- const Protobuf::Message&,
+ const Protobuf::Message& config,
  [[maybe_unused]] Server::Configuration::ServerFactoryContext& context) {
+ const auto& message = MessageUtil::downcastAndValidate<
+ const envoy::extensions::network::socket_interface::v3::DefaultSocketInterface&>(
+ config, context.messageValidationVisitor());
 #ifdef __linux__
- // TODO (soulxu): Add runtime flag here.
- if (Io::isIoUringSupported()) {
+ if (message.enable_io_uring() && Io::isIoUringSupported()) {
  std::shared_ptr<Io::IoUringFactoryImpl> io_uring_factory =
- std::make_shared<Io::IoUringFactoryImpl>(DefaultIoUringSize, UseSubmissionQueuePolling,
- DefaultAcceptSize, DefaultReadBufferSize,
- DefaultWriteTimeoutMs, context.threadLocal());
+ std::make_shared<Io::IoUringFactoryImpl>(
+ PROTOBUF_GET_WRAPPED_OR_DEFAULT(message, io_uring_size, 1000),
+ message.enable_io_uring_submission_queue_polling(),
+ PROTOBUF_GET_WRAPPED_OR_DEFAULT(message, io_uring_accept_backlog, 5),
+ PROTOBUF_GET_WRAPPED_OR_DEFAULT(message, io_uring_read_buffer_size, 8192),
+ PROTOBUF_GET_WRAPPED_OR_DEFAULT(message, io_uring_write_timeout_ms, 1000),
+ context.threadLocal());
  io_uring_factory_ = io_uring_factory;
 
  return std::make_unique<DefaultSocketInterfaceExtension>(*this, io_uring_factory);

source/common/network/socket_interface_impl.h

@@ -44,14 +44,6 @@ class SocketInterfaceImpl : public SocketInterfaceBase {
  absl::optional<int> domain,
  Io::IoUringFactory* io_uring_factory = nullptr);
 
- // TODO (soulxu): making those configurable
- // TODO (soulxu): we should handle when run out of all the entries.
- static constexpr uint32_t DefaultIoUringSize = 300;
- static constexpr uint32_t DefaultAcceptSize = 5;
- static constexpr uint32_t DefaultReadBufferSize = 8192;
- static constexpr uint32_t DefaultWriteTimeoutMs = 1000;
- static constexpr bool UseSubmissionQueuePolling = false;
-
 protected:
  virtual IoHandlePtr makeSocket(int socket_fd, bool socket_v6only, Socket::Type socket_type,
  absl::optional<int> domain) const;

source/server/server.cc

@@ -625,18 +625,13 @@ void InstanceImpl::initialize(Network::Address::InstanceConstSharedPtr local_add
  std::move(safe_actions), std::move(unsafe_actions), api_->threadFactory());
  }
 
- Network::SocketInterface* sock = nullptr;
  if (!bootstrap_.default_socket_interface().empty()) {
  auto& sock_name = bootstrap_.default_socket_interface();
- sock = const_cast<Network::SocketInterface*>(Network::socketInterface(sock_name));
- ASSERT(sock != nullptr);
- Network::SocketInterfaceSingleton::clear();
- Network::SocketInterfaceSingleton::initialize(sock);
- } else {
- auto factory = dynamic_cast<Server::Configuration::BootstrapExtensionFactory*>(
- Network::SocketInterfaceSingleton::getExisting());
- bootstrap_extensions_.push_back(factory->createBootstrapExtension(
- *factory->createEmptyConfigProto(), serverFactoryContext()));
+ auto sock = const_cast<Network::SocketInterface*>(Network::socketInterface(sock_name));
+ if (sock != nullptr) {
+ Network::SocketInterfaceSingleton::clear();
+ Network::SocketInterfaceSingleton::initialize(sock);
+ }
  }
 
  ListenerManagerFactory* listener_manager_factory = nullptr;

test/integration/base_integration_test.cc

@@ -11,6 +11,7 @@
 #include "envoy/buffer/buffer.h"
 #include "envoy/config/bootstrap/v3/bootstrap.pb.h"
 #include "envoy/config/endpoint/v3/endpoint_components.pb.h"
+#include "envoy/extensions/network/socket_interface/v3/default_socket_interface.pb.h"
 #include "envoy/extensions/transport_sockets/quic/v3/quic_transport.pb.h"
 #include "envoy/extensions/transport_sockets/tls/v3/cert.pb.h"
 #include "envoy/service/discovery/v3/discovery.pb.h"
@@ -79,6 +80,18 @@ BaseIntegrationTest::BaseIntegrationTest(const InstanceConstSharedPtrFn& upstrea
  config_helper_.addConfigModifier(
  [&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) -> void { bootstrap.clear_admin(); });
 #endif
+
+#ifdef ENVOY_TEST_IO_URING
+ config_helper_.addConfigModifier([&](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
+ bootstrap.set_default_socket_interface(
+ "envoy.extensions.network.socket_interface.default_socket_interface");
+ auto extension = bootstrap.add_bootstrap_extensions();
+ extension->set_name("envoy.extensions.network.socket_interface.default_socket_interface");
+ envoy::extensions::network::socket_interface::v3::DefaultSocketInterface interface;
+ interface.set_enable_io_uring(true);
+ extension->mutable_typed_config()->PackFrom(interface);
+ });
+#endif
 }
 
 BaseIntegrationTest::BaseIntegrationTest(const InstanceConstSharedPtrFn& upstream_address_fn,

test/integration/buffer_accounting_integration_test.cc

@@ -148,6 +148,13 @@ class Http2BufferWatermarksTest
  setUpstreamProtocol(protocol_test_params.upstream_protocol);
  }
 
+ void SetUp() override {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
+ }
+
 protected:
  // For testing purposes, track >= 4096B accounts.
  std::shared_ptr<Buffer::TrackedWatermarkBufferFactory> buffer_factory_;

test/integration/http2_flood_integration_test.cc

@@ -73,6 +73,13 @@ class Http2FloodMitigationTest
  deferredProcessing(GetParam()) ? "true" : "false");
  }
 
+ void SetUp() override {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
+ }
+
 protected:
  void initializeUpstreamFloodTest();
  std::vector<char> serializeFrames(const Http2Frame& frame, uint32_t num_frames);

test/integration/multiplexed_integration_test.cc

@@ -1239,12 +1239,11 @@ TEST_P(MultiplexedIntegrationTest, IdleTimeoutWithSimultaneousRequests) {
 
 // Test request mirroring / shadowing with an HTTP/2 downstream and a request with a body.
 TEST_P(MultiplexedIntegrationTest, RequestMirrorWithBody) {
- // TODO (soulxu): skip this test for io-uring, since this test depends on the io behavior.
- // After we enable the parameter test for io-uring and
- // default socket, then we should run this test for default socket, and write another version for
- // the io-uring.
- // It is caused by connection order behavior in tests.
+ // TODO(zhxie): io_uring works asynchronously and the connection may not come in a sequence as
+ // expected.
+#ifdef ENVOY_TEST_IO_URING
  GTEST_SKIP();
+#endif
 
  config_helper_.addConfigModifier(
  [&](envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
@@ -2293,13 +2292,12 @@ TEST_P(MultiplexedIntegrationTest, Reset101SwitchProtocolResponse) {
 }
 
 TEST_P(MultiplexedIntegrationTest, PerTryTimeoutWhileDownstreamStopsReading) {
- // TODO (soulxu): skip this test for io-uring, since this test depends on the io behavior.
- // After we enable the parameter test for io-uring and
- // default socket, then we should run this test for default socket, and write another version for
- // the io-uring.
- // It is caused by connection buffer never touch high watermark since the activate write file
- // events of server socket will get handled in the end of the on read of client socket.
+ // TODO(zhxie): io_uring socket is not compatible with watermark. It is caused by connection
+ // buffer never touch high watermark since the activate write file events of server socket will
+ // get handled in the end of the on read of client socket.
+#ifdef ENVOY_TEST_IO_URING
  GTEST_SKIP();
+#endif
 
  if (downstreamProtocol() != Http::CodecType::HTTP2) {
  return;
@@ -2427,6 +2425,13 @@ class SocketSwappableMultiplexedIntegrationTest : public SocketInterfaceSwap,
  : SocketInterfaceSwap(GetParam().downstream_protocol == Http::CodecType::HTTP3
  ? Network::Socket::Type::Datagram
  : Network::Socket::Type::Stream) {}
+
+ void SetUp() override {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
+ }
 };
 
 INSTANTIATE_TEST_SUITE_P(IpVersions, SocketSwappableMultiplexedIntegrationTest,

test/integration/protocol_integration_test.cc

@@ -2840,12 +2840,11 @@ TEST_P(DownstreamProtocolIntegrationTest, ConnDurationTimeoutNoHttpRequest) {
 }
 
 TEST_P(DownstreamProtocolIntegrationTest, TestPreconnect) {
- // TODO (soulxu): skip this test for io-uring, since this test depends on the io behavior.
- // After we enable the parameter test for io-uring and
- // default socket, then we should run this test for default socket, and write another version for
- // the io-uring.
- // It is caused by connection order behavior in tests.
+ // TODO(zhxie): io_uring works asynchronously and the connection may not come in a sequence as
+ // expected.
+#ifdef ENVOY_TEST_IO_URING
  GTEST_SKIP();
+#endif
 
  config_helper_.addConfigModifier([](envoy::config::bootstrap::v3::Bootstrap& bootstrap) {
  auto* cluster = bootstrap.mutable_static_resources()->mutable_clusters(0);
@@ -4289,6 +4288,10 @@ class NoUdpGso : public Api::OsSysCallsImpl {
 };
 
 TEST_P(DownstreamProtocolIntegrationTest, HandleDownstreamSocketFail) {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
  // Make sure for HTTP/3 Envoy will use sendmsg, so the write_matcher will work.
  NoUdpGso reject_gso_;
  TestThreadsafeSingletonInjector<Api::OsSysCallsImpl> os_calls{&reject_gso_};
@@ -4329,6 +4332,10 @@ TEST_P(DownstreamProtocolIntegrationTest, HandleDownstreamSocketFail) {
 }
 
 TEST_P(ProtocolIntegrationTest, HandleUpstreamSocketFail) {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
  SocketInterfaceSwap socket_swap(upstreamProtocol() == Http::CodecType::HTTP3
  ? Network::Socket::Type::Datagram
  : Network::Socket::Type::Stream);

test/integration/shadow_policy_integration_test.cc

@@ -28,6 +28,13 @@ class ShadowPolicyIntegrationTest
  setUpstreamCount(2);
  }
 
+ void SetUp() override {
+ // TODO(zhxie): io_uring is not compatible with SocketInterfaceSwap.
+#ifdef ENVOY_TEST_IO_URING
+ GTEST_SKIP();
+#endif
+ }
+
  // Adds a mirror policy that routes to cluster_header or cluster_name, in that order. Additionally
  // optionally registers an upstream filter on the cluster specified by
  // cluster_with_custom_filter_.