Implement Flow.write and write_from

[haesbaert]

These are the corollary from Flow.read and read_into.

--

Not super sure about write_from naming, but it's the closest to the opposite of read_into I can think of.

[talex5]

With the change above, we can provide both write_all and write_single here, like in File.

[haesbaert]

My first version had both, ha, I'll add it back.

[haesbaert]

The problem is libuv doesn't give us a way to do write_single, it will queue the next write and attempt to write all of them.
If you're curious, it's in:
stream.c: uv_write()->uv_write2()->uv__write()->uv__try_write()-> SHORTCOUNT ?->uv__io_start()...:

    n = uv__try_write(stream,
                      &(req->bufs[req->write_index]),
                      req->nbufs - req->write_index,
                      req->send_handle);

    /* Ensure the handle isn't sent again in case this is a partial write. */
    if (n >= 0) {
      req->send_handle = NULL;
      if (uv__write_req_update(stream, req, n)) { /* haesbaert: true if NOT a shortcount */
        uv__write_req_finish(req);
        return;  /* TODO(bnoordhuis) Start trying to write the next request. */
      }
    } else if (n != UV_EAGAIN)
      break;
    /* If this is a blocking stream, try again. */
    if (stream->flags & UV_HANDLE_BLOCKING_WRITES)
      continue;

    /* We're not done. haesbaert: shortcount falls here */
    uv__io_start(stream->loop, &stream->io_watcher, POLLOUT);

      int uv_write(uv_write_t *req, uv_stream_t *handle, const uv_buf_t bufs[], unsigned int nbufs, uv_write_cb cb)
              Write data to stream. Buffers are written in order. Example:

                 void cb(uv_write_t* req, int status) {
                     /* Logic which handles the write result */
                 }

                 uv_buf_t a[] = {
                     { .base = "1", .len = 1 },
                     { .base = "2", .len = 1 }
                 };

                 uv_buf_t b[] = {
                     { .base = "3", .len = 1 },
                     { .base = "4", .len = 1 }
                 };

                 uv_write_t req1;
                 uv_write_t req2;

                 /* writes "1234" */
                 uv_write(&req1, stream, a, 2, cb);
                 uv_write(&req2, stream, b, 2, cb);

              NOTE:
                 The memory pointed to by the buffers must remain valid until the callback gets called.  This also holds for uv_write2().

And no, it doesn't return bytes written, it's just 0 or error :/

[haesbaert]

So the option is to either lie (write_single = write_all) in libuv, and god knows how many writes were done, or not provide a write_single.

[c-cube]

should copy have a default implementation in terms of write_from?

[talex5]

libuv is callback based. I think the return 0 just means the write operation was scheduled. The callback gets the number of bytes written.

https://github.com/aantron/luv/blob/master/src/stream.mli says:

The second argument passed to the callback is the number of bytes
written. libuv has an internal queue of writes, in part to implement retry.
This means that writes can be partial at the libuv (and Luv) API level, so
it is possible to receive both an [Error] result, and for some data to have
been successfully written.

However, for files it looks like write behaves normally (either returns an error or a byte count, but not both).

So, yeah, maybe we shouldn't provide write_single. But then we don't need a write method at all - Flow.write flow buf can just be Flow.copy (cstruct_source [buf]) flow.

Possibly we just need to add exception Write_error of int * exn to report how many bytes got written, in case anyone cares.

[haesbaert]

libuv is callback based. I think the return 0 just means the write operation was scheduled. The callback gets the number of bytes written.

Even though it's callback based, the code tries to do a write before setting up an event, so we can reason about that only:
uv_write()->uv_write2()->empty_queue->uv__write()->uv__try_write()->uv__writev()->writev(2)

https://github.com/aantron/luv/blob/master/src/stream.mli says:

The second argument passed to the callback is the number of bytes
written. libuv has an internal queue of writes, in part to implement retry.
This means that writes can be partial at the libuv (and Luv) API level, so
it is possible to receive both an [Error] result, and for some data to have
been successfully written.

This is a bit bad worded, what he means is that you can get an error and you also get back how many bytes you successfully wrote before from previous writes, but the callback is only called once. There is no way to distinguish on success if the requested amount occurred in one or N calls.

With this I stand by my previous assessment (humbly), luv does not give us an API to do something in one write or give us control back in-between writes. In a sense they really treat a stream as a stream.

In fact, in order to retrieve the amount of partial bytes on write, he must be reaching into what libuv calls "private", cause the callback for write is not like read:

     typedef void (*uv_write_cb)(uv_write_t *req, int status)
              Callback called after data was written on a stream. status will be 0 in case of success, < 0 otherwise.
       typedef void (*uv_read_cb)(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf)
              Callback called when data was read on a stream.

http://docs.libuv.org/en/v1.x/stream.html#c.uv_write_cb

I can dig up luv later but he is probably getting from:

struct uv_write_s { /* haesbaert: this is write_cb_t without the useless typedef */
  UV_REQ_FIELDS
  uv_write_cb cb;
  uv_stream_t* send_handle; /* TODO: make private and unix-only in v2.x. */
  uv_stream_t* handle;
  UV_WRITE_PRIVATE_FIELDS
};
#define UV_WRITE_PRIVATE_FIELDS   
  void* queue[2]; 
  unsigned int write_index; /* haesbaert: this guy is likely the second argument to stream */
  uv_buf_t* bufs;
  unsigned int nbufs; 
  int error;
  uv_buf_t bufsml[4];

However, for files it looks like write behaves normally (either returns an error or a byte count, but not both).

So, yeah, maybe we shouldn't provide write_single. But then we don't need a write method at all - Flow.write flow buf can just be Flow.copy (cstruct_source [buf]) flow.

Possibly we just need to add exception Write_error of int * exn to report how many bytes got written, in case anyone cares.

I'd be against that, I like the idea of write_all + write_single, one means "I don't care how" and the other means "I need it to be in one".

[haesbaert]

Leme rephrase that, I'm fine with not having single + all, but having the Write_error is not necessary imho, if it's a real stream, no one cares how much was written. If it's a datagram-like FD that never happens and we get EMSGSIZE or ENOBUFS or ENOMEM.

If we don't care about the extra allocations, I agree Flow.write can be Flow.copy (cstruct_source)...

I still think we need something like it, because copy doesn't really guarantee that the source is "atomic".
fallback_copy can end up splitting one chunk into multiple smaller chunks of 4k
Flow.write(16k) being broken into 4x Flow.write(4k) is wrong if the sink is a TAP interface for example.

So Flow.write guarantees we try all in one syscall, but also writes the remaining bytes if we get a short-count (like copy does).

[talex5]

Yeah, no need for Write_error (or maybe Partial_write?) in this PR. I've never needed it, so we'll wait and see if anyone else does.

Note: fallback_copy won't be used if the source is a cstruct_source (it'll use copy_with_rsb instead, which behaves the way you want already). Though the luv backend doesn't do this optimisation yet.

I think it's fine to add a write method as an optimisation for this (would be interesting to know what the speed difference is).

[talex5]

I don't think there's a faster way to do it given the Buffer API. We can just note in the mli that this might not be very efficient (in fact, it already points you in the direction of Buf_read).

[c-cube]

Given that Buffer doesn't expose useful functions for that, it might be a good idea to also expose a better implementation of it ({mutable buf: bytes; mutable len: int } basically), if only to get a string out of a Flow.source easily.

[talex5]

@avsm is thinking about changing Cstruct to be backed by bytes rather than bigarray. If that happens, Eio.Buf_read will work for this case (see #140).

[haesbaert]

@avsm is thinking about changing Cstruct to be backed by bytes rather than bigarray. If that happens, Eio.Buf_read will work for this case (see #140).

Aye, I checked that, I've been thinking of implementing something like Mbuf.t which could be backed by a chain of Bytes or Cstruct, a bit inspired by how mbuf(9) works. With that we can guarantee that the backing Bytes is always > wosize so it goes directly in the major heap, and also, if needed, swap to Cstruct transparently in the future if things change in the GC.

It would be something like "allocate a fixed bytes size, say 256, start writing at say offset 63 so future prepends are free, appending a few words would just write to the existing chunk, appending a longer chunk would link it to the chain". Maybe have a big (>1500) and little ( < 512 ).

We could then make full usage of writing a backing method in Eio to turn that into the iovecs.

I don't think Cstruct or Bytes directly is a good way for people to build packets.

[talex5]

Note: we already have Buf_write, which is similar (though it doesn't allocate room for prepending). It does allow appending cstructs without copying, which results in using writev automatically.

[talex5]

Might be good to note here that this is a low-level API and that using copy may allow optimisations in some cases.

[haesbaert]

I'm not sure I properly understood what you meant, so I'm keeping write as returning unit, changed from write_from -> write, and addressed the comments you mentioned.

[talex5]

Looks good. How about taking a list (writev), since both backends already support that? I guess it's needed for datagram sockets, where writing a vector of two buffers isn't the same as doing two separate writes.

[talex5]

We can provide a default implementation here in terms of copy, then nothing else needs to change except as an optimisation.

[talex5]

I don't think we need this. We can just use the default behaviour of calling copy.

[talex5]

I squashed and rebased the commits, and also marked write as non-virtual in the interface.

If you're happy with that, it's ready for merging.

[haesbaert]

I squashed and rebased the commits, and also marked write as non-virtual in the interface.

If you're happy with that, it's ready for merging.

yep 😄