libsubprocess: pre-exec & post-fork hooks

[chu11]

WIP for #2008 . Ignore most of this code & commits, which are garbage and I haven't even tested really. Mostly posting this to get initial thoughts on the API in subprocess.h. After going back and forth on a few ideas, I think this is not bad, but there's one hairy issue which I detail below.

typedef enum {
    FLUX_SUBPROCESS_HOOK_TYPE_PRE_EXEC,
    FLUX_SUBPROCESS_HOOK_TYPE_POST_FORK,
} flux_subprocess_hook_type_t;

flux_subprocess_hooks_t * flux_subprocess_hooks_create (void);

flux_subprocess_hooks_t * flux_subprocess_hooks_copy (
    const flux_subprocess_hooks_t *hooks);

void flux_subprocess_hooks_destroy (flux_subprocess_hooks_t *hooks);

int flux_subprocess_hooks_add (flux_subprocess_hooks_t *hooks,
                               flux_subprocess_hook_type_t type,
                               flux_subprocess_f cb);

@@ -246,11 +288,13 @@ const char *flux_cmd_getopt (flux_cmd_t *cmd, const char *var);
  */
 flux_subprocess_t *flux_exec (flux_t *h, int flags,
                               const flux_cmd_t *cmd,
-                              flux_subprocess_ops_t *ops);
+                              flux_subprocess_ops_t *ops,
+                              flux_subprocess_hooks_t *hooks);
 
 flux_subprocess_t *flux_local_exec (flux_reactor_t *r, int flags,
                                     const flux_cmd_t *cmd,
-                                    flux_subprocess_ops_t *ops);
+                                    flux_subprocess_ops_t *ops,
+                                    flux_subprocess_hooks_t *hooks);

So basically, we have a new flux_subprocess_hooks_t data structure that users create/setup and pass into flux_exec() or flux_local_exec() as a new arg. Pros / cons I found along the way:

• I considered adding "hooks" to flux_cmd_t, but I did not like the fact that hooks only work against local subprocess and not remote ones. So I felt the the hook API/infrastructure had to be outside of the the cmd part.

• It's a tad inconsistent that we have flux_cmd_t which is a created/destroyed object, flux_subprocess_ops_t, which is just a struct, and now flux_subprocess_hooks_t which is a created/destroyed object. I guess it's not a big deal, but that's there.

• I assume the hook callbacks need to be able to call accessors, such as flux_subprocess_pid()? I think there are racy bits here and there that need to be handled for that case. But that made me wonder, do hook callbacks need access to the flux_subprocess_t structure? If no, that makes things easier.

• The callback set in flux_subprocess_hooks_add() uses the current flux_subprocess_f function prototype (which I picked half randomly), which is just void (*flux_subprocess_f) (flux_subprocess_t *p);. So right now there is no way to pass an extra argument to the callback. Is this important?

  • Right now you can't use flux_subprocess_aux_set(), b/c there's no way to know if a process has been forked yet or not by the time a user calls flux_subprocess_aux_set().
    • in the old subprocess API, b/c there were functions like subprocess_fork() and subprocess_exec() that allowed callers to pause between hook points.
    • This isn't a problem in the libsubprocess API in the general case b/c callbacks are handled via the reactor. So as long as flux_subprocess_aux_set() is called before re-entering the reactor, you're guaranteed to get your aux pointer in any of the ops callbacks.
  • I assume the POST_FORK hook could be handled via a reactor callback, but I'm assuming the PRE_EXEC cannot, but didn't look into this too much.
  • So do we want/need to create a new callback function prototype that adds a void *arg argument to the hook callback? And flux_subprocess_hooks_add() would be changed to something like flux_subprocess_hooks_add (p, type, cb, arg, arg_free)?

[chu11]

I assume the hook callbacks need to be able to call accessors, such as flux_subprocess_pid()? I think there are racy bits here and there that need to be handled for that case. But that made me wonder, do hook callbacks need access to the flux_subprocess_t structure? If no, that makes things easier.

Oh wait, for the PRE_EXEC hook, it's executed in the child. So the flux_subprocess_t is a copy of the parent and probably shouldn't be something the hook callback should be using. Perhaps the hook callback shouldn't be passed a flux_subprocess_t?

[grondo]

Good observations!

I assume the hook callbacks need to be able to call accessors, such as flux_subprocess_pid()? I think there are racy bits here and there that need to be handled for that case. But that made me wonder, do hook callbacks need access to the flux_subprocess_t structure? If no, that makes things easier.

I think yes. One use case might be a post-fork hook in which the parent moves the child task to another cgroup, or adjusts the subprocess before it has a chance to call exec(2). For this, the callback will at least need to get the PID of the child.

Right now you can't use flux_subprocess_aux_set(), b/c there's no way to know if a process has been forked yet or not by the time a user calls flux_subprocess_aux_set().

Hm, yeah that is a bit tricky, but I think it would work for the time being. We could also handle this by allowing optional char *name and void *arg to flux_subprocess_hooks_add(), which could push the void *arg into the subprocess without a race.

I assume the POST_FORK hook could be handled via a reactor callback, but I'm assuming the PRE_EXEC cannot, but didn't look into this too much.

Yeah, possibly POST_FORK is not necessary -- it would be a shame to add all this code only to have one place where it is required, so I think we should think about this further. I'll try to gather some existing use cases and we can maybe go through them to see if there is a much simpler way to handle all cases.

PRE_EXEC is a bit more tricky because this needs to run in the child process not the parent, and I don't think it would be safe to try to use the reactor callbacks in that context.

[grondo]

Oh wait, for the PRE_EXEC hook, it's executed in the child. So the flux_subprocess_t is a copy of the parent and probably shouldn't be something the hook callback should be using. Perhaps the hook callback shouldn't be passed a flux_subprocess_t?

It is probably safe to use the flux_subprocess_t read-only in the child, though besides accessing aux data I can't think of any real use cases for it. Perhaps the safe thing to do would be to add an in_child flag for a flux_subprocess_t that returned errors for many of the calls when in the child.

[grondo]

I assume the POST_FORK hook could be handled via a reactor callback

Regarding this, a key requirement for post_fork is that this callback happens after the fork() but before exec(). I can't remember if this is true for the FLUX_SUBPROCESS_STARTED callback. If so, then we only (might) need the pre_exec hook, which could simplify the design quite a bit...

[grondo]

Maybe I'll see if an implementation of #1948 using FLUX_SUBPROCESS_STARTED is workable and that may give us some insight here.

[chu11]

I think yes. One use case might be a post-fork hook in which the parent moves the child task to another cgroup, or adjusts the subprocess before it has a chance to call exec(2). For this, the callback will at least need to get the PID of the child.

It is probably safe to use the flux_subprocess_t read-only in the child, though besides accessing aux data I can't think of any real use cases for it. Perhaps the safe thing to do would be to add an in_child flag for a flux_subprocess_t that returned errors for many of the calls when in the child.

I like the idea of a in_child flag (Edit: or perhaps it should be a in_hook_cb flag). However, looking through the libsubprocess API, the only useful calls might be the basic getters like flux_subprocess_pid() and flux_subprocess_get_cmd(). Anything related to exit codes isn't relevant and we'd probably want to make all read/write functions (i.e. to stdin/stdout) uncallable by hooks. And if I add a void *arg to flux_subprocess_hooks_add(), I might block the aux functions too b/c that's racy (b/c the caller calls aux_set() after flux_exec() returns).

Regarding this, a key requirement for post_fork is that this callback happens after the fork() but before exec(). I can't remember if this is true for the FLUX_SUBPROCESS_STARTED callback. If so, then we only (might) need the pre_exec hook, which could simplify the design quite a bit...

We'll need POST_FORK then. By the time FLUX_SUBPROCESS_STARTED is sent to the callback, the exec may have already happened.

[grondo]

We'll need POST_FORK then. By the time FLUX_SUBPROCESS_STARTED is sent to the callback, the exec may have already happened.

It seems like this might be fixable for "local" exec. We pause the child process after fork(2) until it is signaled that it is ok to continue and call exec(2). The purpose of this strategy is to allow the parent to manipulate the cihld process before it starts execution.

If it ends up being too difficult to enable this intended behavior for the FLUX_SUBPROCESS_STARTED reactor callback, then we need POST_FORK as you say. However, then I'd suggest that there isn't a real use case for FLUX_SUBPROCESS_STARTED, and we should remove that reactor callback.

[grondo]

And if I add a void *arg to flux_subprocess_hooks_add(), I might block the aux functions too b/c that's racy (b/c the caller calls aux_set() after flux_exec() returns).

Does it seem like we made a misstep in API design by not having a way to create a flux_subprocess_t before calling exec on it?

Also, I'm kind of questioning my original proposal for the design of these subprocess hooks (or maybe just one hook as the case may be) in #2008, especially now that we're discussing the actual implementation. I'm really open to any changes here if anyone has ideas.

For instance, I'm not really sure the libsubprocess interface for these hooks needs to support a stack of functions. The use case I had in mind was to allow the job shell or other components managing processes to be able to support a configurable stack of plugins (e.g. like Slurm spank plugins). However, support for multiple plugins would be more easily supported on top of the pre_exec hook, not by libsubprocess itself.

[chu11]

If it ends up being too difficult to enable this intended behavior for the FLUX_SUBPROCESS_STARTED reactor callback, then we need POST_FORK as you say.

I think it'd be hard at this point to change it. The state change callback mechanism is sort of independent of anything else going on, so it might be hard to synchronize things. And obviously for remote processes, the synchronizing would be even tougher.

However, then I'd suggest that there isn't a real use case for FLUX_SUBPROCESS_STARTED, and we should remove that reactor callback.

I see your point. I think its primary purpose is that it is the "first state" in the state change DAG that the user sees.

 * init -> started
 * started -> exec failed
 * started -> running
 * running -> exited
 * any state -> failed

INIT is never reported to the user, so without STARTED, the first state a user would see is RUNNING or EXEC_FAILED. Which I suppose could be fine. It seems that STARTED is used in one place in cron.

    if (state == FLUX_SUBPROCESS_STARTED) {
        t->started = 1;
        clock_gettime (CLOCK_REALTIME, &t->starttime);
        if (t->timeout >= 0.0)
            cron_task_timeout_start (t);
    }

But this was simply done right after the flux_rexec() before, so that could be tweaked. Let me make an issue on this.

[chu11]

Does it seem like we made a misstep in API design by not having a way to create a flux_subprocess_t before calling exec on it?

Looking back at #1331, the original design did not have aux set/get involved, so I guess it was something added later. I remember discussing it at one point in the development, and since all callbacks were called after re-entering the reactor, it was considered "safe". But that was of course before #2008.

If we supported a flux_subprocess_create(), most of the subprocess API would be unusable until after flux_exec() or similar have been called. It seems like a lot to support a flux_subprocess_create() for just aux set/get support.

I hate to even suggest it, but would something like:

int flux_exec (flux_t *h, ..., flux_subprocess_aux_t *aux);

be a good idea?

[grondo]

I hate to even suggest it, but would something like:

int flux_exec (flux_t *h, ..., flux_subprocess_aux_t *aux);
be a good idea?

Hm, it does solve the immediate problem, though.

Another idea would be to add flux_cmd_aux_set () to flux_cmd_t and allow access to the aux items for the underlying flux_cmd_t for any flux_subprocess_t created from that cmd object.

Another question I had looking at this again is if libsuprocess should be updated to use the common aux.[ch] implementation for auxillary data for consistency with other libflux object interfaces?

[chu11]

Another idea would be to add flux_cmd_aux_set () to flux_cmd_t and allow access to the aux items for the underlying flux_cmd_t for any flux_subprocess_t created from that cmd object.

I thought about that as well, but wasn't sure if we might need to add different auxiliary data for each subprocess? Then the API might be inconvenient.

Another question I had looking at this again is if libsuprocess should be updated to use the common aux.[ch] implementation for auxillary data for consistency with other libflux object interfaces?

Internally it already does.

[grondo]

Internally it already does.

Oops! yeah, I was somehow on an old branch when looking at the code 😞

thought about that as well, but wasn't sure if we might need to add different auxiliary data for each subprocess? Then the API might be inconvenient.

I was thinking that the aux data for the flux_cmd_t parent of a flux_subprocess_t would be overridden by the aux data for each subprocess. That is, kind of like the case with flux_future_fulfill_with the flux_subprocess_t aux list would first check the local aux list, and if the key doesn't exist it would fall back to the underlying flux_cmd_t aux list.

[chu11]

Also, I'm kind of questioning my original proposal for the design of these subprocess hooks (or maybe just one hook as the case may be) in #2008, especially now that we're discussing the actual implementation. I'm really open to any changes here if anyone has ideas.

One subtlety I disliked about having another arg variable in flux_subprocess_hook_add() like:

int flux_subprocess_hooks_add (flux_subprocess_hooks_t *hooks,
                               flux_subprocess_hook_type_t type,
                               flux_subprocess_hook_f cb,
                               void *arg);

is that the user now has to manage the destruction of arg carefully. B/c a flux_subprocess_hooks_t could be used by multiple subprocesses.

If we no longer need to support a stack of hooks, then flux_subprocess_hooks_t could just be a pointer to a struct the user passes in as an argument, just like flux_subprocess_ops_t. It would make the API a lot easier and I think more straight forward. The user could pass in something like this.

struct flux_subproces_hooks {
    flux_subprocess_hook_t pre_exec_cb;
    void *arg_pre_exec_cb;
    flux_subprocess_hook_t post_fork_cb;
    void *arg_post_fork_cb;
};

Edit: I now realize the above may not make much sense. What I meant was:

data = malloc (100);
hooks = flux_subprocess_hooks_create();
flux_subprocess_hooks_add (hooks, type, cb, data);
p = flux_exec (h, ..., hooks);
....
flux_subprocess_hooks_destroy (hooks);
// at this point is it safe to destroy 'data'?
flux_subprocess_destroy (p);

It's a little tricky to determine when its safe to destroy certain data. Do we pass in a free_args kind of callback too? If we have user pass in a plain old struct, I think there's more clarity in how they have to manage their data.

[chu11]

re-pushed trying the approach of having a struct like

/*
 *  flux_subprocess_hooks_t: Hook functions to execute at pre-defined
 *  points.  Hooks can only be executed on local processes.
 */
typedef struct {
    flux_subprocess_hook_f pre_exec_cb;
    void *pre_exec_arg;
    flux_subprocess_hook_f post_fork_cb;
    void *post_fork_arg;
} flux_subprocess_hooks_t;

available. I think the API and implementation should be cleaner and more obvious what's going on, as well as what is the responsibility of the caller.

[grondo]

I think the API and implementation should be cleaner and more obvious what's going on, as well as what is the responsibility of the caller.

Did you consider making the hooks an optional part of the ops structure? Since the hooks are the uncommon case it would be nice not to add an extra NULL parameter for the very common case.

(Note: I didn't look at this it could be a dumb idea)

[chu11]

Did you consider making the hooks an optional part of the ops structure? Since the hooks are the uncommon case it would be nice not to add an extra NULL parameter for the very common case.

I hadn't thought of that It's a good idea. I guess the minor con of doing so is adding hooks into the ops structure which can only be used with a local subprocess and not a remote subprocess. It could make the API a tad more confusing.

I think it's a minor pro/con on either direction.

[chu11]

while trying to add a simple unit test for the pre_exec hook, I hit a snag:

First I did something like:

    flux_subprocess_hooks_t hooks = {
        .pre_exec_cb = count_hook_cb,
        .pre_exec_arg = &hook_count
    };

i.e. have the hook increment a counter. But it didn't work, eventually realizing, "Oh yeah, it's updating a count value that's in the child process memory space". Ok .., lets instead do

    flux_subprocess_hooks_t hooks = {
        .pre_exec_cb = pipe_write_cb,
        .pre_exec_arg = pipe_fds
    };

having the hook just write some data to a pipe to be read back in the parent. This didn't work either and stumbled me for a bit, until I realized, "Oh yeah, libsubprocess doesn't want any file descriptors from the parent falling through to the eventual executed command. So we close all the parent file descriptors.".

• The latter issue above could be rectified easily. I just need to call the pre-exec hook before closing all parent file descriptors. But should I? Obviously this is specifically a unit test case, and I'm not sure it's a general use case we need to be considered? (I eventually solved this in the unit test using shared memory).

• But the above two hiccups made me ponder if passing along an arg is a good idea. As the guy writing the library, I even stumbled twice on this, b/c of how one normally thinks how "callbacks" and "args to callbacks" function.

[grondo]

Use case for pre-exec hook is to have some state created for all children in the parent, then apply it in each child before exec. In all cases the "state" will be read-only and won't need to be modified in the parent.

E.g. in the job shell use case the "state" might be a structure of assigned values for the child process, e.g. local taskid, global taskid, pre-assigned default cpu affinity, etc.

A good test case would be to create a child[] array assigning an identifier to each child and make sure that each spawned subprocess gets the correct identifier (by writing it back out on stdout?)

[chu11]

Use case for pre-exec hook is to have some state created for all children in the parent, then apply it in each child before exec. In all cases the "state" will be read-only and won't need to be modified in the parent.

In that case, I guess the args variable will be fine then. Just got concerned that I even got confused once or twice.

[chu11]

Just pushed a branch that I would consider a reasonable merge candidate, going with the new flux_subprocess_hooks_t structure and it being passed as an argument to flux_exec() and flux_local_exec().

[grondo]

Just pushed a branch that I would consider a reasonable merge candidate, going with the new flux_subprocess_hooks_t structure and it being passed as an argument to flux_exec() and flux_local_exec().

Thanks! Perhaps I'll try an implementation in flux-start using this branch? I think I have most of the code laying around somewhere...

[chu11]

Thanks! Perhaps I'll try an implementation in flux-start using this branch? I think I have most of the code laying around somewhere...

Is this for #1948? I can take a crack at it as well, maybe see how it works. I lack some background for the issue in question though, so could chat about it tomorrow.

[chu11]

re-pushed branch, I lazily added munmap() to a unit test and forgot the length parameter.

[grondo]

Is this for #1948? I can take a crack at it as well, maybe see how it works.

Yes. I already had most of the code written before I realized there was no way to implement it.

[grondo]

One very minor thing I noticed at first:

typedef struct {
    flux_subprocess_hook_f pre_exec_cb;                                        
    void *pre_exec_arg;                                                        
    flux_subprocess_hook_f post_fork_cb;                                       
    void *post_fork_arg;                                                       
} flux_subprocess_hooks_t;                                                   

For consistency with flux_subprocess_ops_t I think we should drop the _cb from the pre_exec_cb and post_fork_cb struct member names.

[grondo]

Proof of concept at least seemed to work out well -- here's the most relevant bit:

@@ -548,6 +631,15 @@ void pmi_server_finalize (void)                           
     pmi_simple_server_destroy (ctx.pmi.srv);                                  
 }                                                                             
                                                                               
+void pre_exec_hook (flux_subprocess_t *p, void *arg)                          
+{                                                                             
+    struct client *cli = arg;                                                 
+    char *s = NULL;                                                           
+    hwloc_bitmap_list_asprintf (&s, cli->cpuset);                             
+    fprintf (stderr, "setting cpubind for client rank=%d: %s\n", cli->rank, s);
+    hwloc_set_cpubind (topology, cli->cpuset, 0);                             
+}                                                                             
+                                                                              
 int client_run (struct client *cli)                                           
 {                                                                             
     flux_subprocess_ops_t ops = {                                             
@@ -557,6 +649,10 @@ int client_run (struct client *cli)                       
         .on_stdout = NULL,                                                    
         .on_stderr = NULL,                                                    
     };                                                                        
+    flux_subprocess_hooks_t hooks = {                                         
+        .pre_exec_cb = pre_exec_hook,                                         
+        .pre_exec_arg = cli,                                                  
+    };                                                                        
     /* We want stdio fallthrough so subprocess can capture tty if             
      * necessary (i.e. an interactive shell)                                  
      */                                                                       
@@ -564,7 +660,7 @@ int client_run (struct client *cli)                        
                                     FLUX_SUBPROCESS_FLAGS_STDIO_FALLTHROUGH,  
                                     cli->cmd,                                 
                                     &ops,                                     
-                                    NULL)))                                   
+                                    &hooks)))                                 
         log_err_exit ("flux_exec");                                           
     if (flux_subprocess_aux_set (cli->p, "cli", cli, NULL) < 0)               
         log_err_exit ("flux_subprocess_aux_set");                             

grondo@asp:~/git/flux-core.git$ src/cmd/flux start -s 2                        
flux-start: warning: setting --bootstrap=selfpmi due to --size option          
setting cpubind for client rank=0: 0-1                                         
setting cpubind for client rank=1: 2-3                                         
grondo@asp:~/git/flux-core.git$ flux hwloc info                                
2 Machines, 4 Cores, 4 PUs                                                     

[chu11]

For consistency with flux_subprocess_ops_t I think we should drop the _cb from the pre_exec_cb and post_fork_cb struct member names.

Re-pushed fixing that nit.

[codecov-io]

Codecov Report

Merging #2152 into master will decrease coverage by <.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master    #2152      +/-   ##
==========================================
- Coverage   80.47%   80.47%   -0.01%     
==========================================
  Files         200      200              
  Lines       31817    31831      +14     
==========================================
+ Hits        25604    25615      +11     
- Misses       6213     6216       +3

Impacted Files	Coverage Δ
src/cmd/builtin/proxy.c	74.41% <ø> (ø)	⬆️
src/cmd/flux-start.c	78.18% <ø> (ø)	⬆️
src/broker/runlevel.c	81% <ø> (ø)	⬆️
src/common/libsubprocess/server.c	70.87% <100%> (ø)	⬆️
src/common/libsubprocess/local.c	76.73% <100%> (+0.92%)	⬆️
src/common/libsubprocess/subprocess.c	87.32% <100%> (+0.07%)	⬆️
src/modules/connector-local/local.c	73.82% <0%> (-1.03%)	⬇️
src/common/libutil/veb.c	98.28% <0%> (-0.58%)	⬇️
src/common/libflux/message.c	80.88% <0%> (+0.25%)	⬆️
src/modules/barrier/barrier.c	80.53% <0%> (+2.01%)	⬆️

[grondo]

You happy with this one @chu11? Happy to merge this as is, any minor tweaks can be done later.

[chu11]

yeah, I'm happy with it, it seems to accomplish what is needed at this point.