Skip to content
Permalink
Browse files

Add interfaces for activating across thread boundaries

Add interfaces that make it easy to send activations across thread
boundaries.

The approach is to give `Activations` an MPSC channel which it checks
for activation requests on every call to `advance`. Other threads can
acquire a handle to the send side of this object via `SyncActivations`
or `SyncActivator`, which are careful to also call unpark on the
associated worker thread for easy interop with `Worker.step_or_park`.
  • Loading branch information...
benesch committed Aug 21, 2019
1 parent b63bea6 commit a8cfadc8a6b38eb96747aca6ce3142d9beebcc7a
Showing with 106 additions and 9 deletions.
  1. +99 −7 timely/src/scheduling/activate.rs
  2. +7 −2 timely/src/scheduling/mod.rs
@@ -2,28 +2,26 @@

use std::rc::Rc;
use std::cell::RefCell;
use std::sync::mpsc::{Sender, Receiver};
use std::thread::Thread;

/// Allocation-free activation tracker.
#[derive(Default)]
pub struct Activations {
clean: usize,
/// `(offset, length)`
bounds: Vec<(usize, usize)>,
slices: Vec<usize>,
buffer: Vec<usize>,
tx: Sender<Vec<usize>>,
rx: Receiver<Vec<usize>>,
}

impl Activations {

/// Creates a new activation tracker.
#[deprecated(since="0.10",note="Type implements Default")]
pub fn new() -> Self {
Self {
clean: 0,
bounds: Vec::new(),
slices: Vec::new(),
buffer: Vec::new(),
}
Self::default()
}

/// Indicates if there are no pending activations.
@@ -40,6 +38,10 @@ impl Activations {
/// Discards the current active set and presents the next active set.
pub fn advance(&mut self) {

while let Ok(path) = self.rx.try_recv() {
self.activate(&path)
}

self.bounds.drain(.. self.clean);

{ // Scoped, to allow borrow to drop.
@@ -94,6 +96,58 @@ impl Activations {
}
});
}

/// Constructs a thread-safe `SyncActivations` handle to this activator.
pub fn sync(&self) -> SyncActivations {
SyncActivations {
tx: self.tx.clone(),
thread: std::thread::current(),
}
}
}

impl Default for Activations {
fn default() -> Self {
let (tx, rx) = std::sync::mpsc::channel();
Self {
clean: 0,
bounds: Vec::new(),
slices: Vec::new(),
buffer: Vec::new(),
tx,
rx,
}
}
}

/// A thread-safe handle to an `Activations`.
pub struct SyncActivations {
tx: Sender<Vec<usize>>,
thread: Thread,
}

impl SyncActivations {
/// Unparks the task addressed by `path` and unparks the associated worker
/// thread.
pub fn activate(&self, path: Vec<usize>) -> Result<(), SyncActivationError> {
self.activate_batch(std::iter::once(path))
}

/// Unparks the tasks addressed by `paths` and unparks the associated worker
/// thread.
///
/// This method can be more efficient than calling `activate` repeatedly, as
/// it only unparks the worker thread after sending all of the activations.
pub fn activate_batch<I>(&self, paths: I) -> Result<(), SyncActivationError>
where
I: IntoIterator<Item = Vec<usize>>
{
for path in paths.into_iter() {
self.tx.send(path).map_err(|_| SyncActivationError)?;
}
self.thread.unpark();
Ok(())
}
}

/// A capability to activate a specific path.
@@ -118,6 +172,44 @@ impl Activator {
}
}

/// A thread-safe version of `Activator`.
pub struct SyncActivator {
path: Vec<usize>,
queue: SyncActivations,
}

impl SyncActivator {
/// Creates a new thread-safe activation handle.
pub fn new(path: &[usize], queue: SyncActivations) -> Self {
Self {
path: path.to_vec(),
queue,
}
}

/// Activates the associated path and unparks the associated worker thread.
pub fn activate(&self) -> Result<(), SyncActivationError> {
self.queue.activate(self.path.clone())
}
}

/// The error returned when activation fails across thread boundaries because
/// the receiving end has hung up.
#[derive(Debug)]
pub struct SyncActivationError;

impl std::fmt::Display for SyncActivationError {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
f.write_str("sync activation error in timely")
}
}

impl std::error::Error for SyncActivationError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
None
}
}

/// A wrapper that unparks on drop.
pub struct ActivateOnDrop<T> {
wrapped: T,
@@ -5,7 +5,7 @@ use std::cell::RefCell;

pub mod activate;

pub use self::activate::{Activations, Activator, ActivateOnDrop};
pub use self::activate::{Activations, Activator, ActivateOnDrop, SyncActivator};

/// A type that can be scheduled.
pub trait Schedule {
@@ -24,9 +24,14 @@ pub trait Schedule {
pub trait Scheduler {
/// Provides a shared handle to the activation scheduler.
fn activations(&self) -> Rc<RefCell<Activations>>;
///
/// Constructs an `Activator` tied to the specified operator address.
fn activator_for(&self, path: &[usize]) -> Activator {
let activations = self.activations().clone();
Activator::new(path, activations)
}
/// Constructs a `SyncActivator` tied to the specified operator address.
fn sync_activator_for(&self, path: &[usize]) -> SyncActivator {
let sync_activations = self.activations().borrow().sync();
SyncActivator::new(path, sync_activations)
}
}

0 comments on commit a8cfadc

Please sign in to comment.
You can’t perform that action at this time.