Magic-Mirror/gecko-dev
1
0
Fork 0
mirror of https://github.com/mozilla/gecko-dev.git synced 2024-11-23 12:51:06 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Bug 1901295 - Use quinn-udp instead of NSPR for QUIC UDP I/O r=necko-reviewers,jesup,supply-chain-reviewers,kershaw

Browse Source 
Differential Revision: https://phabricator.services.mozilla.com/D212959
This commit is contained in:
Max Inden 2024-07-26 08:35:22 +00:00
parent 8991573e43
commit 264edb41cc
21 changed files with 731 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
Cargo.lock generated
View File

@ -4164,6 +4164,7 @@ dependencies = [
"neqo-http3",
"neqo-qpack",
"neqo-transport",
"neqo-udp",
"nserror",
"nsstring",
"qlog",
@ -4811,14 +4812,13 @@ checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
[[package]]
name = "quinn-udp"
version = "0.5.0"
version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb7ad7bc932e4968523fa7d9c320ee135ff779de720e9350fee8728838551764"
checksum = "8bffec3605b73c6f1754535084a85229fa8a30f86014e6c81aeec4abb68b0285"
dependencies = [
"libc",
"once_cell",
"socket2 0.5.7",
"tracing",
"windows-sys 0.52.0",
]

12
modules/libpref/init/StaticPrefList.yaml
View File

@ -13506,6 +13506,12 @@
value: 1048576
mirror: always
# Use NSPR for HTTP3 UDP IO
- name: network.http.http3.use_nspr_for_io
type: RelaxedAtomicBool
value: true
mirror: always
- name: network.http.http3.enable_qlog
type: RelaxedAtomicBool
value: false
@ -13569,12 +13575,14 @@
value: @IS_EARLY_BETA_OR_EARLIER@
mirror: always
# This is for testing purpose. When true, nsUDPSocket::SendWithAddress will
# return NS_ERROR_CONNECTION_REFUSED for address "::1".
# This is for testing purpose. When true, nsUDPSocket::SendWithAddress and
# neqo_http3conn_process_output_and_send will return NS_ERROR_CONNECTION_REFUSED
# for address "::1".
- name: network.http.http3.block_loopback_ipv6_addr
type: RelaxedAtomicBool
value: false
mirror: always
rust: true
# The congestion control algorithm with which to configure neqo.
# 0 => NewReno

9
netwerk/base/nsIUDPSocket.idl
View File

@ -254,6 +254,15 @@ interface nsIUDPSocket : nsISupports
[noscript] void leaveMulticastAddr([const] in NetAddr addr,
[const, optional] in NetAddrPtr iface);
/**
* getFileDescriptor
*
* Get the file descriptor of the socket.
*
* @return The file descriptor.
*/
[noscript, notxpcom] int64_t getFileDescriptor();
/**
* multicastLoopback
*

5
netwerk/base/nsUDPSocket.cpp
View File

@ -21,6 +21,7 @@
#include "nsIOService.h"
#include "prnetdb.h"
#include "prio.h"
#include "private/pprio.h"
#include "nsNetAddr.h"
#include "nsNetSegmentUtils.h"
#include "nsServiceManagerUtils.h"
@ -1194,6 +1195,10 @@ nsUDPSocket::SendWithAddress(const NetAddr* aAddr, const uint8_t* aData,
return NS_OK;
}
int64_t nsUDPSocket::GetFileDescriptor() {
return PR_FileDesc2NativeHandle(mFD);
}
NS_IMETHODIMP
nsUDPSocket::SendBinaryStream(const nsACString& aHost, uint16_t aPort,
nsIInputStream* aStream) {

2
netwerk/docs/http/http3.md
View File

@ -53,7 +53,7 @@ It calls NeqoHttp3Conn::GetEvent which maps to Http3Client::next_event.
The events and their handling will be explained below.
**ProcessOutput**
The function is called when necko has performed some action on neqo, e.g. new HTTP transaction is added, certificate verification is done, etc., or when the timer expires. In both cases, necko wants to check if neqo has data to send or change its state. This function calls NeqoHttp3Conn::ProcessOutput that maps to Http3Client::process_output. NeqoHttp3Conn::ProcessOutput may return a packet that is sent on the socket or a callback timeout. In the Http3Session::ProcessOutput function, NeqoHttp3Conn::ProcessOutput is called repeatedly and packets are sent until a callback timer is returned or a fatal error happens.
The function is called when necko has performed some action on neqo, e.g. new HTTP transaction is added, certificate verification is done, etc., or when the timer expires. In both cases, necko wants to check if neqo has data to send or change its state. This function calls NeqoHttp3Conn::ProcessOutput that maps to Http3Client::process_output. In the Http3Session::ProcessOutput function, NeqoHttp3Conn::ProcessOutput is called repeatedly, which sents packets from neqo to the socket until neqo returns a callback timer, signals being idle, or returns a fatal error.
**Http3Session::RecvData** performs the following steps:
- ProcessSlowConsumers - explained below.

191
netwerk/protocol/http/Http3Session.cpp
View File

@ -92,7 +92,8 @@ static nsresult RawBytesToNetAddr(uint16_t aFamily, const uint8_t* aRemoteAddr,
nsresult Http3Session::Init(const nsHttpConnectionInfo* aConnInfo,
nsINetAddr* aSelfAddr, nsINetAddr* aPeerAddr,
HttpConnectionUDP* udpConn, uint32_t aProviderFlags,
nsIInterfaceRequestor* callbacks) {
nsIInterfaceRequestor* callbacks,
nsIUDPSocket* socket) {
LOG3(("Http3Session::Init %p", this));
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
@ -134,16 +135,34 @@ nsresult Http3Session::Init(const nsHttpConnectionInfo* aConnInfo,
StaticPrefs::network_webtransport_datagrams_enabled()
? StaticPrefs::network_webtransport_datagram_size()
: 0;
nsresult rv = NeqoHttp3Conn::Init(
mConnInfo->GetOrigin(), mConnInfo->GetNPNToken(), selfAddr, peerAddr,
gHttpHandler->DefaultQpackTableSize(),
gHttpHandler->DefaultHttp3MaxBlockedStreams(),
StaticPrefs::network_http_http3_max_data(),
StaticPrefs::network_http_http3_max_stream_data(),
StaticPrefs::network_http_http3_version_negotiation_enabled(),
mConnInfo->GetWebTransport(), gHttpHandler->Http3QlogDir(), datagramSize,
StaticPrefs::network_http_http3_max_accumlated_time_ms(), aProviderFlags,
getter_AddRefs(mHttp3Connection));
mUseNSPRForIO = StaticPrefs::network_http_http3_use_nspr_for_io();
nsresult rv;
if (mUseNSPRForIO) {
rv = NeqoHttp3Conn::InitUseNSPRForIO(
mConnInfo->GetOrigin(), mConnInfo->GetNPNToken(), selfAddr, peerAddr,
gHttpHandler->DefaultQpackTableSize(),
gHttpHandler->DefaultHttp3MaxBlockedStreams(),
StaticPrefs::network_http_http3_max_data(),
StaticPrefs::network_http_http3_max_stream_data(),
StaticPrefs::network_http_http3_version_negotiation_enabled(),
mConnInfo->GetWebTransport(), gHttpHandler->Http3QlogDir(),
datagramSize, StaticPrefs::network_http_http3_max_accumlated_time_ms(),
aProviderFlags, getter_AddRefs(mHttp3Connection));
} else {
rv = NeqoHttp3Conn::Init(
mConnInfo->GetOrigin(), mConnInfo->GetNPNToken(), selfAddr, peerAddr,
gHttpHandler->DefaultQpackTableSize(),
gHttpHandler->DefaultHttp3MaxBlockedStreams(),
StaticPrefs::network_http_http3_max_data(),
StaticPrefs::network_http_http3_max_stream_data(),
StaticPrefs::network_http_http3_version_negotiation_enabled(),
mConnInfo->GetWebTransport(), gHttpHandler->Http3QlogDir(),
datagramSize, StaticPrefs::network_http_http3_max_accumlated_time_ms(),
aProviderFlags, socket->GetFileDescriptor(),
getter_AddRefs(mHttp3Connection));
}
if (NS_FAILED(rv)) {
return rv;
}
@ -381,35 +400,51 @@ Http3Session::~Http3Session() {
// It will not return an error if socket error is
// NS_BASE_STREAM_WOULD_BLOCK.
// A caller of this function will close the Http3 connection
// in case of a error.
// The only callers is:
// HttpConnectionUDP::RecvData ->
// Http3Session::RecvData
void Http3Session::ProcessInput(nsIUDPSocket* socket) {
// in case of an error.
// The only callers is Http3Session::RecvData.
nsresult Http3Session::ProcessInput(nsIUDPSocket* socket) {
MOZ_ASSERT(OnSocketThread(), "not on socket thread");
MOZ_ASSERT(mUdpConn);
LOG(("Http3Session::ProcessInput writer=%p [this=%p state=%d]",
mUdpConn.get(), this, mState));
while (true) {
nsTArray<uint8_t> data;
NetAddr addr{};
// RecvWithAddr actually does not return an error.
nsresult rv = socket->RecvWithAddr(&addr, data);
MOZ_ALWAYS_SUCCEEDS(rv);
if (NS_FAILED(rv) || data.IsEmpty()) {
break;
}
rv = mHttp3Connection->ProcessInput(addr, data);
MOZ_ALWAYS_SUCCEEDS(rv);
if (NS_FAILED(rv)) {
break;
if (mUseNSPRForIO) {
while (true) {
nsTArray<uint8_t> data;
NetAddr addr{};
// RecvWithAddr actually does not return an error.
nsresult rv = socket->RecvWithAddr(&addr, data);
MOZ_ALWAYS_SUCCEEDS(rv);
if (NS_FAILED(rv) || data.IsEmpty()) {
break;
}
rv = mHttp3Connection->ProcessInputUseNSPRForIO(addr, data);
MOZ_ALWAYS_SUCCEEDS(rv);
if (NS_FAILED(rv)) {
break;
}
LOG(("Http3Session::ProcessInput received=%zu", data.Length()));
mTotalBytesRead += static_cast<int64_t>(data.Length());
}
LOG(("Http3Session::ProcessInput received=%zu", data.Length()));
mTotalBytesRead += data.Length();
return NS_OK;
}
// Not using NSPR.
auto rv = mHttp3Connection->ProcessInput();
// Note: WOULD_BLOCK is handled in neqo_glue.
if (NS_FAILED(rv.result)) {
mSocketError = rv.result;
// If there was an error return from here. We do not need to set a timer,
// because we will close the connection.
return rv.result;
}
mTotalBytesRead += rv.bytes_read;
return NS_OK;
}
nsresult Http3Session::ProcessTransactionRead(uint64_t stream_id) {
@ -901,46 +936,69 @@ nsresult Http3Session::ProcessOutput(nsIUDPSocket* socket) {
LOG(("Http3Session::ProcessOutput reader=%p, [this=%p]", mUdpConn.get(),
this));
mSocket = socket;
nsresult rv = mHttp3Connection->ProcessOutputAndSend(
this,
[](void* aContext, uint16_t aFamily, const uint8_t* aAddr, uint16_t aPort,
const uint8_t* aData, uint32_t aLength) {
Http3Session* self = (Http3Session*)aContext;
if (mUseNSPRForIO) {
mSocket = socket;
nsresult rv = mHttp3Connection->ProcessOutputAndSendUseNSPRForIO(
this,
[](void* aContext, uint16_t aFamily, const uint8_t* aAddr,
uint16_t aPort, const uint8_t* aData, uint32_t aLength) {
Http3Session* self = (Http3Session*)aContext;
uint32_t written = 0;
NetAddr addr;
if (NS_FAILED(RawBytesToNetAddr(aFamily, aAddr, aPort, &addr))) {
uint32_t written = 0;
NetAddr addr;
if (NS_FAILED(RawBytesToNetAddr(aFamily, aAddr, aPort, &addr))) {
return NS_OK;
}
LOG3(
("Http3Session::ProcessOutput sending packet with %u bytes to %s "
"port=%d [this=%p].",
aLength, addr.ToString().get(), aPort, self));
nsresult rv =
self->mSocket->SendWithAddress(&addr, aData, aLength, &written);
LOG(("Http3Session::ProcessOutput sending packet rv=%d osError=%d",
static_cast<int32_t>(rv), NS_FAILED(rv) ? PR_GetOSError() : 0));
if (NS_FAILED(rv) && (rv != NS_BASE_STREAM_WOULD_BLOCK)) {
self->mSocketError = rv;
// If there was an error that is not NS_BASE_STREAM_WOULD_BLOCK
// return from here. We do not need to set a timer, because we
// will close the connection.
return rv;
}
self->mTotalBytesWritten += aLength;
self->mLastWriteTime = PR_IntervalNow();
return NS_OK;
}
},
[](void* aContext, uint64_t timeout) {
Http3Session* self = (Http3Session*)aContext;
self->SetupTimer(timeout);
});
mSocket = nullptr;
return rv;
}
LOG3(
("Http3Session::ProcessOutput sending packet with %u bytes to %s "
"port=%d [this=%p].",
aLength, addr.ToString().get(), aPort, self));
// Not using NSPR.
nsresult rv =
self->mSocket->SendWithAddress(&addr, aData, aLength, &written);
LOG(("Http3Session::ProcessOutput sending packet rv=%d osError=%d",
static_cast<int32_t>(rv), NS_FAILED(rv) ? PR_GetOSError() : 0));
if (NS_FAILED(rv) && (rv != NS_BASE_STREAM_WOULD_BLOCK)) {
self->mSocketError = rv;
// If there was an error that is not NS_BASE_STREAM_WOULD_BLOCK
// return from here. We do not need to set a timer, because we
// will close the connection.
return rv;
}
self->mTotalBytesWritten += aLength;
self->mLastWriteTime = PR_IntervalNow();
return NS_OK;
},
[](void* aContext, uint64_t timeout) {
auto rv = mHttp3Connection->ProcessOutputAndSend(
this, [](void* aContext, uint64_t timeout) {
Http3Session* self = (Http3Session*)aContext;
self->SetupTimer(timeout);
});
mSocket = nullptr;
return rv;
// Note: WOULD_BLOCK is handled in neqo_glue.
if (NS_FAILED(rv.result)) {
mSocketError = rv.result;
// If there was an error return from here. We do not need to set a timer,
// because we will close the connection.
return rv.result;
}
if (rv.bytes_written != 0) {
mTotalBytesWritten += rv.bytes_written;
mLastWriteTime = PR_IntervalNow();
}
return NS_OK;
}
// This is only called when timer expires.
@ -1622,7 +1680,10 @@ nsresult Http3Session::RecvData(nsIUDPSocket* socket) {
return rv;
}
ProcessInput(socket);
rv = ProcessInput(socket);
if (NS_FAILED(rv)) {
return rv;
}
rv = ProcessEvents();
if (NS_FAILED(rv)) {

8
netwerk/protocol/http/Http3Session.h
View File

@ -130,7 +130,8 @@ class Http3Session final : public nsAHttpTransaction, public nsAHttpConnection {
Http3Session();
nsresult Init(const nsHttpConnectionInfo* aConnInfo, nsINetAddr* selfAddr,
nsINetAddr* peerAddr, HttpConnectionUDP* udpConn,
uint32_t aProviderFlags, nsIInterfaceRequestor* callbacks);
uint32_t aProviderFlags, nsIInterfaceRequestor* callbacks,
nsIUDPSocket* socket);
bool IsConnected() const { return mState == CONNECTED; }
bool CanSendData() const {
@ -229,7 +230,7 @@ class Http3Session final : public nsAHttpTransaction, public nsAHttpConnection {
bool justKidding);
nsresult ProcessOutput(nsIUDPSocket* socket);
void ProcessInput(nsIUDPSocket* socket);
nsresult ProcessInput(nsIUDPSocket* socket);
nsresult ProcessEvents();
nsresult ProcessTransactionRead(uint64_t stream_id);
@ -310,6 +311,9 @@ class Http3Session final : public nsAHttpTransaction, public nsAHttpConnection {
// True if the mTimer is inited and waiting for firing.
bool mTimerActive{false};
// True if this http3 session uses NSPR for UDP IO.
bool mUseNSPRForIO{true};
RefPtr<HttpConnectionUDP> mUdpConn;
// The underlying socket transport object is needed to propogate some events

2
netwerk/protocol/http/HttpConnectionUDP.cpp
View File

@ -155,7 +155,7 @@ nsresult HttpConnectionUDP::Init(nsHttpConnectionInfo* info,
mPeerAddr = new nsNetAddr(&peerAddr);
mHttp3Session = new Http3Session();
rv = mHttp3Session->Init(mConnInfo, mSelfAddr, mPeerAddr, this, providerFlags,
callbacks);
callbacks, mSocket);
if (NS_FAILED(rv)) {
LOG(
("HttpConnectionUDP::Init mHttp3Session->Init failed "

1
netwerk/socket/neqo_glue/Cargo.toml
View File

@ -9,6 +9,7 @@ license = "MPL-2.0"
name = "neqo_glue"
[dependencies]
neqo-udp = { tag = "v0.8.1", git = "https://github.com/mozilla/neqo" }
neqo-http3 = { tag = "v0.8.1", git = "https://github.com/mozilla/neqo" }
neqo-transport = { tag = "v0.8.1", git = "https://github.com/mozilla/neqo" }
neqo-common = { tag = "v0.8.1", git = "https://github.com/mozilla/neqo" }

42
netwerk/socket/neqo_glue/NeqoHttp3Conn.h
View File

@ -13,6 +13,21 @@ namespace net {
class NeqoHttp3Conn final {
public:
static nsresult InitUseNSPRForIO(
const nsACString& aOrigin, const nsACString& aAlpn,
const NetAddr& aLocalAddr, const NetAddr& aRemoteAddr,
uint32_t aMaxTableSize, uint16_t aMaxBlockedStreams, uint64_t aMaxData,
uint64_t aMaxStreamData, bool aVersionNegotiation, bool aWebTransport,
const nsACString& aQlogDir, uint32_t aDatagramSize,
uint32_t aMaxAccumulatedTime, uint32_t aProviderFlags,
NeqoHttp3Conn** aConn) {
return neqo_http3conn_new_use_nspr_for_io(
&aOrigin, &aAlpn, &aLocalAddr, &aRemoteAddr, aMaxTableSize,
aMaxBlockedStreams, aMaxData, aMaxStreamData, aVersionNegotiation,
aWebTransport, &aQlogDir, aDatagramSize, aMaxAccumulatedTime,
aProviderFlags, (const mozilla::net::NeqoHttp3Conn**)aConn);
}
static nsresult Init(const nsACString& aOrigin, const nsACString& aAlpn,
const NetAddr& aLocalAddr, const NetAddr& aRemoteAddr,
uint32_t aMaxTableSize, uint16_t aMaxBlockedStreams,
@ -20,12 +35,12 @@ class NeqoHttp3Conn final {
bool aVersionNegotiation, bool aWebTransport,
const nsACString& aQlogDir, uint32_t aDatagramSize,
uint32_t aMaxAccumulatedTime, uint32_t aProviderFlags,
NeqoHttp3Conn** aConn) {
int64_t socket, NeqoHttp3Conn** aConn) {
return neqo_http3conn_new(
&aOrigin, &aAlpn, &aLocalAddr, &aRemoteAddr, aMaxTableSize,
aMaxBlockedStreams, aMaxData, aMaxStreamData, aVersionNegotiation,
aWebTransport, &aQlogDir, aDatagramSize, aMaxAccumulatedTime,
aProviderFlags, (const mozilla::net::NeqoHttp3Conn**)aConn);
aProviderFlags, socket, (const mozilla::net::NeqoHttp3Conn**)aConn);
}
void Close(uint64_t aError) { neqo_http3conn_close(this, aError); }
@ -42,14 +57,25 @@ class NeqoHttp3Conn final {
neqo_http3conn_authenticated(this, aError);
}
nsresult ProcessInput(const NetAddr& aRemoteAddr,
const nsTArray<uint8_t>& aPacket) {
return neqo_http3conn_process_input(this, &aRemoteAddr, &aPacket);
nsresult ProcessInputUseNSPRForIO(const NetAddr& aRemoteAddr,
const nsTArray<uint8_t>& aPacket) {
return neqo_http3conn_process_input_use_nspr_for_io(this, &aRemoteAddr,
&aPacket);
}
nsresult ProcessOutputAndSend(void* aContext, SendFunc aSendFunc,
SetTimerFunc aSetTimerFunc) {
return neqo_http3conn_process_output_and_send(this, aContext, aSendFunc,
ProcessInputResult ProcessInput() {
return neqo_http3conn_process_input(this);
}
nsresult ProcessOutputAndSendUseNSPRForIO(void* aContext, SendFunc aSendFunc,
SetTimerFunc aSetTimerFunc) {
return neqo_http3conn_process_output_and_send_use_nspr_for_io(
this, aContext, aSendFunc, aSetTimerFunc);
}
ProcessOutputAndSendResult ProcessOutputAndSend(void* aContext,
SetTimerFunc aSetTimerFunc) {
return neqo_http3conn_process_output_and_send(this, aContext,
aSetTimerFunc);
}

269
netwerk/socket/neqo_glue/src/lib.rs
View File

@ -5,7 +5,9 @@
#[cfg(not(windows))]
use libc::{AF_INET, AF_INET6};
use neqo_common::event::Provider;
use neqo_common::{self as common, qlog::NeqoQlog, qwarn, Datagram, Header, IpTos, Role};
use neqo_common::{
self as common, qdebug, qerror, qlog::NeqoQlog, qwarn, Datagram, Header, IpTos, Role,
};
use neqo_crypto::{init, PRErrorCode};
use neqo_http3::{
features::extended_connect::SessionCloseReason, Error as Http3Error, Http3Client,
@ -28,13 +30,13 @@ use std::fs::OpenOptions;
use std::net::SocketAddr;
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
use std::path::PathBuf;
use std::ptr;
use std::rc::Rc;
use std::slice;
use std::str;
#[cfg(feature = "fuzzing")]
use std::time::Duration;
use std::time::{Duration, Instant};
use std::{io, ptr};
use thin_vec::ThinVec;
use uuid::Uuid;
#[cfg(windows)]
@ -48,6 +50,14 @@ pub struct NeqoHttp3Conn {
refcnt: AtomicRefcnt,
last_output_time: Instant,
max_accumlated_time: Duration,
/// Socket to use for IO.
///
/// When [`None`], NSPR is used for IO.
//
// Use a `BorrowedSocket` instead of e.g. `std::net::UdpSocket`. The latter
// would close the file descriptor on `Drop`. The lifetime of the underlying
// OS socket is managed not by `neqo_glue` but `NSPR`.
socket: Option<neqo_udp::Socket<BorrowedSocket>>,
}
// Opaque interface to mozilla::net::NetAddr defined in DNS.h
@ -104,7 +114,18 @@ type SendFunc = extern "C" fn(
type SetTimerFunc = extern "C" fn(context: *mut c_void, timeout: u64);
#[cfg(unix)]
type BorrowedSocket = std::os::fd::BorrowedFd<'static>;
#[cfg(windows)]
type BorrowedSocket = std::os::windows::io::BorrowedSocket<'static>;
impl NeqoHttp3Conn {
/// Create a new [`NeqoHttp3Conn`].
///
/// Note that [`NeqoHttp3Conn`] works under the assumption that the UDP
/// socket of the connection, i.e. the one provided to
/// [`NeqoHttp3Conn::new`], does not change throughout the lifetime of
/// [`NeqoHttp3Conn`].
fn new(
origin: &nsACString,
alpn: &nsACString,
@ -120,10 +141,46 @@ impl NeqoHttp3Conn {
webtransport_datagram_size: u32,
max_accumlated_time_ms: u32,
provider_flags: u32,
socket: Option<i64>,
) -> Result<RefPtr<NeqoHttp3Conn>, nsresult> {
// Nss init.
init().map_err(|_| NS_ERROR_UNEXPECTED)?;
let socket = socket
.map(|socket| {
#[cfg(unix)]
let borrowed = {
use std::os::fd::{BorrowedFd, RawFd};
if socket == -1 {
qerror!("got invalid socked {}", socket);
return Err(NS_ERROR_INVALID_ARG);
}
let raw: RawFd = socket.try_into().map_err(|e| {
qerror!("got invalid socked {}: {}", socket, e);
NS_ERROR_INVALID_ARG
})?;
unsafe { BorrowedFd::borrow_raw(raw) }
};
#[cfg(windows)]
let borrowed = {
use std::os::windows::io::{BorrowedSocket, RawSocket};
if socket as usize == winapi::um::winsock2::INVALID_SOCKET {
qerror!("got invalid socked {}", socket);
return Err(NS_ERROR_INVALID_ARG);
}
let raw: RawSocket = socket.try_into().map_err(|e| {
qerror!("got invalid socked {}: {}", socket, e);
NS_ERROR_INVALID_ARG
})?;
unsafe { BorrowedSocket::borrow_raw(raw) }
};
neqo_udp::Socket::new(borrowed).map_err(|e| {
qerror!("failed to initialize socket {}: {}", socket, e);
NS_ERROR_FAILURE
})
})
.transpose()?;
let origin_conv = str::from_utf8(origin).map_err(|_| NS_ERROR_INVALID_ARG)?;
let alpn_conv = str::from_utf8(alpn).map_err(|_| NS_ERROR_INVALID_ARG)?;
@ -263,6 +320,7 @@ impl NeqoHttp3Conn {
refcnt: unsafe { AtomicRefcnt::new() },
last_output_time: Instant::now(),
max_accumlated_time: Duration::from_millis(max_accumlated_time_ms.into()),
socket,
}));
unsafe { Ok(RefPtr::from_raw(conn).unwrap()) }
}
@ -308,6 +366,7 @@ pub extern "C" fn neqo_http3conn_new(
webtransport_datagram_size: u32,
max_accumlated_time_ms: u32,
provider_flags: u32,
socket: i64,
result: &mut *const NeqoHttp3Conn,
) -> nsresult {
*result = ptr::null_mut();
@ -327,6 +386,53 @@ pub extern "C" fn neqo_http3conn_new(
webtransport_datagram_size,
max_accumlated_time_ms,
provider_flags,
Some(socket),
) {
Ok(http3_conn) => {
http3_conn.forget(result);
NS_OK
}
Err(e) => e,
}
}
// Allocate a new NeqoHttp3Conn object using NSPR for IO.
#[no_mangle]
pub extern "C" fn neqo_http3conn_new_use_nspr_for_io(
origin: &nsACString,
alpn: &nsACString,
local_addr: *const NetAddr,
remote_addr: *const NetAddr,
max_table_size: u64,
max_blocked_streams: u16,
max_data: u64,
max_stream_data: u64,
version_negotiation: bool,
webtransport: bool,
qlog_dir: &nsACString,
webtransport_datagram_size: u32,
max_accumlated_time_ms: u32,
provider_flags: u32,
result: &mut *const NeqoHttp3Conn,
) -> nsresult {
*result = ptr::null_mut();
match NeqoHttp3Conn::new(
origin,
alpn,
local_addr,
remote_addr,
max_table_size,
max_blocked_streams,
max_data,
max_stream_data,
version_negotiation,
webtransport,
qlog_dir,
webtransport_datagram_size,
max_accumlated_time_ms,
provider_flags,
None,
) {
Ok(http3_conn) => {
http3_conn.forget(result);
@ -340,11 +446,13 @@ pub extern "C" fn neqo_http3conn_new(
* packet holds packet data.
*/
#[no_mangle]
pub unsafe extern "C" fn neqo_http3conn_process_input(
pub unsafe extern "C" fn neqo_http3conn_process_input_use_nspr_for_io(
conn: &mut NeqoHttp3Conn,
remote_addr: *const NetAddr,
packet: *const ThinVec<u8>,
) -> nsresult {
assert!(conn.socket.is_none(), "NSPR IO path");
let remote = match netaddr_to_socket_addr(remote_addr) {
Ok(addr) => addr,
Err(result) => return result,
@ -360,13 +468,67 @@ pub unsafe extern "C" fn neqo_http3conn_process_input(
return NS_OK;
}
#[repr(C)]
pub struct ProcessInputResult {
pub result: nsresult,
pub bytes_read: u32,
}
/// Process input, reading incoming datagrams from the socket and passing them
/// to the Neqo state machine.
#[no_mangle]
pub extern "C" fn neqo_http3conn_process_output_and_send(
pub unsafe extern "C" fn neqo_http3conn_process_input(
conn: &mut NeqoHttp3Conn,
) -> ProcessInputResult {
let mut bytes_read = 0;
loop {
let mut dgrams = match conn
.socket
.as_mut()
.expect("non NSPR IO")
.recv(&conn.local_addr)
{
Ok(dgrams) => dgrams,
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
break;
}
Err(e) => {
qwarn!("failed to receive datagrams: {}", e);
return ProcessInputResult {
result: NS_ERROR_FAILURE,
bytes_read: 0,
};
}
};
if dgrams.is_empty() {
break;
}
bytes_read += dgrams.iter().map(|d| d.len()).sum::<usize>();
// ECN support will be introduced with
// https://bugzilla.mozilla.org/show_bug.cgi?id=1902065.
for dgram in &mut dgrams {
dgram.set_tos(Default::default());
}
conn.conn
.process_multiple_input(dgrams.iter(), Instant::now());
}
return ProcessInputResult {
result: NS_OK,
bytes_read: bytes_read.try_into().unwrap_or(u32::MAX),
};
}
#[no_mangle]
pub extern "C" fn neqo_http3conn_process_output_and_send_use_nspr_for_io(
conn: &mut NeqoHttp3Conn,
context: *mut c_void,
send_func: SendFunc,
set_timer_func: SetTimerFunc,
) -> nsresult {
assert!(conn.socket.is_none(), "NSPR IO path");
let now = Instant::now();
if conn.last_output_time > now {
// The timer fired too early, so reschedule it.
@ -436,6 +598,105 @@ pub extern "C" fn neqo_http3conn_process_output_and_send(
NS_OK
}
#[repr(C)]
pub struct ProcessOutputAndSendResult {
pub result: nsresult,
pub bytes_written: u32,
}
/// Process output, retrieving outgoing datagrams from the Neqo state machine
/// and writing them to the socket.
#[no_mangle]
pub extern "C" fn neqo_http3conn_process_output_and_send(
conn: &mut NeqoHttp3Conn,
context: *mut c_void,
set_timer_func: SetTimerFunc,
) -> ProcessOutputAndSendResult {
let now = Instant::now();
if conn.last_output_time > now {
// The timer fired too early, so reschedule it.
// The 1ms of extra delay is not ideal, but this is a fail
set_timer_func(
context,
u64::try_from((conn.last_output_time - now + conn.max_accumlated_time).as_millis())
.unwrap(),
);
return ProcessOutputAndSendResult {
result: NS_OK,
bytes_written: 0,
};
}
let mut accumulated_time = Duration::from_nanos(0);
let mut bytes_written: usize = 0;
loop {
conn.last_output_time = if accumulated_time.is_zero() {
Instant::now()
} else {
now + accumulated_time
};
match conn.conn.process_output(conn.last_output_time) {
Output::Datagram(mut dg) => {
// ECN support will be introduced with
// https://bugzilla.mozilla.org/show_bug.cgi?id=1902065.
dg.set_tos(Default::default());
if static_prefs::pref!("network.http.http3.block_loopback_ipv6_addr")
&& matches!(dg.destination(), SocketAddr::V6(addr) if addr.ip().is_loopback())
{
qdebug!("network.http.http3.block_loopback_ipv6_addr is set, returning NS_ERROR_CONNECTION_REFUSED for localhost IPv6");
return ProcessOutputAndSendResult {
result: NS_ERROR_CONNECTION_REFUSED,
bytes_written: 0,
};
}
match conn.socket.as_mut().expect("non NSPR IO").send(&dg) {
Ok(()) => {}
Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
qwarn!("dropping datagram as socket would block");
break;
}
Err(e) => {
qwarn!("failed to send datagram: {}", e);
return ProcessOutputAndSendResult {
result: NS_ERROR_FAILURE,
bytes_written: 0,
};
}
}
bytes_written += dg.len();
}
Output::Callback(to) => {
if to.is_zero() {
set_timer_func(context, 1);
break;
}
let timeout = min(to, Duration::from_nanos(u64::MAX - 1));
accumulated_time += timeout;
if accumulated_time >= conn.max_accumlated_time {
let mut timeout = accumulated_time.as_millis() as u64;
if timeout == 0 {
timeout = 1;
}
set_timer_func(context, timeout);
break;
}
}
Output::None => {
set_timer_func(context, std::u64::MAX);
break;
}
}
}
return ProcessOutputAndSendResult {
result: NS_OK,
bytes_written: bytes_written.try_into().unwrap_or(u32::MAX),
};
}
#[no_mangle]
pub extern "C" fn neqo_http3conn_close(conn: &mut NeqoHttp3Conn, error: u64) {
conn.conn.close(

6
supply-chain/audits.toml
View File

@ -3596,6 +3596,12 @@ who = "Kershaw Chang <kershaw@mozilla.com>"
criteria = "safe-to-run"
version = "0.5.0"
[[audits.quinn-udp]]
who = "Max Inden <mail@max-inden.de>"
criteria = "safe-to-deploy"
version = "0.5.4"
notes = "This is a small crate, providing safe wrappers around various low-level networking specific operating system features. Given that the Rust standard library does not provide safe wrappers for these low-level features, safe wrappers need to be build in the crate itself, i.e. `quinn-udp`, thus requiring `unsafe` code."
[[audits.quote]]
who = "Nika Layzell <nika@thelayzells.com>"
criteria = "safe-to-deploy"

2
third_party/rust/quinn-udp/.cargo-checksum.json vendored
View File

@ -1 +1 @@
{"files":{"Cargo.toml":"d0dcdff68dab18eb8770515e182ff2497d8cfa68e70b0633ab51bf2f96ac1dba","LICENSE-APACHE":"c71d239df91726fc519c6eb72d318ec65820627232b2f796219e87dcf35d0ab4","LICENSE-MIT":"4b2d0aca6789fa39e03d6738e869ea0988cceba210ca34ebb59c15c463e93a04","src/cmsg/mod.rs":"c5b3ffc33d05383894bb1aecfd6dce9f85162104b5158a2106ae5b9a13573926","src/cmsg/unix.rs":"138cd32f0861e81555e5da6e47de852594bb02c0b1b3ab7e4759dd51fdbfa80d","src/cmsg/windows.rs":"6fb936ec4a283efc5796872e777441e3039c40589073865644a8ef7936af4f4b","src/fallback.rs":"7fe9666b0bf508d1b5ec0b3690bb7add94c8f213cb51a263c9959e22a5094ad0","src/lib.rs":"f3abbcd52754786ea3a0fb7398253cee1cde952937c318a4a0471e9cda63b753","src/unix.rs":"ebf6a21859bc185b0850ba7b4348991f301c8bf15649f4794b7e1afd1c30df75","src/windows.rs":"7e89b0c8808a422dcbe2c190f0d357e1cd717c6f1e94742d64627f3cd7b8571b","tests/tests.rs":"8a4f9d4a17e12d4f88bfe729a500e91e065dcb9347052a8db0f86735a6639138"},"package":"cb7ad7bc932e4968523fa7d9c320ee135ff779de720e9350fee8728838551764"}
{"files":{"Cargo.toml":"bf505df0c4f9254fa37950bad863cb838a8a7d2be4c8d3f28bdd679f945ef8cf","LICENSE-APACHE":"c71d239df91726fc519c6eb72d318ec65820627232b2f796219e87dcf35d0ab4","LICENSE-MIT":"4b2d0aca6789fa39e03d6738e869ea0988cceba210ca34ebb59c15c463e93a04","benches/throughput.rs":"8f00856e6c6f1dd8c7dd6c8a22b36c6f42dfa4d709edf3348de75d32e22c71fb","src/cmsg/mod.rs":"23d898d72c5aabda93d987526fdd78231bb5907bce2b6b2d292a56bdfd977f86","src/cmsg/unix.rs":"1a4089e5e61536a3c370c6b1bc891036ec2d0e2e78105fbb5b8227705e905d34","src/cmsg/windows.rs":"6fb936ec4a283efc5796872e777441e3039c40589073865644a8ef7936af4f4b","src/fallback.rs":"7fe9666b0bf508d1b5ec0b3690bb7add94c8f213cb51a263c9959e22a5094ad0","src/lib.rs":"72be7f797a3a11e452e7764fadadebc43ae7f9c14ba7fa80aedbbee71aa889c7","src/unix.rs":"fbc9a6ab281cc66500e6afa8b9ebdee73ca281ca14732e8076d9b1f10f431de7","src/windows.rs":"e741a7bdd86d7fcb856db855f9308af01e69387c00e6a726d322f1f4d3046b74","tests/tests.rs":"51bcf6d3f1a3fcf7d481ae966eb679f88341886ff4277f5747df3340ed709d09"},"package":"8bffec3605b73c6f1754535084a85229fa8a30f86014e6c81aeec4abb68b0285"}

24
third_party/rust/quinn-udp/Cargo.toml vendored
View File

@ -13,7 +13,7 @@
edition = "2021"
rust-version = "1.66"
name = "quinn-udp"
version = "0.5.0"
version = "0.5.4"
description = "UDP sockets with ECN information for the QUIC transport protocol"
keywords = ["quic"]
categories = [
@ -29,21 +29,37 @@ all-features = true
[dependencies.libc]
version = "0.2.113"
[dependencies.log]
version = "0.4"
optional = true
[dependencies.socket2]
version = "0.5"
[dependencies.tracing]
version = "0.1.10"
features = ["std"]
optional = true
default-features = false
[dev-dependencies.criterion]
version = "0.5"
[features]
default = ["log"]
default = [
"tracing",
"log",
]
direct-log = ["dep:log"]
log = ["tracing/log"]
[target."cfg(any(target_os = \"linux\", target_os = \"windows\"))"]
[target."cfg(windows)".dependencies.once_cell]
version = "1.19.0"
version = "1.19"
[target."cfg(windows)".dependencies.windows-sys]
version = "0.52.0"
version = "0.52"
features = [
"Win32_Foundation",
"Win32_System_IO",

75
third_party/rust/quinn-udp/benches/throughput.rs vendored Normal file
View File

@ -0,0 +1,75 @@
use criterion::{criterion_group, criterion_main, Criterion};
use quinn_udp::{RecvMeta, Transmit, UdpSocketState};
use std::cmp::min;
use std::{io::IoSliceMut, net::UdpSocket, slice};
pub fn criterion_benchmark(c: &mut Criterion) {
const TOTAL_BYTES: usize = 10 * 1024 * 1024;
// Maximum GSO buffer size is 64k.
const MAX_BUFFER_SIZE: usize = u16::MAX as usize;
const SEGMENT_SIZE: usize = 1280;
let send = UdpSocket::bind("[::1]:0")
.or_else(|_| UdpSocket::bind("127.0.0.1:0"))
.unwrap();
let recv = UdpSocket::bind("[::1]:0")
.or_else(|_| UdpSocket::bind("127.0.0.1:0"))
.unwrap();
let max_segments = min(
UdpSocketState::new((&send).into())
.unwrap()
.max_gso_segments(),
MAX_BUFFER_SIZE / SEGMENT_SIZE,
);
let dst_addr = recv.local_addr().unwrap();
let send_state = UdpSocketState::new((&send).into()).unwrap();
let recv_state = UdpSocketState::new((&recv).into()).unwrap();
// Reverse non-blocking flag set by `UdpSocketState` to make the test non-racy
recv.set_nonblocking(false).unwrap();
let mut receive_buffer = vec![0; MAX_BUFFER_SIZE];
let mut meta = RecvMeta::default();
for gso_enabled in [false, true] {
let mut group = c.benchmark_group(format!("gso_{}", gso_enabled));
group.throughput(criterion::Throughput::Bytes(TOTAL_BYTES as u64));
let segments = if gso_enabled { max_segments } else { 1 };
let msg = vec![0xAB; SEGMENT_SIZE * segments];
let transmit = Transmit {
destination: dst_addr,
ecn: None,
contents: &msg,
segment_size: gso_enabled.then_some(SEGMENT_SIZE),
src_ip: None,
};
group.bench_function("throughput", |b| {
b.iter(|| {
let mut sent: usize = 0;
while sent < TOTAL_BYTES {
send_state.send((&send).into(), &transmit).unwrap();
sent += transmit.contents.len();
let mut received_segments = 0;
while received_segments < segments {
let n = recv_state
.recv(
(&recv).into(),
&mut [IoSliceMut::new(&mut receive_buffer)],
slice::from_mut(&mut meta),
)
.unwrap();
assert_eq!(n, 1);
received_segments += meta.len / meta.stride;
}
assert_eq!(received_segments, segments);
}
})
});
}
}
criterion_group!(benches, criterion_benchmark);
criterion_main!(benches);

4
third_party/rust/quinn-udp/src/cmsg/mod.rs vendored
View File

@ -124,6 +124,10 @@ pub(crate) trait MsgHdr {
fn cmsg_nxt_hdr(&self, cmsg: &Self::ControlMessage) -> *mut Self::ControlMessage;
/// Sets the number of control messages added to this `struct msghdr`.
///
/// Note that this is a destructive operation and should only be done as a finalisation
/// step.
fn set_control_len(&mut self, len: usize);
fn control_len(&self) -> usize;

5
third_party/rust/quinn-udp/src/cmsg/unix.rs vendored
View File

@ -20,6 +20,11 @@ impl MsgHdr for libc::msghdr {
fn set_control_len(&mut self, len: usize) {
self.msg_controllen = len as _;
if len == 0 {
// netbsd is particular about this being a NULL pointer if there are no control
// messages.
self.msg_control = std::ptr::null_mut();
}
}
fn control_len(&self) -> usize {

15
third_party/rust/quinn-udp/src/lib.rs vendored
View File

@ -37,6 +37,9 @@ use std::{
time::{Duration, Instant},
};
#[cfg(all(feature = "direct-log", not(feature = "tracing")))]
use log::warn;
#[cfg(feature = "tracing")]
use tracing::warn;
#[cfg(any(unix, windows))]
@ -84,6 +87,9 @@ pub struct RecvMeta {
/// The Explicit Congestion Notification bits for the datagram(s) in the buffer
pub ecn: Option<EcnCodepoint>,
/// The destination IP address which was encoded in this datagram
///
/// Populated on platforms: Windows, Linux, Android, FreeBSD, OpenBSD, NetBSD, macOS,
/// and iOS.
pub dst_ip: Option<IpAddr>,
}
@ -123,6 +129,7 @@ const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60);
///
/// Logging will only be performed if at least [`IO_ERROR_LOG_INTERVAL`]
/// has elapsed since the last error was logged.
#[cfg(any(feature = "tracing", feature = "direct-log"))]
fn log_sendmsg_error(
last_send_error: &Mutex<Instant>,
err: impl core::fmt::Debug,
@ -138,10 +145,14 @@ fn log_sendmsg_error(
}
}
// No-op
#[cfg(not(any(feature = "tracing", feature = "direct-log")))]
fn log_sendmsg_error(_: &Mutex<Instant>, _: impl core::fmt::Debug, _: &Transmit) {}
/// A borrowed UDP socket
///
/// On Unix, constructible via `From<T: AsRawFd>`. On Windows, constructible via `From<T:
/// AsRawSocket>`.
/// On Unix, constructible via `From<T: AsFd>`. On Windows, constructible via `From<T:
/// AsSocket>`.
// Wrapper around socket2 to avoid making it a public dependency and incurring stability risk
pub struct UdpSockRef<'a>(socket2::SockRef<'a>);

122
third_party/rust/quinn-udp/src/unix.rs vendored
View File

@ -1,4 +1,4 @@
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "openbsd")))]
use std::ptr;
use std::{
io::{self, IoSliceMut},
@ -12,15 +12,27 @@ use std::{
time::Instant,
};
#[cfg(all(feature = "direct-log", not(feature = "tracing")))]
use log::{debug, error};
use socket2::SockRef;
#[cfg(feature = "tracing")]
use tracing::{debug, error};
use super::{
cmsg, log_sendmsg_error, EcnCodepoint, RecvMeta, Transmit, UdpSockRef, IO_ERROR_LOG_INTERVAL,
};
// Defined in netinet6/in6.h on OpenBSD, this is not yet exported by the libc crate
// directly. See https://github.com/rust-lang/libc/issues/3704 for when we might be able to
// rely on this from the libc crate.
#[cfg(any(target_os = "openbsd", target_os = "netbsd"))]
const IPV6_DONTFRAG: libc::c_int = 62;
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
const IPV6_DONTFRAG: libc::c_int = libc::IPV6_DONTFRAG;
#[cfg(target_os = "freebsd")]
type IpTosTy = libc::c_uchar;
#[cfg(not(target_os = "freebsd"))]
#[cfg(not(any(target_os = "freebsd", target_os = "netbsd")))]
type IpTosTy = libc::c_int;
/// Tokio-compatible UDP socket with some useful specializations.
@ -34,7 +46,7 @@ pub struct UdpSocketState {
gro_segments: usize,
may_fragment: bool,
/// True if we have received EINVAL error from `sendmsg` or `sendmmsg` system call at least once.
/// True if we have received EINVAL error from `sendmsg` system call at least once.
///
/// If enabled, we assume that old kernel is used and switch to fallback mode.
/// In particular, we do not use IP_TOS cmsg_type in this case,
@ -48,6 +60,8 @@ impl UdpSocketState {
let mut cmsg_platform_space = 0;
if cfg!(target_os = "linux")
|| cfg!(target_os = "freebsd")
|| cfg!(target_os = "openbsd")
|| cfg!(target_os = "netbsd")
|| cfg!(target_os = "macos")
|| cfg!(target_os = "ios")
|| cfg!(target_os = "android")
@ -73,10 +87,13 @@ impl UdpSocketState {
// mac and ios do not support IP_RECVTOS on dual-stack sockets :(
// older macos versions also don't have the flag and will error out if we don't ignore it
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
if is_ipv4 || !io.only_v6()? {
#[allow(unused_variables)]
if let Err(err) = set_socket_option(&*io, libc::IPPROTO_IP, libc::IP_RECVTOS, OPTION_ON)
{
tracing::debug!("Ignoring error setting IP_RECVTOS on socket: {err:?}",);
#[cfg(any(feature = "tracing", feature = "direct-log"))]
debug!("Ignoring error setting IP_RECVTOS on socket: {err:?}");
}
}
@ -120,7 +137,13 @@ impl UdpSocketState {
)?;
}
}
#[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "ios"))]
#[cfg(any(
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "macos",
target_os = "ios"
))]
// IP_RECVDSTADDR == IP_SENDSRCADDR on FreeBSD
// macOS uses only IP_RECVDSTADDR, no IP_SENDSRCADDR on macOS
// macOS also supports IP_PKTINFO
@ -138,12 +161,8 @@ impl UdpSocketState {
// kernel's path MTU guess, but actually disabling fragmentation requires this too. See
// __ip6_append_data in ip6_output.c.
// Set `may_fragment` to `true` if this option is not supported on the platform.
may_fragment |= !set_socket_option_supported(
&*io,
libc::IPPROTO_IPV6,
libc::IPV6_DONTFRAG,
OPTION_ON,
)?;
may_fragment |=
!set_socket_option_supported(&*io, libc::IPPROTO_IPV6, IPV6_DONTFRAG, OPTION_ON)?;
}
let now = Instant::now();
@ -196,19 +215,29 @@ impl UdpSocketState {
self.may_fragment
}
/// Returns true if we previously got an EINVAL error from `sendmsg` or `sendmmsg` syscall.
/// Returns true if we previously got an EINVAL error from `sendmsg` syscall.
fn sendmsg_einval(&self) -> bool {
self.sendmsg_einval.load(Ordering::Relaxed)
}
/// Sets the flag indicating we got EINVAL error from `sendmsg` or `sendmmsg` syscall.
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
/// Sets the flag indicating we got EINVAL error from `sendmsg` syscall.
#[cfg(not(any(
target_os = "macos",
target_os = "ios",
target_os = "openbsd",
target_os = "netbsd"
)))]
fn set_sendmsg_einval(&self) {
self.sendmsg_einval.store(true, Ordering::Relaxed)
}
}
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
#[cfg(not(any(
target_os = "macos",
target_os = "ios",
target_os = "openbsd",
target_os = "netbsd"
)))]
fn send(
#[allow(unused_variables)] // only used on Linux
state: &UdpSocketState,
@ -260,7 +289,8 @@ fn send(
// Prevent new transmits from being scheduled using GSO. Existing GSO transmits
// may already be in the pipeline, so we need to tolerate additional failures.
if state.max_gso_segments() > 1 {
tracing::error!("got transmit error, halting segmentation offload");
#[cfg(any(feature = "tracing", feature = "direct-log"))]
error!("got transmit error, halting segmentation offload");
state
.max_gso_segments
.store(1, std::sync::atomic::Ordering::Relaxed);
@ -293,7 +323,12 @@ fn send(
}
}
#[cfg(any(target_os = "macos", target_os = "ios"))]
#[cfg(any(
target_os = "macos",
target_os = "ios",
target_os = "openbsd",
target_os = "netbsd"
))]
fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io::Result<()> {
let mut hdr: libc::msghdr = unsafe { mem::zeroed() };
let mut iov: libc::iovec = unsafe { mem::zeroed() };
@ -305,8 +340,10 @@ fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io:
&mut hdr,
&mut iov,
&mut ctrl,
// Only tested on macOS and iOS
cfg!(target_os = "macos") || cfg!(target_os = "ios"),
cfg!(target_os = "macos")
|| cfg!(target_os = "ios")
|| cfg!(target_os = "openbsd")
|| cfg!(target_os = "netbsd"),
state.sendmsg_einval(),
);
let n = unsafe { libc::sendmsg(io.as_raw_fd(), &hdr, 0) };
@ -335,7 +372,7 @@ fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io:
Ok(())
}
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "openbsd")))]
fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result<usize> {
let mut names = [MaybeUninit::<libc::sockaddr_storage>::uninit(); BATCH_SIZE];
let mut ctrls = [cmsg::Aligned(MaybeUninit::<[u8; CMSG_LEN]>::uninit()); BATCH_SIZE];
@ -372,7 +409,7 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) ->
Ok(msg_count as usize)
}
#[cfg(any(target_os = "macos", target_os = "ios"))]
#[cfg(any(target_os = "macos", target_os = "ios", target_os = "openbsd"))]
fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result<usize> {
let mut name = MaybeUninit::<libc::sockaddr_storage>::uninit();
let mut ctrl = cmsg::Aligned(MaybeUninit::<[u8; CMSG_LEN]>::uninit());
@ -401,7 +438,7 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) ->
///
/// It uses [`libc::syscall`] instead of [`libc::recvmmsg`]
/// to avoid linking error on systems where libc does not contain `recvmmsg`.
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "openbsd")))]
unsafe fn recvmmsg_with_fallback(
sockfd: libc::c_int,
msgvec: *mut libc::mmsghdr,
@ -410,7 +447,7 @@ unsafe fn recvmmsg_with_fallback(
let flags = 0;
let timeout = ptr::null_mut::<libc::timespec>();
#[cfg(not(target_os = "freebsd"))]
#[cfg(not(any(target_os = "freebsd", target_os = "netbsd")))]
{
let ret =
libc::syscall(libc::SYS_recvmmsg, sockfd, msgvec, vlen, flags, timeout) as libc::c_int;
@ -419,11 +456,13 @@ unsafe fn recvmmsg_with_fallback(
}
}
// libc on FreeBSD implements `recvmmsg` as a high-level abstraction over `recvmsg`,
// thus `SYS_recvmmsg` constant and direct system call do not exist
#[cfg(target_os = "freebsd")]
// libc on FreeBSD and NetBSD implement `recvmmsg` as a high-level abstraction over
// `recvmsg`, thus `SYS_recvmmsg` constant and direct system call do not exist
#[cfg(any(target_os = "freebsd", target_os = "netbsd"))]
{
let ret = libc::recvmmsg(sockfd, msgvec, vlen as usize, flags, timeout) as libc::c_int;
#[cfg(target_os = "freebsd")]
let vlen = vlen as usize;
let ret = libc::recvmmsg(sockfd, msgvec, vlen, flags, timeout) as libc::c_int;
if ret != -1 {
return ret;
}
@ -442,7 +481,7 @@ unsafe fn recvmmsg_with_fallback(
/// Fallback implementation of `recvmmsg` using `recvmsg`
/// for systems which do not support `recvmmsg`
/// such as Linux <2.6.33.
#[cfg(not(any(target_os = "macos", target_os = "ios")))]
#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "openbsd")))]
unsafe fn recvmmsg_fallback(
sockfd: libc::c_int,
msgvec: *mut libc::mmsghdr,
@ -500,7 +539,10 @@ fn prepare_msg(
|| matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some());
if is_ipv4 {
if !sendmsg_einval {
encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy);
#[cfg(not(target_os = "netbsd"))]
{
encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy);
}
}
} else {
encoder.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn);
@ -524,7 +566,13 @@ fn prepare_msg(
};
encoder.push(libc::IPPROTO_IP, libc::IP_PKTINFO, pktinfo);
}
#[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "ios"))]
#[cfg(any(
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "macos",
target_os = "ios",
))]
{
if encode_src_ip {
let addr = libc::in_addr {
@ -578,8 +626,12 @@ fn decode_recv(
let cmsg_iter = unsafe { cmsg::Iter::new(hdr) };
for cmsg in cmsg_iter {
match (cmsg.cmsg_level, cmsg.cmsg_type) {
(libc::IPPROTO_IP, libc::IP_TOS) => unsafe {
ecn_bits = cmsg::decode::<u8, libc::cmsghdr>(cmsg);
},
// FreeBSD uses IP_RECVTOS here, and we can be liberal because cmsgs are opt-in.
(libc::IPPROTO_IP, libc::IP_TOS) | (libc::IPPROTO_IP, libc::IP_RECVTOS) => unsafe {
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
(libc::IPPROTO_IP, libc::IP_RECVTOS) => unsafe {
ecn_bits = cmsg::decode::<u8, libc::cmsghdr>(cmsg);
},
(libc::IPPROTO_IPV6, libc::IPV6_TCLASS) => unsafe {
@ -601,7 +653,13 @@ fn decode_recv(
pktinfo.ipi_addr.s_addr.to_ne_bytes(),
)));
}
#[cfg(any(target_os = "freebsd", target_os = "macos", target_os = "ios"))]
#[cfg(any(
target_os = "freebsd",
target_os = "openbsd",
target_os = "netbsd",
target_os = "macos",
target_os = "ios",
))]
(libc::IPPROTO_IP, libc::IP_RECVDSTADDR) => {
let in_addr = unsafe { cmsg::decode::<libc::in_addr, libc::cmsghdr>(cmsg) };
dst_ip = Some(IpAddr::V4(Ipv4Addr::from(in_addr.s_addr.to_ne_bytes())));

16
third_party/rust/quinn-udp/src/windows.rs vendored
View File

@ -9,7 +9,11 @@ use std::{
};
use libc::{c_int, c_uint};
#[cfg(all(feature = "direct-log", not(feature = "tracing")))]
use log::{debug, error};
use once_cell::sync::Lazy;
#[cfg(feature = "tracing")]
use tracing::{debug, error};
use windows_sys::Win32::Networking::WinSock;
use crate::{
@ -60,7 +64,8 @@ impl UdpSocketState {
// We don't support old versions of Windows that do not enable access to `WSARecvMsg()`
if WSARECVMSG_PTR.is_none() {
tracing::error!("network stack does not support WSARecvMsg function");
#[cfg(any(feature = "tracing", feature = "direct-log"))]
error!("network stack does not support WSARecvMsg function");
return Err(io::Error::from(io::ErrorKind::Unsupported));
}
@ -387,7 +392,8 @@ const OPTION_ON: u32 = 1;
static WSARECVMSG_PTR: Lazy<WinSock::LPFN_WSARECVMSG> = Lazy::new(|| {
let s = unsafe { WinSock::socket(WinSock::AF_INET as _, WinSock::SOCK_DGRAM as _, 0) };
if s == WinSock::INVALID_SOCKET {
tracing::debug!(
#[cfg(any(feature = "tracing", feature = "direct-log"))]
debug!(
"ignoring WSARecvMsg function pointer due to socket creation error: {}",
io::Error::last_os_error()
);
@ -416,12 +422,14 @@ static WSARECVMSG_PTR: Lazy<WinSock::LPFN_WSARECVMSG> = Lazy::new(|| {
};
if rc == -1 {
tracing::debug!(
#[cfg(any(feature = "tracing", feature = "direct-log"))]
debug!(
"ignoring WSARecvMsg function pointer due to ioctl error: {}",
io::Error::last_os_error()
);
} else if len as usize != mem::size_of::<WinSock::LPFN_WSARECVMSG>() {
tracing::debug!("ignoring WSARecvMsg function pointer due to pointer size mismatch");
#[cfg(any(feature = "tracing", feature = "direct-log"))]
debug!("ignoring WSARecvMsg function pointer due to pointer size mismatch");
wsa_recvmsg_ptr = None;
}

64
third_party/rust/quinn-udp/tests/tests.rs vendored
View File

@ -1,6 +1,8 @@
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6};
use std::{
io::IoSliceMut,
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6, UdpSocket},
net::{IpAddr, Ipv4Addr, Ipv6Addr, UdpSocket},
slice,
};
@ -31,6 +33,46 @@ fn basic() {
#[test]
fn ecn_v6() {
let send = Socket::from(UdpSocket::bind("[::1]:0").unwrap());
let recv = Socket::from(UdpSocket::bind("[::1]:0").unwrap());
for codepoint in [EcnCodepoint::Ect0, EcnCodepoint::Ect1] {
test_send_recv(
&send,
&recv,
Transmit {
destination: recv.local_addr().unwrap().as_socket().unwrap(),
ecn: Some(codepoint),
contents: b"hello",
segment_size: None,
src_ip: None,
},
);
}
}
#[test]
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
fn ecn_v4() {
let send = Socket::from(UdpSocket::bind("127.0.0.1:0").unwrap());
let recv = Socket::from(UdpSocket::bind("127.0.0.1:0").unwrap());
for codepoint in [EcnCodepoint::Ect0, EcnCodepoint::Ect1] {
test_send_recv(
&send,
&recv,
Transmit {
destination: recv.local_addr().unwrap().as_socket().unwrap(),
ecn: Some(codepoint),
contents: b"hello",
segment_size: None,
src_ip: None,
},
);
}
}
#[test]
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
fn ecn_v6_dualstack() {
let recv = socket2::Socket::new(
socket2::Domain::IPV6,
socket2::Type::DGRAM,
@ -72,25 +114,7 @@ fn ecn_v6() {
}
#[test]
fn ecn_v4() {
let send = Socket::from(UdpSocket::bind("127.0.0.1:0").unwrap());
let recv = Socket::from(UdpSocket::bind("127.0.0.1:0").unwrap());
for codepoint in [EcnCodepoint::Ect0, EcnCodepoint::Ect1] {
test_send_recv(
&send,
&recv,
Transmit {
destination: recv.local_addr().unwrap().as_socket().unwrap(),
ecn: Some(codepoint),
contents: b"hello",
segment_size: None,
src_ip: None,
},
);
}
}
#[test]
#[cfg(not(any(target_os = "openbsd", target_os = "netbsd")))]
fn ecn_v4_mapped_v6() {
let send = socket2::Socket::new(
socket2::Domain::IPV6,