Inside Bitcoin

751        dl_start_size = self.node.debug_log_size(encoding="utf-8")
752
753        # Non-blocking send: When the server stops reading from the buffer,
754        # Python will raise an error. That stop could be due to actual
755        # backpressure from the server or any other throttle in the TCP chain.
756        # If the socket was set to blocking, we would have to wait for a timeout.
757        conn.conn.sock.setblocking(False)
758
759        # Use a background thread to push flood data until either an absurd
760        # limit has been reached (server is not throttling) or the foreground
761        # thread has determined that the server is throttling correctly.
762        # We can't rely on TCP backpressure at the client socket to assert
763        # server behavior because different platforms (macos in particular)
764        # like to open and close the TCP window size based on heuristics
765        # out of our control, and delay transmission when windows get too small.
766        flood_stop = threading.Event()
767
768        def send_flood(self, conn):
769            sent_total = 0
770            blocked = False
771            while not flood_stop.is_set():
772                try:
773                    sent = conn.conn.sock.send(flood)
774                    sent_total += sent
775                    blocked = False
776                    self.log.debug(f"Client sent {sent} bytes (total {sent_total})")
777                except BlockingIOError:
778                    if not blocked:
779                        blocked = True
780                        self.log.debug(f"Client socket blocked after sending {sent_total} bytes")
781                if sent_total > len(flood) * 10:
782                    raise Exception(f"Client sent too much data: {sent_total} bytes")
783                flood_stop.wait(0.05)
784
785        executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
786        flood_thread = executor.submit(
787            send_flood,
788            self,
789            conn)
790
791        # Now watch the server-side read counter for this connection. While a
792        # request is in flight the throttled server should stop reading.
793        # Expect a 5-second stall sometime within the timeout factor. That gives
794        # the flood thread plenty of time to trigger the throttle.
795        prev_total = -1
796        total = 0
797        buffered = 0
798
799        def progress_stalled():
800            nonlocal buffered, total, prev_total
801
802            # Check the background thread for assertion errors
803            if flood_thread.done():
804                flood_thread.result()
805
806            with open(self.node.debug_log_path, encoding="utf-8", errors="replace") as dl:
807                # Skip everything written before the flood started.
808                dl.seek(dl_start_size)
809                log = dl.read()
810                matches = re.findall(
811                    rf"Received \d+ bytes from [^\s]*:{local_port} \(id=\d+\): total=(\d+) buffered=(\d+)",
812                    log)
813                total = max((int(m[0]) for m in matches), default=total)
814                buffered = max((int(m[1]) for m in matches), default=buffered)
815                if total == prev_total and total > 0:
816                    return True
817                prev_total = total
818                self.log.debug(f"Server read {total} bytes, {buffered} buffered")
819                return False
820
821        try:
822            self.wait_until(progress_stalled, check_interval=5)
823        finally:
824            # Stop the flooding thread.
825            # Wait for it without a timeout so it can't turn an
826            # already-passed verdict into a spurious TimeoutError. If the flood
827            # thread hit its own assertion (server never throttled), result()
828            # re-raises it here.
829            flood_stop.set()
830            flood_thread.result()
831            executor.shutdown(wait=True)
832
833        self.log.info(
834            f"Pipelined flood: server received {total} bytes, buffered {buffered} bytes "
835            "before the read throttle engaged.")
836
837        # Unblock the client request queue.
838        conn.conn.sock.settimeout(10)
839        generated_block = self.generate(self.node, 1, sync_fun=self.no_op)[0]
840        # First reply is for the blocking request.
841        response = conn.recv_raw().decode()
842        assert generated_block in response
843
844
845    def check_slow_read_throttle(self):
846        self.log.info("Check that request processing is throttled if the client is not draining the socket")
847        self.restart_node(0, extra_args=["-rest=1"])
848        # Generate a big block
849        self.wallet = MiniWallet(self.node)
850        mine_large_block(self, self.wallet, self.node)
851        big_block_hash = self.node.getbestblockhash()
852
853        conn = BitcoinHTTPConnection(self.node)
854
855        # Request the big block JSON once to check its size and establish the connection
856        URI = f"/rest/block/{big_block_hash}.json"
857        response_body_size = len(conn.get(URI).read())
858        assert response_body_size > 7 * 1024 * 1024, f"Big block JSON response size is {response_body_size} bytes"
859
860        # Prepare a batch of requests. We want to fill up at
861        # least one server-side read operation (about 65kB, see HTTPRemoteClient::Receive()).
862        single_req = f"GET {URI} HTTP/1.1\r\nHost: somehost\r\n\r\n"
863        num_req = 0x10000 // len(single_req)
864        batch = single_req * num_req
865        self.log.info(f"Sending {num_req} big block JSON requests")
866
867        # Save a debug log checkpoint
868        dl_start_size = self.node.debug_log_size(encoding="utf-8")
869
870        # Send the batch but do not read any of the responses, leaving
871        # that data on the server-side of the socket.
872        conn.send_raw(batch.encode("ascii"))
873
874        # Open the debug log and count how many of the batch requests were processed.
875        # Expect progress to stall after a few seconds.
876        count = 0
877        prev_count = -1
878        min_count = MAX_BODY_SIZE // response_body_size
879
880        def progress_stalled():
881            nonlocal count, prev_count
882            with open(self.node.debug_log_path, encoding="utf-8", errors="replace") as dl:
883                dl.seek(dl_start_size)
884                count = dl.read().count(f"Received a GET request for {URI} ")
885                if count == prev_count and count > min_count:
886                    self.log.info(f"Response progress stalled after {count} requests were handled.")
887                    assert count < num_req, f"Server handled the whole batch of {num_req}: nothing was throttled"
888                    return True
889                prev_count = count
890                return False
891
892        # Large enough interval, to ensure throttling occurred, rather than
893        # merely slow JSON serialization.
894        self.wait_until(progress_stalled, check_interval=5)
895
896        # Drain the responses that were handled up to the stall point,
897        # plus a few more to confirm that pulling out the cork restores the flow.
898        conn.conn.sock.settimeout(10)
899        num_res = 0
900        while num_res < count + 2:
901            response = conn.conn.sock.recv(10 * 1024 * 1024)
902            num_res += response.count(b"HTTP/1.1 200")
903        self.log.info(f"Drained {num_res} responses")
904
905if __name__ == '__main__':
906    HTTPBasicsTest(__file__).main()
907