DNM: Release plz test

.github/dependabot.yml

@@ -0,0 +1,8 @@
+version: 2
+updates:
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    # Check for updates every Monday
+    schedule:
+      interval: "weekly"
+

.github/workflows/deploy.yml

@@ -7,12 +7,12 @@ on:
 
 jobs:
   deploy:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     steps:
-      - uses: actions/checkout@v2
-      - run: cargo install mdbook --version 0.4.6
+      - uses: actions/checkout@v3
+      - run: cargo install mdbook --version 0.4.27
       - run: cd mdbook && mdbook build
-      - uses: JamesIves/github-pages-deploy-action@4.0.0
+      - uses: JamesIves/github-pages-deploy-action@v4
         with:
           branch: gh-pages
           folder: mdbook/book

.github/workflows/release-plz.yml

@@ -0,0 +1,28 @@
+name: Release-plz
+
+permissions:
+  pull-requests: write
+  contents: write
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+
+jobs:
+  release-plz:
+    name: Release-plz
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - name: Install Rust toolchain
+        uses: actions-rust-lang/setup-rust-toolchain@v1
+      - name: Run release-plz
+        uses: MarcoIeni/[email protected]
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}

.github/workflows/test.yml

@@ -4,10 +4,10 @@ on: [push, pull_request]
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-22.04
     steps:
-    - uses: actions/checkout@v2
-    - run: rustup update 1.50.0 --no-self-update && rustup default 1.50.0
+    - uses: actions/checkout@v3
+    - run: rustup update 1.64 --no-self-update && rustup default 1.64
     - run: cargo build
     - name: test mdBook
       # rustdoc doesn't build dependencies, so it needs to run after `cargo build`,

CHANGELOG.md

@@ -92,7 +92,7 @@ Beneficial fallouts include our ability to remove `RootTimestamp`, as dataflows
 - The `Root` type has been renamed `Worker` and is found in the `::worker` module. The methods of the `ScopeParent` trait are now in the `::worker::AsWorker` trait.
 - The communication `Allocate` trait's main method `allocate` now takes a worker-unique identifier to use for the channel. The allocator may or may not use the information (most often for logging), but they are allowed to be incorrect if one allocates two channels with the same identifier.
 - A `CapabilityRef<T>` now supports `retain_for(usize)` which indicates a specific output port the capability should be retain for use with. The `retain()` method still exists for now and is equivalent to `retain(0)`. This change also comes with the *inability* to use an arbitrary `Capability<T>` with any output; using a capability bound to the wrong output will result in a run-time error.
-- The `unary` and `binary` operators now provide `data` as a `RefOrMut`, which does not implement `DerefMut`. More information on how to port methods can be found [here](https://github.com/frankmcsherry/timely-dataflow/pull/135#issuecomment-418355284).
+- The `unary` and `binary` operators now provide `data` as a `RefOrMut`, which does not implement `DerefMut`. More information on how to port methods can be found [here](https://github.com/TimelyDataflow/timely-dataflow/pull/135#issuecomment-418355284).
 
 
 ### Removed

Cargo.toml

@@ -2,10 +2,9 @@
 members = [
     "bytes",
     "communication",
-    "kafkaesque",
+    "container",
     "logging",
     "timely",
-    "experiments"
 ]
 
 [profile.release]

communication/src/allocator/counters.rs

@@ -2,23 +2,21 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use std::collections::VecDeque;
 
 use crate::{Push, Pull};
-use crate::allocator::Event;
 
 /// The push half of an intra-thread channel.
 pub struct Pusher<T, P: Push<T>> {
     index: usize,
     // count: usize,
-    events: Rc<RefCell<VecDeque<(usize, Event)>>>,
+    events: Rc<RefCell<Vec<usize>>>,
     pusher: P,
     phantom: ::std::marker::PhantomData<T>,
 }
 
 impl<T, P: Push<T>>  Pusher<T, P> {
     /// Wraps a pusher with a message counter.
-    pub fn new(pusher: P, index: usize, events: Rc<RefCell<VecDeque<(usize, Event)>>>) -> Self {
+    pub fn new(pusher: P, index: usize, events: Rc<RefCell<Vec<usize>>>) -> Self {
         Pusher {
             index,
             // count: 0,
@@ -36,7 +34,7 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
         //     if self.count != 0 {
         //         self.events
         //             .borrow_mut()
-        //             .push_back((self.index, Event::Pushed(self.count)));
+        //             .push_back(self.index);
         //         self.count = 0;
         //     }
         // }
@@ -47,7 +45,7 @@ impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
         //       moving information along. Better, but needs cooperation.
         self.events
             .borrow_mut()
-            .push_back((self.index, Event::Pushed(1)));
+            .push(self.index);
 
         self.pusher.push(element)
     }
@@ -59,15 +57,15 @@ use crossbeam_channel::Sender;
 pub struct ArcPusher<T, P: Push<T>> {
     index: usize,
     // count: usize,
-    events: Sender<(usize, Event)>,
+    events: Sender<usize>,
     pusher: P,
     phantom: ::std::marker::PhantomData<T>,
     buzzer: crate::buzzer::Buzzer,
 }
 
 impl<T, P: Push<T>>  ArcPusher<T, P> {
     /// Wraps a pusher with a message counter.
-    pub fn new(pusher: P, index: usize, events: Sender<(usize, Event)>, buzzer: crate::buzzer::Buzzer) -> Self {
+    pub fn new(pusher: P, index: usize, events: Sender<usize>, buzzer: crate::buzzer::Buzzer) -> Self {
         ArcPusher {
             index,
             // count: 0,
@@ -99,7 +97,7 @@ impl<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
         // and finally awaken the thread. Other orders are defective when
         // multiple threads are involved.
         self.pusher.push(element);
-        let _ = self.events.send((self.index, Event::Pushed(1)));
+        let _ = self.events.send(self.index);
             // TODO : Perhaps this shouldn't be a fatal error (e.g. in shutdown).
             // .expect("Failed to send message count");
         self.buzzer.buzz();
@@ -110,14 +108,14 @@ impl<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
 pub struct Puller<T, P: Pull<T>> {
     index: usize,
     count: usize,
-    events: Rc<RefCell<VecDeque<(usize, Event)>>>,
+    events: Rc<RefCell<Vec<usize>>>,
     puller: P,
     phantom: ::std::marker::PhantomData<T>,
 }
 
 impl<T, P: Pull<T>>  Puller<T, P> {
     /// Wraps a puller with a message counter.
-    pub fn new(puller: P, index: usize, events: Rc<RefCell<VecDeque<(usize, Event)>>>) -> Self {
+    pub fn new(puller: P, index: usize, events: Rc<RefCell<Vec<usize>>>) -> Self {
         Puller {
             index,
             count: 0,
@@ -135,7 +133,7 @@ impl<T, P: Pull<T>> Pull<T> for Puller<T, P> {
             if self.count != 0 {
                 self.events
                     .borrow_mut()
-                    .push_back((self.index, Event::Pulled(self.count)));
+                    .push(self.index);
                 self.count = 0;
             }
         }

communication/src/allocator/generic.rs

@@ -5,11 +5,10 @@
 
 use std::rc::Rc;
 use std::cell::RefCell;
-use std::collections::VecDeque;
 
 use crate::allocator::thread::ThreadBuilder;
 use crate::allocator::process::ProcessBuilder as TypedProcessBuilder;
-use crate::allocator::{Allocate, AllocateBuilder, Event, Thread, Process};
+use crate::allocator::{Allocate, AllocateBuilder, Thread, Process};
 use crate::allocator::zero_copy::allocator_process::{ProcessBuilder, ProcessAllocator};
 use crate::allocator::zero_copy::allocator::{TcpBuilder, TcpAllocator};
 
@@ -74,7 +73,7 @@ impl Generic {
             Generic::ZeroCopy(z) => z.release(),
         }
     }
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         match self {
             Generic::Thread(ref t) => t.events(),
             Generic::Process(ref p) => p.events(),
@@ -93,7 +92,7 @@ impl Allocate for Generic {
 
     fn receive(&mut self) { self.receive(); }
     fn release(&mut self) { self.release(); }
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> { self.events() }
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>> { self.events() }
     fn await_events(&self, _duration: Option<std::time::Duration>) {
         match self {
             Generic::Thread(t) => t.await_events(_duration),

communication/src/allocator/mod.rs

@@ -3,7 +3,6 @@
 use std::rc::Rc;
 use std::cell::RefCell;
 use std::time::Duration;
-use std::collections::VecDeque;
 
 pub use self::thread::Thread;
 pub use self::process::Process;
@@ -50,7 +49,7 @@ pub trait Allocate {
     /// drain these events in order to drive their computation. If they
     /// fail to do so the event queue may become quite large, and turn
     /// into a performance problem.
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>>;
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>>;
 
     /// Awaits communication events.
     ///
@@ -92,11 +91,3 @@ pub trait Allocate {
         thread::Thread::new_from(identifier, self.events().clone())
     }
 }
-
-/// A communication channel event.
-pub enum Event {
-    /// A number of messages pushed into the channel.
-    Pushed(usize),
-    /// A number of messages pulled from the channel.
-    Pulled(usize),
-}

communication/src/allocator/process.rs

@@ -5,11 +5,11 @@ use std::cell::RefCell;
 use std::sync::{Arc, Mutex};
 use std::any::Any;
 use std::time::Duration;
-use std::collections::{HashMap, VecDeque};
+use std::collections::{HashMap};
 use crossbeam_channel::{Sender, Receiver};
 
 use crate::allocator::thread::{ThreadBuilder};
-use crate::allocator::{Allocate, AllocateBuilder, Event, Thread};
+use crate::allocator::{Allocate, AllocateBuilder, Thread};
 use crate::{Push, Pull, Message};
 use crate::buzzer::Buzzer;
 
@@ -25,8 +25,8 @@ pub struct ProcessBuilder {
     buzzers_send: Vec<Sender<Buzzer>>,
     buzzers_recv: Vec<Receiver<Buzzer>>,
 
-    counters_send: Vec<Sender<(usize, Event)>>,
-    counters_recv: Receiver<(usize, Event)>,
+    counters_send: Vec<Sender<usize>>,
+    counters_recv: Receiver<usize>,
 }
 
 impl AllocateBuilder for ProcessBuilder {
@@ -63,8 +63,8 @@ pub struct Process {
     // below: `Box<Any+Send>` is a `Box<Vec<Option<(Vec<Sender<T>>, Receiver<T>)>>>`
     channels: Arc<Mutex<HashMap</* channel id */ usize, Box<dyn Any+Send>>>>,
     buzzers: Vec<Buzzer>,
-    counters_send: Vec<Sender<(usize, Event)>>,
-    counters_recv: Receiver<(usize, Event)>,
+    counters_send: Vec<Sender<usize>>,
+    counters_recv: Receiver<usize>,
 }
 
 impl Process {
@@ -174,7 +174,7 @@ impl Allocate for Process {
         (sends, recv)
     }
 
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         self.inner.events()
     }
 
@@ -184,8 +184,8 @@ impl Allocate for Process {
 
     fn receive(&mut self) {
         let mut events = self.inner.events().borrow_mut();
-        while let Ok((index, event)) = self.counters_recv.try_recv() {
-            events.push_back((index, event));
+        while let Ok(index) = self.counters_recv.try_recv() {
+            events.push(index);
         }
     }
 }

communication/src/allocator/thread.rs

@@ -5,7 +5,7 @@ use std::cell::RefCell;
 use std::time::Duration;
 use std::collections::VecDeque;
 
-use crate::allocator::{Allocate, AllocateBuilder, Event};
+use crate::allocator::{Allocate, AllocateBuilder};
 use crate::allocator::counters::Pusher as CountPusher;
 use crate::allocator::counters::Puller as CountPuller;
 use crate::{Push, Pull, Message};
@@ -22,7 +22,7 @@ impl AllocateBuilder for ThreadBuilder {
 /// An allocator for intra-thread communication.
 pub struct Thread {
     /// Shared counts of messages in channels.
-    events: Rc<RefCell<VecDeque<(usize, Event)>>>,
+    events: Rc<RefCell<Vec<usize>>>,
 }
 
 impl Allocate for Thread {
@@ -32,7 +32,7 @@ impl Allocate for Thread {
         let (pusher, puller) = Thread::new_from(identifier, self.events.clone());
         (vec![Box::new(pusher)], Box::new(puller))
     }
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         &self.events
     }
     fn await_events(&self, duration: Option<Duration>) {
@@ -56,12 +56,12 @@ impl Thread {
     /// Allocates a new thread-local channel allocator.
     pub fn new() -> Self {
         Thread {
-            events: Rc::new(RefCell::new(VecDeque::new())),
+            events: Rc::new(RefCell::new(Default::default())),
         }
     }
 
     /// Creates a new thread-local channel from an identifier and shared counts.
-    pub fn new_from<T: 'static>(identifier: usize, events: Rc<RefCell<VecDeque<(usize, Event)>>>)
+    pub fn new_from<T: 'static>(identifier: usize, events: Rc<RefCell<Vec<usize>>>)
         -> (ThreadPusher<Message<T>>, ThreadPuller<Message<T>>)
     {
         let shared = Rc::new(RefCell::new((VecDeque::<Message<T>>::new(), VecDeque::<Message<T>>::new())));

communication/src/allocator/zero_copy/allocator.rs

@@ -10,7 +10,6 @@ use crate::networking::MessageHeader;
 
 use crate::{Allocate, Message, Data, Push, Pull};
 use crate::allocator::AllocateBuilder;
-use crate::allocator::Event;
 use crate::allocator::canary::Canary;
 
 use super::bytes_exchange::{BytesPull, SendEndpoint, MergeQueue};
@@ -229,7 +228,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
 
                     // Increment message count for channel.
                     // Safe to do this even if the channel has been dropped.
-                    events.push_back((header.channel, Event::Pushed(1)));
+                    events.push(header.channel);
 
                     // Ensure that a queue exists.
                     match self.to_local.entry(header.channel) {
@@ -269,7 +268,7 @@ impl<A: Allocate> Allocate for TcpAllocator<A> {
         //     }
         // }
     }
-    fn events(&self) -> &Rc<RefCell<VecDeque<(usize, Event)>>> {
+    fn events(&self) -> &Rc<RefCell<Vec<usize>>> {
         self.inner.events()
     }
     fn await_events(&self, duration: Option<std::time::Duration>) {