Repository: common-workflow-language/cwltool

Issue:
Request for cwltool to support named pipes
## Expected Behavior
Some runners such as Toil might implement file streaming with named pipes when `streamable` flag is true:
https://github.com/DataBiosphere/toil/issues/3469

## Actual Behavior
An exception is raised because a named pipe is not a regular file:
https://github.com/common-workflow-language/cwltool/blob/ed9dd4c3472e940a52dfe90049895f470bfd7329/cwltool/job.py#L180

## Workflow Code
```
#!/usr/bin cwl-runner

cwlVersion: v1.2
class: CommandLineTool
baseCommand: cat

stdout: $(inputs.output_filename)
arguments: ["-A"]

inputs:
  input_file:
    type: File
    streamable: true
    inputBinding:
      position: 1
  output_filename:
    type: string?
    default: out.txt

outputs:
  count:
    type: stdout

```

## Your Environment
* cwltool version: 
3.0.20201203173111 (Linux)


Discussion:
mr-c: Can you add a unit test to prevent regressions?

We had also talked about only allowing named pipes in certain circumstances, so we need a way to only allow them when we are doing streaming, and forbid them otherwise
codecov[bot]: # [Codecov](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language) Report
> Merging [#1469](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language) (c4e9aa5) into [main](https://codecov.io/gh/common-workflow-language/cwltool/commit/ed9dd4c3472e940a52dfe90049895f470bfd7329?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language) (ed9dd4c) will **decrease** coverage by `15.25%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/graphs/tree.svg?width=650&height=150&src=pr&token=xGYp2paYv4&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language)](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language)

```diff
@@             Coverage Diff             @@
##             main    #1469       +/-   ##
===========================================
- Coverage   65.72%   50.47%   -15.26%     
===========================================
  Files          89       45       -44     
  Lines       15608     7832     -7776     
  Branches     3925     1968     -1957     
===========================================
- Hits        10258     3953     -6305     
+ Misses       4258     3339      -919     
+ Partials     1092      540      -552     
```


| [Impacted Files](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language) | Coverage Δ | |
|---|---|---|
| [docker.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-ZG9ja2VyLnB5) | `16.22% <0.00%> (-32.90%)` | :arrow_down: |
| [docker\_id.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-ZG9ja2VyX2lkLnB5) | `25.00% <0.00%> (-22.73%)` | :arrow_down: |
| [checker.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-Y2hlY2tlci5weQ==) | `65.42% <0.00%> (-19.69%)` | :arrow_down: |
| [workflow\_job.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-d29ya2Zsb3dfam9iLnB5) | `67.71% <0.00%> (-12.79%)` | :arrow_down: |
| [job.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-am9iLnB5) | `51.33% <0.00%> (-11.63%)` | :arrow_down: |
| [command\_line\_tool.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-Y29tbWFuZF9saW5lX3Rvb2wucHk=) | `59.18% <0.00%> (-10.37%)` | :arrow_down: |
| [mutation.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-bXV0YXRpb24ucHk=) | `69.23% <0.00%> (-10.26%)` | :arrow_down: |
| [argparser.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-YXJncGFyc2VyLnB5) | `66.51% <0.00%> (-9.51%)` | :arrow_down: |
| [builder.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-YnVpbGRlci5weQ==) | `78.61% <0.00%> (-8.81%)` | :arrow_down: |
| [update.py](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language#diff-dXBkYXRlLnB5) | `55.55% <0.00%> (-7.19%)` | :arrow_down: |
| ... and [52 more](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language). Last update [ed9dd4c...c4e9aa5](https://codecov.io/gh/common-workflow-language/cwltool/pull/1469?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=common-workflow-language).

mhpopescu: > We had also talked about only allowing named pipes in certain circumstances, so we need a way to only allow them when we are doing streaming, and forbid them otherwise

For this check I see the following options:
1) `type` is currently one of the following: `{File, Directory, CreateFile, WritableFile, CreateWritableFile}` and it could be extended with something like `StreamingFile`
2) `MapperEnt` adds a `streamable` attribute
3) have a convention that says that p[3] is the `streamable` flag which may or may not be present
mr-c: > > We had also talked about only allowing named pipes in certain circumstances, so we need a way to only allow them when we are doing streaming, and forbid them otherwise
> 
> For this check I see the following options:
> 
>     1. `type` is currently one of the following: `{File, Directory, CreateFile, WritableFile, CreateWritableFile}` and it could be extended with something like `StreamingFile`
> 
>     2. `MapperEnt` adds a `streamable` attribute
> 
>     3. have a convention that says that p[3] is the `streamable` flag which may or may not be present
> 
>     4. search in `self.joborder` for the file and get the `streamable` attribute from there

@tetron any preferences here?
mr-c: 
> 4) search in `self.joborder` for the file and get the `streamable` attribute from there

How terrible would this be to implement? Is self.joborder available where you made this change in this PR? I guess it could be a bit slow to search for this information if there are many inputs...



Patch:
diff --git a/cwltool/context.py b/cwltool/context.py
index 83c098e1..daf6aa86 100644
--- a/cwltool/context.py
+++ b/cwltool/context.py
@@ -107,6 +107,7 @@ class RuntimeContext(ContextBase):
         self.pull_image = True  # type: bool
         self.rm_container = True  # type: bool
         self.move_outputs = "move"  # type: str
+        self.streaming_allowed: bool = False
 
         self.singularity = False  # type: bool
         self.debug = False  # type: bool
diff --git a/cwltool/job.py b/cwltool/job.py
index f515091e..0f9eb131 100644
--- a/cwltool/job.py
+++ b/cwltool/job.py
@@ -3,6 +3,7 @@ import functools
 import itertools
 import logging
 import os
+import stat
 import re
 import shutil
 import subprocess  # nosec
@@ -175,13 +176,24 @@ class JobBase(HasReqsHints, metaclass=ABCMeta):
         if not os.path.exists(self.outdir):
             os.makedirs(self.outdir)
 
+        def is_streamable(file: str) -> bool:
+            if not runtimeContext.streaming_allowed:
+                return False
+            for inp in self.joborder.values():
+                if isinstance(inp, dict) and inp.get("location", None) == file:
+                    return inp.get("streamable", False)
+            return False
+
         for knownfile in self.pathmapper.files():
             p = self.pathmapper.mapper(knownfile)
             if p.type == "File" and not os.path.isfile(p[0]) and p.staged:
-                raise WorkflowException(
-                    "Input file %s (at %s) not found or is not a regular "
-                    "file." % (knownfile, self.pathmapper.mapper(knownfile)[0])
-                )
+                if not (
+                    is_streamable(knownfile) and stat.S_ISFIFO(os.stat(p[0]).st_mode)
+                ):
+                    raise WorkflowException(
+                        "Input file %s (at %s) not found or is not a regular "
+                        "file." % (knownfile, self.pathmapper.mapper(knownfile)[0])
+                    )
 
         if "listing" in self.generatefiles:
             runtimeContext = runtimeContext.copy()


Test patch:
diff --git a/tests/test_streaming.py b/tests/test_streaming.py
new file mode 100644
index 00000000..e3d3ec75
--- /dev/null
+++ b/tests/test_streaming.py
@@ -0,0 +1,108 @@
+"""Test that files marked as 'streamable' when 'streaming_allowed' can be named pipes."""
+import os
+
+import pytest
+from pathlib import Path
+from typing import cast
+
+from ruamel.yaml.comments import CommentedMap
+from schema_salad.sourceline import cmap
+
+from cwltool.command_line_tool import CommandLineTool
+from cwltool.context import LoadingContext, RuntimeContext
+from cwltool.errors import WorkflowException
+from cwltool.job import JobBase
+from cwltool.update import INTERNAL_VERSION
+from cwltool.utils import CWLObjectType
+from .util import get_data
+
+toolpath_object = cast(
+    CommentedMap,
+    cmap(
+        {
+            "cwlVersion": INTERNAL_VERSION,
+            "class": "CommandLineTool",
+            "inputs": [
+                {
+                    "type": "File",
+                    "id": "inp",
+                    "streamable": True,
+                }
+            ],
+            "outputs": [],
+            "requirements": [],
+        }
+    ),
+)
+
+loading_context = LoadingContext(
+    {
+        "metadata": {
+            "cwlVersion": INTERNAL_VERSION,
+            "http://commonwl.org/cwltool#original_cwlVersion": INTERNAL_VERSION,
+        }
+    }
+)
+
+
+def test_regular_file() -> None:
+    """Test that regular files do not raise any exception when they are checked in job._setup."""
+    clt = CommandLineTool(
+        toolpath_object,
+        loading_context,
+    )
+    runtime_context = RuntimeContext()
+
+    joborder: CWLObjectType = {
+        "inp": {
+            "class": "File",
+            "location": get_data("tests/wf/whale.txt"),
+        }
+    }
+
+    job = next(clt.job(joborder, None, runtime_context))
+    assert isinstance(job, JobBase)
+
+    job._setup(runtime_context)
+
+
+streaming = [
+    (True, True, False),
+    (True, False, True),
+    (False, True, True),
+    (False, False, True),
+]
+
+
+@pytest.mark.parametrize("streamable,streaming_allowed,raise_exception", streaming)
+def test_input_can_be_named_pipe(
+    tmp_path: Path, streamable: bool, streaming_allowed: bool, raise_exception: bool
+) -> None:
+    """Test that input can be a named pipe."""
+    clt = CommandLineTool(
+        toolpath_object,
+        loading_context,
+    )
+
+    runtime_context = RuntimeContext()
+    runtime_context.streaming_allowed = streaming_allowed
+
+    path = tmp_path / "tmp"
+    os.mkfifo(path)
+
+    joborder: CWLObjectType = {
+        "inp": {
+            "class": "File",
+            "location": str(path),
+            "streamable": streamable,
+        }
+    }
+
+    job = next(clt.job(joborder, None, runtime_context))
+    assert isinstance(job, JobBase)
+
+    if raise_exception:
+        with pytest.raises(WorkflowException):
+            job._setup(runtime_context)
+    else:
+        job._setup(runtime_context)