finish removing json_object from public flux api

[garlick]

This is a rework of PR #210 which was rework of PR #204.

This PR abolishes the specialized send/recv functions in earlier attempts. With the enhanced RPC, there seems little justification for wrapping flux_sendmsg(), _recvmsg() / _recvmsg_match() for each message type.

event
Reverted flux_event_send() and flux_event_recv() to be consistent with the other message types.

int flux_event_decode (zmsg_t *zmsg, const char **topic, const char **json_str);
zmsg_t *flux_event_encode (const char *topic, const char *json_str);

request

int flux_request_decode (zmsg_t *zmsg, const char **topic,
                         const char **json_str);
zmsg_t *flux_request_encode (const char *topic, const char *json_str);

response
flux_respond() seems to work out a little nicer in module message handlers than its predecessors. Since it takes both errnum and payload args, ignoring payload if errnum is nonzero, only one call is needed in a handler, and there is less temptation to call the errnum one if the payload one fails, which was likely futile anyway.

int flux_response_decode (zmsg_t *zmsg, const char **topic,
                          const char **json_str);
zmsg_t *flux_response_encode (const char *topic, int errnum,
                              const char *json_str);
int flux_respond (flux_t h, zmsg_t **request, int errnum, const char *json_str);

rpc
Discussed at length previously; main change is flux_rpc_multi() name change.

enum {
    FLUX_RPC_NORESPONSE = 1,
};
typedef struct flux_rpc_struct *flux_rpc_t;
typedef void (*flux_then_f)(flux_rpc_t rpc, void *arg);

flux_rpc_t flux_rpc (flux_t h, const char *topic, const char *json_str,
                     uint32_t nodeid, int flags);
flux_rpc_t flux_rpc_multi (flux_t h, const char *topic, const char *json_str,
                           const char *nodeset, int flags);
void flux_rpc_destroy (flux_rpc_t rpc);
bool flux_rpc_check (flux_rpc_t rpc);
int flux_rpc_get (flux_rpc_t rpc, uint32_t *nodeid, const char **json_str);
int flux_rpc_then (flux_rpc_t rpc, flux_then_f cb, void *arg);
bool flux_rpc_completed (flux_rpc_t rpc);

[garlick]

Here's an example of flux_respond() in a message handler in case my explanation above was too vague:

int echo_cb (flux_t h, int type, zmsg_t **zmsg, void *arg)
{
    int errnum = 0;
    const char *json_str;

    if (flux_request_decode (*zmsg, NULL, &json_str) < 0) {
        errnum = errno;
        goto done;
    }
done:
    if (flux_respond (h, zmsg, errnum, json_str) < 0)
        zmsg_destroy (&zmsg);
    return 0;
}

(commit e78534e makes the reuse of the request-owned payload in the response valid).

So the flow is basically: do stuff, falling through with errnum == 0 and a payload on success, or bail out early and goto done with errnum set nonzero. I think this structure could be applied to most of the message handers I have written, and will simplify them somewhat.

This function doesn't try to log or do anything if the respond fails. It might want to at least try to log that, although depending on what is wrong, that might not work either.

[grondo]

Wow, that seems like a huge improvement at first glance!

So, flux_respond still frees zmsg though? (unless it returns < 0?)

[garlick]

Correct. I guess I backslid there a little.

It could take a const zmsg_t * and dup it before transforming it into the response. Hmm.

[grondo]

So is there anything to do if flux_respond() < 0? Maybe we need a panic type call for this case?

Also, would the new example for flux_respond be more like:

int echo_cb (flux_t h, int type, zmsg_t **zmsg, void *arg)
{
    int rc;
    int errnum = 0;
    const char *json_str;

    if (flux_request_decode (*zmsg, NULL, &json_str) < 0) {
        errnum = errno;
        goto done;
    }
    /* ... */
done:
    if ((rc = flux_respond (h, *zmsg, errnum, json_str)) < 0)
        /* flux_log (LOG_ERR, ...)? */    
    zmsg_destroy (zmsg);
    return rc;
}

Just trying to wrap my head around usage idioms...

[garlick]

Reflecting on this a bit: if the underlying send fails, this is most likely fatal and may indicate that logging to the same handle would fail too. Same goes for ENOMEM errors. On the other hand if there is a message encoding error, this is unremarkable from a server standpoint, and might (at best) be worth a flux_log() message.

I think there are two questions here:

• how to reliably distinguish the error types (promote various message encoding errors to EPROTO, all other errnos are serious?)
• what do recommend on serious error. flux_panic() is inappropriate as it sends a message to the broker. Getting something to stderr for capture by enclosing instance would be appropriate in the comms module context.

Hmm.

[garlick]

In @grondo's example, message manipulation errors resulting from a malformed message are unlikely as the message would have already passed through the dispatch code and maybe some request decoding. Maybe it is reasonable to assume that any error returned by flux_respond() is a fatal one and the following is the right thing to do:

done:
    if ((rc = flux_respond (h, *zmsg, errnum, json_str)) < 0)
        flux_log (h, LOG_ERR, "%s: flux_respond: %s", __FUNCTION__, strerror (errno));
    zmsg_destroy (zmsg);
    return rc; // -1 stops reactor
}

This is a little verbose to have to repeat in every handler. Maybe in reworking the reactor handlers (issue #124) we could say that returning -1 here indicates a "fatal application error", require errno to be set, and then either directly log/terminate reactor as above, or (better) call a user-defined application panic function.

Issue #110 is open to revisit flux_panic(). I will make a note in there about the difficulty of logging fatal handle errors using the handle.

[garlick]

Another idea, not sure if I like it:

Add a mechanism like ferror(3) to store error state in the handle, and then have flux_respond() return void, and the reactor dispatch test the handle error state after a handler returns.

[grondo]

@garlick -- you had also mentioned an idea of registering a fatal error handler with the flux handle. This seemed like a really good idea to me. For testing we could register a handler that ensures test failure (i.e. call bail_out ()), apps that use libflux could use their own internal logging or abort() functionality, and modules could get a purpose-built error handler by default.

Also, as you've said, this could simplify a lot of internal code because instead trying to pass error all the way back up the stack you can call the registered error handler in place.

[garlick]

see commit 8977fbd - is this looking like the right interface?

[grondo]

This is looking pretty good to me.
Will the NORETURN thing work ok in apps that are using Flux via the API, or do we need some way to return the error to them? (I haven't thought about it in detail, so forgive me if I'm missing something obvious)

Also, you may want to have a separate call of out-of-memory situations. It could call the same fatal error handler for now, but in the future, we may want to tie into some other memory-release hook or something (e.g. a callback to plugins or other flux services to free up caches, etc -- hey, you never know). Having the call signature be different in OOM situation would make this easier in the future.

[garlick]

Hmm, yeah, maybe whether the flux_fatal_f should return or not should be up to the application. In 2599afa I made this change, adding a flag to ensure the fatal handler only gets called once (e.g. when unwinding the stack, or otherwise). Then I updated the rpc unit tests in t/loop to call libtap BAIL_OUT on fatal error, and modified the message handlers in those tests to use (void)flux_response(...) as discussed.

[garlick]

I'm not ignoring your OOM suggestion. Just pondering how and whether a flux_fatal_f style handler could accomplish your use case.

[grondo]

Actually probably not quite as necessary for separate OOM handler if flux_fatal can be set to return.
However, we may want to ping Dunlap for his rationalization for separating fatal handlers in lsd-tools just to be sure. (I was thinking it would be nice if OOM conditions could call a handler then optionally retry the allocation, but that would be a special allocator, not a fatal error handler. So you can ignore that.)

[garlick]

Maybe @dun can comment if I'm missing something, but it looks like the two functions in lsd-tools are overriding different behaviors. The lsd_fatal_error() function overrides internal logging. The error is still fatal, e.g. his pthreads functions call abort() if the user's _fatal returns. lsd_nomem_error() inserts a callback but if the callback returns, the error is still propagated to the caller.

The proposed flux_fail_f behaves like the nomem variant I guess. I'm not sure I see a reason at this point to add a separate fail function for OOM.

[grondo]

@garlick, agreed.

In the new rpc tests using flux_fatal is there a way to add a test to ensure fatal error handler does get called when registered? (I guess you'd have to have a way to cause a fatal error, and execute non-fatal handler) If its already there and I missed it, I aplogize.

[garlick]

Shall I squash down the review-commits and submit a new PR?

[grondo]

Yeah, this is looking good to me, though I hope we can get some comment from @trws at least for some more eyes.

I don't think you need to close PR and open another. Can you just force-update the branch referenced here? (Doesn't actually matter to be one way or the other, but is nice to have the entire discussion in one thread)

[trws]

The interface looks great to me, but I'm not sure I understand how the rpc message cleanup is being handled in the case of a continuation. This looks a bit like the zmsg is presumed to be destroyed by get, and if it isn't it loops forever. I guess the questions are, does get collect the message such that a subsequent get would error, and why the loop here?

[garlick]

A subsequent get would block until the next response arrives.

The loop is to handle the case where a new response is received but there was a previous response left in the flux_rpc_t e.g. by a _check() call. In retrospect it seems like I should queue responses to an abitrary depth and not necessarily care if _get() is called in the then handler.

Thoughts on how this should work?

[trws]

I think it depends on the semantics we want for then. If it is meant to consume the request, then it would be reasonable to discard that response whether get is called or not. Is there a reasonable way to grab a handle to the request that the callback is being passed so it can be collected after the call if the call doesn't collect it?

[garlick]

There is an invariant here to be considered: the payload pointer returned by _get() points directly into the message, so flux_rpc_t has to keep the response message around for that pointer to remain valid. What we decided was that the next _get() could invalidate the pointer returned by the previous _get(). If this was not done, then all the responses would have to be kept in there, an unnecessary memory overhead if the payloads are just being converted to JSON objects as they arrive.

The current _then() handling is trying to be level triggered, which is consistent with how the reactor works. We probably want that though it means an infinite loop if a callback is registered that doesn't consume responses.

[trws]

One alternative would be to pass each callback both the RPC and the
actual message intended for it, rather than relying on it to call get
explicitly that moves the get out to a deterministic location in the
caller.

On 4 Jun 2015, at 16:57, Jim Garlick wrote:

return rc;
}

+/* N.B. if the user's 'then' callback doesn't call flux_rpc_get(),

• * the callback will be called again (ad infinitum)
• _/
  +static int rpc_cb (flux_t h, int type, zmsg_t *_zmsg, void *arg)
  +{
• flux_rpc_t rpc = arg;

• assert (rpc->then_cb != NULL);
• do {

•    garbage_collect (rpc);
  

•    if (!rpc->rx_msg)
  

•        rpc_addmsg (rpc, zmsg);
  

•    rpc->then_cb (rpc, rpc->then_arg);
  

• } while (*zmsg != NULL);

There is an invariant here to be considered: the payload pointer
returned by _get() points directly into the message, so flux_rpc_t
has to keep the response message around for that pointer to remain
valid. What we decided was that the next _get() could invalidate
the pointer returned by the previous _get(). If this was not done,
then all the responses would have to be kept in there, an unnecessary
memory overhead if the payloads are just being converted to JSON
objects as they arrive.

The current _then() handling is trying to be level triggered, which
is consistent with how the reactor works. We probably want that
though it means an infinite loop if a callback is registered that
doesn't consume responses.

---

Reply to this email directly or view it on GitHub:
https://github.com/flux-framework/flux-core/pull/215/files#r31777981

[garlick]

At this point I'm not convinced there is a problem here. The infinite loop is what you'd get in most reactor callbacks if you don't clear the condition that triggered the callback.

It feels like the wrong direction to expose the message in the _then() continuation when we've got a fairly nice abstraction that hides messages behind _get().

Keep pushing if you feel strongly on this one, but I'm leaning towards leaving it as is.

[trws]

I don't feel strongly about any of the implementation details really. It
just seems like a source of potentially nasty bugs later to have an
infinite loop dependent on external callback code to continue. That
said, it may just be that I'm being over-cautious.

On 5 Jun 2015, at 10:40, Jim Garlick wrote:

return rc;
}

+/* N.B. if the user's 'then' callback doesn't call flux_rpc_get(),

• * the callback will be called again (ad infinitum)
• _/
  +static int rpc_cb (flux_t h, int type, zmsg_t *_zmsg, void *arg)
  +{
• flux_rpc_t rpc = arg;

• assert (rpc->then_cb != NULL);
• do {

•    garbage_collect (rpc);
  

•    if (!rpc->rx_msg)
  

•        rpc_addmsg (rpc, zmsg);
  

•    rpc->then_cb (rpc, rpc->then_arg);
  

• } while (*zmsg != NULL);

At this point I'm not convinced there is a problem here. The infinite
loop is what you'd get in most reactor callbacks if you don't clear
the condition that triggered the callback.

It feels like the wrong direction to expose the message in the
_then() continuation when we've got a fairly nice abstraction that
hides messages behind _get().

Keep pushing if you feel strongly on this one, but I'm leaning towards
leaving it as is.

---

Reply to this email directly or view it on GitHub:
https://github.com/flux-framework/flux-core/pull/215/files#r31834323

[garlick]

OK, compromise: I opened issue #216 on this. We can revisit before we finalize the first pass through the API. I need to make progress in other areas and for some reason I'm having a hard time getting my head around this one right now, but your point seems like a good one.

[trws]

Come to think of it, is a single then that can't be reset or changed the desired behavior here? I would somewhat expect either to have it be a set mechanism, removing the old `then, or for it to allow a list of them to be specified by calling it multiple times.

[grondo]

Great point. I like the idea of multiple then callbacks, but if that isn't needed then maybe we could return the old callback (kind of like sigaction(2))?

[garlick]

Oh yeah that makes sense, why'd I do that?
Let's return the old then cb.

Let's not do multiple thens unless we have a specific use case in mind.

[trws]

Makes sense to me.

[garlick]

Is returning the old function useful without also returning the old arg? If not, maybe it's OK to just return 0 and allow the cb to be updated...

[grondo]

@trws, very reasonable suggestion imo

[garlick]

Hmm, I'd have been fine with overwriting the arg. The intention (which maybe I've failed to express) is that it's the sole property of the caller, and is only passed through to a continuation callback so the caller can recover its context or whatever.

[garlick]

So do you guys think we need flux_rpc_then_get()? Easy to add.

[trws]

The arg would be property of the caller, but the nature of the interface of the callback is such that the caller may not wait for it to complete right? In a case like that, arg may be implicitly owned by the continuation and require cleanup. If that's not going to be allowed, then just overwriting it would be fine, but we should document the restriction.

[garlick]

OK, for now I will

• modify flux_rpc_then() to allow multiple calls
• document that internal (cb, arg) refs are overwritten by each call
• handle setting cb to NULL (unregister reactor callback)

In addition, I noticed a bug: if _check() is called before _then(), a response may be cached in the flux_rpc_t that the user's continuation won't see until the next response is received. I'll fix that and add a test.

[dun]

Re: lsd_fatal_error() & lsd_nomem_error(), I separated those because a memory allocation failure may not be fatal depending upon the context. The original intent was to provide a generalized list implementation where specific behaviors could be overridden. But it looks like my implementation there could use some work. As @garlick points out, the default implementation of lsd_fatal_error() isn't actually fatal and relies on the caller to terminate. And as @grondo points out, an attempt to reclaim memory upon a failed allocation requires support of a custom allocator.

[garlick]

I'm going to go ahead and force update this PR with the feedback commits so far squashed, then we can iterate some more if necessary.

[garlick]

thanks @dun for commenting on your 13yo code!

[garlick]

Ok, e640d9f (just pushed) should address multiple calls to _then() as discussed, including documenting the overwrite of existing cb, arg.

bf5913e takes the loop out of the reactor callback for continuations, and instead ensures that any unconsumed messages are requeued in the flux_t handle and so the reactor will call the callback again. This does not avoid the infinite loop discussed, but it does place the loop in the reactor where there is at least a hope that the reactor/dispatch will at some point be smart enough to allow other things to run in between continuations for the same rpc. I took the opportunity to simplify the code a bit.

[trws]

Looks good, and it sounds like this is ready to go in.

[grondo]

I agree, just waiting on travis result.

On Fri, Jun 5, 2015 at 2:00 PM, Tom Scogland [email protected]
wrote:

Looks good, and it sounds like this is ready to go in.