Address::InstanceConstSharedPtr instanceOrNull(StatusOr<Address::InstanceConstSharedPtr> address) {

  if (address.ok()) {

    return *address;

  }

  return nullptr;

}

std::string Utility::urlFromDatagramAddress(const Address::Instance& addr) {

  if (addr.ip() != nullptr) {

    return absl::StrCat(UDP_SCHEME, addr.asStringView());

  } else {

    return absl::StrCat(UNIX_SCHEME, addr.asStringView());

  }

}

absl::StatusOr<Address::InstanceConstSharedPtr> Utility::resolveUrl(const std::string& url) {

  Address::InstanceConstSharedPtr address{};

  if (urlIsTcpScheme(url)) {

    address = parseInternetAddressAndPortNoThrow(url.substr(TCP_SCHEME.size()));

  } else if (urlIsUdpScheme(url)) {

    address = parseInternetAddressAndPortNoThrow(url.substr(UDP_SCHEME.size()));

  } else if (urlIsUnixScheme(url)) {

    return Address::PipeInstance::create(url.substr(UNIX_SCHEME.size()));

  } else {

    return absl::InvalidArgumentError(absl::StrCat("unknown protocol scheme: ", url));

  }

  if (!address) {

    return absl::InvalidArgumentError(absl::StrCat("malformed IP address: ", url));

  }

  return address;

}

StatusOr<Socket::Type> Utility::socketTypeFromUrl(const std::string& url) {

  if (urlIsTcpScheme(url)) {

    return Socket::Type::Stream;

  } else if (urlIsUdpScheme(url)) {

    return Socket::Type::Datagram;

  } else if (urlIsUnixScheme(url)) {

    return Socket::Type::Stream;

  } else {

    return absl::InvalidArgumentError(absl::StrCat("unknown protocol scheme: ", url));

  }

}

bool Utility::urlIsTcpScheme(absl::string_view url) { return absl::StartsWith(url, TCP_SCHEME); }

bool Utility::urlIsUdpScheme(absl::string_view url) { return absl::StartsWith(url, UDP_SCHEME); }

bool Utility::urlIsUnixScheme(absl::string_view url) { return absl::StartsWith(url, UNIX_SCHEME); }

                                       const IoHandle::UdpSaveCmsgConfig& save_cmsg_config) {

  auto reservation = buffer->reserveSingleSlice(max_rx_datagram_size);

  Buffer::RawSlice slice = reservation.slice();

  Api::IoCallUint64Result result =

      handle.recvmsg(&slice, 1, local_address.ip()->port(), save_cmsg_config, output);

  if (result.ok()) {

    reservation.commit(std::min(max_rx_datagram_size, result.return_value_));

  }

  return result;

}

                                     absl::optional<std::string> network_namespace) {

  StatusOr<sockaddr_in> sa4 = IpAddressParsing::parseIPv4(ip_address, port);

  if (sa4.ok()) {

    return instanceOrNull(Address::InstanceFactory::createInstancePtr<Address::Ipv4Instance>(

        &sa4.value(), nullptr, network_namespace));

  }

  StatusOr<sockaddr_in6> sa6 = IpAddressParsing::parseIPv6(ip_address, port);

  if (sa6.ok()) {

    return instanceOrNull(Address::InstanceFactory::createInstancePtr<Address::Ipv6Instance>(

        *sa6, v6only, nullptr, network_namespace));

  }

  return nullptr;

}

                                            absl::optional<std::string> network_namespace) {

  if (ip_address.empty()) {

    return nullptr;

  }

  if (ip_address[0] == '[') {

    const auto pos = ip_address.rfind("]:");

    if (pos == std::string::npos) {

      return nullptr;

    }

    const auto ip_str = ip_address.substr(1, pos - 1);

    const auto port_str = ip_address.substr(pos + 2);

    uint64_t port64 = 0;

    if (port_str.empty() || !absl::SimpleAtoi(port_str, &port64) || port64 > 65535) {

      return nullptr;

    }

    StatusOr<sockaddr_in6> sa6 = IpAddressParsing::parseIPv6(ip_str, static_cast<uint16_t>(port64));

    if (sa6.ok()) {

      return instanceOrNull(Address::InstanceFactory::createInstancePtr<Address::Ipv6Instance>(

          *sa6, v6only, nullptr, network_namespace));

    }

    return nullptr;

  }

  const auto pos = ip_address.rfind(':');

  if (pos == std::string::npos) {

    return nullptr;

  }

  const auto ip_str = ip_address.substr(0, pos);

  const auto port_str = ip_address.substr(pos + 1);

  uint64_t port64 = 0;

  if (port_str.empty() || !absl::SimpleAtoi(port_str, &port64) || port64 > 65535) {

    return nullptr;

  }

  StatusOr<sockaddr_in> sa4 = IpAddressParsing::parseIPv4(ip_str, static_cast<uint16_t>(port64));

  if (sa4.ok()) {

    return instanceOrNull(Address::InstanceFactory::createInstancePtr<Address::Ipv4Instance>(

        &sa4.value(), nullptr, network_namespace));

  }

  return nullptr;

}

Address::InstanceConstSharedPtr Utility::copyInternetAddressAndPort(const Address::Ip& ip) {

  if (ip.version() == Address::IpVersion::v4) {

    return std::make_shared<Address::Ipv4Instance>(ip.addressAsString(), ip.port());

  }

  return std::make_shared<Address::Ipv6Instance>(ip.addressAsString(), ip.port());

}

Address::InstanceConstSharedPtr Utility::getLocalAddress(const Address::IpVersion version) {

  Address::InstanceConstSharedPtr ret;

  if (Api::OsSysCallsSingleton::get().supportsGetifaddrs()) {

    Api::InterfaceAddressVector interface_addresses{};

    const Api::SysCallIntResult rc =

        Api::OsSysCallsSingleton::get().getifaddrs(interface_addresses);

    if (rc.return_value_ != 0) {

      ENVOY_LOG_MISC(debug, fmt::format("getifaddrs error: {}", rc.errno_));

    } else {

      for (const auto& interface_address : interface_addresses) {

        if (!isLoopbackAddress(*interface_address.interface_addr_) &&

            interface_address.interface_addr_->ip()->version() == version) {

          ret = interface_address.interface_addr_;

          if (ret->ip()->version() == Address::IpVersion::v6) {

            ret = ret->ip()->ipv6()->addressWithoutScopeId();

          }

          break;

        }

      }

    }

  }

  if (ret == nullptr) {

    if (version == Address::IpVersion::v4) {

      ret = std::make_shared<Address::Ipv4Instance>("127.0.0.1");

    } else if (version == Address::IpVersion::v6) {

  }

  return ret;

}

bool Utility::isSameIpOrLoopback(const ConnectionInfoProvider& connection_info_provider) {

  const auto& remote_address = connection_info_provider.remoteAddress();

  if (remote_address->type() == Address::Type::Pipe || isLoopbackAddress(*remote_address)) {

    return true;

  }

  const auto local_ip = connection_info_provider.localAddress()->ip();

  const auto remote_ip = remote_address->ip();

  if (remote_ip != nullptr && local_ip != nullptr &&

      remote_ip->addressAsString() == local_ip->addressAsString()) {

    return true;

  }

  return false;

}

bool Utility::isInternalAddress(const Address::Instance& address) {

  if (address.type() != Address::Type::Ip) {

    return false;

  }

  if (address.ip()->version() == Address::IpVersion::v4) {

    const uint32_t address4 = address.ip()->ipv4()->address();

    const uint8_t* address4_bytes = reinterpret_cast<const uint8_t*>(&address4);

    if ((address4_bytes[0] == 10) || (address4_bytes[0] == 192 && address4_bytes[1] == 168) ||

        (address4_bytes[0] == 172 && address4_bytes[1] >= 16 && address4_bytes[1] <= 31) ||

        address4 == htonl(INADDR_LOOPBACK)) {

      return true;

    } else {

      return false;

    }

  }

  static_assert(sizeof(absl::uint128) == sizeof(in6addr_loopback),

                "sizeof(absl::uint128) != sizeof(in6addr_loopback)");

  const absl::uint128 address6 = address.ip()->ipv6()->address();

  const uint8_t* address6_bytes = reinterpret_cast<const uint8_t*>(&address6);

  if (address6_bytes[0] == 0xfd ||

      memcmp(&address6, &in6addr_loopback, sizeof(in6addr_loopback)) == 0) {

    return true;

  }

  return false;

}

bool Utility::isLoopbackAddress(const Address::Instance& address) {

  if (address.type() != Address::Type::Ip) {

    return false;

  }

  if (address.ip()->version() == Address::IpVersion::v4) {

    return address.ip()->ipv4()->address() == htonl(INADDR_LOOPBACK);

  } else if (address.ip()->version() == Address::IpVersion::v6) {

    static_assert(sizeof(absl::uint128) == sizeof(in6addr_loopback),

                  "sizeof(absl::uint128) != sizeof(in6addr_loopback)");

    absl::uint128 addr = address.ip()->ipv6()->address();

    return 0 == memcmp(&addr, &in6addr_loopback, sizeof(in6addr_loopback));

  }

}

Address::InstanceConstSharedPtr Utility::getCanonicalIpv4LoopbackAddress() {

  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,

                         new Address::Ipv4Instance("127.0.0.1", 0, nullptr));

}

Address::InstanceConstSharedPtr Utility::getIpv6LoopbackAddress() {

  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,

                         new Address::Ipv6Instance("::1", 0, nullptr));

}

Address::InstanceConstSharedPtr Utility::getIpv4AnyAddress() {

  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,

                         new Address::Ipv4Instance(static_cast<uint32_t>(0)));

}

Address::InstanceConstSharedPtr Utility::getIpv6AnyAddress() {

  CONSTRUCT_ON_FIRST_USE(Address::InstanceConstSharedPtr,

                         new Address::Ipv6Instance(static_cast<uint32_t>(0)));

}

const std::string& Utility::getIpv4CidrCatchAllAddress() {

  CONSTRUCT_ON_FIRST_USE(std::string, "0.0.0.0/0");

}

const std::string& Utility::getIpv6CidrCatchAllAddress() {

  CONSTRUCT_ON_FIRST_USE(std::string, "::/0");

}

                                                            uint32_t port) {

  switch (address.ip()->version()) {

  case Address::IpVersion::v4: {

    sockaddr_in addr = *reinterpret_cast<const sockaddr_in*>(address.sockAddr());

    addr.sin_port = htons(static_cast<uint16_t>(port));

    return std::make_shared<Address::Ipv4Instance>(&addr);

  case Address::IpVersion::v6: {

    sockaddr_in6 addr = *reinterpret_cast<const sockaddr_in6*>(address.sockAddr());

    addr.sin6_port = htons(static_cast<uint16_t>(port));

    return std::make_shared<Address::Ipv6Instance>(addr, address.ip()->ipv6()->v6only());

  }

Address::InstanceConstSharedPtr Utility::getOriginalDst(Socket& sock) {

#ifdef SOL_IP

  if (sock.addressType() != Address::Type::Ip) {

    return nullptr;

  }

  auto ipVersion = sock.ipVersion();

  if (!ipVersion.has_value()) {

    return nullptr;

  }

  SocketOptionName opt_dst;

  SocketOptionName opt_tp;

  if (*ipVersion == Address::IpVersion::v4) {

    opt_dst = ENVOY_SOCKET_SO_ORIGINAL_DST;

    opt_tp = ENVOY_SOCKET_IP_TRANSPARENT;

  } else {

    opt_dst = ENVOY_SOCKET_IP6T_SO_ORIGINAL_DST;

    opt_tp = ENVOY_SOCKET_IPV6_TRANSPARENT;

  }

  sockaddr_storage orig_addr;

  memset(&orig_addr, 0, sizeof(orig_addr));

  socklen_t addr_len = sizeof(sockaddr_storage);

  int status =

      sock.getSocketOption(opt_dst.level(), opt_dst.option(), &orig_addr, &addr_len).return_value_;

  if (status != 0) {

    if (Api::OsSysCallsSingleton::get().supportsIpTransparent(*ipVersion)) {

      socklen_t flag_len = sizeof(int);

      int is_tp;

      status =

          sock.getSocketOption(opt_tp.level(), opt_tp.option(), &is_tp, &flag_len).return_value_;

      if (status == 0 && is_tp) {

        auto address_or_error = sock.ioHandle().localAddress();

        if (address_or_error.status().ok()) {

          return *address_or_error;

        }

      }

    }

    return nullptr;

  }

  return Address::addressFromSockAddrOrDie(orig_addr, 0, -1, true /* default for v6 constructor */);

}

absl::uint128 Utility::Ip6ntohl(const absl::uint128& address) {

#ifdef ABSL_IS_LITTLE_ENDIAN

  return flipOrder(address);

}

absl::uint128 Utility::Ip6htonl(const absl::uint128& address) {

#ifdef ABSL_IS_LITTLE_ENDIAN

  return flipOrder(address);

}

absl::uint128 Utility::flipOrder(const absl::uint128& input) {

  absl::uint128 result{0};

  absl::uint128 data = input;

  for (int i = 0; i < 16; i++) {

    result <<= 8;

    result |= (data & 0x000000000000000000000000000000FF);

    data >>= 8;

  }

  return result;

}

Utility::protobufAddressToAddressNoThrow(const envoy::config::core::v3::Address& proto_address) {

  switch (proto_address.address_case()) {

  case envoy::config::core::v3::Address::AddressCase::kSocketAddress:

    return Utility::parseInternetAddressNoThrow(

        proto_address.socket_address().address(), proto_address.socket_address().port_value(),

        !proto_address.socket_address().ipv4_compat(),

        proto_address.socket_address().network_namespace_filepath());

  case envoy::config::core::v3::Address::AddressCase::kPipe: {

    auto ret_or_error =

        Address::PipeInstance::create(proto_address.pipe().path(), proto_address.pipe().mode());

    if (ret_or_error.status().ok()) {

      return std::move(*ret_or_error);

    }

  }

  case envoy::config::core::v3::Address::AddressCase::kEnvoyInternalAddress:

    return std::make_shared<Address::EnvoyInternalInstance>(

        proto_address.envoy_internal_address().server_listener_name(),

        proto_address.envoy_internal_address().endpoint_id());

  }

                                       envoy::config::core::v3::Address& proto_address) {

  if (address.type() == Address::Type::Pipe) {

    proto_address.mutable_pipe()->set_path(address.asString());

  } else if (address.type() == Address::Type::Ip) {

    auto* socket_address = proto_address.mutable_socket_address();

    socket_address->set_address(address.ip()->addressAsString());

    socket_address->set_port_value(address.ip()->port());

    if (address.networkNamespace().has_value()) {

      socket_address->set_network_namespace_filepath(address.networkNamespace().value());

    }

  } else {

    ASSERT(address.type() == Address::Type::EnvoyInternal);

    auto* internal_address = proto_address.mutable_envoy_internal_address();

    internal_address->set_server_listener_name(address.envoyInternalAddress()->addressId());

    internal_address->set_endpoint_id(address.envoyInternalAddress()->endpointId());

  }

}

Utility::protobufAddressSocketType(const envoy::config::core::v3::Address& proto_address) {

  switch (proto_address.address_case()) {

  case envoy::config::core::v3::Address::AddressCase::kSocketAddress: {

    const auto protocol = proto_address.socket_address().protocol();

    switch (protocol) {

    case envoy::config::core::v3::SocketAddress::TCP:

      return Socket::Type::Stream;

    case envoy::config::core::v3::SocketAddress::UDP:

      return Socket::Type::Datagram;

    }

  }

  case envoy::config::core::v3::Address::AddressCase::kPipe:

    return Socket::Type::Stream;

  case envoy::config::core::v3::Address::AddressCase::kEnvoyInternalAddress:

    return Socket::Type::Stream;

  }

                                               const Address::Instance& peer_address) {

  Buffer::RawSliceVector slices = buffer.getRawSlices();

  return writeToSocket(handle, slices.data(), slices.size(), local_ip, peer_address);

}

                                               const Address::Instance& peer_address) {

  Api::IoCallUint64Result send_result(

      /*rc=*/0, /*err=*/Api::IoError::none());

  const bool is_connected = handle.wasConnected();

  do {

    if (is_connected) {

      send_result = handle.writev(slices, num_slices);

    } else {

      send_result = handle.sendmsg(slices, num_slices, 0, local_ip, peer_address);

    }

  } while (!send_result.ok() &&

           send_result.err_->getErrorCode() == Api::IoError::IoErrorCode::Interrupt);

  if (send_result.ok()) {

    ENVOY_LOG_MISC(trace, "{} bytes {}", is_connected ? "writev" : "sendmsg",

                   send_result.return_value_);

  } else {

    ENVOY_LOG_MISC(debug, "{} failed with error code {}: {}", is_connected ? "writev" : "sendmsg",

                   static_cast<int>(send_result.err_->getErrorCode()),

                   send_result.err_->getErrorDetails());

  }

  return send_result;

}

                            uint8_t tos, Buffer::OwnedImpl saved_cmsg) {

  ENVOY_BUG(peer_addess != nullptr,

            fmt::format("Unable to get remote address on the socket bound to local address: {}.",

                        (local_address == nullptr ? "unknown" : local_address->asString())));

  ENVOY_BUG(peer_addess != nullptr && peer_addess->type() == Address::Type::Ip,

            fmt::format("Unsupported remote address: {} local address: {}, receive size: "

                        "{}",

                        peer_addess->asString(),

                        (local_address == nullptr ? "unknown" : local_address->asString()),

                        bytes_read));

  udp_packet_processor.processPacket(std::move(local_address), std::move(peer_addess),

                                     std::move(buffer), receive_time, tos, std::move(saved_cmsg));

}

                                              uint32_t* num_packets_read) {

  ASSERT(Api::OsSysCallsSingleton::get().supportsUdpGro(),

         "cannot use GRO when the platform doesn't support it.");

  if (num_packets_read != nullptr) {

    *num_packets_read = 0;

  }

  Buffer::InstancePtr buffer = std::make_unique<Buffer::OwnedImpl>();

  IoHandle::RecvMsgOutput output(1, packets_dropped);

  const uint64_t max_rx_datagram_size_with_gro =

      num_packets_read != nullptr

          ? 64 * 1024

          : NUM_DATAGRAMS_PER_RECEIVE * udp_packet_processor.maxDatagramSize();

  ENVOY_LOG_MISC(trace, "starting gro recvmsg with max={}", max_rx_datagram_size_with_gro);

  Api::IoCallUint64Result result =

      receiveMessage(max_rx_datagram_size_with_gro, buffer, output, handle, local_address,

                     udp_packet_processor.saveCmsgConfig());

  if (!result.ok() || output.msg_[0].truncated_and_dropped_) {

    return result;

  }

  const uint64_t gso_size = output.msg_[0].gso_size_;

  ENVOY_LOG_MISC(trace, "gro recvmsg bytes {} with gso_size as {}", result.return_value_, gso_size);

  const MonotonicTime receive_time = time_source.monotonicTime();

  if (gso_size == 0u) {

    if (num_packets_read != nullptr) {

      *num_packets_read += 1;

    }

    passPayloadToProcessor(

        result.return_value_, std::move(buffer), std::move(output.msg_[0].peer_address_),

        std::move(output.msg_[0].local_address_), udp_packet_processor, receive_time,

        output.msg_[0].tos_, std::move(output.msg_[0].saved_cmsg_));

    return result;

  }

  while (buffer->length() > 0) {

    const uint64_t bytes_to_copy = std::min(buffer->length(), gso_size);

    Buffer::InstancePtr sub_buffer = std::make_unique<Buffer::OwnedImpl>();

    sub_buffer->move(*buffer, bytes_to_copy);

    if (num_packets_read != nullptr) {

      *num_packets_read += 1;

    }

    passPayloadToProcessor(bytes_to_copy, std::move(sub_buffer), output.msg_[0].peer_address_,

                           output.msg_[0].local_address_, udp_packet_processor, receive_time,

                           output.msg_[0].tos_, std::move(output.msg_[0].saved_cmsg_));

  }

  return result;

}

                                               uint32_t* num_packets_read) {

  ASSERT(Api::OsSysCallsSingleton::get().supportsMmsg(),

         "cannot use recvmmsg when the platform doesn't support it.");

  const auto max_rx_datagram_size = udp_packet_processor.maxDatagramSize();

  if (num_packets_read != nullptr) {

    *num_packets_read = 0;

  }

  struct BufferAndReservation {

    BufferAndReservation(uint64_t max_rx_datagram_size)

        : buffer_(std::make_unique<Buffer::OwnedImpl>()),

          reservation_(buffer_->reserveSingleSlice(max_rx_datagram_size, true)) {}

    Buffer::InstancePtr buffer_;

    Buffer::ReservationSingleSlice reservation_;

  };

  constexpr uint32_t num_slices_per_packet = 1u;

  absl::InlinedVector<BufferAndReservation, NUM_DATAGRAMS_PER_RECEIVE> buffers;

  RawSliceArrays slices(NUM_DATAGRAMS_PER_RECEIVE,

                        absl::FixedArray<Buffer::RawSlice>(num_slices_per_packet));

  for (uint32_t i = 0; i < NUM_DATAGRAMS_PER_RECEIVE; i++) {

    buffers.push_back(max_rx_datagram_size);

    slices[i][0] = buffers[i].reservation_.slice();

  }

  IoHandle::RecvMsgOutput output(NUM_DATAGRAMS_PER_RECEIVE, packets_dropped);

  ENVOY_LOG_MISC(trace, "starting recvmmsg with packets={} max={}", NUM_DATAGRAMS_PER_RECEIVE,

                 max_rx_datagram_size);

  Api::IoCallUint64Result result = handle.recvmmsg(slices, local_address.ip()->port(),

                                                   udp_packet_processor.saveCmsgConfig(), output);

  if (!result.ok()) {

    return result;

  }

  uint64_t packets_read = result.return_value_;

  ENVOY_LOG_MISC(trace, "recvmmsg read {} packets", packets_read);

  const MonotonicTime receive_time = time_source.monotonicTime();

  for (uint64_t i = 0; i < packets_read; ++i) {

    if (output.msg_[i].truncated_and_dropped_) {

      continue;

    }

    Buffer::RawSlice* slice = slices[i].data();

    const uint64_t msg_len = output.msg_[i].msg_len_;

    ASSERT(msg_len <= slice->len_);

    ENVOY_LOG_MISC(debug, "Receive a packet with {} bytes from {}", msg_len,

                   output.msg_[i].peer_address_->asString());

    buffers[i].reservation_.commit(std::min(max_rx_datagram_size, msg_len));

    if (num_packets_read != nullptr) {

      *num_packets_read += 1;

    }

    passPayloadToProcessor(msg_len, std::move(buffers[i].buffer_), output.msg_[i].peer_address_,

                           output.msg_[i].local_address_, udp_packet_processor, receive_time,

                           output.msg_[i].tos_, std::move(output.msg_[i].saved_cmsg_));

  }

  return result;

}

                                              uint32_t* num_packets_read) {

  if (num_packets_read != nullptr) {

    *num_packets_read = 0;

  }

  Buffer::InstancePtr buffer = std::make_unique<Buffer::OwnedImpl>();

  IoHandle::RecvMsgOutput output(1, packets_dropped);

  ENVOY_LOG_MISC(trace, "starting recvmsg with max={}", udp_packet_processor.maxDatagramSize());

  Api::IoCallUint64Result result =

      receiveMessage(udp_packet_processor.maxDatagramSize(), buffer, output, handle, local_address,

                     udp_packet_processor.saveCmsgConfig());

  if (!result.ok() || output.msg_[0].truncated_and_dropped_) {

    return result;

  }

  ENVOY_LOG_MISC(trace, "recvmsg bytes {}", result.return_value_);

  if (num_packets_read != nullptr) {

    *num_packets_read = 1;

  }

  passPayloadToProcessor(

      result.return_value_, std::move(buffer), std::move(output.msg_[0].peer_address_),

      std::move(output.msg_[0].local_address_), udp_packet_processor, time_source.monotonicTime(),

      output.msg_[0].tos_, std::move(output.msg_[0].saved_cmsg_));

  return result;

}

                        uint32_t* num_packets_read) {

  if (recv_msg_method == UdpRecvMsgMethod::RecvMsgWithGro) {

    return readFromSocketRecvGro(handle, local_address, udp_packet_processor, time_source,

                                 packets_dropped, num_packets_read);

  } else if (recv_msg_method == UdpRecvMsgMethod::RecvMmsg) {

    return readFromSocketRecvMmsg(handle, local_address, udp_packet_processor, time_source,

                                  packets_dropped, num_packets_read);

  }

  return readFromSocketRecvMsg(handle, local_address, udp_packet_processor, time_source,

                               packets_dropped, num_packets_read);

}

                                               bool allow_mmsg, uint32_t& packets_dropped) {

  UdpRecvMsgMethod recv_msg_method = UdpRecvMsgMethod::RecvMsg;

  if (allow_gro && handle.supportsUdpGro()) {

    recv_msg_method = UdpRecvMsgMethod::RecvMsgWithGro;

  } else if (allow_mmsg && handle.supportsMmsg()) {

    recv_msg_method = UdpRecvMsgMethod::RecvMmsg;

  }

  size_t num_packets_to_read = std::min<size_t>(

      MAX_NUM_PACKETS_PER_EVENT_LOOP, udp_packet_processor.numPacketsExpectedPerEventLoop());

  size_t num_reads = std::max<size_t>(1, num_packets_to_read);

  do {

    const uint32_t old_packets_dropped = packets_dropped;

    uint32_t num_packets_processed = 0;

    Api::IoCallUint64Result result =

        Utility::readFromSocket(handle, local_address, udp_packet_processor, time_source,

                                recv_msg_method, &packets_dropped, &num_packets_processed);

    if (!result.ok()) {

      return std::move(result.err_);

    }

    if (packets_dropped != old_packets_dropped) {

      const uint32_t delta =

          (packets_dropped > old_packets_dropped)

              ? (packets_dropped - old_packets_dropped)

              : (packets_dropped + (std::numeric_limits<uint32_t>::max() - old_packets_dropped) +

      ENVOY_LOG_EVERY_POW_2_MISC(

          warn,

          "Kernel dropped {} datagram(s). Consider increasing receive buffer size and/or "

          "max datagram size.",

          delta);

      udp_packet_processor.onDatagramsDropped(delta);

    }

    if (num_packets_to_read <= num_packets_processed) {

      return std::move(result.err_);

    }

    num_packets_to_read -= num_packets_processed;

    --num_reads;

    if (num_reads == 0) {

      return std::move(result.err_);

    }

  } while (true);

}

    : max_rx_datagram_size_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, max_rx_datagram_size,

      prefer_gro_(PROTOBUF_GET_WRAPPED_OR_DEFAULT(config, prefer_gro, prefer_gro_default)) {

  if (prefer_gro_ && !Api::OsSysCallsSingleton::get().supportsUdpGro()) {

    ENVOY_LOG_MISC(

        warn, "GRO requested but not supported by the OS. Check OS config or disable prefer_gro.");

  }

}