Ask
We have now mentioned multiple times that we want to use the “ask”-pattern again, as we did in the introduction briefly already. The “ask”-pattern is simply a mechanism to translate from message-based communication into a thread- or future-based model. It does so by coupling a request message with a future that is to be fulfilled with a response for the request. On the sending thread, the one with the future, the result can then be waited for at some point, when it is needed, via blocking, for example. The receiving Actor, on the other hand, gets a combination of the request message with a promise, an Ask instance, that it can fulfill with the response at any later time. Thus, the Message type for such an actor is not Request but rather Ask<Request, Response>.
Note: The futures returned by Kompact’s “ask”-API conform with Rust’s built-in async/await mechanism. On top of that Kompact offers some convenience methods that can be called on a
KFutureto make the common case of blocking on the result easier.
Manager
The Message type for the manager is thus Ask<Work, WorkResponse>, which we already saw when describing its state. In order to access the actual Work instance in our receive_local(...) implementation, we use the Ask::request() function.
We then must distribute the work more or less evenly over the available workers. If no workers are available, the manager simply does all the work itself. Otherwise we’ll figure out what constitutes and “equal share” (i.e. the stride) and then use it to step through the indices into the data, producing sub-ranges, which we send to each worker immediately. Since it may sometimes happen due to rounding that we have a tiny bit of work left at the end, we just do that at the manager and put it directly into the self.result_accumulator. This extra work, it the reason for the previously mentioned + 1 whenever we are considering the length of the self.result_accumulator. It is simply the manager’s share of the work. In order to keep this length consistent, we will simply push the work.neutral element whenever the manager actually doesn’t do any work. Finally, we need to remember to store the request in self.outstanding_request so we can reply to it later when all responses have arrived.
#![allow(clippy::unused_unit)]
use kompact::prelude::*;
use std::{env, fmt, ops::Range, sync::Arc};
struct Work {
data: Arc<[u64]>,
merger: fn(u64, &u64) -> u64,
neutral: u64,
}
impl Work {
fn with(data: Vec<u64>, merger: fn(u64, &u64) -> u64, neutral: u64) -> Self {
let moved_data: Arc<[u64]> = data.into_boxed_slice().into();
Work {
data: moved_data,
merger,
neutral,
}
}
}
impl fmt::Debug for Work {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"Work{{
data=<data of length={}>,
merger=<function>,
neutral={}
}}",
self.data.len(),
self.neutral
)
}
}
struct WorkPart {
data: Arc<[u64]>,
range: Range<usize>,
merger: fn(u64, &u64) -> u64,
neutral: u64,
}
impl WorkPart {
fn from(work: &Work, range: Range<usize>) -> Self {
WorkPart {
data: work.data.clone(),
range,
merger: work.merger,
neutral: work.neutral,
}
}
}
impl fmt::Debug for WorkPart {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"WorkPart{{
data=<data of length={}>,
range={:?},
merger=<function>,
neutral={}
}}",
self.data.len(),
self.range,
self.neutral
)
}
}
#[derive(Clone, Debug)]
struct WorkResult(u64);
struct WorkerPort;
impl Port for WorkerPort {
type Indication = WorkResult;
type Request = Never;
}
#[derive(ComponentDefinition)]
struct Manager {
ctx: ComponentContext<Self>,
worker_port: RequiredPort<WorkerPort>,
num_workers: usize,
workers: Vec<Arc<Component<Worker>>>,
worker_refs: Vec<ActorRefStrong<WorkPart>>,
outstanding_request: Option<Ask<Work, WorkResult>>,
result_accumulator: Vec<u64>,
}
impl Manager {
fn new(num_workers: usize) -> Self {
Manager {
ctx: ComponentContext::uninitialised(),
worker_port: RequiredPort::uninitialised(),
num_workers,
workers: Vec::with_capacity(num_workers),
worker_refs: Vec::with_capacity(num_workers),
outstanding_request: None,
result_accumulator: Vec::with_capacity(num_workers + 1),
}
}
}
impl ComponentLifecycle for Manager {
fn on_start(&mut self) -> Handled {
// set up our workers
for _i in 0..self.num_workers {
let worker = self.ctx.system().create(Worker::new);
worker.connect_to_required(self.worker_port.share());
let worker_ref = worker.actor_ref().hold().expect("live");
self.ctx.system().start(&worker);
self.workers.push(worker);
self.worker_refs.push(worker_ref);
}
Handled::Ok
}
fn on_stop(&mut self) -> Handled {
// clean up after ourselves
self.worker_refs.clear();
let system = self.ctx.system();
self.workers.drain(..).for_each(|worker| {
system.stop(&worker);
});
Handled::Ok
}
fn on_kill(&mut self) -> Handled {
self.on_stop()
}
}
impl Require<WorkerPort> for Manager {
fn handle(&mut self, event: WorkResult) -> Handled {
if self.outstanding_request.is_some() {
self.result_accumulator.push(event.0);
if self.result_accumulator.len() == (self.num_workers + 1) {
let ask = self.outstanding_request.take().expect("ask");
let work: &Work = ask.request();
let res = self
.result_accumulator
.iter()
.fold(work.neutral, work.merger);
self.result_accumulator.clear();
let reply = WorkResult(res);
ask.reply(reply).expect("reply");
}
} else {
error!(
self.log(),
"Got a response without an outstanding promise: {:?}", event
);
}
Handled::Ok
}
}
impl Actor for Manager {
type Message = Ask<Work, WorkResult>;
fn receive_local(&mut self, msg: Self::Message) -> Handled {
assert!(
self.outstanding_request.is_none(),
"One request at a time, please!"
);
let work: &Work = msg.request();
if self.num_workers == 0 {
// manager gotta work itself -> very unhappy manager
let res = work.data.iter().fold(work.neutral, work.merger);
msg.reply(WorkResult(res)).expect("reply");
} else {
let len = work.data.len();
let stride = len / self.num_workers;
let mut start = 0usize;
let mut index = 0;
while start < len && index < self.num_workers {
let end = len.min(start + stride);
let range = start..end;
info!(self.log(), "Assigning {:?} to worker #{}", range, index);
let msg = WorkPart::from(work, range);
let worker = &self.worker_refs[index];
worker.tell(msg);
start += stride;
index += 1;
}
if start < len {
// manager just does the rest itself
let res = work.data[start..len].iter().fold(work.neutral, work.merger);
self.result_accumulator.push(res);
} else {
// just put a neutral element in there, so our count is right in the end
self.result_accumulator.push(work.neutral);
}
self.outstanding_request = Some(msg);
}
Handled::Ok
}
fn receive_network(&mut self, _msg: NetMessage) -> Handled {
unimplemented!("Still ignoring networking stuff.");
}
}
#[derive(ComponentDefinition)]
struct Worker {
ctx: ComponentContext<Self>,
worker_port: ProvidedPort<WorkerPort>,
}
impl Worker {
fn new() -> Self {
Worker {
ctx: ComponentContext::uninitialised(),
worker_port: ProvidedPort::uninitialised(),
}
}
}
ignore_lifecycle!(Worker);
ignore_requests!(WorkerPort, Worker);
impl Actor for Worker {
type Message = WorkPart;
fn receive_local(&mut self, msg: Self::Message) -> Handled {
let my_slice = &msg.data[msg.range];
let res = my_slice.iter().fold(msg.neutral, msg.merger);
self.worker_port.trigger(WorkResult(res));
Handled::Ok
}
fn receive_network(&mut self, _msg: NetMessage) -> Handled {
unimplemented!("Still ignoring networking stuff.");
}
}
pub fn main() {
let args: Vec<String> = env::args().collect();
assert_eq!(
3,
args.len(),
"Invalid arguments! Must give number of workers and size of the data array."
);
let num_workers: usize = args[1].parse().expect("number");
let data_size: usize = args[2].parse().expect("number");
run_task(num_workers, data_size);
}
fn run_task(num_workers: usize, data_size: usize) {
let system = KompactConfig::default().build().expect("system");
let manager = system.create(move || Manager::new(num_workers));
system.start(&manager);
let manager_ref = manager.actor_ref().hold().expect("live");
let data: Vec<u64> = (1..=data_size).map(|v| v as u64).collect();
let work = Work::with(data, overflowing_sum, 0u64);
println!("Sending request...");
let res = manager_ref.ask(work).wait();
println!("*******\nGot result: {}\n*******", res.0);
assert_eq!(triangular_number(data_size as u64), res.0);
system.shutdown().expect("shutdown");
}
fn triangular_number(n: u64) -> u64 {
(n * (n + 1u64)) / 2u64
}
fn overflowing_sum(lhs: u64, rhs: &u64) -> u64 {
lhs.overflowing_add(*rhs).0
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_workers() {
run_task(3, 1000);
}
}
Sending Work
When sending work to the manager from the main-thread, we can construct the required Ask instance with ActorRef::ask(...). Since we only want to handle a single request at a time, we will immediately wait() for the result of the future.
#![allow(clippy::unused_unit)]
use kompact::prelude::*;
use std::{env, fmt, ops::Range, sync::Arc};
struct Work {
data: Arc<[u64]>,
merger: fn(u64, &u64) -> u64,
neutral: u64,
}
impl Work {
fn with(data: Vec<u64>, merger: fn(u64, &u64) -> u64, neutral: u64) -> Self {
let moved_data: Arc<[u64]> = data.into_boxed_slice().into();
Work {
data: moved_data,
merger,
neutral,
}
}
}
impl fmt::Debug for Work {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"Work{{
data=<data of length={}>,
merger=<function>,
neutral={}
}}",
self.data.len(),
self.neutral
)
}
}
struct WorkPart {
data: Arc<[u64]>,
range: Range<usize>,
merger: fn(u64, &u64) -> u64,
neutral: u64,
}
impl WorkPart {
fn from(work: &Work, range: Range<usize>) -> Self {
WorkPart {
data: work.data.clone(),
range,
merger: work.merger,
neutral: work.neutral,
}
}
}
impl fmt::Debug for WorkPart {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
f,
"WorkPart{{
data=<data of length={}>,
range={:?},
merger=<function>,
neutral={}
}}",
self.data.len(),
self.range,
self.neutral
)
}
}
#[derive(Clone, Debug)]
struct WorkResult(u64);
struct WorkerPort;
impl Port for WorkerPort {
type Indication = WorkResult;
type Request = Never;
}
#[derive(ComponentDefinition)]
struct Manager {
ctx: ComponentContext<Self>,
worker_port: RequiredPort<WorkerPort>,
num_workers: usize,
workers: Vec<Arc<Component<Worker>>>,
worker_refs: Vec<ActorRefStrong<WorkPart>>,
outstanding_request: Option<Ask<Work, WorkResult>>,
result_accumulator: Vec<u64>,
}
impl Manager {
fn new(num_workers: usize) -> Self {
Manager {
ctx: ComponentContext::uninitialised(),
worker_port: RequiredPort::uninitialised(),
num_workers,
workers: Vec::with_capacity(num_workers),
worker_refs: Vec::with_capacity(num_workers),
outstanding_request: None,
result_accumulator: Vec::with_capacity(num_workers + 1),
}
}
}
impl ComponentLifecycle for Manager {
fn on_start(&mut self) -> Handled {
// set up our workers
for _i in 0..self.num_workers {
let worker = self.ctx.system().create(Worker::new);
worker.connect_to_required(self.worker_port.share());
let worker_ref = worker.actor_ref().hold().expect("live");
self.ctx.system().start(&worker);
self.workers.push(worker);
self.worker_refs.push(worker_ref);
}
Handled::Ok
}
fn on_stop(&mut self) -> Handled {
// clean up after ourselves
self.worker_refs.clear();
let system = self.ctx.system();
self.workers.drain(..).for_each(|worker| {
system.stop(&worker);
});
Handled::Ok
}
fn on_kill(&mut self) -> Handled {
self.on_stop()
}
}
impl Require<WorkerPort> for Manager {
fn handle(&mut self, event: WorkResult) -> Handled {
if self.outstanding_request.is_some() {
self.result_accumulator.push(event.0);
if self.result_accumulator.len() == (self.num_workers + 1) {
let ask = self.outstanding_request.take().expect("ask");
let work: &Work = ask.request();
let res = self
.result_accumulator
.iter()
.fold(work.neutral, work.merger);
self.result_accumulator.clear();
let reply = WorkResult(res);
ask.reply(reply).expect("reply");
}
} else {
error!(
self.log(),
"Got a response without an outstanding promise: {:?}", event
);
}
Handled::Ok
}
}
impl Actor for Manager {
type Message = Ask<Work, WorkResult>;
fn receive_local(&mut self, msg: Self::Message) -> Handled {
assert!(
self.outstanding_request.is_none(),
"One request at a time, please!"
);
let work: &Work = msg.request();
if self.num_workers == 0 {
// manager gotta work itself -> very unhappy manager
let res = work.data.iter().fold(work.neutral, work.merger);
msg.reply(WorkResult(res)).expect("reply");
} else {
let len = work.data.len();
let stride = len / self.num_workers;
let mut start = 0usize;
let mut index = 0;
while start < len && index < self.num_workers {
let end = len.min(start + stride);
let range = start..end;
info!(self.log(), "Assigning {:?} to worker #{}", range, index);
let msg = WorkPart::from(work, range);
let worker = &self.worker_refs[index];
worker.tell(msg);
start += stride;
index += 1;
}
if start < len {
// manager just does the rest itself
let res = work.data[start..len].iter().fold(work.neutral, work.merger);
self.result_accumulator.push(res);
} else {
// just put a neutral element in there, so our count is right in the end
self.result_accumulator.push(work.neutral);
}
self.outstanding_request = Some(msg);
}
Handled::Ok
}
fn receive_network(&mut self, _msg: NetMessage) -> Handled {
unimplemented!("Still ignoring networking stuff.");
}
}
#[derive(ComponentDefinition)]
struct Worker {
ctx: ComponentContext<Self>,
worker_port: ProvidedPort<WorkerPort>,
}
impl Worker {
fn new() -> Self {
Worker {
ctx: ComponentContext::uninitialised(),
worker_port: ProvidedPort::uninitialised(),
}
}
}
ignore_lifecycle!(Worker);
ignore_requests!(WorkerPort, Worker);
impl Actor for Worker {
type Message = WorkPart;
fn receive_local(&mut self, msg: Self::Message) -> Handled {
let my_slice = &msg.data[msg.range];
let res = my_slice.iter().fold(msg.neutral, msg.merger);
self.worker_port.trigger(WorkResult(res));
Handled::Ok
}
fn receive_network(&mut self, _msg: NetMessage) -> Handled {
unimplemented!("Still ignoring networking stuff.");
}
}
pub fn main() {
let args: Vec<String> = env::args().collect();
assert_eq!(
3,
args.len(),
"Invalid arguments! Must give number of workers and size of the data array."
);
let num_workers: usize = args[1].parse().expect("number");
let data_size: usize = args[2].parse().expect("number");
run_task(num_workers, data_size);
}
fn run_task(num_workers: usize, data_size: usize) {
let system = KompactConfig::default().build().expect("system");
let manager = system.create(move || Manager::new(num_workers));
system.start(&manager);
let manager_ref = manager.actor_ref().hold().expect("live");
let data: Vec<u64> = (1..=data_size).map(|v| v as u64).collect();
let work = Work::with(data, overflowing_sum, 0u64);
println!("Sending request...");
let res = manager_ref.ask(work).wait();
println!("*******\nGot result: {}\n*******", res.0);
assert_eq!(triangular_number(data_size as u64), res.0);
system.shutdown().expect("shutdown");
}
fn triangular_number(n: u64) -> u64 {
(n * (n + 1u64)) / 2u64
}
fn overflowing_sum(lhs: u64, rhs: &u64) -> u64 {
lhs.overflowing_add(*rhs).0
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_workers() {
run_task(3, 1000);
}
}
Note: For situations where the
Askinstance is nested, for example, into an enum, Kompact offers theActorRef::ask_withfunction. Instead of aRequestvalue,ask_withexpects a function that takes aKPromise<Result>and produces the Actor’sMessagetype. This also allows for customAskvariants with more fields, for example.