diff options
| -rw-r--r-- | docs/research/122-views-off-the-connection-threads.md | 3 | ||||
| -rw-r--r-- | tests/check.py | 50 |
2 files changed, 44 insertions, 9 deletions
diff --git a/docs/research/122-views-off-the-connection-threads.md b/docs/research/122-views-off-the-connection-threads.md index 5af9c09..63a1d8f 100644 --- a/docs/research/122-views-off-the-connection-threads.md +++ b/docs/research/122-views-off-the-connection-threads.md @@ -67,6 +67,9 @@ tail 1`, `tail 2`), 같은 것을 `--view-workers 0`으로 두 번(`control`). � 모듈로 바꾸고(SIGHUP), 이어서 앱을 뺀다. 모든 답이 옛 모듈의 값, 새 모듈의 값, 503 `stopped`, 404 가운데 하나다(섞인 답은 `7`로 보였을 것이다). 그 보기의 스트림이 같은 스트림에서 새 모듈의 값을 받고, 앱이 빠지면 끝난다. 값이 오기 전에 닫은 스트림 열 개가 아무것도 남기지 않는다(멈출 때 구독자 0). `--view-workers 0`이면 보기의 스레드가 없고 같은 답이다. +- 같은 검사에서, 스트림의 보기를 계산하는 동안 쓰기 셋이 오면 그 뒤에 마지막 값이 온다(`id`는 늘기만 한다). 그 길(무리의 `again`)을 + 끄면 스트림이 마지막 값을 받지 못해 검사가 멈춘다. 앱에 `queue=2`, 보기의 스레드 하나에 8연결이 몰리면 503 `busy`가 나오고(36번), + 그 뒤의 `GET`은 200이다(넘긴 일의 자리가 돌아온다). - `server_view_bodies`(64연결이 동시에 묻는 `distance` 1,280개가 모두 모델과 같다)와 `server_push`(계산하는 보기를 미는 것 포함)가 이제 보기의 스레드를 거쳐 그대로 초록이다. - Pure의 `make check`, aup와 Vega의 모든 검사가 초록이다. diff --git a/tests/check.py b/tests/check.py index 7d5648b..d62bb91 100644 --- a/tests/check.py +++ b/tests/check.py @@ -4513,9 +4513,11 @@ def server_view_workers(aup, pure, work): while 8 connections ask a view that loops 3,000,000 times, its app's module changes (SIGHUP) to one with another view in its place and the view tenfold, then the app is removed: every answer is the old module's, - the new one's, 503 stopped or 404 (gone), none mixed. Streams of it - opened and closed at once leave nothing behind. With --view-workers 0 - the connection threads show it, as before.""" + the new one's, 503 stopped or 404 (gone), none mixed. A stream of it + gets the last value of writes made while it is evaluated (ids only + growing); streams opened and closed at once leave nothing behind. Past + the app's queue=2 asks are 503 busy, and the places come back. With + --view-workers 0 the connection threads show it, as before.""" import select import signal import threading @@ -4547,7 +4549,15 @@ def server_view_workers(aup, pure, work): assert stream.head()[0] == 200 stream.close() stream = EventStream(host.port, '/s/views/slow') - assert stream.next()[2] == '3000001' + first, _, value = stream.next() + assert value == '3000001' + for _ in range(3): # written while it is evaluated: the newest snapshot's value after it + assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200 + ids = [first] + while value != '3000004': + ident, _, value = stream.next(timeout=10) + ids.append(ident) + assert ids == sorted(set(ids)), ids def ask(): import http.client @@ -4563,7 +4573,7 @@ def server_view_workers(aup, pure, work): c = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30) continue answers[got] = answers.get(got, 0) + 1 - if got not in ((200, 'none', b'3000001'), (200, 'none', b'3000010'), (503, 'stopped', b'')) and got[0] != 404: + if got not in ((200, 'none', b'3000004'), (200, 'none', b'3000040'), (503, 'stopped', b'')) and got[0] != 404: wrong.append(got) c.close() @@ -4573,7 +4583,7 @@ def server_view_workers(aup, pure, work): time.sleep(0.5) write(tenfold) assert reading()['apps'][0]['did'] == 'changed' - assert stream.next(timeout=10)[2] == '3000010' # the stream, on the new module's view + assert stream.next(timeout=10)[2] == '3000040' # the stream, on the new module's view time.sleep(0.5) write(tenfold, 'q') # s removed reading() @@ -4586,7 +4596,7 @@ def server_view_workers(aup, pure, work): finally: stop.set() stats = host.stop() - old, new = answers.get((200, 'none', b'3000001'), 0), answers.get((200, 'none', b'3000010'), 0) + old, new = answers.get((200, 'none', b'3000004'), 0), answers.get((200, 'none', b'3000040'), 0) assert not wrong and old and new, (wrong[:5], answers) assert stats[0]['view_workers'] == 2 and stats[0]['views_by_workers'] >= old + new and stats[0]['subscribers'] == 0, stats[0] by_workers = stats[0]['views_by_workers'] @@ -4594,12 +4604,34 @@ def server_view_workers(aup, pure, work): host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--view-workers', 0]) try: assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200 - assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000002' # its store kept the first bump + assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000005' # its store kept the bumps finally: stats = host.stop() assert stats[0]['view_workers'] == 0 and stats[0]['views_by_workers'] == 0, stats[0] + conf.write_text(f"s {module} {work / 'slow-s'} queue=2\n") + host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--threads', 1, '--view-workers', 1]) + busy, other = [0], [] + try: + def crowd(): + for _ in range(6): + got = http(host.port, 'GET', '/s/views/slow') + if got[0] == 503 and got[1] == 'busy': + busy[0] += 1 + elif got[0] != 200 or got[3] != b'3000005': + other.append((got[0], got[1], got[3][:20])) + + threads = [threading.Thread(target=crowd) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + assert busy[0] and not other, (busy, other[:5]) + assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000005' # the queue's places came back + finally: + host.stop() return {'old': old, 'new': new, 'stopped': answers.get((503, 'stopped', b''), 0), - 'gone': sum(v for k, v in answers.items() if k[0] == 404), 'by_workers': by_workers} + 'gone': sum(v for k, v in answers.items() if k[0] == 404), 'by_workers': by_workers, 'pushed_ids': len(ids), + 'busy': busy[0]} PUSH_SOURCE = """record Row { |