Rust orchestrator - #433
Conversation
| } | ||
|
|
||
| #[derive(Clone, Debug, PartialEq)] | ||
| pub enum Status { |
There was a problem hiding this comment.
Probably overkill, but what about other states, e.g. created?
Is there a way of indicating the number of retries here / somewhere else
| } | ||
|
|
||
| #[derive(Debug, PartialEq)] | ||
| pub struct WorkerSpec { |
There was a problem hiding this comment.
might be useful to know if a worker is internal (aka from python with @worker.task) or external (a binary we don't control).
Also have a mention about the expected behavior of workers: binary worker_cal_args etc.
| /// The [`AssetStorage`] defines the minimum methods required for Assets to be stored. | ||
| /// | ||
| /// The interface is essentially a key-value store, keyed by [`AssetKey`]s. | ||
| pub trait AssetStorage: Send + Sync { |
There was a problem hiding this comment.
something like invalidate might be useful in the future
There was a problem hiding this comment.
Makes sense, I guess we may also want to be able to do something like specify retention periods or similar?
| /// The names and locations of the inputs to the task. | ||
| pub inputs: HashMap<String, AssetSpec>, | ||
| /// The names of the outputs of the task. | ||
| // TODO: We can find this out from the workers |
There was a problem hiding this comment.
but might be the case that we don't need all of the outputs the worker declares
| /// | ||
| /// If not specified the [Executor] will save the outputs to a default | ||
| /// [`AssetStorage`][crate::asset_storage::AssetStorage] instead. | ||
| pub output_storage_name: Option<String>, |
There was a problem hiding this comment.
implies that all task outputs will be written to the same storage.
Thinking of the ntchem example file outputs could be different from the return value of the binary
There was a problem hiding this comment.
It's something we could do later, I just did the easiest thing
There was a problem hiding this comment.
It also doesn't need to be optional - we do always know what the output storage name should be at this point
| pub struct WorkerSpec { | ||
| /// The name of the Worker. | ||
| pub worker_name: String, | ||
| } |
There was a problem hiding this comment.
I assume this is tbd.? you mention meta data further down but I guess it's supposed to be more than the name? At least in the current design, could point to the api/give an appreviated version of it
There was a problem hiding this comment.
Indeed, I would like to flesh this out with information about which tasks are available and what their signatures are.
There was a problem hiding this comment.
It kind of links in with trying to improve the IDL as well
|
|
||
| /// [`InMemoryEnvironmentSpec`] determines the default execution environment of | ||
| /// [`InMemoryExecutor`] or what is requested as part of a [`TaskPlan`]. | ||
| // TODO: It's perhaps a bit unclear what this means for in-memory execution? |
There was a problem hiding this comment.
I'm not sure I understand. I guess this would be something like available modules?
But if a worker is installed this would mean everything should be acaialble, else how would it run?
And I assume we don't target stuff like numa information
| } | ||
|
|
||
| /// The [Executor] defines the minimum methods required for Task execution. | ||
| pub trait Executor: Send + Sync { |
There was a problem hiding this comment.
I think we will need something like register_worker
There was a problem hiding this comment.
Could be useful, but it might depend a bit about how long we expect Executor objects to live, something to discuss in person tomorrow perhaps?
| // This guarantees the tasks will complete in order as far as the runtime | ||
| // is concerned. |
There was a problem hiding this comment.
is this a necessary requrirement? I thought only the dispatch order matters (i.e. all inputs must be ready)
There was a problem hiding this comment.
It is not a requirement and I'm also pretty inconsistent about it, I guess ideally we might want to pass the task dependencies into the executor as well but I didn't think enough about the correct way to do this.
| pub struct SubprocessExecutor { | ||
| event_sender: mpsc::Sender<Event>, | ||
| event_receiver: Mutex<Option<mpsc::Receiver<Event>>>, | ||
| cancel_senders: Mutex<HashMap<u32, oneshot::Sender<()>>>, |
There was a problem hiding this comment.
whats the reason this is oneshot and not mpsc?
There was a problem hiding this comment.
There's a slight inconsistency in the way I implemented the subprocess and the inmemory executors that I can probably fix where the subprocess executor futures are just spawned into the background and they handle their own cancellation, whereas the inmemory one has a central cancellation channel. As such cancellation is performed with a oneshot channel for the subprocess one as only one message needs to be sent.
| let asset_storage_registry_lock = asset_storage_registry | ||
| .read() | ||
| .map_err(|err| miette!("Failed to lock AssetStorageRegistry for reading: {err}"))?; | ||
| if !asset_storage_registry_lock.contains_key(default_storage_name) { | ||
| return Err(miette!("default_storage_name not in registry")); | ||
| } | ||
|
|
||
| if !executor_registry.contains_key(default_executor_name) { | ||
| return Err(miette!("default_executor_name not in registry")); | ||
| } |
There was a problem hiding this comment.
depending how often we need this we could change this to a function on the registry?
There was a problem hiding this comment.
Sure, right now it just a type alias, but we could wrap it up into a struct if there's a small number of methods that we use like this.
| let default_executor_name = &self.default_executor_name; | ||
| let executor = self | ||
| .executor_registry | ||
| .get(&self.default_executor_name) |
There was a problem hiding this comment.
do I understand correctly that we always choose the same executor?
There was a problem hiding this comment.
Currently yes, I would like to be able to set the executor on the node but I am not sure what the best way to do this is. As in, should be metadata or similar?
| } | ||
| Node::Output {} => { | ||
| // Notify that the outputs are ready | ||
| let mut event_sender = self.event_sender.clone(); |
There was a problem hiding this comment.
who is consuming this?
There was a problem hiding this comment.
All of the events end up in the stream provided by listen. Ideally I want to then use this stream to build up the execution state somewhere which then drives the next round of execution.
| .into_diagnostic()?; | ||
| } | ||
| Node::Eval {} => { | ||
| // spawn a sub-orchestrator? |
There was a problem hiding this comment.
I don't think this one possible way but then we need to deal with nested orchestrators
There was a problem hiding this comment.
Yeah I've been thinking about this today and I think we want to avoided nested orchestrators but it needs some thinking about how we handle locations.
| fn load(&self, key: &AssetKey) -> miette::Result<Vec<u8>> { | ||
| let location = self.location(key); | ||
| let file = File::open(self.location(key)) | ||
| let mut file = File::open(self.location(key)) |
There was a problem hiding this comment.
why is this mut now?
There was a problem hiding this comment.
Because the read trait needs it to change where it is looking at in the file, I think before we were passing ownership to serde_json which was doing it's own mutation
2 participants
No description provided.