diff --git a/docs/usage.md b/docs/usage.md index a4ec38d..28e9449 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -130,6 +130,16 @@ with db.stream(match_event_handler=on_match, context=2345) as stream: stream.scan(b'qux', match_event_handler=on_qux_match) ``` +A stream keeps its database, callback, context, and optional scratch object alive +until the stream itself is released. Entering the context opens the native stream; +scanning before entry or after closure raises `RuntimeError`. Closing is +idempotent, so explicitly closing inside a `with` block is supported. The same +stream object can be entered again after closure, but cannot be entered or +reinitialized while already open. + +Use explicit closure or a context manager to deliver end-of-data matches. +Releasing an unclosed stream frees its native state without invoking callbacks. + ### Vectored Mode ```python diff --git a/src/hyperscan/extension.c b/src/hyperscan/extension.c index e072d10..2130fb8 100644 --- a/src/hyperscan/extension.c +++ b/src/hyperscan/extension.c @@ -851,9 +851,8 @@ static PyObject *Database_stream(Database *self, PyObject *args, PyObject *kwds) HS_LOCK_RETURN_NULL(); PyObject *stream = PyObject_CallFunction( (PyObject *)&StreamType, "OIOO", (PyObject *)self, flags, ocallback, octx); - if (PyErr_Occurred()) + if (stream == NULL) HS_LOCK_RETURN_NULL(); - Py_INCREF(stream); HS_LOCK_RETURN(stream); } @@ -986,10 +985,37 @@ static PyTypeObject DatabaseType = { Database_new, /* tp_new */ }; +static int Stream_traverse(Stream *self, visitproc visit, void *arg) +{ + Py_VISIT(self->database); + Py_VISIT(self->scratch); + if (self->cctx != NULL) { + Py_VISIT(self->cctx->callback); + Py_VISIT(self->cctx->ctx); + } + return 0; +} + +static int Stream_clear(Stream *self) +{ + if (self->identifier != NULL) { + hs_close_stream(self->identifier, NULL, NULL, NULL); + self->identifier = NULL; + } + Py_CLEAR(self->database); + Py_CLEAR(self->scratch); + if (self->cctx != NULL) { + Py_CLEAR(self->cctx->callback); + Py_CLEAR(self->cctx->ctx); + } + return 0; +} + static void Stream_dealloc(Stream *self) { - if (self->cctx != NULL) - free(self->cctx); + PyObject_GC_UnTrack(self); + Stream_clear(self); + free(self->cctx); Py_TYPE(self)->tp_free((PyObject *)self); } @@ -1000,8 +1026,6 @@ static PyObject *Stream_new(PyTypeObject *type, PyObject *args, PyObject *kwds) self = (Stream *)type->tp_alloc(type, 0); if (self != NULL) { self->flags = 0; - self->database = Py_None; - self->scratch = Py_None; } return (PyObject *)self; @@ -1017,24 +1041,38 @@ static int Stream_init(Stream *self, PyObject *args, PyObject *kwds) "scratch", NULL, }; - self->cctx = malloc(sizeof(py_scan_callback_ctx)); + PyObject *database, *scratch = Py_None; + PyObject *callback = Py_None, *context = Py_None; + uint32_t flags = 0; if (!PyArg_ParseTupleAndKeywords( args, kwds, - "O|IOOO!", + "O!|IOOO!", kwlist, - &self->database, - &self->flags, - &self->cctx->callback, - &self->cctx->ctx, - &self->scratch, - &ScratchType)) + &DatabaseType, + &database, + &flags, + &callback, + &context, + &ScratchType, + &scratch)) return -1; - if (!PyObject_IsInstance(self->database, (PyObject *)&DatabaseType)) { - PyErr_SetString( - PyExc_TypeError, "database must be a hyperscan.Database instance"); + if (self->identifier != NULL) { + PyErr_SetString(PyExc_RuntimeError, "cannot reinitialize an open stream"); return -1; } + if (self->cctx == NULL) { + self->cctx = calloc(1, sizeof(py_scan_callback_ctx)); + if (self->cctx == NULL) { + PyErr_NoMemory(); + return -1; + } + } + Py_XSETREF(self->database, Py_NewRef(database)); + Py_XSETREF(self->scratch, Py_NewRef(scratch)); + Py_XSETREF(self->cctx->callback, Py_NewRef(callback)); + Py_XSETREF(self->cctx->ctx, Py_NewRef(context)); + self->flags = flags; return 0; } @@ -1047,29 +1085,33 @@ static PyObject *Stream_close(Stream *self, PyObject *args, PyObject *kwds) PyObject *oscratch = Py_None, *ocallback = Py_None, *octx = Py_None; static char *kwlist[] = {"scratch", "match_event_handler", "context", NULL}; if (!PyArg_ParseTupleAndKeywords( - args, - kwds, - "|O!OO", - kwlist, - &oscratch, - &ScratchType, - &ocallback, - &octx)) + args, kwds, "|OOO", kwlist, &oscratch, &ocallback, &octx)) HS_LOCK_RETURN_NULL(); + if (self->identifier == NULL) + HS_LOCK_RETURN(Py_NewRef(Py_None)); Database *db = (Database *)self->database; Scratch *scratch; - if (PyObject_Not(oscratch)) - oscratch = ((Database *)self->database)->scratch; - cctx.callback = PyObject_IsTrue(ocallback) ? ocallback : self->cctx->callback; - cctx.ctx = PyObject_IsTrue(octx) ? octx : self->cctx->ctx; - if (PyObject_IsTrue(oscratch) && cctx.callback != NULL) - scratch = (Scratch *)oscratch; - else - scratch = (Scratch *)db->scratch; + if (oscratch == Py_None) + oscratch = self->scratch == Py_None ? db->scratch : self->scratch; + if (!PyObject_TypeCheck(oscratch, &ScratchType)) { + PyErr_SetString( + PyExc_TypeError, "scratch must be a hyperscan.Scratch instance"); + HS_LOCK_RETURN_NULL(); + } + cctx.callback = ocallback == Py_None ? self->cctx->callback : ocallback; + cctx.ctx = octx == Py_None ? self->cctx->ctx : octx; + scratch = (Scratch *)oscratch; hs_scratch_t *hs_scratch = scratch->hs_scratch; hs_error_t hs_err = hs_close_stream( - self->identifier, hs_scratch, hs_match_handler, (void *)&cctx); + self->identifier, + hs_scratch, + cctx.callback == Py_None ? NULL : hs_match_handler, + (void *)&cctx); + if (hs_err == HS_SUCCESS || hs_err == HS_UNKNOWN_ERROR) + self->identifier = NULL; + if (PyErr_Occurred()) + HS_LOCK_RETURN_NULL(); HANDLE_HYPERSCAN_ERR(hs_err, NULL); HS_LOCK_RETURN(Py_NewRef(Py_None)); @@ -1099,13 +1141,17 @@ static PyObject *Stream_enter(Stream *self) Stream *stream = (Stream *)self; Database *db = (Database *)stream->database; + if (self->identifier != NULL) { + PyErr_SetString(PyExc_RuntimeError, "stream is already open"); + HS_LOCK_RETURN_NULL(); + } if (db->chimera) { PyErr_SetString(PyExc_RuntimeError, "chimera does not support streams"); HS_LOCK_RETURN_NULL(); } hs_error_t err = hs_open_stream(db->hs_db, 0, &self->identifier); HANDLE_HYPERSCAN_ERR(err, NULL); - HS_LOCK_RETURN((PyObject *)self); + HS_LOCK_RETURN(Py_NewRef((PyObject *)self)); } static PyObject *Stream_exit(Stream *self) @@ -1113,9 +1159,10 @@ static PyObject *Stream_exit(Stream *self) HS_LOCK_DECLARE(); HS_LOCK_ACQUIRE_OR_RETURN_NULL(); - PyObject_CallMethod((PyObject *)self, "close", NULL); - if (PyErr_Occurred()) + PyObject *result = PyObject_CallMethod((PyObject *)self, "close", NULL); + if (result == NULL) HS_LOCK_RETURN_NULL(); + Py_DECREF(result); HS_LOCK_RETURN(Py_NewRef(Py_None)); } @@ -1124,6 +1171,11 @@ static PyObject *Stream_scan(Stream *self, PyObject *args, PyObject *kwds) HS_LOCK_DECLARE(); HS_LOCK_ACQUIRE_OR_RETURN_NULL(); + if (self->identifier == NULL) { + PyErr_SetString(PyExc_RuntimeError, "stream is not open"); + HS_LOCK_RETURN_NULL(); + } + Py_buffer view; uint32_t flags = 0; PyObject *ocallback = Py_None, *octx = Py_None, *oscratch = Py_None; @@ -1143,16 +1195,17 @@ static PyObject *Stream_scan(Stream *self, PyObject *args, PyObject *kwds) HS_LOCK_RETURN_NULL(); } - if (PyObject_Not(ocallback)) + if (ocallback == Py_None) ocallback = self->cctx->callback; - if (PyObject_Not(octx)) + if (octx == Py_None) octx = self->cctx->ctx; Database *db = (Database *)self->database; Scratch *scratch; - if (PyObject_Not(oscratch)) - scratch = (Scratch *)db->scratch; + if (oscratch == Py_None) + scratch = + (Scratch *)(self->scratch == Py_None ? db->scratch : self->scratch); else { if (!PyObject_IsInstance(oscratch, (PyObject *)&ScratchType)) { PyErr_SetString( @@ -1261,7 +1314,7 @@ static PyTypeObject StreamType = { 0, /* tp_getattro */ 0, /* tp_setattro */ 0, /* tp_as_buffer */ - Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, /* tp_flags */ + Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC, /* tp_flags */ "Stream(database=None, flags=0, match_event_handler=None)\n\n" " Provides a context manager for scanning streams of text.\n\n" " Args:\n" @@ -1272,24 +1325,24 @@ static PyTypeObject StreamType = { " which is invoked for each match result, and passed the\n" " expression id, start offset, end offset, flags, and a\n" " context object." - "\n\n", /* tp_doc */ - 0, /* tp_traverse */ - 0, /* tp_clear */ - 0, /* tp_richcompare */ - 0, /* tp_weaklistoffset */ - 0, /* tp_iter */ - 0, /* tp_iternext */ - Stream_methods, /* tp_methods */ - Stream_members, /* tp_members */ - 0, /* tp_getset */ - 0, /* tp_base */ - 0, /* tp_dict */ - 0, /* tp_descr_get */ - 0, /* tp_descr_set */ - 0, /* tp_dictoffset */ - (initproc)Stream_init, /* tp_init */ - 0, /* tp_alloc */ - Stream_new, /* tp_new */ + "\n\n", /* tp_doc */ + (traverseproc)Stream_traverse, /* tp_traverse */ + (inquiry)Stream_clear, /* tp_clear */ + 0, /* tp_richcompare */ + 0, /* tp_weaklistoffset */ + 0, /* tp_iter */ + 0, /* tp_iternext */ + Stream_methods, /* tp_methods */ + Stream_members, /* tp_members */ + 0, /* tp_getset */ + 0, /* tp_base */ + 0, /* tp_dict */ + 0, /* tp_descr_get */ + 0, /* tp_descr_set */ + 0, /* tp_dictoffset */ + (initproc)Stream_init, /* tp_init */ + 0, /* tp_alloc */ + Stream_new, /* tp_new */ }; static void Scratch_dealloc(Scratch *self) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py new file mode 100644 index 0000000..2551e20 --- /dev/null +++ b/tests/test_stream_lifecycle.py @@ -0,0 +1,266 @@ +"""Run native lifetime regressions in children so a crash does not kill pytest.""" + +import subprocess +import sys +import textwrap + +import pytest + + +def run_stream_case(code): + setup = """ + import gc + import sys + import weakref + import hyperscan + + db = hyperscan.Database(mode=hyperscan.HS_MODE_STREAM) + db.compile(expressions=[b"a$"]) + """ + result = subprocess.run( + [sys.executable, "-c", textwrap.dedent(setup) + textwrap.dedent(code)], + capture_output=True, + text=True, + timeout=10, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +@pytest.mark.parametrize("argument", ["database", "callback", "context", "scratch"]) +def test_stream_retains_constructor_objects(argument): + run_stream_case( + f""" + class Context: + pass + + callback = lambda *args: None + context = Context() + scratch = hyperscan.Scratch(db) + objects = dict(database=db, callback=callback, context=context, scratch=scratch) + value = objects[{argument!r}] + kwargs = dict(match_event_handler=callback, context=context) + if {argument!r} == "scratch": + kwargs["scratch"] = scratch + before = sys.getrefcount(value) + stream = hyperscan.Stream(db, **kwargs) + assert sys.getrefcount(value) == before + 1 + del stream + assert sys.getrefcount(value) == before + """ + ) + + +def test_enter_returns_owned_reference(): + run_stream_case( + """ + stream = db.stream(match_event_handler=None) + before = sys.getrefcount(stream) + entered = stream.__enter__() + assert entered is stream + assert sys.getrefcount(stream) == before + 1 + stream.close() + del entered + assert sys.getrefcount(stream) == before + """ + ) + + +def test_factory_returns_one_owned_reference(): + run_stream_case( + """ + manager = db.stream(match_event_handler=None) + assert sys.getrefcount(manager) == 2 + """ + ) + + +def test_exit_releases_close_result(): + run_stream_case( + """ + marker = object() + class CustomStream(hyperscan.Stream): + def close(self): + return marker + manager = CustomStream(db) + before = sys.getrefcount(marker) + assert manager.__exit__(None, None, None) is None + assert sys.getrefcount(marker) == before + """ + ) + + +@pytest.mark.parametrize("invalid", ["database", "scratch"]) +def test_invalid_constructor_argument_is_rejected(invalid): + run_stream_case( + f""" + try: + if {invalid!r} == "database": + hyperscan.Stream(object()) + else: + hyperscan.Stream(db, scratch=object()) + except TypeError: + pass + else: + raise AssertionError("invalid constructor argument was accepted") + """ + ) + + +def test_stream_survives_context_exit_and_can_reopen(): + run_stream_case( + """ + matches = [] + manager = db.stream(match_event_handler=lambda *args: matches.append(args)) + for _ in range(2): + with manager as stream: + stream.scan(b"a") + assert stream is manager + assert stream.database is db + assert len(matches) == 2 + """ + ) + + +def test_stream_retains_temporary_database_and_callback(): + run_stream_case( + """ + matches = [] + def make_stream(): + database = hyperscan.Database(mode=hyperscan.HS_MODE_STREAM) + database.compile(expressions=[b"a$"]) + return database.stream( + match_event_handler=lambda *args: matches.append(args) + ) + manager = make_stream() + gc.collect() + with manager as stream: + stream.scan(b"a") + assert len(matches) == 1 + """ + ) + + +@pytest.mark.parametrize("opened", [False, True]) +def test_stream_context_cycle_is_collectible(opened): + run_stream_case( + f""" + class Context: + pass + context = Context() + context_ref = weakref.ref(context) + manager = db.stream(match_event_handler=None, context=context) + if {opened!r}: + entered = manager.__enter__() + del entered + context.manager = manager + del context, manager + gc.collect() + assert context_ref() is None + """ + ) + + +@pytest.mark.parametrize("phase", ["before_enter", "after_close"]) +def test_scan_requires_open_stream(phase): + run_stream_case( + f""" + manager = db.stream(match_event_handler=None) + if {phase!r} == "after_close": + entered = manager.__enter__() + del entered + manager.close() + try: + manager.scan(b"a") + except RuntimeError as exc: + assert "not open" in str(exc) + else: + raise AssertionError("scan accepted a closed stream") + """ + ) + + +def test_close_is_idempotent_and_flushes_once(): + run_stream_case( + """ + matches = [] + manager = db.stream(match_event_handler=lambda *args: matches.append(args)) + manager.close() + with manager as stream: + stream.scan(b"a") + stream.close() + stream.close() + assert len(matches) == 1 + """ + ) + + +@pytest.mark.parametrize("operation", ["enter", "init"]) +def test_open_stream_rejects_reinitialization(operation): + run_stream_case( + f""" + manager = db.stream(match_event_handler=None) + with manager: + try: + if {operation!r} == "enter": + manager.__enter__() + else: + manager.__init__(db) + except RuntimeError: + pass + else: + raise AssertionError("open stream was replaced") + manager.scan(b"a") + """ + ) + + +def test_close_preserves_callback_exception_and_consumes_stream(): + run_stream_case( + """ + def callback(*args): + raise ValueError("end-of-data callback") + manager = db.stream(match_event_handler=callback) + entered = manager.__enter__() + del entered + manager.scan(b"a") + try: + manager.close() + except ValueError as exc: + assert str(exc) == "end-of-data callback" + else: + raise AssertionError("callback error was lost") + manager.close() + """ + ) + + +def test_stored_and_explicit_scratch(): + run_stream_case( + """ + scratch = hyperscan.Scratch(db) + matches = [] + manager = hyperscan.Stream( + db, scratch=scratch, match_event_handler=lambda *args: matches.append(args) + ) + with manager as stream: + stream.scan(b"a", scratch=scratch) + stream.close(scratch=scratch) + assert len(matches) == 1 + """ + ) + + +def test_invalid_close_scratch_keeps_stream_open(): + run_stream_case( + """ + manager = db.stream(match_event_handler=None) + with manager as stream: + try: + stream.close(scratch=object()) + except TypeError: + pass + else: + raise AssertionError("invalid scratch was accepted") + stream.scan(b"a") + """ + )