Experimental use of crossbeam AtomicCell to achieve different scheduling properties.

Cargo.lock

@@ -446,6 +446,30 @@ dependencies = [
  "itertools",
 ]
 
+[[package]]
+name = "crossbeam"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2801af0d36612ae591caa9568261fddce32ce6e08a7275ea334a06a4ad021a2c"
+dependencies = [
+ "cfg-if 1.0.0",
+ "crossbeam-channel",
+ "crossbeam-deque",
+ "crossbeam-epoch",
+ "crossbeam-queue",
+ "crossbeam-utils",
+]
+
+[[package]]
+name = "crossbeam-channel"
+version = "0.5.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a33c2bf77f2df06183c3aa30d1e96c0695a313d4f9c453cc3762a6db39f99200"
+dependencies = [
+ "cfg-if 1.0.0",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-deque"
 version = "0.8.3"
@@ -470,6 +494,16 @@ dependencies = [
  "scopeguard",
 ]
 
+[[package]]
+name = "crossbeam-queue"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1cfb3ea8a53f37c40dea2c7bedcbd88bdfae54f5e2175d6ecaff1c988353add"
+dependencies = [
+ "cfg-if 1.0.0",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.8.16"
@@ -924,15 +958,6 @@ dependencies = [
  "bytes",
 ]
 
-[[package]]
-name = "instant"
-version = "0.1.12"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7a5bbe824c507c5da5956355e86a746d82e0e1464f65d862cc5e71da70e94b2c"
-dependencies = [
- "cfg-if 1.0.0",
-]
-
 [[package]]
 name = "iovec"
 version = "0.1.4"
@@ -1344,27 +1369,25 @@ dependencies = [
 
 [[package]]
 name = "parking_lot"
-version = "0.11.2"
+version = "0.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
+checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
 dependencies = [
- "instant",
  "lock_api",
  "parking_lot_core",
 ]
 
 [[package]]
 name = "parking_lot_core"
-version = "0.8.6"
+version = "0.9.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60a2cfe6f0ad2bfc16aefa463b497d5c7a5ecd44a23efa72aa342d90177356dc"
+checksum = "4c42a9226546d68acdd9c0a280d17ce19bfe27a46bf68784e4066115788d008e"
 dependencies = [
  "cfg-if 1.0.0",
- "instant",
  "libc",
- "redox_syscall 0.2.16",
+ "redox_syscall 0.4.1",
  "smallvec",
- "winapi 0.3.9",
+ "windows-targets",
 ]
 
 [[package]]
@@ -1599,15 +1622,6 @@ dependencies = [
  "crossbeam-utils",
 ]
 
-[[package]]
-name = "redox_syscall"
-version = "0.2.16"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a"
-dependencies = [
- "bitflags 1.3.2",
-]
-
 [[package]]
 name = "redox_syscall"
 version = "0.3.5"
@@ -1930,6 +1944,7 @@ dependencies = [
  "blake2",
  "bytes",
  "criterion",
+ "crossbeam",
  "futures",
  "getrandom 0.2.10",
  "hmac",
@@ -1939,7 +1954,7 @@ dependencies = [
  "preserves",
  "preserves-schema",
  "tokio",
- "tokio-util",
+ "tokio-util 0.7.10",
  "tracing",
  "tracing-futures",
  "tracing-subscriber",
@@ -1974,7 +1989,7 @@ dependencies = [
  "tikv-jemallocator",
  "tokio",
  "tokio-tungstenite",
- "tokio-util",
+ "tokio-util 0.6.10",
  "tracing",
  "tracing-futures",
  "tracing-subscriber",
@@ -2146,6 +2161,19 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-util"
+version = "0.7.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5419f34732d9eb6ee4c3578b7989078579b7f039cbbb9ca2c4da015749371e15"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "tracing"
 version = "0.1.40"

syndicate-macros/examples/ring.rs

@@ -111,6 +111,6 @@ async fn main() -> ActorResult {
             Ok(())
         });
         Ok(())
-    }).await??;
+    }).await?;
     Ok(())
 }

syndicate-server/src/main.rs

@@ -213,7 +213,7 @@ async fn main() -> ActorResult {
         }));
 
         Ok(())
-    }).await??;
+    }).await?;
 
     wait_for_all_actors_to_stop(std::time::Duration::from_secs(2)).await;
 

syndicate-server/src/services/config_watcher.rs

@@ -187,7 +187,7 @@ fn run(
     let facet = t.facet.clone();
     let trace_collector = t.trace_collector();
     let span = tracing::Span::current();
-    thread::spawn(move || {
+    tokio::task::spawn_blocking(move || {
         let _entry = span.enter();
 
         let mut path_state: Map<PathBuf, FacetId> = Map::new();

syndicate/Cargo.toml

@@ -19,17 +19,18 @@ preserves-schema = "4.991"
 preserves = "4.991"
 preserves-schema = "4.991"
 
-tokio = { version = "1.10", features = ["io-util", "macros", "rt", "rt-multi-thread", "time"] }
-tokio-util = "0.6"
-bytes = "1.0"
+tokio = { version = "1.33", features = ["io-util", "macros", "rt", "rt-multi-thread", "time"] }
+tokio-util = "0.7"
+bytes = "1.5"
 
 futures = "0.3"
 
 blake2 = "0.10"
+crossbeam = "0.8"
 getrandom = "0.2"
 hmac = "0.12"
 lazy_static = "1.4"
-parking_lot = "0.11"
+parking_lot = "0.12"
 
 tracing = "0.1"
 tracing-subscriber = "0.2"

syndicate/src/supervise.rs

@@ -40,7 +40,7 @@ pub struct Supervisor {
     boot_fn: Boot,
     restarts: VecDeque<Instant>,
     state: Arc<Field<State>>,
-    ac_ref: Option<ActorRef>,
+    ac_ref: Option<Actor>,
 }
 
 impl Default for SupervisorConfiguration {

syndicate/src/trace.rs

@@ -27,7 +27,7 @@ pub struct TraceCollector {
 impl<M> From<&Ref<M>> for Target {
     fn from(v: &Ref<M>) -> Target {
         Target {
-            actor: ActorId(AnyValue::new(v.mailbox.actor_id)),
+            actor: ActorId(AnyValue::new(v.mailbox.actor.actor_id)),
             facet: FacetId(AnyValue::new(u64::from(v.facet_id))),
             oid: Oid(AnyValue::new(v.oid())),
         }