io_uring support

.github/workflows/ci.yml

@@ -16,7 +16,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest, macos-latest]
+        os: [ubuntu-latest]
     steps:
     - uses: actions/checkout@v3
     - uses: dtolnay/rust-toolchain@nightly

Cargo.toml

@@ -1,7 +1,8 @@
 [package]
 name          = "heph"
 description   = "Heph is an actor framework based on asynchronous functions."
-version       = "0.4.0"
+version       = "0.5.0"
+publish       = false # In development.
 authors       = ["Thomas de Zeeuw <thomasdezeeuw@gmail.com>"]
 license       = "MIT"
 documentation = "https://docs.rs/heph"
@@ -18,15 +19,15 @@ test = ["getrandom"]
 
 [dependencies]
 heph-inbox        = { version = "0.2.3", path = "./inbox", default-features = false }
-log               = { version = "0.4.16", default-features = false, features = ["kv_unstable", "kv_unstable_std"] }
+log               = { version = "0.4.17", default-features = false, features = ["kv_unstable", "kv_unstable_std"] }
 
 # Optional dependencies, enabled by features.
 # Required by the `test` feature.
 getrandom         = { version = "0.2.2", default-features = false, features = ["std"], optional = true }
 
 [dev-dependencies]
 getrandom         = { version = "0.2.2", default-features = false, features = ["std"] }
-heph-rt           = { version = "0.4.0", path = "./rt", default-features = false, features = ["test"] }
+heph-rt           = { version = "0.5.0", path = "./rt", default-features = false, features = ["test"] }
 std-logger        = { version = "0.5.0", default-features = false, features = ["log-panic", "nightly"] }
 
 [[test]]
@@ -42,7 +43,7 @@ required-features = ["test"]
 
 [workspace]
 members = [
-  #"http", # Stuck on 2021-11-01, also enable in Makefile.
+  "http",
   "inbox",
   "remote",
   "rt",

LICENSE

@@ -1,4 +1,4 @@
-Copyright (C) 2017-2022 Thomas de Zeeuw
+Copyright (C) 2017-2023 Thomas de Zeeuw
 
 
 Permission is hereby granted, free of charge, to any person obtaining a copy of

Makefile

@@ -1,7 +1,7 @@
 include Makefile.include
 
 # Crates in this repo.
-CRATES := ./ inbox rt remote # http, stuck on old nightly.
+CRATES := ./ inbox rt remote http
 # Target that run the target in all $CRATES.
 TARGETS := test_all test_sanitizers_all test_sanitizer_all check_all clippy_all
 

Makefile.include

@@ -39,10 +39,11 @@ $(TARGETS):
 
 # Reasons to allow lints:
 # `cargo-common-metadata`: for `benches` and `tools`.
+# `doc-markdown`: too many false positives.
 # `equatable-if-let`: bad lint.
-# `manual-let-else`: not really a fan of this.
-# `match-bool`, `single-match-else`: often less lines of code and I find that
-# use `match` generally strictly better then `if`s.
+# `future-not-send`: we don't want to require all generic parameters to be `Send`.
+# `match-bool`: often less lines of code and I find that use `match` generally
+# strictly better then `if`s.
 # `missing-const-for-fn`: See https://github.com/rust-lang/rust-clippy/issues/4979.
 # `module-name-repetitions`: we re-export various names.
 # `needless-lifetimes`: lifetime serves as documentation.
@@ -68,9 +69,10 @@ clippy:
 		--deny clippy::nursery \
 		--deny clippy::cargo \
 		--allow clippy::cargo-common-metadata \
+		--allow clippy::doc-markdown \
 		--allow clippy::enum-glob-use \
 		--allow clippy::equatable-if-let \
-		--allow clippy::manual-let-else \
+		--allow clippy::future-not-send \
 		--allow clippy::match-bool \
 		--allow clippy::missing-const-for-fn \
 		--allow clippy::missing-errors-doc \
@@ -81,7 +83,6 @@ clippy:
 		--allow clippy::option-if-let-else \
 		--allow clippy::redundant-pub-crate \
 		--allow clippy::significant-drop-tightening \
-		--allow clippy::single-match-else \
 		--allow clippy::use-self \
 
 doc:

README.md

@@ -73,7 +73,7 @@ Second, Heph needs to be added as a dependency.
 
 ```toml
 [dependencies]
-heph = "0.4.0"
+heph = "0.5.0"
 ```
 
 Now, you're ready to starting writing your application! Next, you can look at

http/Cargo.toml

@@ -2,6 +2,7 @@
 name          = "heph-http"
 description   = "Heph-HTTP is a HTTP library build on top of Heph."
 version       = "0.1.0"
+publish       = false # In development.
 authors       = ["Thomas de Zeeuw <thomasdezeeuw@gmail.com>"]
 license       = "MIT"
 documentation = "https://docs.rs/heph-http"
@@ -13,15 +14,21 @@ include       = ["/Cargo.toml", "/src/**/*.rs", "/README.md", "/LICENSE"]
 edition       = "2021"
 
 [dependencies]
-heph     = { version = "0.4.0", path = "../", default-features = false, features = ["runtime"] }
-httparse = { version = "1.5.1", default-features = false }
-httpdate = { version = "1.0.0", default-features = false }
-log      = { version = "0.4.8", default-features = false }
-itoa     = { version = "0.4.7", default-features = false }
+heph     = { version = "0.5.0", default-features = false, path = "../" }
+heph-rt  = { version = "0.5.0", default-features = false, path = "../rt" }
+httparse = { version = "1.8.0", default-features = false }
+httpdate = { version = "1.0.2", default-features = false }
+log      = { version = "0.4.17", default-features = false }
+itoa     = { version = "1.0.6", default-features = false }
 
 [dev-dependencies]
 std-logger = { version = "0.5.0", default-features = false, features = ["log-panic", "nightly"] }
 
 [dev-dependencies.heph]
 path     = "../"
 features = ["test"]
+
+
+[dev-dependencies.heph-rt]
+path     = "../rt"
+features = ["test"]

http/LICENSE

@@ -1,4 +1,4 @@
-Copyright (C) 2021 Thomas de Zeeuw
+Copyright (C) 2021-2023 Thomas de Zeeuw
 
 
 Permission is hereby granted, free of charge, to any person obtaining a copy of

http/examples/my_ip.rs

@@ -2,31 +2,30 @@
 
 use std::borrow::Cow;
 use std::io;
-use std::net::SocketAddr;
 use std::time::Duration;
 
-use heph::actor::{self, Actor, NewActor};
-use heph::net::TcpStream;
-use heph::rt::{self, Runtime, ThreadLocal};
-use heph::supervisor::{Supervisor, SupervisorStrategy};
-use heph::timer::Deadline;
+use heph::actor;
+use heph::supervisor::SupervisorStrategy;
 use heph_http::body::OneshotBody;
-use heph_http::{self as http, Header, HeaderName, Headers, HttpServer, Method, StatusCode};
+use heph_http::{self as http, server, Header, HeaderName, Headers, Method, StatusCode};
+use heph_rt::net::TcpStream;
 use heph_rt::spawn::options::{ActorOptions, Priority};
+use heph_rt::timer::Deadline;
+use heph_rt::{Runtime, ThreadLocal};
 use log::{debug, error, info, warn};
 
-fn main() -> Result<(), rt::Error> {
+fn main() -> Result<(), heph_rt::Error> {
     std_logger::Config::logfmt().init();
 
-    let actor = http_actor as fn(_, _, _) -> _;
+    let actor = http_actor as fn(_, _) -> _;
     let address = "127.0.0.1:7890".parse().unwrap();
-    let server = HttpServer::setup(address, conn_supervisor, actor, ActorOptions::default())
-        .map_err(rt::Error::setup)?;
+    let server = server::setup(address, conn_supervisor, actor, ActorOptions::default())
+        .map_err(heph_rt::Error::setup)?;
 
     let mut runtime = Runtime::setup().use_all_cores().build()?;
     runtime.run_on_workers(move |mut runtime_ref| -> io::Result<()> {
         let options = ActorOptions::default().with_priority(Priority::LOW);
-        let server_ref = runtime_ref.try_spawn_local(ServerSupervisor, server, (), options)?;
+        let server_ref = runtime_ref.spawn_local(server_supervisor, server, (), options);
 
         runtime_ref.receive_signals(server_ref.try_map());
         Ok(())
@@ -35,37 +34,20 @@ fn main() -> Result<(), rt::Error> {
     runtime.start()
 }
 
-/// Our supervisor for the TCP server.
-#[derive(Copy, Clone, Debug)]
-struct ServerSupervisor;
-
-impl<NA> Supervisor<NA> for ServerSupervisor
-where
-    NA: NewActor<Argument = (), Error = io::Error>,
-    NA::Actor: Actor<Error = http::server::Error<!>>,
-{
-    fn decide(&mut self, err: http::server::Error<!>) -> SupervisorStrategy<()> {
-        use http::server::Error::*;
-        match err {
-            Accept(err) => {
-                error!("error accepting new connection: {err}");
-                SupervisorStrategy::Restart(())
-            }
-            NewActor(_) => unreachable!(),
+fn server_supervisor(err: server::Error<!>) -> SupervisorStrategy<()> {
+    match err {
+        // When we hit an error accepting a connection we'll drop the old
+        // server and create a new one.
+        server::Error::Accept(err) => {
+            error!("error accepting new connection: {err}");
+            SupervisorStrategy::Restart(())
         }
-    }
-
-    fn decide_on_restart_error(&mut self, err: io::Error) -> SupervisorStrategy<()> {
-        error!("error restarting the TCP server: {err}");
-        SupervisorStrategy::Stop
-    }
-
-    fn second_restart_error(&mut self, err: io::Error) {
-        error!("error restarting the actor a second time: {err}");
+        // Async function never return an error creating a new actor.
+        server::Error::NewActor(_) => unreachable!(),
     }
 }
 
-fn conn_supervisor(err: io::Error) -> SupervisorStrategy<(TcpStream, SocketAddr)> {
+fn conn_supervisor(err: io::Error) -> SupervisorStrategy<TcpStream> {
     error!("error handling connection: {err}");
     SupervisorStrategy::Stop
 }
@@ -75,18 +57,22 @@ const ALIVE_TIMEOUT: Duration = Duration::from_secs(120);
 const WRITE_TIMEOUT: Duration = Duration::from_secs(10);
 
 async fn http_actor(
-    mut ctx: actor::Context<!, ThreadLocal>,
+    ctx: actor::Context<!, ThreadLocal>,
     mut connection: http::Connection,
-    address: SocketAddr,
 ) -> io::Result<()> {
+    let address = connection.peer_addr()?;
     info!("accepted connection: source={address}");
     connection.set_nodelay(true)?;
 
     let mut read_timeout = READ_TIMEOUT;
     let mut headers = Headers::EMPTY;
     loop {
-        let fut = Deadline::after(&mut ctx, read_timeout, connection.next_request());
-        let (code, body, should_close) = match fut.await? {
+        let fut = Deadline::after(
+            ctx.runtime_ref().clone(),
+            read_timeout,
+            connection.next_request(),
+        );
+        let (code, body, should_close) = match fut.await {
             Ok(Some(request)) => {
                 info!("received request: {request:?}: source={address}");
                 if request.path() != "/" {
@@ -121,9 +107,9 @@ async fn http_actor(
         }
 
         debug!("sending response: code={code}, body='{body}', source={address}");
-        let body = OneshotBody::new(body.as_bytes());
+        let body = OneshotBody::new(body);
         let write_response = connection.respond(code, &headers, body);
-        Deadline::after(&mut ctx, WRITE_TIMEOUT, write_response).await?;
+        Deadline::after(ctx.runtime_ref().clone(), WRITE_TIMEOUT, write_response).await?;
 
         if should_close {
             warn!("closing connection: source={address}");