Is there a max buffer size for ev/write?

[nstgc]

I have a process started with (def cat (spawn ["cat"] :p {:in :pipe})) and am writing to it with (:write (cat :in) data) with (length data) returning 14833896. It seems to be hanging: it prints the file's contents (binary) to the terminal and then stops.

When I try this with something smaller, such as "hello world" it works as expected.

Is there a maximum data size for ev/write?

Note: cat is meant as a minimal working example, and the actual process I care about is zstd which behaves the same way in that small inputs work as expected and large ones hang without any apparent work being done.

[sogaiu]

I think it's possible you're encountering deadlock due to pipe buffer limits.

Possibly you'll find some of the content here that was harvested from recent issues and discussions to be of interest.

Some of us have been working on documentation for process management recently and there is a PR for it here -- it's based on the notes I linked earlier. That's not in rendered form but may be you'll find it to be of interest.

[nstgc]

Ah, I see. So this is a known limitation?

That's not in rendered form but may be you'll find it to be of interest.
You mean the following?
janet-lang/janet-lang.org@8ed77b5#diff-1e50b1cf30bff9a769fc83a48097143185215b5b3b799e1d9f216d934e3cdffbR1

That's nice. I had read Ian's Janet Book, which covers shell scripting, but it mostly leans on Andrew's sh package.

[llmII]

Not a Janet limitation, an operating system limitation (across all known sensible operating systems...).

[llmII]

@nstgc Do you have :out :pipe or similar? Could you provide a full minimal example?

I have the following example code working:

(def cat (os/spawn ["cat"] :p {:in :pipe}))
(def buf (buffer/new-filled 14833896))
(:write (cat :in) buf)
(os/proc-close cat)

The key being that if you have pipes for input and output, you have to keep in mind that the OS has a buffer limitation on how much data can be in a pipe that a process hasn't read off. If you write into the input and the process fails to read it you'll lose the capability to write to it's input until it reads it out. If the program writes to it's output and you have a custom pipe for it, it will block on writing if you fail to read it's output.

What @sogaiu linked details some of this, but in general, you must pump the output pipes during your writes. The easiest way is with ev/gather as mentioned in there.

[nstgc]

The key being that if you have pipes for input and output, you have to keep in mind that the OS has a buffer limitation on how much data can be in a pipe that a process hasn't read off.
Yeah, that's what I figured the issue was.

Could you provide a full minimal example?

I can't call this a minimal working example, but it's also not the full script, and I cut out some bits and bobs not necessary to capture what's going on.

I'm trying to make a sort of homegrown stress tester to compliment MPrime, StressAppTest, and MemTest86+.

(import spork/crc :as crc)

(def crc32c (crc/named-variant :crc32c))

(defn mk-zstd []
  (os/spawn ["zstd"] :p 
            {:in :pipe :out :pipe}))

(defn test-data [a]
  (let [data (apply string a)
        zstd (mk-zstd)]
    (:write (zstd :in) data)
    (:close (zstd :in))
    (let [compd  (:read (zstd :out) :all)
          hashed (crc32c (string data compd))
          exit   (:wait zstd)]
      {:exit exit :hash hashed})))

The a in (defn test-data [a] is an array of buffers created via (map slurp file-list).

For anyone curious, it "works" by compiling some code (I'm using OpenBLAS, but suggestions are very welcome), slurping those output files which are reproducible, hashing the data as a single chunk, and storing that in a table as a key, then repeating. Each time the process repeats, it either increments an existing hash, initializes a new one, or increments an error counter. Recently I've added the compression bit just to give it something else to do.

The easiest way is with ev/gather as mentioned in there.

Unfortunately, I've not quite been able to grasp the event loop stuff. I've tried several times, but each time come away bewildered.

I do suppose a decent work around would be to pipe to an external program, such as rhash which would be able to continually consume the output side of the pipe.

[llmII]

Try wrapping the lines below in an ev/spawn

    (:write (zstd :in) data)
    (:close (zstd :in))

Your issue does lie in that the output of the subprocess zstd is not yet read.

You might have to create a different question pertaining to how the event loop works but in essence, each thing doing IO will yield the fiber to the event loop (this is an oversimplification if you take into account core/file but to keep this simple we're not addressing that). The way your code is done presently it is a single fiber that calls write, close, read in succession. You need your read and write on separate fibers or to read in a loop a certain size chunk. The problem is that zstd reads it's input, then writes it's output, but it hangs on writing output because the pipe buffer (OS level) is full, and thus fails to read it's input any further. Your program (as per the example) writes the input, up till zstd stops reading due to it's hung writes, and this yields the current fiber out to the event loop because it's waiting on zstd's input pipe to be ready for writes. Your read follows the write but it's too late, your fiber is already yielded waiting for write events, not read events, and will never be woken up again. The solution is for one or the other or both to be put on a separate event loop managed fiber ("tasks") via ev/spawn or similar.

The reason this "works" for small inputs is because the pipe buffer (OS level) is not full. Once your input to zstd exceeds the OS pipe buffer limitations... you get hung on writes because the subprocess is hung waiting on you to read the output pipe.

[ianthehenry]

It's not too hard to consume the output side of the pipe in your own process. It sounds like you understand the issue, but here's a more minimal example:

# on my system, at least
(def pipe-buffer-size (* 1024 64))
(def proc (os/spawn ["cat"] :p {:in :pipe :out :pipe}))
(def buf (buffer/new-filled (+ 1 pipe-buffer-size)))
(ev/write (proc :in) buf)
(ev/close (proc :in))
(os/proc-wait proc)

Once cat fills up its own output buffer, it will stop reading from its input, which means that the write we made will never complete. So we need to consume cats output "while" the write is still in progress. (Note that without the + 1, this completes just fine, because cat is able to read all of its input before it fills up its write buffer.)

We can't just ev/read :all, because cat won't say "okay I'm done writing" until it sees the end of our input pipe. But we don't close our input pipe until all the bytes have been written. So we have a deadlock.

So we need to read cat's output. But the "obvious" thing doesn't actually work:

(def pipe-buffer-size (* 1024 64))
(def proc (os/spawn ["cat"] :p {:in :pipe :out :pipe}))
(def buf (buffer/new-filled (+ 1 pipe-buffer-size)))
(ev/write (proc :in) buf)
(ev/close (proc :in))
(def output (ev/read (proc :out) :all))
(os/proc-wait proc)

Because we never get to the ev/read call, because the write hangs forever. But if we do it in the opposite order, it still doesn't work, because we never get to the ev/write call while we're waiting for the ev/read to finish...

So we want to do both "at the same time." Which we can do like this:

(def pipe-buffer-size (* 1024 64))
(def proc (os/spawn ["cat"] :p {:in :pipe :out :pipe}))

(def output-chan (ev/chan))
(def read-fiber (ev/spawn
  (ev/give output-chan (ev/read (proc :out) :all))))

(def write-fiber (ev/spawn
  (def buf (buffer/new-filled (+ 1 pipe-buffer-size)))
  (ev/write (proc :in) buf)
  (ev/close (proc :in))))

(os/proc-wait proc)
(pp (length (ev/take output-chan)))

We use ev/spawn to create two tasks which will execute concurrently. The first reads chunks from cat's output and then writes the results to a channel once it's read all of them. The second does the write that we were doing in the main fiber before. But while that write still "hangs" until cat makes progress, the read fiber has a chance to read from cat in order to unstick the process.

Finally we pull the value off of the channel in the main process. (I don't think it matters what order we do os/proc-wait and ev/take in.) As a general rule, the way that asynchronous event-loopy fibers communicate with one another is by reading and writing from channels, not by yielding values like "normal" fibers do. So this channel is how the reader "subfiber" sends values back to the "main fiber" running at the top-level of our program.

You can also do this more concisely with ev/gather. ev/gather effectively creates a channel for every fiber and then takes from it on the main fiber. So the code would look like:

(def pipe-buffer-size (* 1024 64))
(def proc (os/spawn ["cat"] :p {:in :pipe :out :pipe}))

(def [output _] (ev/gather
  (ev/read (proc :out) :all)
  (do
    (def buf (buffer/new-filled (+ 1 pipe-buffer-size)))
    (ev/write (proc :in) buf)
    (ev/close (proc :in)))))

(os/proc-wait proc)
(pp (length output))

(The [output _] is because ev/gather returns results from all of the fibers.)

Edit: oh yeah, as @llmII points out, in the case that you only have one subprocess like this, you can use your main fiber as your "reader fiber." You don't actually need to move that to work to a separate fiber. Even simpler:

(def pipe-buffer-size (* 1024 64))
(def proc (os/spawn ["cat"] :p {:in :pipe :out :pipe}))

(ev/spawn
  (def buf (buffer/new-filled (+ 1 pipe-buffer-size)))
  (ev/write (proc :in) buf)
  (ev/close (proc :in)))

(pp (length (ev/read (proc :out) :all)))
(os/proc-wait proc)

(Now it is important to do your os/proc-wait after the ev/read.)