split: fix round-robin filtering to handle closed filter processes correctly

- Track closed state for output writers to stop writing to them when the underlying process/file is closed
- Ensure all files are created when not in elide-empty-files mode, even if no data is written
- Flush writers in round-robin mode to ensure data is sent to filter processes immediately
- Add integration tests for round-robin filtering and early closure handling

Author: naoNao89

src/uu/split/src/split.rs

@@ -923,6 +923,7 @@ struct OutFile {
     filename: String,
     maybe_writer: Option<BufWriter<Box<dyn Write>>>,
     is_new: bool,
+    is_closed: bool,
 }
 
 /// A set of output files
@@ -994,6 +995,7 @@ impl ManageOutFiles for OutFiles {
                 filename,
                 maybe_writer,
                 is_new: true,
+                is_closed: false,
             });
         }
         Ok(out_files)
@@ -1198,7 +1200,7 @@ where
                 None => {
                     let idx = (i - 1) as usize;
                     let writer = out_files.get_writer(idx, settings)?;
-                    writer.write_all(buf)?;
+                    custom_write_all(buf, writer, settings)?;
                 }
             }
         } else {
@@ -1393,13 +1395,21 @@ where
     // Create one writer for each chunk.
     // This will create each of the underlying files
     // or stdin pipes to child shell/command processes if in `--filter` mode
+    // If in N chunks mode
+    // Create one writer for each chunk.
+    // This will create each of the underlying files
+    // or stdin pipes to child shell/command processes if in `--filter` mode.
+    // We use lazy initialization (is_writer_optional = true) to avoid hitting
+    // system resource limits (like EMFILE or EAGAIN on macOS) prematurely,
+    // especially when many filter processes would be spawned at once.
     if kth_chunk.is_none() {
-        out_files = OutFiles::init(num_chunks, settings, settings.elide_empty_files)?;
+        out_files = OutFiles::init(num_chunks, settings, true)?;
     }
 
-    let num_chunks: usize = num_chunks.try_into().unwrap();
+    let num_chunks_size: usize = num_chunks.try_into().unwrap();
     let sep = settings.separator;
-    let mut closed_writers = 0;
+    let mut any_writer_open = true;
+    let mut wrote_in_current_round = false;
 
     let mut i = 0;
     loop {
@@ -1413,22 +1423,56 @@ where
 
         let bytes = line.as_slice();
         if let Some(chunk_number) = kth_chunk {
-            if (i % num_chunks) == (chunk_number - 1) as usize {
+            if (i % num_chunks_size) == (chunk_number - 1) as usize {
                 stdout_writer.write_all(bytes)?;
             }
         } else {
-            let writer = out_files.get_writer(i % num_chunks, settings)?;
-            let writer_stdin_open = custom_write_all(bytes, writer, settings)?;
-            if !writer_stdin_open {
-                closed_writers += 1;
+            let chunk_idx = i % num_chunks_size;
+            if !out_files[chunk_idx].is_closed {
+                let writer = out_files.get_writer(chunk_idx, settings)?;
+                let writer_stdin_open = custom_write_all(bytes, writer, settings)?;
+                if writer_stdin_open {
+                    wrote_in_current_round = true;
+                    // Ensure data is sent to filter processes immediately in round-robin mode
+                    if let Err(e) = writer.flush() {
+                        if ignorable_io_error(&e, settings) {
+                            out_files[chunk_idx].is_closed = true;
+                        } else {
+                            return Err(e.into());
+                        }
+                    }
+                } else {
+                    out_files[chunk_idx].is_closed = true;
+                }
             }
         }
         i += 1;
-        if closed_writers == num_chunks {
+
+        // After completing a full round of chunks
+        if i % num_chunks_size == 0 && kth_chunk.is_none() {
+            if !wrote_in_current_round {
+                any_writer_open = false;
+            }
+            wrote_in_current_round = false;
+        }
+
+        if !any_writer_open {
             // all writers are closed - stop reading
             break;
         }
     }
+
+    // GNU coreutils compatibility: Ensure all files are created if !elide_empty_files,
+    // even if no data was written to them.
+    if kth_chunk.is_none() && !settings.elide_empty_files {
+        for chunk_idx in 0..num_chunks_size {
+            // We only need to "touch" the file (instantiate the writer).
+            // get_writer will create the file/process if it doesn't exist.
+            // If it already exists, it does nothing.
+            let _ = out_files.get_writer(chunk_idx, settings);
+        }
+    }
+
     Ok(())
 }
 

tests/by-util/test_split.rs

@@ -2091,3 +2091,32 @@ fn test_split_directory_already_exists() {
         .no_stdout()
         .stderr_is("split: xaa: Is a directory\n");
 }
+
+#[test]
+fn test_round_robin_to_files() {
+    let (at, mut ucmd) = at_and_ucmd!();
+    at.write("input", "line1\nline2\nline3\nline4\n");
+
+    ucmd.args(&["-n", "r/2", "input", "out"]).succeeds();
+
+    assert_eq!(at.read("outaa"), "line1\nline3\n");
+    assert_eq!(at.read("outab"), "line2\nline4\n");
+}
+
+#[test]
+#[cfg(not(windows))]
+fn test_round_robin_filter_closes_early() {
+    // This test exercises the wrote_in_current_round logic
+    // We use a filter that only accepts one line then closes.
+    let (at, mut ucmd) = at_and_ucmd!();
+    at.write("input", "line1\nline2\nline3\nline4\n");
+
+    // Filter 'head -n 1' will close its stdin after reading one line.
+    // In round-robin, this should cause split to mark that chunk as closed.
+    ucmd.args(&["--filter", "head -n 1 > $FILE", "-n", "r/2", "input", "out"])
+        .succeeds();
+
+    assert_eq!(at.read("outaa"), "line1\n");
+    // outab also only gets one line because 'head -n 1' closes for it too.
+    assert_eq!(at.read("outab"), "line2\n");
+}