diff --git a/changelog.d/20260731_163747_kurtmckee_rm_quadratic_behavior.rst b/changelog.d/20260731_163747_kurtmckee_rm_quadratic_behavior.rst
index a2cd390..01412de 100644
--- a/changelog.d/20260731_163747_kurtmckee_rm_quadratic_behavior.rst
+++ b/changelog.d/20260731_163747_kurtmckee_rm_quadratic_behavior.rst
@@ -1,7 +1,9 @@
Fixed
-----
-* Fix quadratic search behavior in ``SGMLParser.feed()``.
+* Fix quadratic CPU usage in ``SGMLParser.feed()``.
This can occur when a large document with an unfinished token
(like `` None:
"""Initialize and reset this instance."""
self.verbose = verbose
- # These variables help guard against quadratic search behavior.
+ # These variables help guard against quadratic CPU usage behavior
+ # without significantly increasing memory usage.
+ #
# Previously, each call to `.feed()` resulted in string-concatenation
- # and a regular expression matching from the start of the raw data.
+ # and a regular expression matching from the start of the raw data
+ # even though it was known that the existing raw data wouldn't match.
# For large unclosed tokens (like ` None:
"""Reset this instance. Loses all unprocessed data."""
self.__starttag_text: str | None = None
self._rawdata = ""
- self._pending = []
- self._pending_content_length = 0
+ self._pending = StringIO()
self._retry_at = 0
self.stack: list[str] = []
self.lasttag = "???"
@@ -105,16 +111,14 @@ def rawdata(self) -> str:
@rawdata.setter
def rawdata(self, value: str) -> None:
self._rawdata = value
- self._pending = []
- self._pending_content_length = 0
+ self._pending = StringIO()
def _flush_pending(self) -> None:
"""Merge any chunks queued by `feed()` into `self._rawdata`."""
- if self._pending:
- self._rawdata += "".join(self._pending)
- self._pending = []
- self._pending_content_length = 0
+ if self._pending.tell():
+ self._rawdata += self._pending.getvalue()
+ self._pending = StringIO()
def setnomoretags(self) -> None:
"""Enter literal mode (CDATA) till EOF.
@@ -138,9 +142,8 @@ def feed(self, data: str) -> None:
all the processing is done by goahead().)
"""
- self._pending.append(data)
- self._pending_content_length += len(data)
- if len(self._rawdata) + self._pending_content_length < self._retry_at:
+ self._pending.write(data)
+ if len(self._rawdata) + self._pending.tell() < self._retry_at:
# The last attempt to resolve the pending token failed.
# Skip re-scanning until enough new data has arrived
# to make another attempt worthwhile.
diff --git a/src/feedparser_sgmllib/_compat.py b/src/feedparser_sgmllib/_compat.py
new file mode 100644
index 0000000..2f67667
--- /dev/null
+++ b/src/feedparser_sgmllib/_compat.py
@@ -0,0 +1,38 @@
+# Manage compatibility across supported Python versions.
+
+import sys
+
+__all__ = [
+ "StringIO",
+]
+
+
+if sys.version_info >= (3, 12):
+ from io import StringIO
+else:
+ # In Python >=3.12, `io.StringIO` has good memory allocation performance,
+ # which makes it a good candidate for collecting unparsed input data
+ # (see the comment in `SGMLParser.__init__()` about CPU and memory usage).
+ #
+ # However, the performance of `io.StringIO` is poor in Python 3.11 and below.
+ # For these lower Python versions, bytearray exhibits good performance
+ # but requires manual tracking of the length of the decoded string
+ # as well as calls to `.encode()` and `.decode()`.
+ #
+ # Therefore, the `io.StringIO` interface is mimicked here
+ # and backed by a bytearray implementation.
+ class StringIO:
+ def __init__(self) -> None:
+ self._buffer = bytearray()
+ self._length = 0
+
+ def write(self, text: str) -> int:
+ self._buffer.extend(text.encode())
+ self._length += len(text)
+ return len(text)
+
+ def getvalue(self) -> str:
+ return self._buffer.decode()
+
+ def tell(self) -> int:
+ return self._length
diff --git a/tests/test_sgmllib.py b/tests/test_sgmllib.py
index 0db0563..348903a 100644
--- a/tests/test_sgmllib.py
+++ b/tests/test_sgmllib.py
@@ -1,6 +1,7 @@
import io
import pathlib
import time
+import tracemalloc
import pytest
@@ -535,9 +536,9 @@ def test_endtag_with_no_valid_tag_name_does_not_close_open_tag(nested_tag_collec
pytest.param("