Changeset 6dd2b2d in trunk

Timestamp:

2022-06-27T21:00:24Z (3 years ago)

Author:

Itamar Turner-Trauring <itamar@…>

Branches:

master

Children:

75f3302

Parents:

06eca79

git-author:

Itamar Turner-Trauring <itamar@…> (2022-06-27 18:44:51)

git-committer:

Itamar Turner-Trauring <itamar@…> (2022-06-27 21:00:24)

Message:

More streaming, with tests passing again.

Location:

src/allmydata

Files:

• storage/http_server.py (modified) (5 diffs)
• test/test_storage_http.py (modified) (16 diffs)

src/allmydata/storage/http_server.py ¶

@@ -282 +282 @@
 @implementer(IPullProducer)
 @define
-class _ReadProducer:
-    """
-    Producer that calls a read function, and writes to a request.
+class _ReadAllProducer:
+    """
+    Producer that calls a read function repeatedly to read all the data, and
+    writes to a request.
     """
 
@@ -293 +294 @@
 
     def resumeProducing(self):
-        data = self.read_data(self.start, self.start + 65536)
+        data = self.read_data(self.start, 65536)
         if not data:
             self.request.unregisterProducer()
@@ -310 +311 @@
 
 
+@implementer(IPullProducer)
+@define
+class _ReadRangeProducer:
+    """
+    Producer that calls a read function to read a range of data, and writes to
+    a request.
+    """
+
+    request: Request
+    read_data: Callable[[int, int], bytes]
+    result: Deferred
+    start: int
+    remaining: int
+    first_read: bool = field(default=True)
+
+    def resumeProducing(self):
+        to_read = min(self.remaining, 65536)
+        data = self.read_data(self.start, to_read)
+        assert len(data) <= to_read
+        if self.first_read and data:
+            # For empty bodies the content-range header makes no sense since
+            # the end of the range is inclusive.
+            self.request.setHeader(
+                "content-range",
+                ContentRange("bytes", self.start, self.start + len(data)).to_header(),
+            )
+        self.request.write(data)
+
+        if not data or len(data) < to_read:
+            self.request.unregisterProducer()
+            d = self.result
+            del self.result
+            d.callback(b"")
+            return
+
+        self.start += len(data)
+        self.remaining -= len(data)
+        assert self.remaining >= 0
+
+    def pauseProducing(self):
+        pass
+
+    def stopProducing(self):
+        pass
+
+
 def read_range(request: Request, read_data: Callable[[int, int], bytes]) -> None:
     """
@@ -325 +372 @@
     The resulting data is written to the request.
     """
+
+    def read_data_with_error_handling(offset: int, length: int) -> bytes:
+        try:
+            return read_data(offset, length)
+        except _HTTPError as e:
+            request.setResponseCode(e.code)
+            # Empty read means we're done.
+            return b""
+
     if request.getHeader("range") is None:
         d = Deferred()
-        request.registerProducer(_ReadProducer(request, read_data, d), False)
+        request.registerProducer(
+            _ReadAllProducer(request, read_data_with_error_handling, d), False
+        )
         return d
 
@@ -340 +398 @@
 
     offset, end = range_header.ranges[0]
-
-    # TODO limit memory usage
-    # https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3872
-    data = read_data(offset, end - offset)
-
     request.setResponseCode(http.PARTIAL_CONTENT)
-    if len(data):
-        # For empty bodies the content-range header makes no sense since
-        # the end of the range is inclusive.
-        request.setHeader(
-            "content-range",
-            ContentRange("bytes", offset, offset + len(data)).to_header(),
-        )
-    request.write(data)
-    request.finish()
+    d = Deferred()
+    request.registerProducer(
+        _ReadRangeProducer(
+            request, read_data_with_error_handling, d, offset, end - offset
+        ),
+        False,
+    )
+    return d
 
 

src/allmydata/test/test_storage_http.py ¶

@@ -11 +11 @@
 from pycddl import ValidationError as CDDLValidationError
 from hypothesis import assume, given, strategies as st
-from fixtures import Fixture, TempDir
+from fixtures import Fixture, TempDir, MockPatch
 from treq.testing import StubTreq
 from klein import Klein
@@ -315 +315 @@
         self.clock = Clock()
         self.tempdir = self.useFixture(TempDir())
+        self.mock = self.useFixture(
+            MockPatch(
+                "twisted.internet.task._theCooperator",
+                Cooperator(scheduler=lambda c: self.clock.callLater(0.000001, c)),
+            )
+        )
         self.storage_server = StorageServer(
             self.tempdir.path, b"\x00" * 20, clock=self.clock
@@ -325 +331 @@
             treq=self.treq,
         )
+
+    def result_of_with_flush(self, d):
+        for i in range(100):
+            self.clock.advance(0.001)
+        self.treq.flush()
+        return result_of(d)
 
 
@@ -549 +561 @@
         # We can now read:
         for offset, length in [(0, 100), (10, 19), (99, 1), (49, 200)]:
-            downloaded = result_of(
+            downloaded = self.http.result_of_with_flush(
                 self.imm_client.read_share_chunk(storage_index, 1, offset, length)
             )
@@ -624 +636 @@
         # The upload of share 1 succeeded, demonstrating that second create()
         # call didn't overwrite work-in-progress.
-        downloaded = result_of(
+        downloaded = self.http.result_of_with_flush(
             self.imm_client.read_share_chunk(storage_index, 1, 0, 100)
         )
@@ -754 +766 @@
         )
         self.assertEqual(
-            result_of(self.imm_client.read_share_chunk(storage_index, 1, 0, 10)),
+            self.http.result_of_with_flush(
+                self.imm_client.read_share_chunk(storage_index, 1, 0, 10)
+            ),
             b"1" * 10,
         )
         self.assertEqual(
-            result_of(self.imm_client.read_share_chunk(storage_index, 2, 0, 10)),
+            self.http.result_of_with_flush(
+                self.imm_client.read_share_chunk(storage_index, 2, 0, 10)
+            ),
             b"2" * 10,
         )
@@ -922 +938 @@
         self.assertEqual(
             uploaded_data,
-            result_of(
+            self.http.result_of_with_flush(
                 self.imm_client.read_share_chunk(
                     storage_index,
@@ -987 +1003 @@
         """
         storage_index, _, _ = self.create_upload()
-        data0 = result_of(self.mut_client.read_share_chunk(storage_index, 0, 1, 7))
-        data1 = result_of(self.mut_client.read_share_chunk(storage_index, 1, 0, 8))
+        data0 = self.http.result_of_with_flush(
+            self.mut_client.read_share_chunk(storage_index, 0, 1, 7)
+        )
+        data1 = self.http.result_of_with_flush(
+            self.mut_client.read_share_chunk(storage_index, 1, 0, 8)
+        )
         self.assertEqual((data0, data1), (b"bcdef-0", b"abcdef-1"))
 
@@ -1016 +1036 @@
         )
         # But the write did happen:
-        data0 = result_of(self.mut_client.read_share_chunk(storage_index, 0, 0, 8))
-        data1 = result_of(self.mut_client.read_share_chunk(storage_index, 1, 0, 8))
+        data0 = self.http.result_of_with_flush(
+            self.mut_client.read_share_chunk(storage_index, 0, 0, 8)
+        )
+        data1 = self.http.result_of_with_flush(
+            self.mut_client.read_share_chunk(storage_index, 1, 0, 8)
+        )
         self.assertEqual((data0, data1), (b"aXYZef-0", b"abcdef-1"))
 
@@ -1058 +1082 @@
         self.assertTrue(result.success)
         self.assertEqual(
-            result_of(self.mut_client.read_share_chunk(storage_index, 0, 0, 8)),
+            self.http.result_of_with_flush(
+                self.mut_client.read_share_chunk(storage_index, 0, 0, 8)
+            ),
             b"aXYZef-0",
         )
@@ -1095 +1121 @@
         # The write did not happen:
         self.assertEqual(
-            result_of(self.mut_client.read_share_chunk(storage_index, 0, 0, 8)),
+            self.http.result_of_with_flush(
+                self.mut_client.read_share_chunk(storage_index, 0, 0, 8)
+            ),
             b"abcdef-0",
         )
@@ -1195 +1223 @@
         """
         with assert_fails_with_http_code(self, http.NOT_FOUND):
-            result_of(
+            self.http.result_of_with_flush(
                 self.client.read_share_chunk(
                     b"1" * 16,
@@ -1210 +1238 @@
         storage_index, _, _ = self.upload(1)
         with assert_fails_with_http_code(self, http.NOT_FOUND):
-            result_of(
+            self.http.result_of_with_flush(
                 self.client.read_share_chunk(
                     storage_index,
@@ -1236 +1264 @@
                 self, http.REQUESTED_RANGE_NOT_SATISFIABLE
             ):
-                result_of(
+                self.http.result_of_with_flush(
                     client.read_share_chunk(
                         storage_index,
@@ -1265 +1293 @@
         A read with no range returns the whole mutable/immutable.
         """
-        self.patch(
-            task,
-            "_theCooperator",
-            Cooperator(scheduler=lambda c: self.http.clock.callLater(0.000001, c)),
-        )
-
-        def result_of_with_flush(d):
-            for i in range(100):
-                self.http.clock.advance(0.001)
-            self.http.treq.flush()
-            return result_of(d)
-
         storage_index, uploaded_data, _ = self.upload(1, data_length)
-        response = result_of_with_flush(
+        response = self.http.result_of_with_flush(
             self.http.client.request(
                 "GET",
@@ -1299 +1315 @@
             headers = Headers()
             headers.setRawHeaders("range", [requested_range])
-            response = result_of(
+            response = self.http.result_of_with_flush(
                 self.http.client.request(
                     "GET",