Skip to content

Speed up MultipartParser callback dispatch #337

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
John-Cusack wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Speed up MultipartParser callback dispatch #337

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1117af5
Speed up MultipartParser callback dispatch
Oct 14, 2026
e7af72a
Add CHANGELOG entry for #337
Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
## Unreleased

* Speed up querystring callback dispatch [#316](https://github.com/Kludex/python-multipart/pull/316).
* Speed up `MultipartParser` callback dispatch [#337](https://github.com/Kludex/python-multipart/pull/337).

## 0.0.32 (2026-06-04)

Expand Down
77 changes: 55 additions & 22 deletions python_multipart/multipart.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1140,6 +1140,39 @@ def _internal_write(self, data: bytes, length: int) -> int:
current_header_count = self._current_header_count
current_header_size = self._current_header_size

# Hoist callback dispatch out of the per-byte loop (same shape as the
# QuerystringParser optimization in #316): one dict lookup per write

@cubic-dev-ai cubic-dev-ai Bot Oct 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P3: This comment says there is “one dict lookup per write”, but the block performs nine .get() calls per write. Say “one lookup per callback per write” to keep the performance description accurate.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At python_multipart/multipart.py, line 1144:

<comment>This comment says there is “one dict lookup per write”, but the block performs nine `.get()` calls per write. Say “one lookup per callback per write” to keep the performance description accurate.</comment>

<file context>
@@ -1140,6 +1140,39 @@ def _internal_write(self, data: bytes, length: int) -> int:
         current_header_size = self._current_header_size
 
+        # Hoist callback dispatch out of the per-byte loop (same shape as the
+        # QuerystringParser optimization in #316): one dict lookup per write
+        # instead of a `"on_" + name` concatenation, dict lookup and `cast`
+        # per emitted event.
</file context>
Suggested change
# QuerystringParser optimization in #316): one dict lookup per write
# QuerystringParser optimization in #316): one lookup per callback per write

# instead of a `"on_" + name` concatenation, dict lookup and `cast`
# per emitted event.
callbacks = cast("MultipartCallbacks", self.callbacks)

@cubic-dev-ai cubic-dev-ai Bot Oct 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P3: This snapshot changes BaseParser.set_callback() semantics without documenting the operation boundary: a callback registered during on_part_begin no longer affects later events in the same write(). Document that callback updates made during a write take effect from the next write() (while updates between writes apply normally).\n\n(Based on your team's feedback about callback update timing.) [d68ae4b8-e1a0-4dee-a221-824cce4f1a50]

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At python_multipart/multipart.py, line 1147:

<comment>This snapshot changes `BaseParser.set_callback()` semantics without documenting the operation boundary: a callback registered during `on_part_begin` no longer affects later events in the same `write()`. Document that callback updates made during a write take effect from the next `write()` (while updates between writes apply normally).\n\n(Based on your team's feedback about callback update timing.) [d68ae4b8-e1a0-4dee-a221-824cce4f1a50]</comment>

<file context>
@@ -1140,6 +1140,39 @@ def _internal_write(self, data: bytes, length: int) -> int:
+        # QuerystringParser optimization in #316): one dict lookup per write
+        # instead of a `"on_" + name` concatenation, dict lookup and `cast`
+        # per emitted event.
+        callbacks = cast("MultipartCallbacks", self.callbacks)
+        on_part_begin = callbacks.get("on_part_begin")
+        on_part_data = callbacks.get("on_part_data")
</file context>

on_part_begin = callbacks.get("on_part_begin")
on_part_data = callbacks.get("on_part_data")
on_part_end = callbacks.get("on_part_end")
on_header_begin = callbacks.get("on_header_begin")
on_header_field = callbacks.get("on_header_field")
on_header_value = callbacks.get("on_header_value")
on_header_end = callbacks.get("on_header_end")
on_headers_finished = callbacks.get("on_headers_finished")
on_end = callbacks.get("on_end")
if on_part_begin is None:
on_part_begin = _noop_event
if on_part_data is None:
on_part_data = _noop_data
if on_part_end is None:
on_part_end = _noop_event
if on_header_begin is None:
on_header_begin = _noop_event
if on_header_field is None:
on_header_field = _noop_data
if on_header_value is None:
on_header_value = _noop_data
if on_header_end is None:
on_header_end = _noop_event
if on_headers_finished is None:
on_headers_finished = _noop_event
if on_end is None:
on_end = _noop_event

# Our index defaults to 0.
i = 0

Expand All @@ -1162,7 +1195,7 @@ def delete_mark(name: str, reset: bool = False) -> None:
# end of the buffer, and reset the mark, instead of deleting it. This
# is used at the end of the function to call our callbacks with any
# remaining data in this chunk.
def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> None:
def data_callback(name: CallbackName, end_i: int, func: Callable[..., Any], remaining: bool = False) -> None:
marked_index = self.marks.get(name)
if marked_index is None:
return
Expand All @@ -1174,26 +1207,26 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
pass
elif marked_index >= 0:
# We are emitting data from the local buffer.
self.callback(name, data, marked_index, end_i)
func(data, marked_index, end_i)
else:
# Some of the data comes from a partial boundary match.
# and requires look-behind.
# We need to use self.flags (and not flags) because we care about
# the state when we entered the loop.
lookbehind_len = -marked_index
if lookbehind_len <= boundary_length:
self.callback(name, boundary, 0, lookbehind_len)
func(boundary, 0, lookbehind_len)
elif self.flags & FLAG_PART_BOUNDARY:
lookback = boundary + b"\r\n"
self.callback(name, lookback, 0, lookbehind_len)
func(lookback, 0, lookbehind_len)
elif self.flags & FLAG_LAST_BOUNDARY:
lookback = boundary + b"--\r\n"
self.callback(name, lookback, 0, lookbehind_len)
func(lookback, 0, lookbehind_len)
else: # pragma: no cover (error case)
self.logger.warning("Look-back buffer error")

if end_i > 0:
self.callback(name, data, 0, end_i)
func(data, 0, end_i)
# If we're getting remaining data, we have got all the data we
# can be certain is not a boundary, leaving only a partial boundary match.
if remaining:
Expand Down Expand Up @@ -1253,7 +1286,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
index = 0

# Callback for the start of a part.
self.callback("part_begin")
on_part_begin()
current_header_count = 0
current_header_size = 0

Expand Down Expand Up @@ -1289,7 +1322,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
# to stop parsing headers in the MultipartState.HEADER_FIELD state,
# below.
if c != CR:
self.callback("header_begin")
on_header_begin()

# Move to parsing header fields.
state = HEADER_FIELD
Expand Down Expand Up @@ -1335,7 +1368,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No

# Call our callback with the header field.
i = colon
data_callback("header_field", i)
data_callback("header_field", i, on_header_field)

# Move to parsing the header value.
state = HEADER_VALUE_START
Expand All @@ -1362,8 +1395,8 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
advance_header_size(end - i)
if cr != -1:
i = cr
data_callback("header_value", i)
self.callback("header_end")
data_callback("header_value", i, on_header_value)
on_header_end()
current_header_size = 0
state = HEADER_VALUE_ALMOST_DONE
else:
Expand All @@ -1390,7 +1423,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
self.logger.warning(msg)
raise MultipartParseError(msg, offset=i)

self.callback("headers_finished")
on_headers_finished()
state = PART_DATA_START

elif state == PART_DATA_START:
Expand Down Expand Up @@ -1477,11 +1510,11 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
flags &= ~FLAG_PART_BOUNDARY

# We have identified a boundary, callback for any data before it.
data_callback("part_data", i - index)
data_callback("part_data", i - index, on_part_data)
# Callback indicating that we've reached the end of
# a part, and are starting a new one.
self.callback("part_end")
self.callback("part_begin")
on_part_end()
on_part_begin()
current_header_count = 0
current_header_size = 0

Expand All @@ -1502,11 +1535,11 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
# We need a second hyphen here.
if c == HYPHEN:
# We have identified a boundary, callback for any data before it.
data_callback("part_data", i - index)
data_callback("part_data", i - index, on_part_data)
# Callback to end the current part, and then the
# message.
self.callback("part_end")
self.callback("end")
on_part_end()
on_end()
state = END
else:
# No match, so reset index.
Expand All @@ -1531,7 +1564,7 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
self.logger.warning(msg)
raise MultipartParseError(msg, offset=i)
index += 1
self.callback("end")
on_end()
state = END

elif state == END:
Expand All @@ -1557,9 +1590,9 @@ def data_callback(name: CallbackName, end_i: int, remaining: bool = False) -> No
# that we haven't yet reached the end of this 'thing'. So, by setting
# the mark to 0, we cause any data callbacks that take place in future
# calls to this function to start from the beginning of that buffer.
data_callback("header_field", length, True)
data_callback("header_value", length, True)
data_callback("part_data", length - index, True)
data_callback("header_field", length, on_header_field, True)
data_callback("header_value", length, on_header_value, True)
data_callback("part_data", length - index, on_part_data, True)

# Save values to locals.
self.state = state
Expand Down
95 changes: 95 additions & 0 deletions tests/test_multipart.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -644,6 +644,101 @@ def test_invalid_max_size(self) -> None:
q = OctetStreamParser(max_size="foo") # type: ignore[arg-type]


class TestMultipartParser(unittest.TestCase):
BOUNDARY = b"boundary"
BODY = (
b"--boundary\r\n"
b'Content-Disposition: form-data; name="a"\r\n'
b"\r\n"
b"foo\r\n"
b"--boundary\r\n"
b'Content-Disposition: form-data; name="b"; filename="b.txt"\r\n'
b"Content-Type: text/plain\r\n"
b"\r\n"
b"barbaz\r\n"
b"--boundary--\r\n"
)

def parse(self, callbacks: Any = None, chunk_size: int | None = None) -> list[tuple[Any, ...]]:
events: list[tuple[Any, ...]] = []

def make_event(name: str) -> Any:
def on_event() -> None:
events.append((name,))

return on_event

def make_data(name: str) -> Any:
def on_data(data: bytes, start: int, end: int) -> None:
events.append((name, data[start:end]))

return on_data

if callbacks is None:
callbacks = {
"on_part_begin": make_event("part_begin"),
"on_part_data": make_data("part_data"),
"on_part_end": make_event("part_end"),
"on_header_begin": make_event("header_begin"),
"on_header_field": make_data("header_field"),
"on_header_value": make_data("header_value"),
"on_header_end": make_event("header_end"),
"on_headers_finished": make_event("headers_finished"),
"on_end": make_event("end"),
}
parser = MultipartParser(self.BOUNDARY, callbacks)
if chunk_size is None:
self.assertEqual(parser.write(self.BODY), len(self.BODY))
else:
for i in range(0, len(self.BODY), chunk_size):
parser.write(self.BODY[i : i + chunk_size])
parser.finalize()
return events

def test_no_callbacks(self) -> None:
parser = MultipartParser(self.BOUNDARY)

self.assertEqual(parser.write(self.BODY), len(self.BODY))
parser.finalize()

def test_none_callbacks(self) -> None:
callbacks: Any = {
"on_part_begin": None,
"on_part_data": None,
"on_part_end": None,
"on_header_begin": None,
"on_header_field": None,
"on_header_value": None,
"on_header_end": None,
"on_headers_finished": None,
"on_end": None,
}
parser = MultipartParser(self.BOUNDARY, callbacks)

self.assertEqual(parser.write(self.BODY), len(self.BODY))
parser.finalize()

def test_events_match_across_chunkings(self) -> None:
# Chunk boundaries legitimately split data callbacks, so compare the
# joined payload per channel plus the notification-event sequence.
def summarize(events: list[tuple[Any, ...]]) -> tuple[Any, ...]:
joined: dict[str, list[bytes]] = {}
notifications: list[str] = []
for event in events:
if len(event) == 2:
joined.setdefault(event[0], []).append(event[1])
else:
notifications.append(event[0])
return ({name: b"".join(payloads) for name, payloads in joined.items()}, notifications)
Comment on lines +725 to +732

@cubic-dev-ai cubic-dev-ai Bot Oct 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: This comparison drops callback ordering: moving part_data after part_end still produces the same summary. Normalize adjacent data fragments while preserving every event’s position, then compare the normalized sequences.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At tests/test_multipart.py, line 725:

<comment>This comparison drops callback ordering: moving `part_data` after `part_end` still produces the same summary. Normalize adjacent data fragments while preserving every event’s position, then compare the normalized sequences.</comment>

<file context>
@@ -644,6 +644,101 @@ def test_invalid_max_size(self) -> None:
+        # Chunk boundaries legitimately split data callbacks, so compare the
+        # joined payload per channel plus the notification-event sequence.
+        def summarize(events: list[tuple[Any, ...]]) -> tuple[Any, ...]:
+            joined: dict[str, list[bytes]] = {}
+            notifications: list[str] = []
+            for event in events:
</file context>
Suggested change
joined: dict[str, list[bytes]] = {}
notifications: list[str] = []
for event in events:
if len(event) == 2:
joined.setdefault(event[0], []).append(event[1])
else:
notifications.append(event[0])
return ({name: b"".join(payloads) for name, payloads in joined.items()}, notifications)
normalized: list[tuple[Any, ...]] = []
for event in events:
if (
len(event) == 2
and normalized
and len(normalized[-1]) == 2
and normalized[-1][0] == event[0]
):
previous = normalized[-1]
normalized[-1] = (event[0], previous[1] + event[1])
else:
normalized.append(event)
return tuple(normalized)

@cubic-dev-ai cubic-dev-ai Bot Oct 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2: This comparison can miss a broken callback stream: it derives its expected result from the optimized parser and discards ordering between notifications and data callbacks. Compare against an explicit expected sequence, merging only adjacent chunks from the same data callback.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At tests/test_multipart.py, line 732:

<comment>This comparison can miss a broken callback stream: it derives its expected result from the optimized parser and discards ordering between notifications and data callbacks. Compare against an explicit expected sequence, merging only adjacent chunks from the same data callback.</comment>

<file context>
@@ -644,6 +644,101 @@ def test_invalid_max_size(self) -> None:
+                    joined.setdefault(event[0], []).append(event[1])
+                else:
+                    notifications.append(event[0])
+            return ({name: b"".join(payloads) for name, payloads in joined.items()}, notifications)
+
+        whole = self.parse()
</file context>


whole = self.parse()
self.assertIn(("part_data", b"foo"), whole)
self.assertIn(("part_data", b"barbaz"), whole)
expected = summarize(whole)
for chunk_size in (1, 3, 7, 64):
self.assertEqual(summarize(self.parse(chunk_size=chunk_size)), expected)


class TestBase64Decoder(unittest.TestCase):
# Note: base64('foobar') == 'Zm9vYmFy'
def setUp(self) -> None:
Expand Down
Loading