Uh oh!
There was an error while loading. Please reload this page.
FEAT: async POC - #677
Draft
Subrata (subrata-ms) wants to merge 8 commits into
Draft
Conversation
Introduces the foundational C++ building blocks for the upcoming Cursor.execute_async / Cursor.fetch_async work. No user-facing API is exposed yet; sync execute/fetch paths are functionally unchanged. Helpers added in ddbc_bindings.cpp: - struct PollingConfig (initial_ms=0.5, max_ms=20, multiplier=1.5) - class AsyncEnableGuard: RAII sets/clears SQL_ATTR_ASYNC_ENABLE on an HSTMT so the async flag never leaks onto a handle reused by later sync calls. - template<Fn> ExecuteWithPolling(callOnce, asyncMode, cfg): sync passthrough when asyncMode=false; polling loop with exponential backoff when asyncMode=true. Caller owns GIL release. - PrepareAndBind(...): factored out of SQLExecute_wrap (SQLPrepare under released GIL, DescribeCache clear, isStmtPrepared flag update, BindParameters). Enables reuse from upcoming async execute wrap. SQLExecute_wrap refactored to route both branches through the helpers with asyncMode=false (semantic no-op vs. the previous inline calls). DAE loop unchanged. Capability probe: - Connection::isAsyncCapable() / ConnectionHandle::isAsyncCapable() probes SQLGetInfo(SQL_ASYNC_MODE), caches the value in an atomic int (SQL_AM_STATEMENT / SQL_AM_CONNECTION => true). Exposed to Python as Connection.is_async_capable(). Build: clean under -Werror -Wattributes -Wint-to-pointer-cast.
📊 Code Coverage Report
Diff CoverageDiff: main...HEAD, staged and unstaged changes
Summary
mssql_python/cursor.pyLines 3526-3534 3526perconnection.
3527 """
3528 conn = self._connection._conn
3529 if not conn.is_async_capable():
! 3530 raise NotSupportedError(
3531 driver_error=(
3532 "Async execution is not supported by this ODBC driver. "
3533 "SQLGetInfo(SQL_ASYNC_MODE) reports SQL_AM_NONE."
3534 ),Lines 3537-3545 35373538def_acquire_async_slot(self) ->None:
3539"""Take the cursor-busy slot; raise ProgrammingError if already held."""3540ifself._async_in_flight:
! 3541raiseProgrammingError(
3542driver_error=(
3543"Cursor is busy: another async operation is already in flight "3544"on this cursor. One Cursor == one HSTMT, so async ops on the "3545"same cursor must be serialized. Create a second cursor for "Lines 3575-3590 3575ifreset_cursor:
3576ifself.hstmt:
3577self._soft_reset_cursor()
3578else:
! 3579self._reset_cursor()
3580else:
! 3581ifself.hstmt:
3582logger.debug(
3583"execute_async: Closing cursor for re-execution (reset_cursor=False)"3584 )
! 3585self.hstmt._close_cursor()
! 3586self._clear_rownumber()
35873588# Clear any previous messages3589self.messages= []Lines 3590-3603 35903591# Parameter unwrap (same rules as sync execute).3592ifparameters:
3593ifisinstance(parameters, tuple) andlen(parameters) ==1:
! 3594ifisinstance(parameters[0], (tuple, list, dict)):
! 3595actual_params=parameters[0]
! 3596elifisinstance(parameters[0], Row):
! 3597actual_params=tuple(parameters[0])
3598else:
! 3599actual_params=parameters3600else:
3601actual_params=parameters36023603ifoperation==self.last_executed_stmtandisinstance(Lines 3602-3610 36023603ifoperation==self.last_executed_stmtandisinstance(
3604actual_params, (tuple, list)
3605 ):
! 3606parameters=list(actual_params)
3607else:
3608operation, converted_params=detect_and_convert_parameters(
3609operation, actual_params3610 )Lines 3609-3617 3609operation, actual_params3610 )
3611parameters=list(converted_params)
3612else:
! 3613parameters= []
36143615encoding_settings=self._get_encoding_settings()
36163617logger.debug("execute_async: Creating parameter type list")Lines 3618-3627 3618param_info=ddbc_bindings.ParamInfo3619parameters_type: List[Any] = []
36203621ifparametersandself._inputsizes:
! 3622iflen(self._inputsizes) !=len(parameters):
! 3623warnings.warn(
3624f"Number of input sizes ({len(self._inputsizes)}) does not match "3625f"number of parameters ({len(parameters)}). "3626f"This may lead to unexpected behavior.",
3627Warning,Lines 3657-3668 3657DELIBERATEDUPLICATEofthecorrespondinglogicin``execute()``.
3658 """
3659 try:
3660 check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
! 3661 except Exception as e: # pylint: disable=broad-exception-caught
3662 logger.warning("execute_async failed, resetting cursor: %s", e)
! 3663self._reset_cursor()
! 3664raise36653666self._capture_diagnostics(ret)
3667self.last_executed_stmt=operation3668self.rowcount=ddbc_bindings.DDBCSQLRowCount(self.hstmt)Lines 3670-3680 3670column_metadata: List[Any] = []
3671try:
3672ddbc_bindings.DDBCSQLDescribeCol(self.hstmt, column_metadata)
3673self._initialize_description(column_metadata)
! 3674exceptException: # pylint: disable=broad-exception-caught3675# If describe fails, it's likely there are no results (e.g. INSERT)
! 3676self.description=None36773678ifself.description:
3679self.rowcount=-13680self._reset_rownumber()Lines 3688-3701 3688 )
3689self._cached_converter_map=self._build_converter_map()
3690self._uuid_str_indices=self._compute_uuid_str_indices()
3691else:
! 3692self.rowcount=ddbc_bindings.DDBCSQLRowCount(self.hstmt)
! 3693self._clear_rownumber()
! 3694self._cached_column_map=None
! 3695self._cached_column_map_lower=None
! 3696self._cached_converter_map=None
! 3697self._uuid_str_indices=None36983699self._reset_inputsizes()
37003701def_wrap_row_async(self, row_data: List[Any]) ->Row:Lines 3711-3721 3711 )
37123713def_wrap_rows_async(self, rows_data: List[List[Any]]) ->List[Row]:
3714"""Async-only mirror of fetchmany() / fetchall()'s Row-construction block."""
! 3715column_map, converter_map, column_map_lower=self._get_column_and_converter_maps()
! 3716uuid_idx=self._uuid_str_indices
! 3717return [
3718Row(
3719row_data,
3720column_map,
3721cursor=self,Lines 3847-3855 3847self._check_closed()
3848self._check_async_capable()
38493850ifnotself._has_result_setandself.description:
! 3851self._reset_rownumber()
38523853char_decoding=self._get_decoding_settings(ddbc_sql_const.SQL_CHAR.value)
3854wchar_decoding=self._get_decoding_settings(ddbc_sql_const.SQL_WCHAR.value)
3855char_enc=char_decoding.get("encoding", "utf-16le")Lines 3883-3897 3883 )
38843885ifret==ddbc_sql_const.SQL_NO_DATA.value:
3886ifself._next_row_index==0andself.descriptionisnotNone:
! 3887self.rowcount=03888returnNone38893890# rownumber tracking (mirrors sync fetchone)3891ifself._skip_increment_for_next_fetch:
! 3892self._skip_increment_for_next_fetch=False
! 3893self._next_row_index+=13894else:
3895self._increment_rownumber()
3896self.rowcount=self._next_row_index3897returnself._wrap_row_async(row_data)Lines 3895-3908 3895self._increment_rownumber()
3896self.rowcount=self._next_row_index3897returnself._wrap_row_async(row_data)
3898 ! 3899ifsize==-1:
3900# ---- fetchall-async ----
! 3901rows_data: List[List[Any]] = []
! 3902self._acquire_async_slot()
! 3903try:
! 3904ret=awaitloop.run_in_executor(
3905None,
3906ddbc_bindings.DDBCSQLFetchAllAsync,
3907self.hstmt,
3908rows_data,Lines 3912-3946 3912poll_initial_ms,
3913poll_max_ms,
3914 )
3915finally:
! 3916self._release_async_slot()
3917 ! 3918check_error(ddbc_sql_const.SQL_HANDLE_STMT.value, self.hstmt, ret)
3919 ! 3920ifself.hstmt:
! 3921self.messages.extend(
3922ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt)
3923 )
3924 ! 3925ifrows_dataandself._has_result_set:
! 3926self._next_row_index+=len(rows_data)
! 3927self._rownumber=self._next_row_index-13928 ! 3929iflen(rows_data) ==0andself._next_row_index==0:
! 3930self.rowcount=03931else:
! 3932self.rowcount=self._next_row_index3933 ! 3934returnself._wrap_rows_async(rows_data)
39353936# ---- fetchmany-async (size > 0 or size <= 0 sentinel) ----
! 3937ifsize<=0:
! 3938return []
! 3939rows_data= []
! 3940self._acquire_async_slot()
! 3941try:
! 3942ret=awaitloop.run_in_executor(
3943None,
3944ddbc_bindings.DDBCSQLFetchManyAsync,
3945self.hstmt,
3946rows_data,Lines 3951-3972 3951poll_initial_ms,
3952poll_max_ms,
3953 )
3954finally:
! 3955self._release_async_slot()
3956 ! 3957ifself.hstmt:
! 3958self.messages.extend(
3959ddbc_bindings.DDBCSQLGetAllDiagRecords(self.hstmt)
3960 )
3961 ! 3962ifrows_dataandself._has_result_set:
! 3963self._next_row_index+=len(rows_data)
! 3964self._rownumber=self._next_row_index-13965 ! 3966iflen(rows_data) ==0andself._next_row_index==0:
! 3967self.rowcount=03968else:
! 3969self.rowcount=self._next_row_index3970 ! 3971returnself._wrap_rows_async(rows_data)mssql_python/pybind/connection/connection.cppLines 642-655 642 }
643644// Async POC: probe SQL_ASYNC_MODE and cache the result. Returns true iff the645// driver supports statement-level async (SQL_AM_STATEMENT) or connection-level
! 646// async (SQL_AM_CONNECTION). Called from Cursor.execute_async as a capability
! 647// gate before configuring SQL_ATTR_ASYNC_ENABLE on the statement handle.648boolConnection::isAsyncCapable() const {
! 649int cached = _asyncModeCache.load(std::memory_order_acquire);
! 650if (cached < 0) {
! 651if (!_dbcHandle) {
652returnfalse;
653 }
654if (!SQLGetInfo_ptr) {
655LOG("isAsyncCapable: SQLGetInfo not initialized, loading driver");Lines 653-665 653 }
654if (!SQLGetInfo_ptr) {
655LOG("isAsyncCapable: SQLGetInfo not initialized, loading driver");
656DriverLoader::getInstance().loadDriver();
! 657 }
658SQLUSMALLINT asyncMode = 0;
! 659SQLSMALLINT outLen = 0;
! 660SQLRETURN ret = SQLGetInfo_ptr(_dbcHandle->get(), SQL_ASYNC_MODE, &asyncMode,
! 661 sizeof(asyncMode), &outLen);
662if (!SQL_SUCCEEDED(ret)) {
663LOG("isAsyncCapable: SQLGetInfo(SQL_ASYNC_MODE) failed - SQLRETURN=%d", ret);
664// Cache as SQL_AM_NONE so we don't re-probe on every call.665 _asyncModeCache.store(static_cast<int>(SQL_AM_NONE), std::memory_order_release);Lines 668-677 668 cached = static_cast<int>(asyncMode);
669 _asyncModeCache.store(cached, std::memory_order_release);
670LOG("isAsyncCapable: SQL_ASYNC_MODE=%d (cached)", cached);
671 }
! 672return cached == static_cast<int>(SQL_AM_STATEMENT) ||
! 673 cached == static_cast<int>(SQL_AM_CONNECTION);
674 }
675676boolConnectionHandle::isAsyncCapable() const {
677if (!_conn) {Lines 675-684 675676boolConnectionHandle::isAsyncCapable() const {
677if (!_conn) {
678ThrowStdException("Connection object is not initialized");
! 679 }
! 680return _conn->isAsyncCapable();
681 }
682683voidConnectionHandle::setAttr(int attribute, py::object value) {
684if (!_conn) {mssql_python/pybind/ddbc_bindings.cppLines 1845-1855 1845 }
1846 }
18471848~AsyncEnableGuard() {
! 1849if (_enabled && _hStmt && SQLSetStmtAttr_ptr) {
! 1850SQLRETURN rc = SQLSetStmtAttr_ptr(
! 1851 _hStmt, SQL_ATTR_ASYNC_ENABLE,
1852reinterpret_cast<SQLPOINTER>(static_cast<uintptr_t>(SQL_ASYNC_ENABLE_OFF)),
18530);
1854if (!SQL_SUCCEEDED(rc)) {
1855LOG("AsyncEnableGuard: SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE, OFF) "Lines 1855-1864 1855LOG("AsyncEnableGuard: SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE, OFF) "1856"failed - SQLRETURN=%d, hStmt=%p",
1857 rc, (void*)_hStmt);
1858 }
! 1859 }
! 1860 }
18611862boolenabled() const { return _enabled; }
18631864AsyncEnableGuard(const AsyncEnableGuard&) = delete;Lines 1881-1891 1881// `callOnce` is any callable returning SQLRETURN (typically a lambda that1882// captures the statement handle and calls SQLExecute / SQLExecDirect / SQLFetch).1883template <typename Fn>
1884inlineSQLRETURNExecuteWithPolling(Fn callOnce, bool asyncMode,
! 1885const PollingConfig& cfg = PollingConfig{}) {
! 1886SQLRETURN ret = callOnce();
! 1887if (!asyncMode) {
1888return ret;
1889 }
1890double sleep_ms = cfg.initial_ms;
1891while (ret == SQL_STILL_EXECUTING) {Lines 1889-1898 1889 }
1890double sleep_ms = cfg.initial_ms;
1891while (ret == SQL_STILL_EXECUTING) {
1892std::this_thread::sleep_for(
! 1893std::chrono::microseconds(static_cast<longlong>(sleep_ms * 1000.0)));
! 1894 sleep_ms = std::min(sleep_ms * cfg.multiplier, cfg.max_ms);
1895 ret = callOnce();
1896 }
1897return ret;
1898 }Lines 1899-1911 18991900// Prepares the statement (if requested) and binds all parameters. Returns1901// the SQLRETURN from the last ODBC call; on non-success the caller should1902// short-circuit before invoking SQLExecute.
! 1903//
! 1904// `paramBuffers` is an OUT parameter: it accumulates heap-owned buffers that
! 1905// MUST remain in scope until SQLExecute completes, because ODBC keeps raw
! 1906// pointers into them across the SQLBindParameter / SQLExecute boundary.
! 1907//1908// GIL: SQLPrepare is called with the GIL released (matches previous inline1909// behavior); BindParameters requires the GIL (inspects py::list contents).1910SQLRETURNPrepareAndBind(SqlHandlePtr statementHandle, SQLHANDLE hStmt, SQLWCHAR* queryPtr,
1911const py::list& params, std::vector<ParamInfo>& paramInfos,Lines 1910-1919 1910SQLRETURNPrepareAndBind(SqlHandlePtr statementHandle, SQLHANDLE hStmt, SQLWCHAR* queryPtr,
1911const py::list& params, std::vector<ParamInfo>& paramInfos,
1912 py::list& isStmtPrepared, bool usePrepare,
1913 std::vector<std::shared_ptr<void>>& paramBuffers,
! 1914const std::string& charEncoding) {
! 1915// isStmtPrepared is a single-element list carrying a bool by reference1916// (Python bools are immutable, so we can't pass the raw bool by ref).1917assert(isStmtPrepared.size() == 1);
19181919SQLRETURN rc = SQL_SUCCESS;Lines 2017-2026 2017// Release the GIL during the blocking ODBC call. In async mode2018// ExecuteWithPolling re-invokes SQLExecDirect while it returns2019// SQL_STILL_EXECUTING, with backoff sleeps between polls.2020 py::gil_scoped_release release;
! 2021 rc = ExecuteWithPolling(
! 2022 [&]() { returnSQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS); },
2023 asyncMode, pollCfg);
2024 }
2025if (!SQL_SUCCEEDED(rc) && rc != SQL_NO_DATA) {
2026LOG("SQLExecute: Direct execution failed (non-parameterized query) "Lines 2034-2046 2034if (encodingSettings.contains("encoding")) {
2035 charEncoding = encodingSettings["encoding"].cast<std::string>();
2036 }
2037 ! 2038// This vector manages the heap memory allocated for parameter buffers.
! 2039// It must remain in scope until SQLExecute (and any DAE loop) completes.2040 std::vector<std::shared_ptr<void>> paramBuffers;
! 2041 rc = PrepareAndBind(statementHandle, hStmt, queryPtr, params, paramInfos,
! 2042 isStmtPrepared, usePrepare, paramBuffers, charEncoding);
2043if (!SQL_SUCCEEDED(rc)) {
2044return rc;
2045 }Lines 2044-2057 2044return rc;
2045 }
20462047 {
! 2048// Release the GIL during the blocking SQLExecute network call. In
! 2049// async mode ExecuteWithPolling handles SQL_STILL_EXECUTING with
! 2050// backoff sleeps between polls.2051 py::gil_scoped_release release;
! 2052 rc = ExecuteWithPolling([&]() { returnSQLExecute_ptr(hStmt); },
! 2053 asyncMode, pollCfg);
2054 }
2055if (rc == SQL_NEED_DATA) {
2056LOG("SQLExecute: SQL_NEED_DATA received - Starting DAE "2057"(Data-At-Execution) loop for large parameter streaming");Lines 2197-2208 2197 py::list& isStmtPrepared, constbool usePrepare,
2198const py::dict& encodingSettings) {
2199returnSQLExecute_impl(statementHandle, query, params, paramInfos, isStmtPrepared,
2200 usePrepare, encodingSettings,
! 2201/*asyncMode=*/false, PollingConfig{});
! 2202 }
! 2203 ! 2204// Async wrapper (exposed as DDBCSQLExecuteAsync). Called from Cursor.execute_async2205// via loop.run_in_executor so the polling loop runs on a background thread with2206// the GIL released, keeping the asyncio event loop responsive.2207SQLRETURNSQLExecuteAsync_wrap(const SqlHandlePtr statementHandle, const std::u16string& query,
2208const py::list& params, std::vector<ParamInfo>& paramInfos,Lines 2210-2218 2210const py::dict& encodingSettings, double poll_initial_ms,
2211double poll_max_ms) {
2212 PollingConfig cfg;
2213 cfg.initial_ms = poll_initial_ms;
! 2214 cfg.max_ms = poll_max_ms;
2215// multiplier stays at the PollingConfig default (1.5)2216returnSQLExecute_impl(statementHandle, query, params, paramInfos, isStmtPrepared,
2217 usePrepare, encodingSettings, /*asyncMode=*/true, cfg);
2218 }Lines 4175-4186 4175 }
41764177// Fetch rows in batches4178// TODO: Move to anonymous namespace, since it is not used outside this file
! 4179//
! 4180// Async POC: when asyncMode=true, the SQLFetchScroll network call is driven
! 4181// through ExecuteWithPolling so it can return SQL_STILL_EXECUTING. The caller
! 4182// is responsible for having installed SQL_ATTR_ASYNC_ENABLE on hStmt (via4183// AsyncEnableGuard) before calling this function.4184SQLRETURNFetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& columnNames,
4185 py::list& rows, SQLUSMALLINT numCols, SQLULEN& numRowsFetched,
4186const std::vector<SQLUSMALLINT>& lobColumns,Lines 4766-4775 4766 cfg.initial_ms = poll_initial_ms;
4767 cfg.max_ms = poll_max_ms;
4768return FetchMany_impl(StatementHandle, rows, fetchSize, charEncoding, wcharEncoding,
4769 charCtype, /*asyncMode=*/true, cfg);
! 4770 }
! 47714772// GetDataVar - Progressively fetches variable-length column data using SQLGetData.4773//4774// Calls SQLGetData repeatedly, reallocating the buffer as needed, until all data is retrieved.4775// Handles both fixed-size and unknown-size (SQL_NO_TOTAL) responses from the driver.Lines 6027-6036 6027 }
60286029// Sync wrapper (exposed as DDBCSQLFetchOne). Existing pybind binding target;6030// keeps the original signature and default arguments unchanged.
! 6031SQLRETURNFetchOne_wrap(SqlHandlePtr StatementHandle, py::list& row,
! 6032const std::string& charEncoding = "utf-16le",
6033const std::string& wcharEncoding = "utf-16le",
6034int charCtype = SQL_C_WCHAR) {
6035returnFetchOne_impl(StatementHandle, row, charEncoding, wcharEncoding, charCtype,
6036/*asyncMode=*/false, PollingConfig{});📋 Files Needing Attention📉 Files with overall lowest coverage (click to expand)mssql_python.pybind.logger_bridge.cpp: 59.2%
mssql_python.pybind.ddbc_bindings.h: 59.9%
mssql_python.pybind.logger_bridge.hpp: 70.8%
mssql_python.pybind.connection.connection.cpp: 75.4%
mssql_python.pybind.ddbc_bindings.cpp: 76.4%
mssql_python.__init__.py: 77.3%
mssql_python.row.py: 77.6%
mssql_python.ddbc_bindings.py: 79.6%
mssql_python.connection.py: 83.6%
mssql_python.logging.py: 85.5%🔗 Quick Links
|
Adds the four statement-level async pybind entry points that Cursor.execute_async / Cursor.fetch_async will call from Python via loop.run_in_executor. Sync entry points (DDBCSQLExecute / DDBCSQLFetchOne|Many|All) are unchanged in signature and behavior. Refactor pattern for each of the four sync wraps: - *_wrap body extracted into static *_impl(..., bool asyncMode, const PollingConfig& pollCfg). - _impl installs AsyncEnableGuard(hStmt, asyncMode) at the top so SQL_ATTR_ASYNC_ENABLE is toggled on entry and reset on exit; the guard is a no-op when asyncMode=false. - Every SQLExecute / SQLExecDirect / SQLFetch / SQLFetchScroll call site now goes through ExecuteWithPolling(cb, asyncMode, pollCfg). Sync callers pass asyncMode=false which collapses to a straight one-shot call (semantic identity with previous inline behavior). - *_wrap becomes a thin sync forwarder that calls _impl with asyncMode=false. Existing pybind bindings and callers unchanged. - *Async_wrap is a new thin forwarder that builds PollingConfig from the Python-supplied poll_initial_ms / poll_max_ms and calls _impl with asyncMode=true. Additional plumbing: - FetchBatchData (internal helper shared by FetchMany_impl and FetchAll_impl non-LOB paths) gained defaulted asyncMode + pollCfg params; its SQLFetchScroll_ptr call is now routed through ExecuteWithPolling. All existing sync callers pass the defaults. - SQLExecute_impl rejects Data-At-Execution (isDAE=true) parameters up-front when asyncMode=true - the SQL_NEED_DATA loop is not safe to drive alongside SQL_STILL_EXECUTING polling. Callers must fall back to sync execute for large VARBINARY(MAX) / NVARCHAR(MAX) values. New pybind exports (each with trailing poll_initial_ms=0.5, poll_max_ms=20.0 float args): - DDBCSQLExecuteAsync - DDBCSQLFetchOneAsync - DDBCSQLFetchManyAsync - DDBCSQLFetchAllAsync Out of scope for POC (still sync-only): SQLGetData_wrap (LOB streaming), DDBCSQLFetchArrowBatch. Documented inline in the _impl bodies. Build: clean under -Werror -Wattributes -Wint-to-pointer-cast. All four async bindings + all existing sync bindings verified importable.
Adds the Python-facing async API on top of the C++ statement-level async
scaffolding delivered in steps 1 and 2. The change is 100% additive to
cursor.py (+483 lines, 0 removed): the sync execute / fetchone /
fetchmany / fetchall implementations are byte-identical to main and are
not touched.
Design decision: DUPLICATE, don't refactor.
- The prep and finalize logic that the async path needs is deliberately
duplicated from the sync execute() implementation instead of extracted
into shared helpers. Rationale: the sync methods are hot, well-tested,
and touching them for an experimental POC risks subtle regressions
that only surface in edge cases. Long-term de-duplication is deferred
until the async POC stabilizes.
Additions on Cursor:
- self._async_in_flight: bool guard added to __init__ (only sync-code
change, purely additive) - prevents overlapping async ops on the same
HSTMT (one Cursor == one HSTMT).
- _check_async_capable() - raises NotSupportedError when the driver
reports SQL_AM_NONE (uses cached is_async_capable() from step 1).
- _acquire_async_slot() / _release_async_slot() - raise ProgrammingError
on cursor-busy overlap.
- _prepare_execute_state_async(op, parameters, use_prepare, reset_cursor)
- deliberate duplicate of execute()'s pre-ODBC prep block.
- _finalize_execute_async(ret, operation) - deliberate duplicate of
execute()'s post-ODBC finalize block.
- _wrap_row_async(row_data) / _wrap_rows_async(rows_data) - deliberate
duplicates of the Row-construction blocks in fetchone / fetchmany /
fetchall.
Public async surface:
- async execute_async(operation, *parameters, use_prepare=True,
reset_cursor=True, poll_initial_ms=0.5, poll_max_ms=20.0) -> Cursor
offloads DDBCSQLExecuteAsync via loop.run_in_executor. asyncio is
imported lazily inside the method so sync-only users pay nothing.
- async fetch_async(size=None, *, poll_initial_ms=0.5, poll_max_ms=20.0)
-> Row | List[Row] | None
size=None -> DDBCSQLFetchOneAsync -> Row or None
size=-1 -> DDBCSQLFetchAllAsync -> List[Row]
size>0 -> DDBCSQLFetchManyAsync -> List[Row]
size<=0 -> [] (matches sync fetchmany semantics)
Concurrency contract:
- Same-cursor async ops are serialized via _async_in_flight; overlap
raises ProgrammingError('Cursor is busy...').
- Cross-cursor async ops on the same connection are allowed (each cursor
has its own HSTMT).
Not covered in this POC (documented in the tail block header):
- Data-At-Execution parameters (rejected up-front in C++).
- asyncio.CancelledError propagation (needs SQLCancelHandle).
- executemany_async / arrow_batch_async / catalog method async.
Verification:
- Diff is 100% additive (+483 / -0). Sync methods byte-identical to main.
- Module imports cleanly; all 9 new members present on Cursor.
- execute_async and fetch_async both verified as coroutine functions.
- Signatures match the spec exactly.
Not run: functional end-to-end tests (no DB in this env).…etch_async Adds tests/test_030_async_execute_fetch.py with three tests covering the Python-facing async surface introduced by the previous three commits (is_async_capable probe, DDBCSQL*Async pybind wrappers, and the public Cursor.execute_async / Cursor.fetch_async methods). Tests: 1. test_capability_probe_returns_true Sanity check on Connection.is_async_capable() - fails loudly on a regression in the C++ SQLGetInfo(SQL_ASYNC_MODE) probe rather than silently skipping the whole module. 2. test_single_execute_async_and_fetch_async End-to-end smoke: one execute_async with two parameters, one fetch_async that returns the row, and a second fetch_async that must return None. 3. test_100_concurrent_async_selects Fires 100 concurrent execute_async + fetch_async pairs through asyncio.gather with a bounded Semaphore(16). One dedicated connection per task so we don't hit SQL Server's default no-MARS restriction. Each task sends its index as a parameter and asserts the returned row matches - proves per-HSTMT isolation (no cross-wiring between concurrent statements) and that the AsyncEnableGuard toggling on/off around each call does not leak across cursors. Design decisions: - No pytest-asyncio dependency; each test uses asyncio.run(...) inline. Keeps requirements.txt unchanged. - Skips cleanly when DB_CONNECTION_STRING is unset OR the driver reports SQL_ASYNC_MODE == SQL_AM_NONE. - Concurrency knobs env-overridable for CI tuning: ASYNC_TEST_CONCURRENCY (default 100), ASYNC_TEST_MAX_INFLIGHT (default 16), ASYNC_TEST_WALL_BUDGET (default 120s). Run against SQL Server 2022 (mcr.microsoft.com/mssql/server:2022-latest) in a local Docker container: all 3 tests pass, 100 concurrent async queries complete in 0.75s wall clock.
…ntly skips) Complements test_100_concurrent_async_selects (one connection per task) with a same-connection MARS variant that opens ONE connection with MultipleActiveResultSets=Yes and creates N cursors on it. Both variants share the same correctness assertion (each task must return its own index), but exercise different bug classes: * connection-per-task proves cross-DBC async isolation and true network-level parallelism. * MARS same-connection proves per-HSTMT AsyncEnableGuard toggling doesn't cross-wire result sets between cursors that share a single DBC. Current status: the mssql-python connection-string parser rejects both MultipleActiveResultSets and MARS_Connection - neither keyword is in _ALLOWED_CONNECTION_STRING_PARAMS (mssql_python/constants.py). The test therefore skips today with a clear message pointing at the exact underlying error. When MARS is added to the allowlist the test will start running automatically with no further changes. Also added a _with_mars() helper that appends the keyword unless the caller already set MARS explicitly (either enabled or disabled). Verified against SQL Server 2022 in local Docker: 3 pass, 1 skips cleanly.
…mment While investigating why the MARS variant test skips, we prototyped adding MultipleActiveResultSets / MARS_Connection to _ALLOWED_CONNECTION_STRING_PARAMS to see if the test would pass. It did not — the combination MARS + SQL_ATTR_ASYNC_ENABLE + concurrent execute across multiple HSTMTs on one DBC (concurrency >= 2) crashes the interpreter with SIGSEGV inside the ODBC driver / C++ layer. Reproduces reliably at concurrency=2, N=5 on msodbcsql18 against SQL Server 2022. Sequential MARS-async and Semaphore=1 (serialized) both work fine — the failure is specific to multi-threaded concurrent SQLExecute on HSTMTs that share a single DBC under async mode. The MARS allowlist prototype was reverted (constants.py untouched — 0-line delta vs main). This commit updates the WARNING block in the MARS test to document the finding so a future maintainer knows why the keyword must NOT be added to the mssql-python allowlist until either: * the driver-side race is root-caused and fixed, OR * mssql-python detects a MARS DBC in execute_async and either serializes per-DBC via a lock, or raises NotSupportedError. Until then, the connection-per-task pattern (test 3) remains the supported way to run concurrent async workloads. Test suite behavior unchanged: 3 pass, 1 skip.
Adds a second wave of functional tests exercising the async surface across long queries, large result sets, event-loop non-blocking behavior, multi-batch fetches, and sequential vs concurrent invocation patterns. All safe patterns use one connection per concurrent task (see root-cause investigation on why cross-cursor concurrent async on a shared DBC is unsafe until per-Connection serialization is added). New tests: Case 1: test_execute_async_long_running_query 2-second server-side WAITFOR through execute_async; verifies the polling loop survives many SQL_STILL_EXECUTING iterations without losing the result set. Case 2: test_event_loop_progresses_during_execute_async Background heartbeat coroutine (10ms tick) runs while a 2s WAITFOR query executes. Requires >= 60 ticks. Local run: 185 ticks. Proves the event loop is NOT starved by the polling loop. Case 4: test_1000_async_executes_on_different_connections Scale test_100_concurrent_async_selects up to 1000 tasks. Marked @pytest.mark.stress (excluded from default run per pytest.ini). Local run: 1000 queries in 2.44s under Semaphore(16). Case 6: test_fetch_async_returns_batch_of_rows fetch_async(size=5) returns exactly 5 Row objects with correct values. Case 7: test_fetch_async_large_result_set 5000-row SELECT via CROSS JOIN of sys.all_objects; fetch_async(-1) returns all rows with correct sequential numbering. Case 8: test_event_loop_progresses_during_fetch_async Symmetric to Case 2 but on the fetch path. Uses HB_FETCH_ROWS (default 50000) so the fetch is long enough for meaningful heartbeat ticks. Skips inconclusively if the fetch finishes in <50ms (fast local DB). Case 9: test_multiple_concurrent_execute_async_small_batch 10-task variant of test_100_concurrent_async_selects; kept small for quick smoke runs. Case 10: test_execute_async_followed_by_fetch_async All three fetch_async modes (size=None, positive int, -1) on the same cursor after independent executes. Case 11: test_multiple_concurrent_fetch_async_across_connections Two-phase: (1) N=20 connections each execute a query sequentially, (2) fetch_async fired concurrently across all N via asyncio.gather. Verifies the fetch code path under concurrent gather. Case 12: test_sequential_execute_async_on_same_cursor Five sequential execute_async calls on the same cursor. Verifies that AsyncEnableGuard correctly toggles SQL_ATTR_ASYNC_ENABLE OFF at the end of each call so subsequent executes are not affected by leftover state. Env-overridable knobs added: ASYNC_TEST_LARGE_ROWS rows for the large-result-set test (5000) ASYNC_TEST_HB_FETCH_ROWS rows for fetch-heartbeat test (50000) ASYNC_TEST_STRESS_CONCURRENCY N for the 1000-task stress test (1000) ASYNC_TEST_WAITFOR_SECONDS server-side delay for long-query tests (2) Verified against SQL Server 2022 in local Docker: 12 passed, 1 skipped (MARS), 1 deselected (stress) in 4.58s; stress test explicitly: 1 passed in 2.47s.
…tability test Root-cause fix for the MARS variant test that was skipping. Investigation found that the SQL Server-native connection-string keyword 'MultipleActiveResultSets' is SILENTLY IGNORED by the bundled msodbcsql18 driver (empirically verified: keyword is accepted but MARS is never actually enabled). Only the ODBC-standard alias 'MARS_Connection' turns MARS on. The earlier segfault we saw when prototyping MARS in the allowlist was because MARS was silently off — 100 concurrent async statements on a non-MARS DBC hit the 'connection busy' error path, and concurrent DDBCSQLCheckError calls raced on shared DBC diagnostic state. With MARS genuinely on (this commit), concurrent async on shared DBC works cleanly. 100 truly-concurrent execute_async + fetch_async pairs on ONE MARS connection completes in ~130ms with no crash, no error, no cross-wiring. Verified across 10 iterations on the same connection. Changes: mssql_python/constants.py (+13 lines) Adds "mars_connection": "MARS_Connection" to _ALLOWED_CONNECTION_STRING_PARAMS. Detailed comment warns future maintainers NOT to add 'multipleactiveresultsets' without confirming driver honors it (currently doesn't). This also unblocks MARS for sync code — users can now do multiple active cursors on one connection with 'MARS_Connection=Yes'. tests/test_030_async_execute_fetch.py (+193/-35) * _with_mars() helper now appends 'MARS_Connection=Yes' (the working keyword) instead of 'MultipleActiveResultSets=Yes' (silently ignored). * Block comment above test_100_concurrent_async_selects_on_single_mars_connection rewritten: replaces the outdated 'MARS segfault' warning with a history / root-cause fix note explaining what was actually broken. * Test docstring updated to remove 'skipped today' language. * NEW test_mars_stability_100_cursors_high_concurrency_multi_iteration (@pytest.mark.stress): 10 iterations of 100 concurrent async cursors on ONE MARS connection, no semaphore, verifying per-iteration correctness AND long-term stability. Local run: 1.33s total, per-iteration timing consistent 126-160ms across all 10 rounds (no drift, no leak, no cross-wiring). * Env knobs: ASYNC_TEST_MARS_STABILITY_N (default 100), ASYNC_TEST_MARS_STABILITY_ITERS (default 10). Key finding for future work: MARS gives client-side result-set multiplexing over ONE TCP socket, but SQL Server assigns one worker thread per session (SPID). So N concurrent statements on one MARS connection queue up server-side, not truly parallel. Real network-level parallelism still requires N separate connections (test_100_concurrent_async_selects pattern). Verified against SQL Server 2022 in local Docker: 15/15 tests pass (13 non-stress + 2 stress) in 10.17s total. MARS variant test: 100 concurrent async in 0.13s (was skipping). MARS stability test: 10 iters × 100 cursors in 1.33s (new).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for freeto join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This pull request adds support for detecting and caching ODBC driver async capabilities, which is a foundational step for enabling asynchronous operations (such as
execute_async) on SQL Server connections. The changes ensure that the driver’s support for asynchronous execution is probed once per connection and cached for efficient future checks. Additionally, the connection string parsing is updated to clarify and safely handle the enabling of Multiple Active Result Sets (MARS).Async capability detection and caching:
_asyncModeCache) in theConnectionclass to store the result of probingSQLGetInfo(SQL_ASYNC_MODE), minimizing redundant driver queries and improving performance for async operations.isAsyncCapable()method in bothConnectionandConnectionHandle, which checks and caches whether the driver supports statement-level or connection-level async, and is used as a capability gate before enabling async features. [1][2][3]Connection string handling improvements:
constants.pyto explicitly document and support the ODBC-standardMARS_Connectionparameter, while warning against using the SQL Server-nativeMultipleActiveResultSetsalias unless driver support is confirmed, preventing user confusion and unpredictable behavior.Work Item / Issue Reference
Summary