Switch the JSON byte producer from a pull to a push producer. (#8116)

This commit is contained in:
Patrick Cloke 2020-08-19 08:07:57 -04:00 committed by GitHub
parent cfeb37f039
commit f594e434c3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 52 additions and 45 deletions

1
changelog.d/8116.feature Normal file
View File

@ -0,0 +1 @@
Iteratively encode JSON to avoid blocking the reactor.

View File

@ -500,7 +500,7 @@ class RootOptionsRedirectResource(OptionsResource, RootRedirect):
pass pass
@implementer(interfaces.IPullProducer) @implementer(interfaces.IPushProducer)
class _ByteProducer: class _ByteProducer:
""" """
Iteratively write bytes to the request. Iteratively write bytes to the request.
@ -515,29 +515,40 @@ class _ByteProducer:
): ):
self._request = request self._request = request
self._iterator = iterator self._iterator = iterator
self._paused = False
def start(self) -> None: # Register the producer and start producing data.
self._request.registerProducer(self, False) self._request.registerProducer(self, True)
self.resumeProducing()
def _send_data(self, data: List[bytes]) -> None: def _send_data(self, data: List[bytes]) -> None:
""" """
Send a list of strings as a response to the request. Send a list of bytes as a chunk of a response.
""" """
if not data: if not data:
return return
self._request.write(b"".join(data)) self._request.write(b"".join(data))
def pauseProducing(self) -> None:
self._paused = True
def resumeProducing(self) -> None: def resumeProducing(self) -> None:
# We've stopped producing in the meantime (note that this might be # We've stopped producing in the meantime (note that this might be
# re-entrant after calling write). # re-entrant after calling write).
if not self._request: if not self._request:
return return
self._paused = False
# Write until there's backpressure telling us to stop.
while not self._paused:
# Get the next chunk and write it to the request. # Get the next chunk and write it to the request.
# #
# The output of the JSON encoder is coalesced until min_chunk_size is # The output of the JSON encoder is buffered and coalesced until
# reached. (This is because JSON encoders produce a very small output # min_chunk_size is reached. This is because JSON encoders produce
# per iteration.) # very small output per iteration and the Request object converts
# each call to write() to a separate chunk. Without this there would
# be an explosion in bytes written (e.g. b"{" becoming "1\r\n{\r\n").
# #
# Note that buffer stores a list of bytes (instead of appending to # Note that buffer stores a list of bytes (instead of appending to
# bytes) to hopefully avoid many allocations. # bytes) to hopefully avoid many allocations.
@ -561,6 +572,7 @@ class _ByteProducer:
self._send_data(buffer) self._send_data(buffer)
def stopProducing(self) -> None: def stopProducing(self) -> None:
# Clear a circular reference.
self._request = None self._request = None
@ -620,8 +632,7 @@ def respond_with_json(
if send_cors: if send_cors:
set_cors_headers(request) set_cors_headers(request)
producer = _ByteProducer(request, encoder(json_object)) _ByteProducer(request, encoder(json_object))
producer.start()
return NOT_DONE_YET return NOT_DONE_YET

View File

@ -62,8 +62,7 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
"identifier": {"type": "m.id.user", "user": "kermit" + str(i)}, "identifier": {"type": "m.id.user", "user": "kermit" + str(i)},
"password": "monkey", "password": "monkey",
} }
request_data = json.dumps(params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
request, channel = self.make_request(b"POST", LOGIN_URL, request_data)
self.render(request) self.render(request)
if i == 5: if i == 5:
@ -76,14 +75,13 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
# than 1min. # than 1min.
self.assertTrue(retry_after_ms < 6000) self.assertTrue(retry_after_ms < 6000)
self.reactor.advance(retry_after_ms / 1000.0) self.reactor.advance(retry_after_ms / 1000.0 + 1.0)
params = { params = {
"type": "m.login.password", "type": "m.login.password",
"identifier": {"type": "m.id.user", "user": "kermit" + str(i)}, "identifier": {"type": "m.id.user", "user": "kermit" + str(i)},
"password": "monkey", "password": "monkey",
} }
request_data = json.dumps(params)
request, channel = self.make_request(b"POST", LOGIN_URL, params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
self.render(request) self.render(request)
@ -111,8 +109,7 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
"identifier": {"type": "m.id.user", "user": "kermit"}, "identifier": {"type": "m.id.user", "user": "kermit"},
"password": "monkey", "password": "monkey",
} }
request_data = json.dumps(params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
request, channel = self.make_request(b"POST", LOGIN_URL, request_data)
self.render(request) self.render(request)
if i == 5: if i == 5:
@ -132,7 +129,6 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
"identifier": {"type": "m.id.user", "user": "kermit"}, "identifier": {"type": "m.id.user", "user": "kermit"},
"password": "monkey", "password": "monkey",
} }
request_data = json.dumps(params)
request, channel = self.make_request(b"POST", LOGIN_URL, params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
self.render(request) self.render(request)
@ -160,8 +156,7 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
"identifier": {"type": "m.id.user", "user": "kermit"}, "identifier": {"type": "m.id.user", "user": "kermit"},
"password": "notamonkey", "password": "notamonkey",
} }
request_data = json.dumps(params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
request, channel = self.make_request(b"POST", LOGIN_URL, request_data)
self.render(request) self.render(request)
if i == 5: if i == 5:
@ -174,14 +169,13 @@ class LoginRestServletTestCase(unittest.HomeserverTestCase):
# than 1min. # than 1min.
self.assertTrue(retry_after_ms < 6000) self.assertTrue(retry_after_ms < 6000)
self.reactor.advance(retry_after_ms / 1000.0) self.reactor.advance(retry_after_ms / 1000.0 + 1.0)
params = { params = {
"type": "m.login.password", "type": "m.login.password",
"identifier": {"type": "m.id.user", "user": "kermit"}, "identifier": {"type": "m.id.user", "user": "kermit"},
"password": "notamonkey", "password": "notamonkey",
} }
request_data = json.dumps(params)
request, channel = self.make_request(b"POST", LOGIN_URL, params) request, channel = self.make_request(b"POST", LOGIN_URL, params)
self.render(request) self.render(request)

View File

@ -160,7 +160,7 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase):
else: else:
self.assertEquals(channel.result["code"], b"200", channel.result) self.assertEquals(channel.result["code"], b"200", channel.result)
self.reactor.advance(retry_after_ms / 1000.0) self.reactor.advance(retry_after_ms / 1000.0 + 1.0)
request, channel = self.make_request(b"POST", self.url + b"?kind=guest", b"{}") request, channel = self.make_request(b"POST", self.url + b"?kind=guest", b"{}")
self.render(request) self.render(request)
@ -186,7 +186,7 @@ class RegisterRestServletTestCase(unittest.HomeserverTestCase):
else: else:
self.assertEquals(channel.result["code"], b"200", channel.result) self.assertEquals(channel.result["code"], b"200", channel.result)
self.reactor.advance(retry_after_ms / 1000.0) self.reactor.advance(retry_after_ms / 1000.0 + 1.0)
request, channel = self.make_request(b"POST", self.url + b"?kind=guest", b"{}") request, channel = self.make_request(b"POST", self.url + b"?kind=guest", b"{}")
self.render(request) self.render(request)

View File

@ -353,6 +353,7 @@ class CleanupExtremDummyEventsTestCase(HomeserverTestCase):
self.event_creator_handler._rooms_to_exclude_from_dummy_event_insertion[ self.event_creator_handler._rooms_to_exclude_from_dummy_event_insertion[
"3" "3"
] = 300000 ] = 300000
self.event_creator_handler._expire_rooms_to_exclude_from_dummy_event_insertion() self.event_creator_handler._expire_rooms_to_exclude_from_dummy_event_insertion()
# All entries within time frame # All entries within time frame
self.assertEqual( self.assertEqual(
@ -362,7 +363,7 @@ class CleanupExtremDummyEventsTestCase(HomeserverTestCase):
3, 3,
) )
# Oldest room to expire # Oldest room to expire
self.pump(1) self.pump(1.01)
self.event_creator_handler._expire_rooms_to_exclude_from_dummy_event_insertion() self.event_creator_handler._expire_rooms_to_exclude_from_dummy_event_insertion()
self.assertEqual( self.assertEqual(
len( len(