tobz

pub trait Router<T: RequestProcessor> {
    type Future;

    fn route(&self, OrderedMessages<T::Message>) -> Self::Future;
}

impl<T> Router<T> for FixedRouter<T>
where
    T: RequestProcessor + Clone + Send + 'static,
    T::Message: Keyed + Clone + Send + 'static,

tobz

#!/usr/bin/env run-cargo-script
// cargo-deps: tokio, tokio-codec, tokio-io, futures-cpupool

extern crate tokio;
extern crate tokio_codec;

use tokio_codec::{Framed, LinesCodec};
use tokio::net::TcpListener;
use tokio::prelude::*;

tobz

# make sure redis-server is running, listening on localhost or 0.0.0.0:6379
# make sure you have redis-benchmark in your path

git clone https://github.com/nuclearfurnace/synchrotron.git
cargo build --release
target/release/synchrotron

# in another window
redis-benchmark -p 6380 -t get -c 1 -n 10000000

tobz

se std::collections::VecDeque;
use crossbeam_arccell::ArcCell;
use crossbeam_channel::{Sender as ChannelSender, Receiver as ChannelReceiver};
use futures::task;
use futures::task::Task;

pub struct WorkSender<T> {
    work_tx: ChannelSender<T>,
    work_rx: ChannelReceiver<T>,
    wait_tx: ChannelSender<ArcCell<Option<Task>>>,

tobz

* thread #2, name = 'synchrotron-worker-5', stop reason = EXC_BAD_ACCESS (code=1, address=0x8)
    frame #0: 0x00000001002d4d7d synchrotron`core::sync::atomic::atomic_sub::h06cef389ea519679(dst=&0x8, val=1, order=Release) at atomic.rs:2005
Target 0: (synchrotron) stopped.
(lldb) bt
error: need to add support for DW_TAG_base_type '()' encoded with DW_ATE = 0x7, bit_size = 0
error: need to add support for DW_TAG_base_type '()' encoded with DW_ATE = 0x7, bit_size = 0
* thread #2, name = 'synchrotron-worker-5', stop reason = EXC_BAD_ACCESS (code=1, address=0x8)
  * frame #0: 0x00000001002d4d7d synchrotron`core::sync::atomic::atomic_sub::h06cef389ea519679(dst=&0x8, val=1, order=Release) at atomic.rs:2005
    frame #1: 0x00000001002d5103 synchrotron`core::sync::atomic::AtomicUsize::fetch_sub::h923c9ac699a646a4(self=&0x8, val=1, order=Release) at atomic.rs:1506
    frame #2: 0x00000001002d7098 synchrotron`_$LT$alloc..sync..Arc$LT$T$GT$$u20$as$u20$core..ops..drop..Drop$GT$::drop::h054ba5716e670600(self=&0x103a1e0b0) 

tobz

extern crate tokio; // 0.1.8;
extern crate futures; // 0.1.24;

use futures::sync::mpsc;
use futures::{task, Async, Future, Sink, Stream};

pub fn invoke_bpm_battle_warm_up( // this method is called from the poll parent method
    receiver: &mut Option<mpsc::Receiver<bool>>,
    msg: BattleWarmUp,
    addr: Addr<Bpm>,

tobz

pub fn connect_async<P, S>(
  plugin: P,
  destination: &str,
) -> impl Future<Item = Connection, Error = ()>
where
  P: Plugin + 'static,
  S: Session,
{
    let destination = destination.to_owned();

tobz

diff --git a/src/backend/backend.rs b/src/backend/backend.rs
index fc64df4..c2fbe40 100644
--- a/src/backend/backend.rs
+++ b/src/backend/backend.rs
@@ -25,12 +25,34 @@ use futures::{
     future::{ok, Either, Shared},
     prelude::*,
     sync::mpsc,
+    Poll,
 };

tobz

fn launch_supervisor(supervisor_rx: mpsc::UnboundedReceiver<SupervisorCommand>, shutdown_tx: oneshot::Sender<()>) {
    let turnstyle = Turnstyle::new();
    let supervisor = supervisor_rx
        .fold(turnstyle, |ts, command| {
            match command {
                SupervisorCommand::Launch => {
                    let closer = ts.join();
                    if let Err(e) = launch_listeners(closer) {
                        return err::<Turnstyle, CreationError>(e)
                    }

tobz

# in one window, checkout and build synchrotron
git clone https://github.com/nuclearfurnace/synchrotron.git
git checkout 7b5c96e808955235f5441f74a68d07e128e0c697 # working commit
cargo build --release
target/release/synchrotron

# in another window, install redis and run the benchmark
brew install redis
brew services start redis
redis-cli -p 6379 ping # make sure redis is running