libflux: store fully decoded message in flux_msg_t

[chu11]

Per discussion in #3617, this is part 1 to remove the libczmq dependency on libflux-core. In this code we rework the flux_msg_t internals to store data in its own native data structure rather than using zmsg_t.

Just wanted to get some initial thoughts.

Should we consider this a mergable PR? Or would we want to complete more of #3617 in one PR? It is a lot of code changes.

The only places that zmsg_t and zframe_t remain are in flux_msg_encode_size(), flux_msg_encode(), flux_msg_decode(), flux_msg_sendzsock(), and flux_msg_recvzsock(). We have some msg_to_zmsg() and zmsg_to_msg() internal functions to handle the transition for the time being. These functions will eventually be migrated out of libflux-core. To librouter? Or a helper lib? This is perhaps still TBD.

Note that both ccan and czmq define a streq() macro ... ugh. So decided to change the macro in ccan for the time being. We can (hopefully) revert this commit at a later time when things are no longer coupled.

Should the format of flux_msg_fprint() change? With the removal of internal zmsg_t storage, we are no longer dependent on formatting output like zframe_fprint(). We can output whatever we like. This can be a follow on PR of course.

[garlick]

After a quick skim (since this is WIP), I have a couple quick comments inline and one general one:

Suggestion: dispense with the internal "proto block accessors", and instead just add a couple of functions that can be used as helpers by msg_to_zmsg() and zmsg_to_msg(), e.g.

static int proto_encode (void *buf, int len, const flux_msg_t *msg);
static int proto_decode (const void *buf, int len, flux_msg_t *msg);

I'd suggest adding another generic union name to the two u32's like aux1 and aux2 and, in those functions, just encode/decode from/to those instead of picking a union member based on the message type.

[garlick]

I noticed that before but I figured it would just be easier to do stuff like:

#include <czmq.h>
#ifdef streq
#undef streq
#endif
#include "src/common/libccan/ccan/str/str.h"

in places where czmq.h and str.h are both used?

If that doesn't feel right, then it seems like just conditionally defining streq in str.h would be better than a rename, since the macros are equivalently defined.

[garlick]

magic and version could probably be left out of the struct and just added/checked when the message is encoded/decoded?

[garlick]

Don't need to set r to NULL right before you set it to the result of calloc(). Let's not make gratuitous initialization a thing.

[garlick]

strndup?

[garlick]

Or make it one malloc for the struct plus the string and use set id = r + 1?

[chu11]

re-pushed, did some more cleanup per some comments above. I suddenly recalled the:

struct route_id {
    struct list_node route_id_node;
    char id[0];
};

trick for the route id. Wondering if there are other places in flux-core we could do it to save a malloc.

Could this be a potential merge-able candidate? Or should we consider making more progress on #3617 within the PR?

[garlick]

Oh heh, id[0] saves sizeof (void *) per struct compared to using a pointer and initializing route->id = route + 1 which is what I had done elsewhere. Neat!

Could this be a potential merge-able candidate? Or should we consider making more progress on #3617 within the PR?

I'd be tempted to suggest that we try to get these functions out of the public API in this PR:

/* Send message to zeromq socket.
 * Returns 0 on success, -1 on failure with errno set.
 */
int flux_msg_sendzsock (void *dest, const flux_msg_t *msg);
int flux_msg_sendzsock_ex (void *dest, const flux_msg_t *msg, bool nonblock);

/* Receive a message from zeromq socket.
 * Returns message on success, NULL on failure with errno set.
 */
flux_msg_t *flux_msg_recvzsock (void *dest);

Then the PR could be re-titled with release notes in mind, reflecting the outwardly visible change, like:

libflux/message: drop zsock functions from public API

To accomplish that i think it means we would need to move the struct flux_msg definition to a message_private.h header or similar, and relocate those functions to a sendzsock.c in librouter, analogous to sendfd.c. Then we'd need to drop the flux_ prefix from those functions and update users. What do you think, does that make sense?

[garlick]

Just ran a 40K job throughput test a couple times and it seems like these changes are somewhat helpful?

number of jobs: 40000
submit time:    10.270s (3894.8 job/s)
script runtime: 939.603s
job runtime:    201.295s
throughput:     198.7 job/s (script:  42.6 job/s)

number of jobs: 40000
submit time:    10.317s (3877.1 job/s)
script runtime: 982.660s
job runtime:    202.157s
throughput:     197.9 job/s (script:  40.7 job/s)

compared with e.g. master

number of jobs: 40000
submit time:    10.897s (3670.9 job/s)
script runtime: 1154.583s
job runtime:    212.699s
throughput:     188.1 job/s (script:  34.6 job/s)

[chu11]

Oh heh, id[0] saves sizeof (void *) per struct compared to using a pointer and initializing route->id = route + 1 which is what I had done elsewhere. Neat!

Yeah, I just recalled it suddenly. I guess it was the right context for remembering it since struct route_id only has 2 fields.

I'd be tempted to suggest that we try to get these functions out of the public API in this PR:

Sounds good. What about the encode/decode functions? I feel I can "hand" encode data in the encode functions (probably will for performance), but feel decode will still need zmsg (I lazily used zmsg in encode b/c of the latter case).

Just ran a 40K job throughput test a couple times and it seems like these changes are somewhat helpful?

Nice, I hadn't run a benchmark yet b/c I was waiting to do more work (possibly remove some additional to/from zmsg/msg).

[garlick]

What about the encode/decode functions? I feel I can "hand" encode data in the encode functions (probably will for performance), but feel decode will still need zmsg (I lazily used zmsg in encode b/c of the latter case).

librouter/sendfd.c is the only user so probably it belongs in librouter

That would mean that message.c would be czmq free (!) but as soon as one opens the broker, the local connector would pull it in. So progress I guess :-)

[garlick]

I'm fuzzy on how message parts are delimited when czmq encodes the message to a buffer.

[chu11]

ugh, one builder hit a lot of

   message.c:326:42: error: variable 'r' is uninitialized when used here [-Werror,-Wuninitialized]
          list_for_each_rev (&msg->routes, r, route_id_node) {

   message.c:1075:34: error: variable 'r' is uninitialized when used here [-Werror,-Wuninitialized]
      list_for_each (&msg->routes, r, route_id_node)

   message.c:1092:34: error: variable 'r' is uninitialized when used here [-Werror,-Wuninitialized]
      list_for_each (&msg->routes, r, route_id_node) {

   message.c:1465:42: error: variable 'r' is uninitialized when used here [-Werror,-Wuninitialized]
          list_for_each_rev (&msg->routes, r, route_id_node) {

not sure what subtlety the compiler is seeing, but i guess should initialize everywhere.

[garlick]

I can't remember the details (or if I spleunked far enough into the CCAN list implementation to find them) but the need to initialize that variable seems to be a property of the CCAN list_for_each_* macros.

[chu11]

I made an attempt to move the czmq-requiring functions from libflux/message to librouter by putting flux_msg_t into a new header called message_private.h. The attempt went poorly b/c of the number of common things between the czmq-requiring and non-czmq-requiring halves. Eventually I did an experiment, splitting libflux/message into message.[ch], message_private.[ch], and 'message_czmq.[ch] just to see what would have to be "common" between the czmq and non-czmq halves.

I'm now thinking that a utility library might be best to share between libflux and librouter. We'd do something like:

struct flux_proto {
    // optional route list, if FLUX_MSGFLAG_ROUTE
    struct list_head routes;
    int routes_len;     /* to avoid looping */

    // optional topic frame, if FLUX_MSGFLAG_TOPIC
    char *topic;

    // optional payload frame, if FLUX_MSGFLAG_PAYLOAD
    void *payload;
    size_t payload_size;

    // required proto frame data
    uint8_t type;
    uint8_t flags;
    uint32_t userid;
    uint32_t rolemask;
    union {
        uint32_t nodeid;  // request
        uint32_t sequence; // event
        uint32_t errnum; // response, keepalive
        uint32_t aux1; // common accessor
    };
    union {
        uint32_t matchtag; // request, response
        uint32_t status; // keepalive
        uint32_t aux2; // common accessor
    };
};

in the helper lib (libfluxproto?, that's the best I can come up with right now)

and flux_msg_t becomes:

struct flux_msg {
    struct flux_proto proto;
    json_t *json;
    char *lasterr;
    struct aux_item *aux;
    int refcount;
};

As an aside, during this experiment I also split libflux/test/message.c and it exposed some helper libs that just assume libflux will provide the linkage to libczmq (such as libtestutil). So that's something to work on as well. That's probably another issue to deal with after this PR, removing -lczmq from libflux-core and then dealing w/ the fallout of the change.

[garlick]

I think it would be a good step to try to do whatever refactoring is necessary to eliminate message_private.c. Going through those functions one by one, my reaction is:

• msg_create_common() - just absorb into flux_msg_create() and allow the type message to be FLUX_MSGTYPE_ANY
• msg_proto_setup(), proto_set_u32() - these have to do with encoding and so should migrate to librouter. Looks like it's only shared for purposes of flux_msg_fprint(). Perhaps just change the output format - showing the encoding in hex is not too useful anyway.
• route_id_create(), route_id_destroy() - these should live in message.c and encode/decode should use a higher level interface exported by message.c, e.g. push/pop/first/last. Add a next for iteration? (Why include id_len parameter in route_id_create()? Call strlen() internally?)

What's accomplished by splitting the struct? I think it is fine if the whole thing is exposed in message_private.h. If there's a good reason to split then I would change name of the new struct since "proto" is the name of one of the message frames.

[garlick]

Having second thoughts about suggesting that all message encode/decode be pushed to librouter.

Maybe we should keep message.[ch] a complete implementation of RFC 3. That is, perform encoding in there down to the level of frames. The encoding that would go in librouter in support of sendfd/sendzmq would translate an opaque list of frames to the underlying wire protocol.

That would mean adding some interfaces to message.[ch] for the frame based encoding/decoding.

We could reinvent zframe for this, but I would think it might be a little nicer to use something like an array of iovecs, similar to readv(2), writev(2).

Any thoughts? I'm wondering if we ought to make some time to chat about this in a coffee meeting to get on the same page?

[chu11]

re-pushed w/ my experimental libmsgproto library I loosely described above and eventually moved all of the encode/decode/zsock to librouter (somewhat lazily, there's a lot of cut & paste). The only "message_private.h" functions are:

struct msgproto *flux_msg_get_proto (flux_msg_t *msg);
flux_msg_t *flux_msg_create_common (void);

which worked out pretty well. We can eliminate flux_msg_create_common() easily.

This was mostly done before @garlick comments above as I experimented to see how this would work.

msg_create_common() - just absorb into flux_msg_create() and allow the type message to be FLUX_MSGTYPE_ANY

I believe flux_msg_create_common() was initially created to avoid unnecessary initialization on common paths. I suppose setting a few fields isn't that big of a deal (the original initialization involved setting up the protoframe, which was far more costly).

shared for purposes of flux_msg_fprint(). Perhaps just change the output format - showing the encoding in hex is not too useful anyway.

agreed

route_id_create(), route_id_destroy() - these should live in message.c and encode/decode should use a higher level interface exported by message.c, e.g. push/pop/first/last. Add a next for iteration?

Will definitely need a next for iteration, and I think a reverse next as well. Thinking about this, I might need to add a append too. The more we add, we sort of break the simple abstraction.

Aside, do we have a code style pattern for "reverse iterator", like "foo_next_rev ()"? Or perhaps keeping track if the user called first() or last() when starting the iterating?

(Edit: it appears zlistx() uses zlistx_prev(), so we could use that pattern).

(Why include id_len parameter in route_id_create()? Call strlen() internally?)

The id_len is because zframe_data() may not include the NUL byte at the end of a string. So sometimes its called with route_id_create (zframe_data (zf), zframe_size (zf)). We didn't have to worry about this before b/c zframe_strdup() would add a trailing NUL byte for you.

What's accomplished by splitting the struct? I think it is fine if the whole thing is exposed in message_private.h.

Its simply for abstraction purposes, limiting access to only the fields necessary. If we can eliminate some of the "common functions" like you said above, then the benefit may not be there anymore.

[chu11]

Any thoughts? I'm wondering if we ought to make some time to chat about this in a coffee meeting to get on the same page?

Hmmm, the idea is intriguing. I think the only thing is I don't know if it's a big win at this point. I don't know the zsock code that well so I don't know how much work with would be to change a lot of the zsock code. Basically:

• In my latest push, the encode functions no longer need zmsg. So there's no need for a "flux frames" data structure there.-

• The decode function uses a zmsg internally mostly as a convenience data structure. So we could remove zmsg use there if we had a flux_frames internal lib. In fact, I might even be able to get away with just using a zlist instead.

• For sendzsock we require zmsg b/c we call zframe_send(). Looking at the code in czmq for zframe_send(), there's a lot of logic in that function.

• For recvzsock right now we call zmsg_recv() and thus it returns a zmsg.

So if we implemented an internal zframes like data structure, it could only be used in 1 of the 4 places above.

I imagine you were thinking we could do something like:

flux_frames_t *flux_msg_to_fluxframes (flux_msg_t *msg);

and flux_frames_t could be used with direct calls to zsock so people wouldn't have to call flux_msg_sendzsock()?

The value could be there, but I don't think I know the zsock code well enough to make a judgement.

[chu11]

Per offline discussion, we've decided that this PR with just the remove of zmsg_t from the internal flux_msg_t is a good "checkpoint" for this work. If we move things into librouter, the PR will just get too large.

Re-pushed with what I believe is my final cleanup. Some cleanup tradeoffs I made:

Because of the fact field values are easily accessible in the flux_msg_t now, I debated if internal calls to getters/setters like:

    if (flux_msg_get_rolemask (msg, &cred->rolemask) < 0)
        return -1;

should be removed and replaced with direct access to the field in question. i.e. just read msg->rolemask directly in the case above. Or for setters, just set the value instead of calling flux_msg_set_FOO().

Generally speaking I left the calls in, b/c its good abstraction and who knows how things could change in the future.

The two exceptions I made were to access msg->type and msg->flags. They are used everywhere and it seems excessive to call flux_msg_get_type() and flux_msg_get_flags() so often. I also hated this pattern:

    uint8_t flags;
    if (flux_msg_get_flags (msg, &flags) < 0)
       return -1;
    if (!(flags & FOOBAR))
       return -1;
    // do something;

which of course just reading msg->flags is faster/easier/makes more sense.

In this PR there is a zmsg_to_msg() function used by flux_msg_decode() and flux_msg_recvzsock(). We discussed removing zmsg requirements from flux_msg_decode() so that zmsg is only used by the zsock functions.

I actually have a branch that removed zmsg needs from flux_msg_decode(), but b/c flux_msg_recvzsock() still needs zmsg_to_msg(), it was a lot of change for questionable needs. So I reversed it until we know exactly how we plan to move forward with what will/will not move into librouter.

We also discussed adding flux_msg_route_first/last/next/prev() iterators on the routes code so we don't have to call list_for_each() in some many places. Decided to not add that code to this PR, b/c its new functionality. I'll do that in another PR.

We also discussed re-formatting flux_msg_fprint() to do further cleanup, but going to punt that to follow up PR since it changes present behavior.

A small throughput performance test:

Master branch

Running throughput.py 2048 jobs per iteration, 10 iterations
throughput:     31.1 job/s (script:  30.9 job/s)
1: 66 seconds
throughput:     30.2 job/s (script:  30.0 job/s)
2: 135 seconds
throughput:     28.5 job/s (script:  28.4 job/s)
3: 207 seconds
throughput:     28.1 job/s (script:  27.9 job/s)
4: 281 seconds
throughput:     27.8 job/s (script:  27.6 job/s)
5: 355 seconds
throughput:     28.5 job/s (script:  28.3 job/s)
6: 428 seconds
throughput:     29.2 job/s (script:  29.0 job/s)
7: 499 seconds
throughput:     28.0 job/s (script:  27.8 job/s)
8: 573 seconds
throughput:     28.4 job/s (script:  28.2 job/s)
9: 645 seconds
throughput:     28.3 job/s (script:  28.1 job/s)
10: 718 seconds
Total of 718 seconds for everything

and this branch

Running throughput.py 2048 jobs per iteration, 10 iterations
throughput:     33.0 job/s (script:  32.8 job/s)
1: 62 seconds
throughput:     30.6 job/s (script:  30.4 job/s)
2: 130 seconds
throughput:     30.6 job/s (script:  30.4 job/s)
3: 198 seconds
throughput:     29.2 job/s (script:  29.0 job/s)
4: 268 seconds
throughput:     29.4 job/s (script:  29.3 job/s)
5: 338 seconds
throughput:     28.3 job/s (script:  28.2 job/s)
6: 411 seconds
throughput:     29.1 job/s (script:  29.0 job/s)
7: 482 seconds
throughput:     28.4 job/s (script:  28.3 job/s)
8: 555 seconds
throughput:     29.0 job/s (script:  28.8 job/s)
9: 626 seconds
throughput:     28.5 job/s (script:  28.3 job/s)
10: 699 seconds
Total of 699 seconds for everything

ran several times, but the results are largely the same. There's a small incremental gain on throughput performance. The most important number is probably the one at the end, how long everything took to run.

[chu11]

re-pushed, fixing up some compiler warnings a builder found

[garlick]

Is this ready for another review?

Suggestion: change title to

libflux: store fully decoded message in flux_msg_t

[chu11]

@garlick yup, ready for review

[chu11]

looking through code coverage, there are alot of new paths being missed due to changes. I'm going to re-org the tests and add a bunch more coverage. But I think the main message.c file is ready for review.

[chu11]

re-pushed, adding some additional tests. I am going to split off all of the cleanup commits into a new PR, b/c they are starting to add up after a test/message.c re-org of code (this is PR #3745)

[chu11]

just an FYI, the order of PRs at the moment is

#3745
#3701
#3742
#3746

[garlick]

Maybe this is an intentional baby step, but there seems to be a disparity between flux_msg_encode() which is zmsg/zframe free, and flux_msg_decode() which is not. Is there a good reason not to go all the way and convert the decode path as well?

[chu11]

I actually have a branch that did remove zmsg use from flux_msg_decode(), using a zlist instead (b/c that's all I'm really using zmsg for in flux_msg_decode(), just as a list) and I created a zlist_to_msg() kinda function.

But zmsg_to_msg() is still needed for flux_msg_recvzsock(). And a (good) common function between zlist_to_msg() and zmsg_to_msg() was annoying enough I just abandoned the zlist_to_msg() until the longer term plan was established.

We could go ahead and add zlist_to_msg() here b/c I think we're going to end up having two functions in the end anyways (one in libflux and one in librouter).

[garlick]

In the interest of keeping things moving forward, maybe we should leave as is. This is a big change and there will be opportunities to refine it later.

[garlick]

I'm not spotting any problems that we aren't confident we can resolve a few steps down the road. As noted earlier, just making the accessors more direct has a positive performance impact, so let's just go full speed ahead.

[chu11]

@garlick thanks, i'll set MWP

[garlick]

bionic builder seems to have been aborted while running the shell input sharness test. Restarted.

[chu11]

hmmm centos7 builder failed with a ton of errors ... investigating. unsure if issue with rebase i did.

Edit: gonna just rebase & retry, something may have gone really wrong w/ that builder. 25 test fails over 6 resource related sharness test files.

[codecov]

Codecov Report

Merging #3701 (2fb0d86) into master (12bc8d7) will increase coverage by 0.05%.
The diff coverage is 89.77%.

@@            Coverage Diff             @@
##           master    #3701      +/-   ##
==========================================
+ Coverage   83.21%   83.27%   +0.05%     
==========================================
  Files         333      333              
  Lines       50763    50793      +30     
==========================================
+ Hits        42244    42298      +54     
+ Misses       8519     8495      -24     

Impacted Files	Coverage Δ
src/common/libflux/message.c	91.38% <89.77%> (+3.60%)	⬆️
src/broker/module.c	76.80% <0.00%> (-1.25%)	⬇️
src/modules/job-info/guest_watch.c	76.00% <0.00%> (-0.62%)	⬇️
src/broker/broker.c	77.44% <0.00%> (+0.13%)	⬆️
src/common/libkvs/kvs_txn_compact.c	72.51% <0.00%> (+0.76%)	⬆️