Check lent views: whole versions, and every loan given back

Views of a 1 MiB note are pipelined on one connection, more than the
kernel takes, while the note is replaced: each answer is one whole
version, the store mapping the waiting answers hold stays (through a
compaction too) and is unmapped once they are read, or once the reader
leaves without reading, and nothing stays charged. On one thread and
with an owner.

1 files changed, 89 insertions, 0 deletions

diff --git a/tests/check.py b/tests/check.py
index e7b7c8b..c7833df 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -943,6 +943,94 @@ def server_lent_budget(aup, pure, work):
     return budget
 
 
+NOTE_SOURCE = """state note: text
+input write = message {
+  text: text
+}
+on write { note = write.text }
+view page = note
+view size = note.bytes
+"""
+
+
+def server_lent_views(aup, pure, work, threads=1):
+    """A view that shows a text state as it is lends the state's stored
+    extents. Answers pipelined on one connection and left waiting hold the
+    store mapping they were read from while the note is replaced: each comes
+    out whole in one version. When a reader goes without reading, the
+    mappings its answers held are unmapped and nothing stays charged. With
+    several threads the owner lends and a connection thread gives back."""
+    import socket
+    import time
+    module, source = work / 'note.pmod', work / 'note.aup'
+    source.write_text(NOTE_SOURCE)
+    compile_module(aup, source, module)
+    store = work / f'lent-views-{threads}'
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads])
+
+    def write(text):
+        body = json.dumps({'text': text}).encode()
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
+        connection.request('POST', '/inputs/write', body, {'Content-Type': 'application/json'})
+        assert connection.getresponse().status == 200
+        connection.close()
+
+    def mappings():
+        # A compaction replaces the file: a mapping of the old one shows as "(deleted)".
+        path = str((store / 'pure.tree').resolve())
+        return sum(1 for line in open(f'/proc/{host.process.pid}/maps') if path in line)
+
+    def settle(to):
+        for _ in range(100):  # an answer's ticket is released on its thread
+            if mappings() <= to: return mappings()
+            time.sleep(0.02)
+        return mappings()
+
+    def pipeline(count):
+        # Far more than the kernel's buffers take (a few MiB): an answer waits, lent.
+        sock = socket.create_connection(('127.0.0.1', host.port), timeout=30)
+        sock.sendall(b'GET /views/page HTTP/1.1\r\nHost: x\r\n\r\n' * count)
+        time.sleep(0.3)
+        return sock
+
+    size = 1 << 20
+    try:
+        write('x' * size)
+        assert http(host.port, 'GET', '/views/size')[3] == str(size).encode()
+        alone = mappings()
+        # Whole versions, however the note changes while answers wait.
+        sock = pipeline(32)
+        for round_ in range(5):
+            write(chr(ord('A') + round_) * size)
+        held = mappings()
+        data = b''
+        while data.count(b'HTTP/1.1 200') < 32 or len(data) < 32 * size:
+            chunk = sock.recv(1 << 20)
+            assert chunk, len(data)
+            data += chunk
+        sock.close()
+        answers = [part.split(b'\r\n\r\n', 1) for part in data.split(b'HTTP/1.1 200 OK\r\n')[1:]]
+        bodies = [body[:size] for _, body in answers]
+        assert all(b'text/plain' in head for head, _ in answers) and bodies[0] == b'x' * size, [b[:1] for b in bodies]
+        assert all(len(set(body)) == 1 and len(body) == size for body in bodies), [len(b) for b in bodies]
+        after_reading = settle(alone)
+        # A reader that goes without reading gives back what its answers held.
+        sock = pipeline(32)
+        for round_ in range(3):
+            write(chr(ord('a') + round_) * size)
+        held_again = mappings()
+        sock.close()
+        after_leaving = settle(alone)
+        assert held > alone and held_again > alone and after_reading <= alone and after_leaving <= alone, \
+            (alone, held, after_reading, held_again, after_leaving)
+        assert http(host.port, 'GET', '/views/page')[3] == b'c' * size
+    finally:
+        stats = host.stop()
+    assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats
+    return {'alone': alone, 'while_lent': max(held, held_again), 'after': after_leaving}
+
+
 def server_bursts(aup, pure, work, clients=8, edits=30, shards=1):
     """Concurrent edits share commits, and every acknowledged edit survives a
     kill -9 of the server."""
@@ -1831,6 +1919,7 @@ def main():
         report['server_pipelined_answers'] = server_pipelining(aup, args.pure.resolve(), work)
         report['server_lent_read_bytes'] = server_lent_reads(aup, args.pure.resolve(), work)
         report['server_lent_budget_bytes'] = server_lent_budget(aup, args.pure.resolve(), work)
+        report['server_lent_views'] = [server_lent_views(aup, args.pure.resolve(), work, threads) for threads in (1, 3)]
         report['server_bursts'] = server_bursts(aup, args.pure.resolve(), work)
         report['server_bursts_three_threads'] = server_bursts(aup, args.pure.resolve(), work, shards=3)
         report['server_app_requests'], report['server_app_acknowledged_before_kill'] = \