Provide for worker-configuration parameters #350
Conversation
Tweak the execution API so that worker configuration parameters can be specified along with communication fabric configuration parameters. In particular, this permits the DEFAULT_PROGRESS_MODE environment variable to be replaced with a typed ProgressMode enum, and for its value to be controllable with a command-line option. The new WorkerConfig struct also admits arbitrary typed configuration parameters, which downstream libraries (e.g., differential) can use to store their own worker configuration variables.
communication/src/initialize.rs
Outdated
| pub enum Configuration { | ||
| pub enum CommunicationConfig { |
There was a problem hiding this comment.
I feel like the "recommended style" is leaving this as Configuration but then importing it as CommunicationConfig in timely. I don't think it matters too much outside of this crate (as it will be CommunicationConfig by timely users). Any thoughts?
| /// The progress mode to use. | ||
| pub(crate) progress_mode: ProgressMode, | ||
| /// A map from parameter name to typed parameter values. | ||
| registry: HashMap<String, Arc<dyn Any + Send + Sync>>, |
There was a problem hiding this comment.
Is there a non-obvious reason that this is an Arc instead of a Box?
There was a problem hiding this comment.
Alternately (if it is like, setting up the configs for everyone and I've missed that) is the Send required for Arc?
There was a problem hiding this comment.
Yeah, what we actually need is a clonable Box<dyn Any>, but I don't know how to make that happen easily types besides Arc<dyn Any + Send + Sync>, which is a bit overspecific. Specifically, I don't think Sync is important, but Arc<T> doesn't implement Send unless the underlying type implements Sync too.
I'm sure you could make something work with https://docs.rs/dyn-clone/1.0.4/dyn_clone/, but it's always a big pain in the butt in my experience.
| where | ||
| A: AllocateBuilder+'static, | ||
| T: Send+'static, | ||
| F: Fn(&mut Worker<<A as AllocateBuilder>::Allocator>)->T+Send+Sync+'static { | ||
| initialize_from(builders, others, move |allocator| { | ||
| let mut worker = Worker::new(allocator); | ||
| let mut worker = Worker::new(worker_config.clone(), allocator); |
There was a problem hiding this comment.
This looks like the Arc moment!
timely/src/execute.rs
Outdated
| @@ -269,8 +302,65 @@ pub fn execute_from_args<I, T, F>(iter: I, func: F) -> Result<WorkerGuards<T>,St | |||
| where I: Iterator<Item=String>, | |||
| T:Send+'static, | |||
| F: Fn(&mut Worker<Allocator>)->T+Send+Sync+'static, { | |||
| let configuration = Configuration::from_args(iter)?; | |||
| execute(configuration, func) | |||
| let mut opts = getopts::Options::new(); | |||
There was a problem hiding this comment.
The opts parsing .. I was sort of imagining as part of ExecuteConfig. That way e.g. someone could parse up some arguments, and then maybe flip a few, read out some and print them on screen, stuff like that. Is there a reason to stash this logic here, vs in some constructor/initialization for ExecuteConfig?
Add some methods to the various config structs that makes it possible to assemble a getopts::Options out of various different config structs.
|
I overhauled the naming of these structs so that everything is called |
|
Thanks very much! I think this all looks great! |
|
Woohoo! I think all that's missing then is documentation of what the different progress modes actually do. |
|
I'm happy to add that in post, if that works for you. |
|
It most certainly does! |
|
Documentation added in bafcd7d |
|
|
||
| initialize_from(allocators, other, move |allocator| { | ||
|
|
||
| let mut worker = Worker::new(allocator); | ||
| let mut worker = Worker::new(WorkerConfig::default(), allocator); |
There was a problem hiding this comment.
@frankmcsherry Should this take the value from config.worker instead? I tried configuring idle_merge_effort in DD as below, but it does not look like the config is actually set in Worker.
let mut config = timely::Config::from_args(std::env::args()).expect("error");
differential_dataflow::configure(&mut config.worker, &differential_dataflow::Config::default().idle_merge_effort(Some(100000)));
timely::execute(config, move |worker| {
// code
}Or is this not how it's meant to be used?
There was a problem hiding this comment.
Probably! Lemme look further. Sorry, this stuff is definitely under-exercised.
There was a problem hiding this comment.
Yes, it should be fixed now, with a test in place. Thanks very much for noticing!
@frankmcsherry, as discussed on Slack! No promises that any of this is appealing!
Tweak the execution API so that worker configuration parameters can be
specified along with communication fabric configuration parameters. In
particular, this permits the DEFAULT_PROGRESS_MODE environment variable
to be replaced with a typed ProgressMode enum, and for its value to be
controllable with a command-line option.
The new WorkerConfig struct also admits arbitrary typed configuration
parameters, which downstream libraries (e.g., differential) can use to
store their own worker configuration variables.