Add ebuf data structure

[chu11]

Per discussion in #1052, add a simple buffer api that will execute a callback when data in the buffer goes above/below a certain mark or when a line is ready.

Internally, I use the cbuf library. Some API decisions were made under the assumption I may swap out cbuf for another library in the future (evbuffer, vrb etc.).

For example, some libraries expose their internal buffers so users can read data without an additional copy. cbuf doesn't support this, but I wanted to prepare for a potential future change. So ebuf_read() presently returns a pointer to a malloced buffer for the user to read from, but does not expect the user to free this pointer.

Also, at most one callback is supported right now. I did not want to allow the user to have both read & write callbacks, which could lead to cascading infinite callback calls.

[codecov-io]

Codecov Report

Merging #1487 into master will increase coverage by 0.08%.
The diff coverage is 92.45%.

@@            Coverage Diff             @@
##           master    #1487      +/-   ##
==========================================
+ Coverage   78.69%   78.78%   +0.08%     
==========================================
  Files         164      165       +1     
  Lines       30405    30564     +159     
==========================================
+ Hits        23928    24079     +151     
- Misses       6477     6485       +8

Impacted Files	Coverage Δ
src/common/libflux/ebuf.c	92.45% <92.45%> (ø)
src/common/libutil/blobref.c	97.22% <0%> (-1.39%)	⬇️
src/common/libflux/msg_handler.c	86.64% <0%> (-0.37%)	⬇️
src/cmd/flux-module.c	85.06% <0%> (-0.31%)	⬇️
src/common/libflux/message.c	81.25% <0%> (-0.24%)	⬇️
src/modules/kvs/kvs.c	65.87% <0%> (-0.17%)	⬇️
src/common/libflux/future.c	88.78% <0%> (ø)	⬆️
src/common/libflux/request.c	88.46% <0%> (ø)	⬆️
src/broker/overlay.c	74.13% <0%> (+0.31%)	⬆️
... and 6 more

[coveralls]

Coverage increased (+0.1%) to 79.097% when pulling 851a2a7 on chu11:issue1052-ebuf into 01e2403 on flux-framework:master.

[grondo]

Some initial comments, maybe to get discussion going:

• would it make more sense to put this in libutil until we have a solid interface for libflux, at which point we can rename or add symbol alias for flux_ebuf_t or similar?
• we'll probably want a size parameter to ebuf_create()
• one issue with cbuf is that it will always grow to maxsize until you've put that much data in, even if you reach each byte out as you write it in (which might be surprising to callers). I'd say for your ebufs we probably want to avoid that and at first just set min, max of all cbufs to the size of ebuf_create

[chu11]

would it make more sense to put this in libutil until we have a solid interface for libflux, at which point we can rename or add symbol alias for flux_ebuf_t or similar?

Sounds like a good idea. I stuck it in libflux somewhat randomly.

we'll probably want a size parameter to ebuf_create()

I had this at first, but took it out b/c some other buf libs don't have such a configuration. Perhaps I was thinking too far out and we should just add it for the current implementation.

one issue with cbuf is that it will always grow to maxsize until you've put that much data in, even if you reach each byte out as you write it in (which might be surprising to callers). I'd say for your ebufs we probably want to avoid that and at first just set min, max of all cbufs to the size of ebuf_create

I was not aware of that, so definitely should adjust that.

[chu11]

re-pushed with adjustments per comments above. Also added a test to check for ENOSPC if writes go above max buffer size. Went ahead and squashed everything since changes are relatively tiny.

[chu11]

Hit a #1150, re-start a builder.

[grondo]

Sorry I haven't had another chance to finish a review this yet @chu11. It would be nice if the interface had at least one user before we merge it, but I realize that will be difficult for this particular case, so I'm willing to merge if it helps things move forward.

Talking with @garlick and he had a good question about whether a reactor based buffer watcher could be built from this interface. I seem to remember I've had definite opinions on this before, but I'm not remembering the details. At the very least, the simpler ebufs you have here could most likely be used to build buffer watchers, and are easier to unit test. It is also more flexible, because we aren't constrained to the flux_watcher_t interface (e.g. single flux_watcher_f callback)

I guess one question is whether a watcher interface for these ebufs would be useful, and if so, does the ebuf interface support it? This could probably use some discussion.

[chu11]

Yeah, the path between now and #1052 completion isn't 100% in my head at the moment. I think the main steps I had in my head were:

1. ebuf
2. reader & writer watchers in reactor using ebuf, most notably line buffered ones.
   • which that makes me realize now that I did not add a line writer callback in ebuf, I should do that.
3. using reader & writer buffered watchers, can replace zio or some other code in some places
   • this step is unnecessary, but is a good test to see things are working as desired. It could be done but need not be committed.
4. create composite reader/writer watcher

In my mind #1 and #2 are clear, #3 is sort of optional. #4 is the one I can't see clearly. I'm sort of hoping it becomes more clear as this is worked on.

As long as you feel ebuf looks "decent", no need to merge. Perhaps it'd be best to move on to steps 2 & 3 and see how things go, updating as needed. I was mulling over the reactor interface for this for a bit, so perhaps some obvious fallout / problems will appear.

[chu11]

I prototyped a:

flux_watcher_t *flux_iobuffer_read_line_watcher_create (flux_reactor_t *r, int fd,                                                     
                                                        int size, flux_watcher_f cb,                                                   
                                                        void *arg);

Basically it's a read watcher on a fd, but only calls the callback when a line is ready.

I'm not sure the ebuf library (even after I tweaked it a bit compared to what is in this PR) may be that useful afterall.

Basically, converting from the ebuf's callback to the user's callback is somewhat cumbersome. You need to stuff whatever you need to carry along in the arg argument and carry it all the way through.

I think just semi-manually doing a:

read data
put data into buffer
while (buffer has lines in it)
    call user's callback function

would be far easier.

Which then begs the question, if I feel the callback mechanisms of ebuf aren't that useful, perhaps we don't ebuf at all. The rest of ebuf is mostly just wrappers around the cbuf library. It's primary benefit would be the ability to convert the underlying buffer implementation (evbuffer or vrb) in the future.

[chu11]

The other subtlety I realized is that we probably need to have a public facing flux_event_buffer or similar data structure. Because we don't want to have some collection of:

flux_iobuffer_wacher_read ();
flux_iobuffer_watcher_write ();
flux_iobuffer_watcher_read_line ();
flux_iobuffer_watcher_write_line ();
etc.

set of functions for watchers type. Instead, the user should just call:

flux_event_buffer_t *flux_iobuffer_watcher_get_buffer (flux_watcher_t *w);

and go from there. This is similar to how bufferevents does it.

[chu11]

talked to @grondo, since we're going to need a public flux_buffer api, i'm going to redo this PR to provide that rather than this ebuf API. Closing the PR.