From 0ad46603d88eccaefd250870109186f823e1845a Mon Sep 17 00:00:00 2001 From: eunhhu Date: Thu, 1 Oct 2026 07:03:14 +0100 Subject: Hand sends to another app of the host with no socket (017) send=OUTPUT=APP/INPUT in --apps wires an output to another app of the same host through a local outlet: no socket, seal or packet. The sender's turn queues its saved sends there, under the window links and wires have, and hands them, one batch at a time, to the receiving app's queue laid out as a link delivers records (first number, count, source APP.OUTPUT, the records as stored, and the sender's fields when the input takes them by name, which pure_module_takes checks at start). The receiver applies them after its last number heard; its save over, it tells the outlet what it heard and the sender's next turn trims its outbox so far. What it did not store is queued again from the outbox, after 100 ms when it had no room. An output now goes one way only: two outlets would trim each other's sends. server_apps_links: an app sends each of 400 messages to an app of another host over a link and to three apps of its own host (one of the same shape, one taking the fields by name, one with a memory budget that fills), with a kill -9 of the sending host between two requests; all arrive once and in order, and the full app's sends wait in the sender's outbox until it lets go of what it holds. --- tests/check.py | 84 ++++++++++++++++++++++++++++++++++++++++++++++------------ 1 file changed, 67 insertions(+), 17 deletions(-) (limited to 'tests/check.py') diff --git a/tests/check.py b/tests/check.py index f75204a..bb7ec25 100644 --- a/tests/check.py +++ b/tests/check.py @@ -3851,9 +3851,15 @@ twin {work / 'apps.pmod'} {work / 'store-a'} LINKS_SOURCE = """state n = 0 input put = message { k: u32 } output out = message { k: u32 } +output here = message { k: u32 } +output named = message { k: u32 } +output small = message { k: u32 } on put { n = n + 1 send out { k: put.k } + send here { k: put.k } + send named { k: put.k } + send small { k: put.k } } view count = n """ @@ -3875,17 +3881,34 @@ view count = count(g in got where g.k > 0) view sum = sum(g.k for g in got) view in_order = ordered """ +# The same, its input laid out otherwise (a field before k): it takes the sends by name. +LINKS_NAMED = LINKS_TAKER.replace('input take = message { k: u32 }', 'input take = message {\n note: text\n k: u32\n}') +# The same, and it can let go of what it got (and what it heard stays). +LINKS_SMALL = LINKS_TAKER + """input clear = message { all: u32 } +on clear { + for g in got { + got.remove(g) + } +} +view held = count(g in got where g.k > 0) +""" def server_apps_links(aup, pure, work): - """Apps of a host send (docs/design/017, stage 2): an app's send= wires - its output, its sends' source APP.OUTPUT, sealed with the host's key; an - app of another host takes them on that host's one --link-port, named - APP/INPUT. Every send arrives once and in order, and the stop line counts - the link's packets.""" + """Apps of a host send (docs/design/017, stage 2). An app's send= wires + each output: to an app of another host over a link (its sends' source + APP.OUTPUT, sealed with the host's key, taken on that host's one + --link-port as APP/INPUT), and to apps of its own host with no socket + (send=OUTPUT=APP/INPUT): one taking the same shape, one taking the + fields by name, one whose memory budget fills. Every send arrives once + and in order, across a kill -9 of the sending host between requests; + the full app's sends wait in the sender's outbox until it lets go of + what it holds, and none is lost.""" + import signal import socket import time - for name, text in (('links-src', LINKS_SOURCE), ('links-taker', LINKS_TAKER)): + for name, text in (('links-src', LINKS_SOURCE), ('links-taker', LINKS_TAKER), ('links-named', LINKS_NAMED), + ('links-small', LINKS_SMALL)): (work / f'{name}.aup').write_text(text) compile_module(aup, work / f'{name}.aup', work / f'{name}.pmod') serve = pure / 'build/pure-serve' @@ -3899,27 +3922,54 @@ def server_apps_links(aup, pure, work): keys[side] = host.info['public_key'] host.stop() (work / 'links-far.conf').write_text(f"far {work / 'links-taker.pmod'} {work / 'links-far'}\n") - (work / 'links-near.conf').write_text(f"src {work / 'links-src.pmod'} {work / 'links-src'} " - f"send=out=pure://127.0.0.1:{link_port}/far/take seal=out={keys['receiver']}\n") + (work / 'links-near.conf').write_text( + f"src {work / 'links-src.pmod'} {work / 'links-src'} send=out=pure://127.0.0.1:{link_port}/far/take " + f"seal=out={keys['receiver']} send=here=here/take send=named=byname/take send=small=small/take\n" + f"here {work / 'links-taker.pmod'} {work / 'links-here'}\n" + f"byname {work / 'links-named.pmod'} {work / 'links-byname'}\n" + f"small {work / 'links-small.pmod'} {work / 'links-small'} memory=1200\n") far = HttpHost([serve, '--apps', work / 'links-far.conf', '--key', work / 'links-receiver.key', '--link-port', link_port, '--sender', f"src.out={keys['sender']}"]) - near = HttpHost([serve, '--apps', work / 'links-near.conf', '--key', work / 'links-sender.key']) - sent = 300 + near_args = [serve, '--apps', work / 'links-near.conf', '--key', work / 'links-sender.key', '--threads', 2] + near = HttpHost(near_args) + sent, killed_at = 400, 150 try: for k in range(1, sent + 1): + if k == killed_at: # between requests: every send before it is stored, some not yet handed on + near.process.send_signal(signal.SIGKILL) + near.process.wait() + near = HttpHost(near_args) status, why, _, _ = http(near.port, 'POST', f'/src/inputs/put?k={k}') assert status == 200, (k, status, why) + + def views(port, app): + return [http(port, 'GET', f'/{app}/views/{view}')[3] for view in ('count', 'sum', 'in_order')] + + whole = [str(sent).encode(), str(sent * (sent + 1) // 2).encode(), b'true'] deadline = time.time() + 30 - while time.time() < deadline and http(far.port, 'GET', '/far/views/count')[3] != str(sent).encode(): + while time.time() < deadline and (views(far.port, 'far') != whole or views(near.port, 'here') != whole or + views(near.port, 'byname') != whole): time.sleep(0.05) - assert http(far.port, 'GET', '/far/views/count')[3] == str(sent).encode() - assert http(far.port, 'GET', '/far/views/sum')[3] == str(sent * (sent + 1) // 2).encode() - assert http(far.port, 'GET', '/far/views/in_order')[3] == b'true' + assert views(far.port, 'far') == whole, views(far.port, 'far') + assert views(near.port, 'here') == whole, views(near.port, 'here') + assert views(near.port, 'byname') == whole, views(near.port, 'byname') + # The small app is full: it holds some, the rest wait in src's outbox; it lets go, and they come. + held = int(http(near.port, 'GET', '/small/views/held')[3]) + assert 0 < held < sent, held + taken = held + for _ in range(sent): + assert http(near.port, 'POST', '/small/inputs/clear?all=1')[0] == 200 + time.sleep(0.3) # a held-back batch goes again after 100 ms + now = int(http(near.port, 'GET', '/small/views/held')[3]) + if not now: + break + taken += now + assert taken == sent and http(near.port, 'GET', '/small/views/in_order')[3] == b'true', taken finally: - packets = near.stop()[-1]['link_packets'] + near_stats = near.stop()[-1] far.stop() - assert packets > 0, packets - return {'sent': sent, 'link_packets': packets} + assert near_stats['handed'] > 0 and near_stats['link_packets'] > 0, near_stats + return {'sent': sent, 'killed_at': killed_at, 'handed': near_stats['handed'], 'small_held': held} def main(): -- cgit v1.2.3