From d6e365c0f59346a9401e1b2cc2c58246c7b5a9ba Mon Sep 17 00:00:00 2001 From: Ralph Castain Date: Tue, 4 Aug 2026 11:54:17 -0600 Subject: [PATCH] Make the iof's XON/XOFF actually stop the producer The iof has carried the shape of stdin back-pressure for years and none of it did anything. A daemon whose stdin sink passed PRTE_IOF_MAX_INPUT_BUFFERS told the HNP so, the HNP logged the message and returned, and the sink went on queueing - bounded only by iof_base_output_limit, which defaults to INT_MAX. The review that landed in #2610 documented that rather than fixing it, because the fix was not ours to make: stdin originates in a PMIx server's read of its own stdin or in a tool's PMIx_IOF_push, and neither end honored a refusal, so returning an error would have dropped bytes rather than slowed anybody. PMIx has since grown PMIx_server_IOF_flow_control, which suspends the read at the source - the bytes stay in the producer's own input stream and the OS applies the back-pressure, so nothing is buffered on behalf of a suspended stream and nothing is lost. Wire both halves to it. A daemon's XON/XOFF, which prte_iof_hnp_recv already had to screen ahead of everything else, now becomes a wildcard flow-control call. Wildcard because the message says only that this daemon is behind and never which producer filled it, so every process feeding us stdin is suspended; that is the conservative reading, and any of them may be the one responsible. The HNP's own local procs never involve the RML at all - push_stdin returns PRTE_ERR_OUT_OF_RESOURCE when a local sink crosses the same threshold, and pmix_server_stdin_push discarded it. It now turns that into PMIX_ERR_IOF_XOFF on the push_stdin completion, which PMIx reads as "I have the data, suspend the stream" and which is deliberately not a failure - the tool is not told anything is wrong, because nothing is. The obligation that comes with it is the pairing. PMIx has no status meaning "resume", so an XOFF we never release leaves a producer suspended. prte_mca_iof_hnp_component.xoff is the latch, and release_flow_control() runs from every path a backed-up sink can leave that state by: the check: and finish: arms of the write handler, and hnp_close/hnp_complete, which release a stdin sink directly and never reach the write handler at all. That last pair is the case that matters - a sink is backed up precisely when its proc stopped reading, which is the proc a teardown is most likely to be retiring. The helper is a no-op when nothing is outstanding, so it is safe to call from anywhere, which is what lets it be called from everywhere it must be. Worth being precise about what an unpaired XOFF costs, because it bounds how bad this can get: PMIx keeps no sticky suspension state. pmix_iof_flow_control pushes to the peers that exist at the moment of the call, and the only durable state is the xoff flag on the read event inside the producer. The producer is prun, which is per-job, so the suspension dies with it and a later prun starts reading normally. So the worst case is a hung job, not a hung DVM - the next job is unaffected, and the first sink to drain clears the stale latch. That is a property of PMIx's present design rather than something it owes us, so the release paths above are still the contract. Several procs taking stdin at different rates will oscillate: one draining proc turns the producers back on while another is still behind, and that one asserts XOFF again on its next write. That is the intended behavior - the alternative to oscillating is stalling, and nothing is dropped either way. The prted module's CHECK block has carried the same caveat in its own comment since ORTE. All of it is gated on PRTE_PMIX_IOF_FLOW_CONTROL, from a new PRTE_CHECK_PMIX_CAP([IOF_FLOW_CONTROL]). Against an older PMIx every path compiles away and the previous behavior returns exactly: the message is consumed quietly and the sink queues. Verified by building both ways. The dockerswarm suite gains a pass that counts the two halves in the HNP's log rather than grepping for them, because an implementation that asserts fifty times and releases once is indistinguishable from a correct one under a presence test - and is the hang. It fails on a mismatch, on flow control never engaging at all (which would mean the reader was too fast and the case proved nothing), and on any byte lost across the suspensions. Verified against a live DVM two ways. Throughput: 1 MB of stdin into a reader taking 2 KB every 3 ms produced 37 XOFF assertions and 37 XON releases, exactly paired, with all 1048576 bytes delivered and the checksum intact. Teardown: a job whose proc stops reading and then exits while its sink is still backed up released both of its XOFFs, and the two jobs run after it on the same persistent DVM each took their full 256 KB of stdin byte-identically. Signed-off-by: Ralph Castain --- config/prte_setup_pmix.m4 | 18 ++++ contrib/dockerswarm/AGENTS.md | 12 +++ contrib/dockerswarm/run-tests.sh | 37 ++++++++ src/mca/iof/AGENTS.md | 140 ++++++++++++++++++---------- src/mca/iof/hnp/AGENTS.md | 8 +- src/mca/iof/hnp/iof_hnp.c | 59 +++++++++++- src/mca/iof/hnp/iof_hnp.h | 7 ++ src/mca/iof/hnp/iof_hnp_component.c | 3 +- src/mca/iof/hnp/iof_hnp_receive.c | 38 ++++++-- src/prted/pmix/pmix_server_gen.c | 28 ++++++ 10 files changed, 285 insertions(+), 65 deletions(-) diff --git a/config/prte_setup_pmix.m4 b/config/prte_setup_pmix.m4 index a49f974bcd..02ac93d231 100644 --- a/config/prte_setup_pmix.m4 +++ b/config/prte_setup_pmix.m4 @@ -309,6 +309,24 @@ AC_DEFUN([PRTE_CHECK_PMIX],[ [$prte_pmix_iof_deliver_local], [Whether PMIx honors PMIX_IOF_LOCAL_OUTPUT on an IOF delivery]) + dnl The iof has always had the shape of XON/XOFF back-pressure on stdin - + dnl a daemon whose stdin sink passes PRTE_IOF_MAX_INPUT_BUFFERS tells the + dnl HNP so - but the HNP could not act on it. Stdin originates in a PMIx + dnl server's read of its own stdin or in a tool's PMIx_IOF_push, and until + dnl PMIx grew PMIx_server_IOF_flow_control there was no way to stop either + dnl at the source: a refusal anywhere would have dropped bytes rather than + dnl slowed the producer. Without this the HNP consumes the message and + dnl the daemon's sink simply queues, bounded only by iof_base_output_limit. + AC_MSG_CHECKING([for PMIx IOF stdin flow control]) + PRTE_CHECK_PMIX_CAP([IOF_FLOW_CONTROL], + [AC_MSG_RESULT([yes]) + prte_pmix_iof_flow_control=1], + [AC_MSG_RESULT([no]) + prte_pmix_iof_flow_control=0]) + AC_DEFINE_UNQUOTED([PRTE_PMIX_IOF_FLOW_CONTROL], + [$prte_pmix_iof_flow_control], + [Whether PMIx supports PMIx_server_IOF_flow_control]) + AC_MSG_CHECKING([for LTO compatibility]) PRTE_CHECK_PMIX_CAP([LTO], [PRTE_PMIX_LTO_CAPABILITY=1 diff --git a/contrib/dockerswarm/AGENTS.md b/contrib/dockerswarm/AGENTS.md index 423ddc796b..f345a08cda 100644 --- a/contrib/dockerswarm/AGENTS.md +++ b/contrib/dockerswarm/AGENTS.md @@ -418,6 +418,18 @@ unpack error. Do not weaken the reader's pace: at `cat` speed, or even at the pace the two cases above use, the backlog may never cross 50 and the case passes without having tested anything. +A fourth pass runs the same reader with `--prtemca iof_base_verbose 1` and +**counts** the two halves of the flow control in the HNP's log. That is +the case that covers `PMIx_server_IOF_flow_control` actually being +reached, and it is a count rather than a grep on purpose: PMIx has no +status meaning "resume", so an implementation that asserts XOFF fifty +times and releases once looks identical to a correct one under a presence +test — and the job it produces is *hung*, not slow. The case fails if the +counts differ, if flow control never engaged at all (which would mean the +reader was too fast and the case proved nothing), or if a byte went +missing across the suspensions. See the framework guide's *Flow control* +section for where each half lives. + **Grow** (`elastic grow node2:2,node3:2`): phase-1 `PMIX_SUCCESS`, then phase-2 `PMIX_DVM_IS_READY`, and `prted` now running on node2 and node3. diff --git a/contrib/dockerswarm/run-tests.sh b/contrib/dockerswarm/run-tests.sh index edd2e3a1bd..f0065748f0 100755 --- a/contrib/dockerswarm/run-tests.sh +++ b/contrib/dockerswarm/run-tests.sh @@ -5007,6 +5007,43 @@ gcc -o /root/staged_marker /root/staged_marker.c' >/dev/null 2>&1 || ok "no unpack error at the HNP while stdin was backed up" ON 2 'rm -f /tmp/iof_xoff_out.txt' >/dev/null 2>&1 + # The same reader again, this time with the iof talking, to assert + # that flow control actually engaged AND that every XOFF was paired + # with an XON. + # + # The pairing is the part that matters. PMIx has no status meaning + # "resume", so a suspension the HNP asserts and forgets to release + # stalls the producer for the life of the job - a hang, not a + # slowdown. On the HNP side that release lives in + # release_flow_control(), called from every path a backed-up sink + # can leave that state by, including the ones where the sink is torn + # down rather than drained. + # + # Counting rather than merely grepping is deliberate: an + # implementation that asserts XOFF once and releases once looks + # identical to a correct one under a presence test, and so does one + # that asserts fifty times and releases once. + RUN 'cp /tmp/prte.out /tmp/prte.out.mark2 2>/dev/null || : ' >/dev/null 2>&1 + out=$(RUN 'cd /tmp && timeout 300 prun --prtemca iof_base_verbose 1 -n 1 \ + '"$SC"' /tmp/iof_pair_out.txt 2048 3000 < iof_stdin_in.txt \ + 2>iof_pair_err.txt; echo "rc=$?"') + rc=$(echo "$out" | sed -n 's/^rc=//p') + got=$(echo "$out" | sed -n 's/^SLOWCAT-BYTES //p') + pairlog=$(RUN 'diff /tmp/prte.out.mark2 /tmp/prte.out 2>/dev/null | grep "^>" \ + || cat /tmp/prte.out 2>/dev/null') + nxoff=$(echo "$pairlog" | grep -c 'buffer backed up - holding') + nxon=$(echo "$pairlog" | grep -c 'releasing stdin flow control') + [ "$nxoff" -gt 0 ] \ + && ok "stdin flow control engaged ($nxoff XOFF)" \ + || bad "stdin flow control never engaged - the reader was not slow enough, or the backlog never crossed PRTE_IOF_MAX_INPUT_BUFFERS" + [ "$nxoff" = "$nxon" ] \ + && ok "every XOFF was paired with an XON ($nxon)" \ + || bad "unpaired stdin flow control: $nxoff XOFF vs $nxon XON - a producer is left suspended" + [ "$rc" = 0 ] && [ "$got" = "$insz" ] \ + && ok "delivery stayed exact across $nxoff suspensions" \ + || bad "flow-controlled stdin lost data (rc=$rc, sent=$insz received=$got)" + RUN 'rm -f /tmp/iof_pair_out.txt' >/dev/null 2>&1 + # And the same slow reader on the HNP node, where push_stdin writes # into the proc sink directly instead of going out over the RML: the # HNP and the daemon carry separate copies of the write handler, so a diff --git a/src/mca/iof/AGENTS.md b/src/mca/iof/AGENTS.md index 7e3ed0418a..a9e1977727 100644 --- a/src/mca/iof/AGENTS.md +++ b/src/mca/iof/AGENTS.md @@ -419,56 +419,98 @@ tree or an old commit is **retired**. --- -## Flow control — read this before you trust it - -stdin can outrun a slow reader, and the framework has the *shape* of -XON/XOFF back-pressure keyed on `PRTE_IOF_MAX_INPUT_BUFFERS` (50 queued -chunks). **It does not actually throttle anything today.** Know that -before you reason about a stdin backlog, and do not "fix" a symptom by -leaning on it. - -What is really wired: - -- On a daemon, when `prte_iof_base_write_output` reports the stdin sink - backlog has crossed 50 (or a write errors out), - `prte_iof_prted_send_xonxoff(PRTE_IOF_XOFF)` sends the HNP a buffer - holding nothing but the tag, latched by - `prte_mca_iof_prted_component.xoff`; when the backlog drains below 50 it - sends `PRTE_IOF_XON`. -- The HNP **recognizes** those messages (`prte_iof_hnp_recv` screens the - `PRTE_IOF_XON | PRTE_IOF_XOFF` mask ahead of everything else) and traces - them, but takes no action. -- On the HNP, `push_stdin` to a *local* proc returns - `PRTE_ERR_OUT_OF_RESOURCE` when its own sink passes the same threshold. - -Why nothing happens, and why that is currently correct: stdin originates -in the PMIx server's read of its own stdin, or in a tool's -`PMIx_IOF_push`. Neither end honors a refusal. PRRTE's own glue -(`pmix_server_stdin_push` in -[`src/prted/pmix/pmix_server_gen.c`](../../prted/pmix/pmix_server_gen.c)) -discards whatever `prte_iof.push_stdin` returns and reports -`PMIX_SUCCESS`; PMIx's completion handler for the server-side read -(`opcbfn` in its `src/common/pmix_iof.c`) explicitly discards the status -and re-arms the read regardless. So a refusal anywhere in this framework -would not slow the producer — it would only **drop bytes**. Queueing is -the only lossless behavior available, and queueing is what happens. - -The consequence worth knowing: a daemon's stdin sink is bounded only by +## Flow control + +stdin can outrun a slow reader, so the framework applies XON/XOFF +back-pressure keyed on `PRTE_IOF_MAX_INPUT_BUFFERS` (50 queued chunks). +**It reaches all the way back to the producer, and it did not always** — +until PMIx grew `PMIx_server_IOF_flow_control` the whole mechanism was a +signal the HNP logged and discarded. If you are reading an older tree, or +an older copy of this guide, that is what it is describing. + +The producer is never ours. stdin originates in a PMIx server's read of +its own stdin, or in a tool's `PMIx_IOF_push`, so the only way to slow it +is to ask PMIx to stop it at the source. That is what +`PMIx_server_IOF_flow_control` does: it leaves the read un-armed, so the +bytes stay in the producer's own input stream and the OS applies the +back-pressure. **Nothing is buffered on behalf of a suspended stream and +nothing is dropped** — an XOFF is not permission to lose data. + +The two halves, which are separate mechanisms that happen to share a +vocabulary: + +- **A daemon** whose stdin sink crosses 50 (or whose write errors out) + calls `prte_iof_prted_send_xonxoff(PRTE_IOF_XOFF)`, latched by + `prte_mca_iof_prted_component.xoff`, and sends `PRTE_IOF_XON` when the + backlog drains. `prte_iof_hnp_recv` screens the + `PRTE_IOF_XON | PRTE_IOF_XOFF` mask ahead of everything else and turns + it into a wildcard `PMIx_server_IOF_flow_control` call. Wildcard because + the message says only *that* this daemon is behind, never which producer + filled it — so every process feeding us stdin is suspended, which is the + conservative reading. +- **The HNP's own local procs** never involve the RML at all. + `push_stdin` returns `PRTE_ERR_OUT_OF_RESOURCE` when a local sink passes + the same threshold, latched by `prte_mca_iof_hnp_component.xoff`, and + the glue in + [`src/prted/pmix/pmix_server_gen.c`](../../prted/pmix/pmix_server_gen.c) + turns that into `PMIX_ERR_IOF_XOFF` on the `push_stdin` completion — + which PMIx reads as "I have the data, suspend the stream". The matching + release is `release_flow_control()` in + [`hnp/iof_hnp.c`](hnp/iof_hnp.c). + +**Every XOFF must be paired with an XON, and that is on us.** PMIx has no +status meaning "resume": a suspension persists until somebody calls the +API with `xoff` false. So a release has to run on *every* path a +backed-up sink can leave that state by — not just the one where it +drains, but the ones where it is torn down. `release_flow_control()` is +called from the `check:` and `finish:` arms of `stdin_write_handler` +**and** from `hnp_close`/`hnp_complete`, which release a stdin sink +directly and never reach the write handler at all — which is exactly the +case that matters, since a sink is backed up precisely when its proc +stopped reading, and that is the proc a teardown is likely to be +retiring. It is a no-op when no XOFF is outstanding, so it is safe to +call from anywhere, which is what lets it be called from everywhere it +must be. + +**The blast radius is the job, not the DVM** — know why, because it is +not obvious and it is what makes the failure tolerable. PMIx keeps no +sticky suspension state: `pmix_iof_flow_control` walks the peers that +exist *at the moment of the call* and pushes the request to them, and +the only durable state is the `xoff` flag on the read event inside the +producer itself. The producer is `prun`, which is per-job, so a +suspension it is still carrying dies when it does. A `prun` that +connects afterwards has never been told anything and starts reading +normally. So the worst an unreleased XOFF can do is hang **that** job; +the next one is unaffected, and the first sink to drain clears the stale +latch. Do not read that as license to skip a release — a hung job is +still a bug, and the reasoning above is a property of PMIx's current +design rather than a guarantee it owes us. The dockerswarm suite counts +the two and fails on a mismatch rather than merely checking that flow +control happened. + +The oscillation is expected. Several procs take stdin at different rates, +so one draining proc can turn the producers back on while another is +still behind; the one still behind asserts XOFF again on its next write. +That is the intended failure mode — the alternative to oscillating is +stalling, and nothing is dropped either way. The prted module's `CHECK` +block carries the same caveat in its own comment. + `prte_iof_base_output_limit` (MCA param `iof_base_output_limit`, default -`INT_MAX`, i.e. unbounded). Pipe a very large file into a process that -reads it very slowly and the daemon's backlog grows without limit. Set -`iof_base_output_limit` to a finite value and the write handler declares -IOF hopelessly behind and fires `PRTE_JOB_STATE_FORCED_EXIT` instead. - -Making XOFF *mean* something requires a way to stop the read at the -source, which is a PMIx-side change (honor the `push_stdin` completion -status and hold the stdin read event until the host says go). Until that -exists, leave the latch alone: it is a signal the HNP logs, not a control -loop. What is **not** optional is that the HNP keep screening the tag -first — a flow-control message carries no proc and no payload, so falling -through to the output unpack reports the daemon's XOFF to the user as a -corrupted message, which is exactly what used to happen every time a -process read its stdin slowly. +`INT_MAX`) is still the harder ceiling underneath all of this: if a +sink's backlog exceeds it, the write handler concludes something is +permanently wedged and fires `PRTE_JOB_STATE_FORCED_EXIT`. With flow +control working, reaching it means the producer ignored the suspension, +not merely that the reader is slow. + +**Against a PMIx that predates the capability** (`PRTE_PMIX_IOF_FLOW_CONTROL` +is 0, from `PRTE_CHECK_PMIX_CAP([IOF_FLOW_CONTROL])` in +`config/prte_setup_pmix.m4`) every one of these paths compiles away and +the old behavior returns: the message is consumed quietly, the sink +queues, and `iof_base_output_limit` is the only bound. What is **not** +conditional is that the HNP screen the tag first — a flow-control message +carries no proc and no payload, so falling through to the output unpack +reports the daemon's XOFF to the user as a corrupted message, which is +what used to happen every time a process read its stdin slowly. --- diff --git a/src/mca/iof/hnp/AGENTS.md b/src/mca/iof/hnp/AGENTS.md index 828bf3f83b..e95140e321 100644 --- a/src/mca/iof/hnp/AGENTS.md +++ b/src/mca/iof/hnp/AGENTS.md @@ -103,9 +103,11 @@ before the proc unpack.** The order in the handler is deliberate: `PMIx_Data_unpack(…PMIX_PROC)` fails and reports the daemon's XOFF to the user as a corrupted message, which is what happened every time a process read its stdin slowly. We recognize it by mask (the control - bits are disjoint from every stream bit), trace it, and return. We do - not act on it — see the framework guide's *Flow control* section for - why acting on it would drop bytes rather than slow the producer. + bits are disjoint from every stream bit), trace it, and hand it to + `PMIx_server_IOF_flow_control` — wildcard, because the message says + only *that* this daemon is behind and never which producer filled it. + See the framework guide's *Flow control* section, and note the pairing + obligation it describes: an XOFF we never release hangs the producer. 2. **Relayed stdin.** A leading `PRTE_IOF_STDIN` is stdin a daemon is relaying on behalf of a tool attached to *it* rather than to us, and the proc that follows is the intended **recipient**, not a source. That diff --git a/src/mca/iof/hnp/iof_hnp.c b/src/mca/iof/hnp/iof_hnp.c index f59d179a46..8fdcc9433d 100644 --- a/src/mca/iof/hnp/iof_hnp.c +++ b/src/mca/iof/hnp/iof_hnp.c @@ -75,6 +75,7 @@ static void hnp_complete(const prte_job_t *jdata); static int finalize(void); static int push_stdin(const pmix_proc_t *dst_name, uint8_t *data, size_t sz); +static void release_flow_control(void); /* The API's in this module are solely used to support LOCAL * procs - i.e., procs that are co-located to the HNP. Remote @@ -273,10 +274,14 @@ static int push_stdin(const pmix_proc_t *dst_name, uint8_t *data, size_t sz) if (PRTE_IOF_MAX_INPUT_BUFFERS < prte_iof_base_write_output(&proct->name, PRTE_IOF_STDIN, data, sz, proct->stdinev->wev)) { - /* getting too backed up - stop the read event for now if it is still active */ - + /* getting too backed up - the data is queued, so this is + * "taken, now slow down" rather than a refusal. Latch it + * so stdin_write_handler knows to send the matching XON + * when this sink drains; without that pairing the + * producer would stay suspended forever */ PMIX_OUTPUT_VERBOSE((1, prte_iof_base_framework.framework_output, "buffer backed up - holding")); + prte_mca_iof_hnp_component.xoff = true; return PRTE_ERR_OUT_OF_RESOURCE; } } @@ -359,6 +364,12 @@ static int hnp_close(const pmix_proc_t *peer, prte_iof_tag_t source_tag) if (PMIX_CHECK_PROCID(&proct->name, peer)) { if (PRTE_IOF_STDIN & source_tag) { if (NULL != proct->stdinev) { + /* a sink released here never reaches the write handler, + * so this is the only chance to release an XOFF it was + * responsible for. A sink is backed up precisely when + * its proc stopped reading, which is precisely the proc + * a close is likely to be tearing down */ + release_flow_control(); PMIX_RELEASE(proct->stdinev); } proct->stdinev = NULL; @@ -409,6 +420,10 @@ static void hnp_complete(const prte_job_t *jdata) * killed does not, and that is the case this exists for. */ if (NULL != proct->stdinev) { + /* as in close(): this sink will never reach the write + * handler, so an XOFF it was responsible for has to be + * released here or it outlives the job that caused it */ + release_flow_control(); PMIX_RELEASE(proct->stdinev); proct->stdinev = NULL; } @@ -436,6 +451,37 @@ static int finalize(void) /* this function is called by the event library and thus * can access information global to the state machine */ +/* Release an XOFF we asserted because one of our own local procs' stdin + * sinks was backed up. A no-op unless we actually asserted one, so it is + * safe to call from every path a sink can leave the backed-up state by - + * including the paths where the sink is torn down rather than drained, + * which would otherwise leave the producers suspended forever. + * + * Against a PMIx that predates PMIx_server_IOF_flow_control there is + * nothing to release: push_stdin's PRTE_ERR_OUT_OF_RESOURCE is discarded + * by the glue rather than turned into an XOFF, so nothing was suspended. */ +static void release_flow_control(void) +{ +#if PRTE_PMIX_IOF_FLOW_CONTROL + pmix_status_t prc; + + if (!prte_mca_iof_hnp_component.xoff) { + return; + } + /* clear the latch first: the API completes inline, and a re-entrant + * assert would otherwise be lost */ + prte_mca_iof_hnp_component.xoff = false; + PMIX_OUTPUT_VERBOSE((1, prte_iof_base_framework.framework_output, + "%s iof:hnp releasing stdin flow control", + PRTE_NAME_PRINT(PRTE_PROC_MY_NAME))); + prc = PMIx_server_IOF_flow_control(NULL, PMIX_FWD_STDIN_CHANNEL, false, + NULL, 0, NULL, NULL); + if (PMIX_SUCCESS != prc && PMIX_OPERATION_SUCCEEDED != prc) { + PMIX_ERROR_LOG(prc); + } +#endif +} + static void stdin_write_handler(int fd, short event, void *cbdata) { prte_iof_sink_t *sink = (prte_iof_sink_t *) cbdata; @@ -525,6 +571,11 @@ static void stdin_write_handler(int fd, short event, void *cbdata) PRTE_IOF_SINK_ACTIVATE(wev); check: + if (pmix_list_get_size(&wev->outputs) < PRTE_IOF_MAX_INPUT_BUFFERS) { + /* this proc has absorbed enough to justify restarting the producers + * we suspended */ + release_flow_control(); + } if (sink->closed && 0 == pmix_list_get_size(&wev->outputs)) { /* the sink has already been closed and everything was written, time to release it */ PMIX_RELEASE(sink); @@ -532,6 +583,10 @@ static void stdin_write_handler(int fd, short event, void *cbdata) return; finish: + /* this sink is going away, so it will never drain - if it is the one + * that suspended the producers, they have to be let go here or they + * stay suspended for the life of the job */ + release_flow_control(); PMIX_RELEASE(wev); sink->wev = NULL; return; diff --git a/src/mca/iof/hnp/iof_hnp.h b/src/mca/iof/hnp/iof_hnp.h index e7fddc5789..187eaf6604 100644 --- a/src/mca/iof/hnp/iof_hnp.h +++ b/src/mca/iof/hnp/iof_hnp.h @@ -64,6 +64,13 @@ BEGIN_C_DECLS struct prte_mca_iof_hnp_component_t { prte_iof_base_component_t super; pmix_list_t procs; + /* set when we have told PMIx to stop the processes feeding us stdin + * because one of OUR OWN local procs' stdin sinks passed + * PRTE_IOF_MAX_INPUT_BUFFERS. This is the HNP's counterpart to + * prte_mca_iof_prted_component.xoff - a daemon tells us over the RML, + * but we have no one above us to tell, so we tell PMIx directly. It + * latches so the XON is sent exactly once, when the sink drains */ + bool xoff; }; typedef struct prte_mca_iof_hnp_component_t prte_mca_iof_hnp_component_t; diff --git a/src/mca/iof/hnp/iof_hnp_component.c b/src/mca/iof/hnp/iof_hnp_component.c index cf992fb2dc..ecf9699a24 100644 --- a/src/mca/iof/hnp/iof_hnp_component.c +++ b/src/mca/iof/hnp/iof_hnp_component.c @@ -63,7 +63,8 @@ prte_mca_iof_hnp_component_t prte_mca_iof_hnp_component = { .pmix_mca_open_component = prte_iof_hnp_open, .pmix_mca_close_component = prte_iof_hnp_close, .pmix_mca_query_component = prte_iof_hnp_query, - } + }, + .xoff = false }; PMIX_MCA_BASE_COMPONENT_INIT(prte, iof, hnp) diff --git a/src/mca/iof/hnp/iof_hnp_receive.c b/src/mca/iof/hnp/iof_hnp_receive.c index 0924da549a..fabac6a76d 100644 --- a/src/mca/iof/hnp/iof_hnp_receive.c +++ b/src/mca/iof/hnp/iof_hnp_receive.c @@ -95,16 +95,26 @@ void prte_iof_hnp_recv(int status, pmix_proc_t *sender, pmix_data_buffer_t *buff * unpack below reads off the end of the buffer and reports the daemon's * XOFF as a corrupted message. * - * We cannot act on it. Stdin originates in the PMIx server's read of its - * own stdin (or a tool's PMIx_IOF_push), and neither end honors a refusal: - * pmix_server_stdin_push discards what prte_iof.push_stdin returns, and - * PMIx's own completion handler discards the status and re-arms the read - * regardless. Returning an error here would therefore not slow the source - * down - it would only drop the bytes. So the daemon's sink queues, and - * iof_base_output_limit is the only ceiling. Making XOFF mean something - * requires a way to stop the read at the source; until there is one, the - * right behavior is to consume the message quietly rather than to log an - * unpack failure at the user every time a proc reads its stdin slowly. + * Acting on it means reaching the producer, and the producer is not ours: + * stdin originates in a PMIx server's read of its own stdin, or in a + * tool's PMIx_IOF_push. PMIx_server_IOF_flow_control is how we ask PMIx + * to stop them - it suspends any stdin the library is reading here and + * relays the request to every tool that has pushed stdin to us, leaving + * the unread bytes in the producer's own input stream where the OS + * applies the back-pressure. Nothing is buffered on behalf of a + * suspended stream and nothing is dropped; an XOFF is not permission to + * lose data. + * + * We do not know which producer a given daemon's backlog came from - the + * message says only that this daemon is behind - so the request is made + * wildcard, against every process feeding us stdin. That is the correct + * conservative reading: any of them may be the one filling that sink. + * + * Against a PMIx too old to have the API we do what we always did - + * consume the message quietly. The daemon's sink then queues, bounded + * only by iof_base_output_limit, which is the pre-existing behavior and + * is still better than logging an unpack failure at the user every time + * a proc reads its stdin slowly. */ if ((PRTE_IOF_XON | PRTE_IOF_XOFF) & stream) { PMIX_OUTPUT_VERBOSE((1, prte_iof_base_framework.framework_output, @@ -112,6 +122,14 @@ void prte_iof_hnp_recv(int status, pmix_proc_t *sender, pmix_data_buffer_t *buff PRTE_NAME_PRINT(PRTE_PROC_MY_NAME), (PRTE_IOF_XON & stream) ? "xon" : "xoff", PRTE_NAME_PRINT(sender))); +#if PRTE_PMIX_IOF_FLOW_CONTROL + prc = PMIx_server_IOF_flow_control(NULL, PMIX_FWD_STDIN_CHANNEL, + (PRTE_IOF_XOFF & stream) ? true : false, + NULL, 0, NULL, NULL); + if (PMIX_SUCCESS != prc && PMIX_OPERATION_SUCCEEDED != prc) { + PMIX_ERROR_LOG(prc); + } +#endif goto CLEAN_RETURN; } diff --git a/src/prted/pmix/pmix_server_gen.c b/src/prted/pmix/pmix_server_gen.c index 4ddde6d26f..18c2622398 100644 --- a/src/prted/pmix/pmix_server_gen.c +++ b/src/prted/pmix/pmix_server_gen.c @@ -1095,6 +1095,9 @@ static void pmix_server_stdin_push(int sd, short args, void *cbdata) pmix_byte_object_t *bo = (pmix_byte_object_t *) cd->server_object; uint8_t *bytes; size_t nbytes, n; +#if PRTE_PMIX_IOF_FLOW_CONTROL + bool backed_up = false; +#endif PRTE_HIDE_UNUSED_PARAMS(sd, args); /* a client that pushed no data at all leaves us no byte object - PMIx @@ -1125,11 +1128,36 @@ static void pmix_server_stdin_push(int sd, short args, void *cbdata) PRTE_NAME_PRINT(PRTE_PROC_MY_NAME), PRTE_NAME_PRINT(&cd->procs[n]), nbytes)); +#if PRTE_PMIX_IOF_FLOW_CONTROL + if (PRTE_ERR_OUT_OF_RESOURCE == prte_iof.push_stdin(&cd->procs[n], bytes, nbytes)) { + /* a sink here has passed PRTE_IOF_MAX_INPUT_BUFFERS. The data + * was still queued - the module refuses nothing - so this is + * "taken, now slow down", not a failure. One backed-up target + * is enough to hold the whole push: we have no way to tell the + * producer to keep feeding the other targets and not this one, + * and the alternative is to keep taking data for a sink that is + * already unbounded. Keep pushing to the remaining targets + * first, so no target is starved by the order of this loop */ + backed_up = true; + } +#else prte_iof.push_stdin(&cd->procs[n], bytes, nbytes); +#endif } if (NULL == bytes || 0 == nbytes) { cd->cbfunc(PMIX_ERR_IOF_COMPLETE, cd->cbdata); +#if PRTE_PMIX_IOF_FLOW_CONTROL + } else if (backed_up) { + /* PMIx reads this as "I have the data, suspend the stream" and + * stops the producer at its source. It is not reported to the + * tool as an error, and nothing has been dropped. The XON that + * releases it comes from the hnp module's write handler when the + * sink drains - see release_flow_control() there. Against an older + * PMIx there is no such status, so we report success exactly as we + * always did and the sink simply queues */ + cd->cbfunc(PMIX_ERR_IOF_XOFF, cd->cbdata); +#endif } else { cd->cbfunc(PMIX_SUCCESS, cd->cbdata); }