Enables piped output when "stream" defined job file

Author: spirali

CHANGELOG.md

@@ -25,6 +25,7 @@
 
 * Fixed the issue of possible ignoring idle timeout when time request is used.
 * Worker process terminated because of idle timeout now returns zero exit code.
+* Fixes broken streaming when job file is used.
 
 ## 0.23.0
 

crates/hyperqueue/src/client/commands/submit/jobfile.rs

@@ -31,10 +31,10 @@ pub struct JobSubmitFileOpts {
     job: Option<JobId>,
 }
 
-fn create_stdio(def: Option<StdioDefInput>, default: &str, is_log: bool) -> StdioDef {
+fn create_stdio(def: Option<StdioDefInput>, default: &str, has_streaming: bool) -> StdioDef {
     match def {
         None => {
-            if is_log {
+            if has_streaming {
                 StdioDef::Pipe
             } else {
                 StdioDef::File {
@@ -54,14 +54,14 @@ fn create_stdio(def: Option<StdioDefInput>, default: &str, is_log: bool) -> Stdi
     }
 }
 
-fn build_task_description(cfg: TaskConfigDef) -> TaskDescription {
+fn build_task_description(cfg: TaskConfigDef, has_streaming: bool) -> TaskDescription {
     TaskDescription {
         kind: TaskKind::ExternalProgram(TaskKindProgram {
             program: ProgramDefinition {
                 args: cfg.command.into_iter().map(|x| x.into()).collect(),
                 env: cfg.env,
-                stdout: create_stdio(cfg.stdout, DEFAULT_STDOUT_PATH, false),
-                stderr: create_stdio(cfg.stderr, DEFAULT_STDERR_PATH, false),
+                stdout: create_stdio(cfg.stdout, DEFAULT_STDOUT_PATH, has_streaming),
+                stderr: create_stdio(cfg.stderr, DEFAULT_STDERR_PATH, has_streaming),
                 stdin: cfg.stdin.map(|s| s.as_bytes().into()).unwrap_or_default(),
                 cwd: cfg.cwd.map(|x| x.into()).unwrap_or_else(get_current_dir),
             },
@@ -89,6 +89,7 @@ fn build_task(
     tdef: TaskDef,
     max_id: &mut JobTaskId,
     data_flags: TaskDataFlags,
+    has_streaming: bool,
 ) -> TaskWithDependencies {
     let id = tdef.id.unwrap_or_else(|| {
         *max_id = JobTaskId::new(max_id.as_num() + 1);
@@ -97,13 +98,13 @@ fn build_task(
     TaskWithDependencies {
         id,
         data_flags,
-        task_desc: build_task_description(tdef.config),
+        task_desc: build_task_description(tdef.config, has_streaming),
         task_deps: tdef.deps,
         data_deps: tdef.data_deps,
     }
 }
 
-fn build_job_desc_array(array: ArrayDef) -> JobTaskDescription {
+fn build_job_desc_array(array: ArrayDef, has_streaming: bool) -> JobTaskDescription {
     let ids = array
         .ids
         .unwrap_or_else(|| IntArray::from_range(0, array.entries.len() as JobTaskCount));
@@ -121,13 +122,14 @@ fn build_job_desc_array(array: ArrayDef) -> JobTaskDescription {
     JobTaskDescription::Array {
         ids,
         entries,
-        task_desc: build_task_description(array.config),
+        task_desc: build_task_description(array.config, has_streaming),
     }
 }
 
 fn build_job_desc_individual_tasks(
     tasks: Vec<TaskDef>,
     data_flags: TaskDataFlags,
+    has_streaming: bool,
 ) -> crate::Result<JobTaskDescription> {
     let mut max_id: JobTaskId = tasks
         .iter()
@@ -143,7 +145,7 @@ fn build_job_desc_individual_tasks(
     let mut in_degrees = Map::new();
     let mut consumers: Map<JobTaskId, Vec<_>> = Map::new();
     for task in tasks {
-        let t = build_task(task, &mut max_id, data_flags);
+        let t = build_task(task, &mut max_id, data_flags, has_streaming);
         if in_degrees.insert(t.id, t.task_deps.len()).is_some() {
             return Err(crate::Error::GenericError(format!(
                 "Task {} is defined multiple times",
@@ -190,13 +192,13 @@ fn build_job_desc_individual_tasks(
 
 fn build_job_submit(jdef: JobDef, job_id: Option<JobId>) -> crate::Result<SubmitRequest> {
     let task_desc = if let Some(array) = jdef.array {
-        build_job_desc_array(array)
+        build_job_desc_array(array, jdef.stream.is_some())
     } else {
         let mut data_flags = TaskDataFlags::empty();
         if jdef.data_layer {
             data_flags.insert(TaskDataFlags::ENABLE_DATA_LAYER);
         }
-        build_job_desc_individual_tasks(jdef.tasks, data_flags)?
+        build_job_desc_individual_tasks(jdef.tasks, data_flags, jdef.stream.is_some())?
     };
     Ok(SubmitRequest {
         job_desc: JobDescription {

tests/test_jobfile.py

@@ -21,13 +21,14 @@ def test_job_file_submit_minimal(hq_env: HqEnv, tmp_path):
 
 
 def test_job_file_submit_maximal(hq_env: HqEnv, tmp_path):
+    # This test does not set "stream", to avoid redirecting outputs.
+    # This is tested separately in test_job_file_stream
     hq_env.start_server()
     hq_env.start_workers(3, cpus=4, args=["--resource", "gpus=[0,1]"])
     os.mkdir("output")
     tmp_path.joinpath("job.toml").write_text(
         """
 name = "test-job"
-stream = "output"
 max_fails = 11
 
 [[task]]
@@ -421,3 +422,21 @@ def test_job_file_attach(hq_env: HqEnv, tmp_path):
 
     table = hq_env.command(["job", "info", "1"], as_table=True)
     table.check_row_value("Tasks", "4; Ids: 1,3-5")
+
+
+def test_job_file_stream(hq_env: HqEnv, tmp_path):
+    hq_env.start_server()
+    hq_env.start_worker()
+    tmp_path.joinpath("job.toml").write_text(
+        """
+stream = "output"
+
+[[task]]
+id = 1
+command = ["bash", "-c", "echo 'Hello'"]
+"""
+    )
+    hq_env.command(["job", "submit-file", "job.toml"])
+    wait_for_job_state(hq_env, 1, "FINISHED")
+    result = hq_env.command(["output-log", "output", "cat", "1", "stdout"])
+    assert result == "Hello\n"