Removed some very old and obsolete code from Tako

crates/hyperqueue/src/server/client/submit.rs

@@ -307,8 +307,6 @@ fn build_tasks_array(
             resources: task_desc.resources.clone(),
             n_outputs: 0,
             time_limit: task_desc.time_limit,
-            keep: false,
-            observe: true,
             priority: task_desc.priority,
             crash_limit: task_desc.crash_limit,
         }],
@@ -362,8 +360,6 @@ fn build_tasks_graph(
                     n_outputs: 0,
                     time_limit: task.time_limit,
                     priority: task.priority,
-                    keep: false,
-                    observe: true,
                     crash_limit: task.crash_limit,
                 });
                 index

crates/tako/src/gateway.rs

@@ -106,12 +106,6 @@ pub struct SharedTaskConfiguration {
     #[serde(default)]
     pub priority: Priority,
 
-    #[serde(default)]
-    pub keep: bool,
-
-    #[serde(default)]
-    pub observe: bool,
-
     pub crash_limit: u32,
 }
 
@@ -136,11 +130,6 @@ pub struct NewTasksMessage {
     pub adjust_instance_id: Map<TaskId, InstanceId>,
 }
 
-#[derive(Deserialize, Serialize, Debug)]
-pub struct ObserveTasksMessage {
-    pub tasks: Vec<TaskId>,
-}
-
 #[derive(Deserialize, Serialize, Debug)]
 pub struct TaskInfoRequest {
     pub tasks: Vec<TaskId>, // If empty, then all tasks are assumed
@@ -190,7 +179,6 @@ pub struct NewWorkerQuery {
 #[serde(tag = "op")]
 pub enum FromGatewayMessage {
     NewTasks(NewTasksMessage),
-    ObserveTasks(ObserveTasksMessage),
     CancelTasks(CancelTasks),
     GetTaskInfo(TaskInfoRequest),
     ServerInfo,
@@ -234,10 +222,6 @@ impl Serialize for TaskState {
     }
 }
 
-/* User can receive this updates when task is registered with "observe flag"
-  Note: Error state is NOT there, it is sent separately as TaskFail,
-  because task fail is received even without "observe" flag.
-*/
 #[derive(Serialize, Deserialize, Debug)]
 pub struct TaskUpdate {
     pub id: TaskId,

crates/tako/src/internal/server/client.rs

@@ -3,16 +3,15 @@ use tokio::sync::mpsc::UnboundedSender;
 
 use crate::gateway::{
     CancelTasksResponse, FromGatewayMessage, NewTasksMessage, NewTasksResponse,
-    SharedTaskConfiguration, TaskInfo, TaskState, TaskUpdate, TasksInfoResponse, ToGatewayMessage,
+    SharedTaskConfiguration, TaskInfo, TaskState, TasksInfoResponse, ToGatewayMessage,
 };
+use crate::internal::common::resources::request::ResourceRequestEntry;
 use crate::internal::messages::worker::ToWorkerMessage;
+use crate::internal::scheduler::query::compute_new_worker_query;
 use crate::internal::server::comm::{Comm, CommSender, CommSenderRef};
 use crate::internal::server::core::{Core, CoreRef};
-use crate::internal::server::reactor::{on_cancel_tasks, on_new_tasks, on_set_observe_flag};
+use crate::internal::server::reactor::{on_cancel_tasks, on_new_tasks};
 use crate::internal::server::task::{Task, TaskConfiguration, TaskInput, TaskRuntimeState};
-//use crate::internal::transfer::transport::make_protocol_builder;
-use crate::internal::common::resources::request::ResourceRequestEntry;
-use crate::internal::scheduler::query::compute_new_worker_query;
 use std::rc::Rc;
 use thin_vec::ThinVec;
 
@@ -107,30 +106,6 @@ pub(crate) async fn process_client_message(
     message: FromGatewayMessage,
 ) -> Option<String> {
     match message {
-        FromGatewayMessage::ObserveTasks(msg) => {
-            let mut core = core_ref.get_mut();
-            let mut comm = comm_ref.get_mut();
-            for task_id in msg.tasks {
-                log::debug!("Client start observing task={}", task_id);
-                if !on_set_observe_flag(&mut core, &mut *comm, task_id, true) {
-                    log::debug!(
-                        "Client ask for observing of invalid (old?) task={}",
-                        task_id
-                    );
-                    client_sender
-                        .send(ToGatewayMessage::TaskUpdate(TaskUpdate {
-                            id: task_id,
-                            state: if core.is_used_task_id(task_id) {
-                                TaskState::Finished
-                            } else {
-                                TaskState::Invalid
-                            },
-                        }))
-                        .unwrap();
-                };
-            }
-            None
-        }
         FromGatewayMessage::NewTasks(msg) => handle_new_tasks(
             &mut core_ref.get_mut(),
             &mut comm_ref.get_mut(),
@@ -243,14 +218,12 @@ fn handle_new_tasks(
         .into_iter()
         .map(|c| {
             assert_eq!(c.n_outputs, 0); // TODO: Implementation for more outputs
-            let keep = c.keep;
-            let observe = c.observe;
-            (Rc::new(create_task_configuration(core, c)), keep, observe)
+            Rc::new(create_task_configuration(core, c))
         })
         .collect();
 
     for cfg in &configurations {
-        if let Err(e) = cfg.0.resources.validate() {
+        if let Err(e) = cfg.resources.validate() {
             return Some(format!("Invalid task request {e:?}"));
         }
     }
@@ -264,13 +237,13 @@ fn handle_new_tasks(
         if idx >= configurations.len() {
             return Some(format!("Invalid configuration index {idx}"));
         }
-        let (conf, keep, observe) = &configurations[idx];
+        let conf = &configurations[idx];
         let inputs: ThinVec<_> = task
             .task_deps
             .iter()
             .map(|&task_id| TaskInput::new_task_dependency(task_id))
             .collect();
-        let task = Task::new(task.id, inputs, conf.clone(), task.body, *keep, *observe);
+        let task = Task::new(task.id, inputs, conf.clone(), task.body);
         tasks.push(task);
     }
     if !msg.adjust_instance_id.is_empty() {

crates/tako/src/internal/server/reactor.rs

@@ -241,9 +241,7 @@ pub(crate) fn on_task_running(
             }
         };
 
-        if task.is_observed() {
-            comm.send_client_task_started(task_id, task.instance_id, worker_ids, context);
-        }
+        comm.send_client_task_started(task_id, task.instance_id, worker_ids, context);
     }
 }
 
@@ -302,10 +300,7 @@ pub(crate) fn on_task_finished(
 
             task.state = TaskRuntimeState::Finished(FinishInfo {});
             comm.ask_for_scheduling();
-
-            if task.is_observed() {
-                comm.send_client_task_finished(task.id);
-            }
+            comm.send_client_task_finished(task.id);
         } else {
             log::debug!("Unknown task finished id={}", msg.id);
             return;
@@ -409,30 +404,6 @@ pub(crate) fn on_steal_response(
     }
 }
 
-#[cfg(test)] // The current version of HQ does not use it, it is now used only in tests
-pub(crate) fn on_reset_keep_flag(core: &mut Core, comm: &mut impl Comm, task_id: TaskId) {
-    let task = core.get_task_mut(task_id);
-    task.set_keep_flag(false);
-    remove_task_if_possible(core, comm, task_id);
-}
-
-pub(crate) fn on_set_observe_flag(
-    core: &mut Core,
-    comm: &mut impl Comm,
-    task_id: TaskId,
-    value: bool,
-) -> bool {
-    if let Some(task) = core.find_task_mut(task_id) {
-        if value && task.is_finished() {
-            comm.send_client_task_finished(task_id);
-        }
-        task.set_observed_flag(value);
-        true
-    } else {
-        false
-    }
-}
-
 fn fail_task_helper(
     core: &mut Core,
     comm: &mut impl Comm,
@@ -529,7 +500,6 @@ pub(crate) fn on_cancel_tasks(
             continue;
         }
 
-        let mut remove = false;
         {
             let (tasks, workers) = core.split_tasks_workers_mut();
             let task = tasks.get_task(task_id);
@@ -564,17 +534,10 @@ pub(crate) fn on_cancel_tasks(
                     running_ids.entry(from_id).or_default().push(task_id);
                 }
                 TaskRuntimeState::Finished(_) => {
-                    if task.is_keeped() {
-                        remove = true;
-                    }
                     already_finished.push(task_id);
                 }
             };
         }
-        if remove {
-            core.get_task_mut(task_id).set_keep_flag(false);
-            remove_task_if_possible(core, comm, task_id);
-        }
     }
 
     for &task_id in &to_unregister {
@@ -609,7 +572,6 @@ fn remove_task_if_possible(core: &mut Core, _comm: &mut impl Comm, task_id: Task
     if !core.get_task(task_id).is_removable() {
         return;
     }
-
     match core.remove_task(task_id) {
         TaskRuntimeState::Finished(_finfo) => { /* Ok */ }
         _ => unreachable!(),

crates/tako/src/internal/server/task.rs

@@ -47,13 +47,10 @@ impl fmt::Debug for TaskRuntimeState {
 bitflags::bitflags! {
     #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
     pub struct TaskFlags: u32 {
-        const KEEP    = 0b00000001;
-        const OBSERVE = 0b00000010;
-
-        const FRESH   = 0b00000100;
+        const FRESH   = 0b00000001;
 
         // This is utilized inside scheduler, it has no meaning between scheduler calls
-        const TAKE   = 0b00001000;
+        const TAKE   = 0b00000010;
     }
 }
 
@@ -129,14 +126,10 @@ impl Task {
         inputs: ThinVec<TaskInput>,
         configuration: Rc<TaskConfiguration>,
         body: Box<[u8]>,
-        keep: bool,
-        observe: bool,
     ) -> Self {
         log::debug!("New task {} {:?}", id, &configuration.resources);
 
         let mut flags = TaskFlags::empty();
-        flags.set(TaskFlags::KEEP, keep);
-        flags.set(TaskFlags::OBSERVE, observe);
         flags.set(TaskFlags::FRESH, true);
 
         Self {
@@ -211,36 +204,16 @@ impl Task {
         &self.consumers
     }
 
-    #[inline]
-    pub(crate) fn set_keep_flag(&mut self, value: bool) {
-        self.flags.set(TaskFlags::KEEP, value);
-    }
-
     #[inline]
     pub(crate) fn set_take_flag(&mut self, value: bool) {
         self.flags.set(TaskFlags::TAKE, value);
     }
 
-    #[inline]
-    pub(crate) fn set_observed_flag(&mut self, value: bool) {
-        self.flags.set(TaskFlags::OBSERVE, value);
-    }
-
     #[inline]
     pub(crate) fn set_fresh_flag(&mut self, value: bool) {
         self.flags.set(TaskFlags::FRESH, value);
     }
 
-    #[inline]
-    pub(crate) fn is_observed(&self) -> bool {
-        self.flags.contains(TaskFlags::OBSERVE)
-    }
-
-    #[inline]
-    pub(crate) fn is_keeped(&self) -> bool {
-        self.flags.contains(TaskFlags::KEEP)
-    }
-
     #[inline]
     pub(crate) fn is_fresh(&self) -> bool {
         self.flags.contains(TaskFlags::FRESH)
@@ -253,7 +226,7 @@ impl Task {
 
     #[inline]
     pub(crate) fn is_removable(&self) -> bool {
-        self.consumers.is_empty() && !self.is_keeped() && self.is_finished()
+        self.consumers.is_empty() && self.is_finished()
     }
 
     pub(crate) fn collect_consumers(&self, taskmap: &TaskMap) -> Set<TaskId> {

crates/tako/src/internal/tests/integration/test_basic.rs

@@ -27,11 +27,7 @@ async fn test_submit_simple_task_ok() {
         let ids = handler
             .submit(
                 GraphBuilder::default()
-                    .task(
-                        TaskConfigBuilder::default()
-                            .args(simple_args(&["/bin/hostname"]))
-                            .keep(true),
-                    )
+                    .task(TaskConfigBuilder::default().args(simple_args(&["/bin/hostname"])))
                     .simple_task(&["/bin/hostname"])
                     .task(
                         TaskConfigBuilder::default()
@@ -134,8 +130,6 @@ async fn test_cancel_error_task() {
 
         let response = cancel(&mut handle, &[1]).await;
         assert_eq!(response.already_finished, vec![1].to_ids());
-
-        assert!(handle.wait(&[1]).await.get(1).is_invalid());
     })
     .await;
 }

crates/tako/src/internal/tests/integration/utils/server.rs

@@ -10,8 +10,8 @@ use tokio::task::{JoinHandle, LocalSet};
 use tokio::time::timeout;
 
 use crate::gateway::{
-    FromGatewayMessage, NewTasksMessage, NewTasksResponse, ObserveTasksMessage,
-    SharedTaskConfiguration, StopWorkerRequest, TaskConfiguration, ToGatewayMessage,
+    FromGatewayMessage, NewTasksMessage, NewTasksResponse, SharedTaskConfiguration,
+    StopWorkerRequest, TaskConfiguration, ToGatewayMessage,
 };
 use crate::internal::common::{Map, Set};
 use crate::internal::server::client::process_client_message;
@@ -205,10 +205,6 @@ impl ServerHandle {
     }
 
     pub async fn wait<T: Into<TaskId> + Copy>(&mut self, tasks: &[T]) -> TaskWaitResultMap {
-        let msg = ObserveTasksMessage {
-            tasks: tasks.iter().map(|&v| v.into()).collect(),
-        };
-        self.send(FromGatewayMessage::ObserveTasks(msg)).await;
         timeout(WAIT_TIMEOUT, wait_for_tasks(self, tasks.to_vec()))
             .await
             .unwrap()

crates/tako/src/internal/tests/integration/utils/task.rs

@@ -76,8 +76,6 @@ pub fn build_task_def_from_config(
 ) -> (TaskConfiguration, SharedTaskConfiguration) {
     let TaskConfig {
         id,
-        keep,
-        observe,
         time_limit,
         resources,
         args,
@@ -115,8 +113,6 @@ pub fn build_task_def_from_config(
         n_outputs: 0,
         time_limit,
         priority: 0,
-        keep,
-        observe: observe.unwrap_or(true),
         crash_limit: 5,
     };
     (
@@ -136,11 +132,6 @@ pub struct TaskConfig {
     #[builder(default)]
     pub id: Option<u64>,
 
-    #[builder(default)]
-    keep: bool,
-    #[builder(default)]
-    observe: Option<bool>,
-
     #[builder(default)]
     time_limit: Option<Duration>,