Skip to content

Add basic windows support #61

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
coco-speed wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Add basic windows support #61

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions pingora-core/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@ http = { workspace = true }
log = { workspace = true }
h2 = { workspace = true }
lru = { workspace = true }
nix = "~0.24.3"
clap = { version = "3.2.25", features = ["derive"] }
once_cell = { workspace = true }
serde = { version = "1.0", features = ["derive"] }
Expand All @@ -46,7 +45,6 @@ libc = "0.2.70"
chrono = { version = "~0.4.31", features = ["alloc"], default-features = false }
thread_local = "1.0"
prometheus = "0.13"
daemonize = "0.5.0"
sentry = { version = "0.26", features = [
"backtrace",
"contexts",
Expand All @@ -69,16 +67,25 @@ tokio-test = "0.4"
zstd = "0"
httpdate = "1"

[target.'cfg(unix)'.dependencies]
daemonize = "0.5.0"
nix = "~0.24.3"

[target.'cfg(windows)'.dependencies]
windows-sys = { version = "0.59.0", features = ["Win32_Networking_WinSock"] }

[target.'cfg(unix)'.dev-dependencies]
hyperlocal = "0.8"
jemallocator = "0.5"

[dev-dependencies]
matches = "0.1"
env_logger = "0.9"
reqwest = { version = "0.11", features = ["rustls"], default-features = false }
hyperlocal = "0.8"
hyper = "0.14"
jemallocator = "0.5"

[features]
default = ["openssl"]
openssl = ["pingora-openssl"]
boringssl = ["pingora-boringssl"]
patched_http1 = []
patched_http1 = []
10 changes: 5 additions & 5 deletions pingora-core/src/connectors/http/v2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ use super::HttpSession;
use crate::connectors::{ConnectorOptions, TransportConnector};
use crate::protocols::http::v1::client::HttpSession as Http1Session;
use crate::protocols::http::v2::client::{drive_connection, Http2Session};
use crate::protocols::{Digest, Stream};
use crate::protocols::{Digest, Stream, UniqueIDType};
use crate::upstreams::peer::{Peer, ALPN};

use bytes::Bytes;
Expand Down Expand Up @@ -47,7 +47,7 @@ pub(crate) struct ConnectionRefInner {
connection_stub: Stub,
closed: watch::Receiver<bool>,
ping_timeout_occurred: Arc<AtomicBool>,
id: i32,
id: UniqueIDType,
// max concurrent streams this connection is allowed to create
max_streams: usize,
// how many concurrent streams already active
Expand All @@ -69,7 +69,7 @@ impl ConnectionRef {
send_req: SendRequest<Bytes>,
closed: watch::Receiver<bool>,
ping_timeout_occurred: Arc<AtomicBool>,
id: i32,
id: UniqueIDType,
max_streams: usize,
digest: Digest,
) -> Self {
Expand Down Expand Up @@ -98,7 +98,7 @@ impl ConnectionRef {
self.0.current_streams.fetch_sub(1, Ordering::SeqCst);
}

pub fn id(&self) -> i32 {
pub fn id(&self) -> UniqueIDType {
self.0.id
}

Expand Down Expand Up @@ -196,7 +196,7 @@ impl InUsePool {

// release a h2_stream, this functional will cause an ConnectionRef to be returned (if exist)
// the caller should update the ref and then decide where to put it (in use pool or idle)
fn release(&self, reuse_hash: u64, id: i32) -> Option<ConnectionRef> {
fn release(&self, reuse_hash: u64, id: UniqueIDType) -> Option<ConnectionRef> {
let pools = self.pools.read();
if let Some(pool) = pools.get(&reuse_hash) {
pool.remove(id)
Expand Down
35 changes: 31 additions & 4 deletions pingora-core/src/connectors/l4.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,15 @@ use log::debug;
use pingora_error::{Context, Error, ErrorType::*, OrErr, Result};
use rand::seq::SliceRandom;
use std::net::SocketAddr as InetSocketAddr;
#[cfg(unix)]
use std::os::unix::io::AsRawFd;
#[cfg(windows)]
use std::os::windows::io::AsRawSocket;

#[cfg(unix)]
use crate::protocols::l4::ext::connect_uds;
use crate::protocols::l4::ext::{
connect_uds, connect_with as tcp_connect, set_dscp, set_recv_buf, set_tcp_fastopen_connect,
connect_with as tcp_connect, set_dscp, set_recv_buf, set_tcp_fastopen_connect,
};
use crate::protocols::l4::socket::SocketAddr;
use crate::protocols::l4::stream::Stream;
Expand All @@ -39,9 +44,12 @@ where
P: Peer + Send + Sync,
{
if peer.get_proxy().is_some() {
#[cfg(unix)]
return proxy_connect(peer)
.await
.err_context(|| format!("Fail to establish CONNECT proxy: {}", peer));
#[cfg(windows)]
panic!("peer proxy not supported on windows")
}
let peer_addr = peer.address();
let mut stream: Stream =
Expand All @@ -51,16 +59,21 @@ where
match peer_addr {
SocketAddr::Inet(addr) => {
let connect_future = tcp_connect(addr, bind_to.as_ref(), |socket| {
#[cfg(unix)]
let raw = socket.as_raw_fd();
#[cfg(windows)]
let raw = socket.as_raw_socket();

if peer.tcp_fast_open() {
set_tcp_fastopen_connect(socket.as_raw_fd())?;
set_tcp_fastopen_connect(raw)?;
}
if let Some(recv_buf) = peer.tcp_recv_buf() {
debug!("Setting recv buf size");
set_recv_buf(socket.as_raw_fd(), recv_buf)?;
set_recv_buf(raw, recv_buf)?;
}
if let Some(dscp) = peer.dscp() {
debug!("Setting dscp");
set_dscp(socket.as_raw_fd(), dscp)?;
set_dscp(raw, dscp)?;
}
Ok(())
});
Expand All @@ -86,6 +99,7 @@ where
}
}
}
#[cfg(unix)]
SocketAddr::Unix(addr) => {
let connect_future = connect_uds(
addr.as_pathname()
Expand Down Expand Up @@ -128,7 +142,10 @@ where
}
stream.set_nodelay()?;

#[cfg(unix)]
let digest = SocketDigest::from_raw_fd(stream.as_raw_fd());
#[cfg(windows)]
let digest = SocketDigest::from_raw_socket(stream.as_raw_socket());
digest
.peer_addr
.set(Some(peer_addr.clone()))
Expand Down Expand Up @@ -164,12 +181,14 @@ pub(crate) fn bind_to_random<P: Peer>(
InetSocketAddr::V4(_) => bind_to_ips(v4_list),
InetSocketAddr::V6(_) => bind_to_ips(v6_list),
},
#[cfg(unix)]
SocketAddr::Unix(_) => None,
}
}

use crate::protocols::raw_connect;

#[cfg(unix)]
async fn proxy_connect<P: Peer>(peer: &P) -> Result<Stream> {
// safe to unwrap
let proxy = peer.get_proxy().unwrap();
Expand Down Expand Up @@ -217,6 +236,7 @@ mod tests {
use std::collections::BTreeMap;
use std::path::PathBuf;
use tokio::io::AsyncWriteExt;
#[cfg(unix)]
use tokio::net::UnixListener;

#[tokio::test]
Expand Down Expand Up @@ -285,6 +305,7 @@ mod tests {
assert!(new_session.is_ok());
}

#[cfg(unix)]
#[tokio::test]
async fn test_connect_proxy_fail() {
let mut peer = HttpPeer::new("1.1.1.1:80".to_string(), false, "".to_string());
Expand All @@ -302,9 +323,11 @@ mod tests {
assert!(!e.retry());
}

#[cfg(unix)]
const MOCK_UDS_PATH: &str = "/tmp/test_unix_connect_proxy.sock";

// one-off mock server
#[cfg(unix)]
async fn mock_connect_server() {
let _ = std::fs::remove_file(MOCK_UDS_PATH);
let listener = UnixListener::bind(MOCK_UDS_PATH).unwrap();
Expand All @@ -316,6 +339,7 @@ mod tests {
let _ = std::fs::remove_file(MOCK_UDS_PATH);
}

#[cfg(unix)]
#[tokio::test(flavor = "multi_thread")]
async fn test_connect_proxy_work() {
tokio::spawn(async {
Expand All @@ -336,10 +360,12 @@ mod tests {
assert!(new_session.is_ok());
}

#[cfg(unix)]
const MOCK_BAD_UDS_PATH: &str = "/tmp/test_unix_bad_connect_proxy.sock";

// one-off mock bad proxy
// closes connection upon accepting
#[cfg(unix)]
async fn mock_connect_bad_server() {
let _ = std::fs::remove_file(MOCK_BAD_UDS_PATH);
let listener = UnixListener::bind(MOCK_BAD_UDS_PATH).unwrap();
Expand All @@ -350,6 +376,7 @@ mod tests {
let _ = std::fs::remove_file(MOCK_BAD_UDS_PATH);
}

#[cfg(unix)]
#[tokio::test(flavor = "multi_thread")]
async fn test_connect_proxy_conn_closed() {
tokio::spawn(async {
Expand Down
22 changes: 22 additions & 0 deletions pingora-core/src/connectors/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,11 +195,29 @@ impl TransportConnector {
let mut stream = l.into_inner();
// test_reusable_stream: we assume server would never actively send data
// first on an idle stream.
#[cfg(unix)]
if peer.matches_fd(stream.id()) && test_reusable_stream(&mut stream) {
Some(stream)
} else {
None
}
#[cfg(windows)]
{
use std::os::windows::io::{AsRawSocket, RawSocket};
struct WrappedRawSocket(RawSocket);
impl AsRawSocket for WrappedRawSocket {
fn as_raw_socket(&self) -> RawSocket {
self.0
}
}
if peer.matches_sock(WrappedRawSocket(stream.id() as RawSocket))
&& test_reusable_stream(&mut stream)
{
Some(stream)
} else {
None
}
}
}
Err(_) => {
error!("failed to acquire reusable stream");
Expand Down Expand Up @@ -372,6 +390,7 @@ mod tests {
use crate::tls::ssl::SslMethod;
use crate::upstreams::peer::BasicPeer;
use tokio::io::AsyncWriteExt;
#[cfg(unix)]
use tokio::net::UnixListener;

// 192.0.2.1 is effectively a black hole
Expand Down Expand Up @@ -403,9 +422,11 @@ mod tests {
assert!(reused);
}

#[cfg(unix)]
const MOCK_UDS_PATH: &str = "/tmp/test_unix_transport_connector.sock";

// one-off mock server
#[cfg(unix)]
async fn mock_connect_server() {
let _ = std::fs::remove_file(MOCK_UDS_PATH);
let listener = UnixListener::bind(MOCK_UDS_PATH).unwrap();
Expand All @@ -416,6 +437,7 @@ mod tests {
}
let _ = std::fs::remove_file(MOCK_UDS_PATH);
}
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread")]
async fn test_connect_uds() {
tokio::spawn(async {
Expand Down
Loading