Uh oh!
There was an error while loading. Please reload this page.
ASYNC Probe - #676
Draft
Subrata (subrata-ms) wants to merge 1 commit into
Draft
Conversation
| """Probe the ODBC driver's async capability and verify polling works end-to-end. | ||
| Run: | ||
| export DB_CONNECTION_STRING="Server=localhost,1433;Database=master;UID=sa;PWD=...;TrustServerCertificate=Yes;Encrypt=Yes" |
| "Ratios BELOW ~0.5x would indicate a driver-side global lock — " | ||
| "that is what we") | ||
| _row(" ", | ||
| "actually gate on. Any value between 0.5x and ~1.5x on localhost " |
📊 Code Coverage Report
Diff CoverageDiff: main...HEAD, staged and unstaged changes
Summary
mssql_python/pybind/connection/connection.cppLines 652-696 652// enabling SQL_ATTR_ASYNC_ENABLE on a fresh statement handle.653//654// The existing generic Connection::getInfo returns raw bytes; for these655// numeric infotypes we want the value as an int, decoded on the C++ side.
! 656// -----------------------------------------------------------------------------657658// SQL_ASYNC_STMT_FUNCTIONS is defined by the Microsoft ODBC headers but is
! 659// missing from some unixODBC releases. The numeric ID is stable across the
! 660// spec (10005), so provide a local fallback.
! 661 #ifndef SQL_ASYNC_STMT_FUNCTIONS
! 662 #define SQL_ASYNC_STMT_FUNCTIONS10005
! 663 #endif
! 664 ! 665 py::dict Connection::getAsyncCapability() const {
! 666if (!_dbcHandle) {
! 667ThrowStdException("Connection handle not allocated");
668 }
! 669if (!SQLGetInfo_ptr || !SQLAllocHandle_ptr || !SQLSetStmtAttr_ptr ||
! 670 !SQLFreeHandle_ptr) {
671LOG("getAsyncCapability: driver function pointers not initialized, loading driver");
672DriverLoader::getInstance().loadDriver();
! 673 }
! 674 ! 675SQLHDBC hDbc = _dbcHandle->get();
! 676 py::dict result;
! 677 ! 678// --- SQL_ASYNC_MODE (SQLUSMALLINT) -----------------------------------
! 679 {
! 680SQLUSMALLINT asyncMode = 0;
! 681SQLSMALLINT outLen = 0;
! 682SQLRETURN ret =
! 683SQLGetInfo_ptr(hDbc, SQL_ASYNC_MODE, &asyncMode, sizeof(asyncMode), &outLen);
! 684if (SQL_SUCCEEDED(ret)) {
! 685 result["async_mode"] = static_cast<unsignedint>(asyncMode);
! 686constchar* name = "SQL_AM_NONE";
! 687if (asyncMode == SQL_AM_CONNECTION) {
! 688 name = "SQL_AM_CONNECTION";
! 689 } elseif (asyncMode == SQL_AM_STATEMENT) {
! 690 name = "SQL_AM_STATEMENT";
! 691 }
! 692 result["async_mode_name"] = std::string(name);
693 } else {
694LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_MODE) failed - "695"SQLRETURN=%d",
696 ret);Lines 694-727 694LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_MODE) failed - "695"SQLRETURN=%d",
696 ret);
697 result["async_mode"] = py::none();
! 698 result["async_mode_name"] = py::none();
! 699 }
! 700 }
! 701 ! 702// --- SQL_ASYNC_STMT_FUNCTIONS (SQLUINTEGER bitmask) ------------------
! 703// We return the raw bitmask value. Interpreting the bitmask across
! 704// drivers is unreliable, so the functional check below is the real
! 705// gate. This value is exposed for informational logging only.
! 706 {
! 707SQLUINTEGER stmtFns = 0;
! 708SQLSMALLINT outLen = 0;
! 709SQLRETURN ret = SQLGetInfo_ptr(hDbc, SQL_ASYNC_STMT_FUNCTIONS, &stmtFns,
710sizeof(stmtFns), &outLen);
711if (SQL_SUCCEEDED(ret)) {
! 712 result["async_stmt_functions_bitmask"] = static_cast<unsignedint>(stmtFns);
! 713 } else {
! 714LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_STMT_FUNCTIONS) not "
! 715"reported by driver - SQLRETURN=%d",
! 716 ret);
! 717 result["async_stmt_functions_bitmask"] = py::none();
! 718 }
! 719 }
! 720 ! 721// --- SQL_ASYNC_DBC_FUNCTIONS (SQLUINTEGER bitmask, ODBC 3.8+) --------
! 722 {
! 723SQLUINTEGER dbcFns = 0;
724SQLSMALLINT outLen = 0;
725SQLRETURN ret =
726SQLGetInfo_ptr(hDbc, SQL_ASYNC_DBC_FUNCTIONS, &dbcFns, sizeof(dbcFns), &outLen);
727if (SQL_SUCCEEDED(ret)) {Lines 726-747 726SQLGetInfo_ptr(hDbc, SQL_ASYNC_DBC_FUNCTIONS, &dbcFns, sizeof(dbcFns), &outLen);
727if (SQL_SUCCEEDED(ret)) {
728 result["async_dbc_functions_bitmask"] = static_cast<unsignedint>(dbcFns);
729 } else {
! 730LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_DBC_FUNCTIONS) not "
! 731"supported by driver - SQLRETURN=%d",
! 732 ret);
! 733 result["async_dbc_functions_bitmask"] = py::none();
! 734 }
! 735 }
! 736 ! 737// --- SQL_ASYNC_NOTIFICATION (SQLUINTEGER, ODBC 3.8+) -----------------
! 738// Returns SQL_ASYNC_NOTIFICATION_CAPABLE (1) or SQL_ASYNC_NOTIFICATION_NOT_CAPABLE (0).
! 739// Tells us whether the driver supports the event-driven / callback
! 740// completion path (SQL_ATTR_ASYNC_STMT_EVENT / SQL_ATTR_ASYNC_STMT_PCALLBACK)
! 741// as an alternative to polling.
! 742 {
! 743SQLUINTEGER notify = 0;
744SQLSMALLINT outLen = 0;
745SQLRETURN ret =
746SQLGetInfo_ptr(hDbc, SQL_ASYNC_NOTIFICATION, ¬ify, sizeof(notify), &outLen);
747if (SQL_SUCCEEDED(ret)) {Lines 751-856 751LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_NOTIFICATION) not "752"reported by driver - SQLRETURN=%d",
753 ret);
754 result["async_notification"] = py::none();
! 755 result["async_notification_capable"] = py::none();
! 756 }
! 757 }
758 ! 759// --- Functional smoke test: run WAITFOR + SELECT under polling --------
! 760// This is the real gate. We allocate an HSTMT, enable statement-level
! 761// async, and drive a short server-side delay through the polling loop:
! 762// 1. SQLExecDirect returns SQL_STILL_EXECUTING while the server sleeps
! 763// 2. Re-invoke SQLExecDirect until it returns SQL_SUCCESS
! 764// 3. SQLFetch may also return SQL_STILL_EXECUTING; poll it too
! 765// A pass here proves polling works end-to-end on this OS + driver.
! 766// mssql-python loads msodbcsql directly via dlopen / LoadLibraryW (see
! 767// LoadDriverLibrary in ddbc_bindings.cpp), so this test is independent
! 768// of any ODBC Driver Manager (unixODBC / iODBC / Windows DM).
! 769 {
! 770 py::dict smoke;
! 771 smoke["ran"] = false;
! 772 ! 773if (!SQLExecDirect_ptr || !SQLFetch_ptr || !SQLFreeStmt_ptr) {
! 774 smoke["error"] = std::string(
! 775"polling smoke test skipped: required driver function pointers not initialized");
! 776 result["polling_smoke_test"] = smoke;
! 777 } else {
! 778SQLHANDLE hStmt = nullptr;
! 779SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, hDbc, &hStmt);
780if (!SQL_SUCCEEDED(ret) || hStmt == nullptr) {
781 smoke["error"] = std::string("SQLAllocHandle(SQL_HANDLE_STMT) failed");
! 782 smoke["sqlreturn"] = static_cast<int>(ret);
! 783 result["polling_smoke_test"] = smoke;
! 784 } else {
785 ret = SQLSetStmtAttr_ptr(hStmt, SQL_ATTR_ASYNC_ENABLE,
786 (SQLPOINTER)(uintptr_t)SQL_ASYNC_ENABLE_ON, 0);
! 787 smoke["set_async_enable_sqlreturn"] = static_cast<int>(ret);
! 788if (!SQL_SUCCEEDED(ret)) {
! 789 smoke["error"] =
! 790std::string("SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE) failed");
! 791SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
! 792 result["polling_smoke_test"] = smoke;
! 793 } else {
! 794// 1-second server-side sleep guarantees the initial call
! 795// returns SQL_STILL_EXECUTING at least once.
! 796 std::u16string queryStr =
! 797u"WAITFOR DELAY '00:00:01'; SELECT 1 AS async_probe";
! 798SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(queryStr);
! 799 ! 800// --- SQLExecDirect polling loop -----------------------
! 801using Clock = std::chrono::steady_clock;
! 802auto execStart = Clock::now();
! 803unsignedint execPollCount = 0;
! 804SQLRETURN execRet;
! 805 {
! 806 py::gil_scoped_release release;
807 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
808while (execRet == SQL_STILL_EXECUTING) {
809 ++execPollCount;
! 810std::this_thread::sleep_for(std::chrono::milliseconds(10));
811 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
812 }
! 813 }
! 814auto execMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 815Clock::now() - execStart)
! 816 .count();
! 817 smoke["execute_sqlreturn"] = static_cast<int>(execRet);
! 818 smoke["execute_ok"] = SQL_SUCCEEDED(execRet);
! 819 smoke["execute_poll_count"] = execPollCount;
! 820 smoke["execute_elapsed_ms"] = static_cast<longlong>(execMs);
! 821// observed_still_executing is the critical evidence: if
! 822// this is false, either the driver blocked internally or
! 823// the query completed too quickly (unlikely with WAITFOR).
! 824 smoke["execute_observed_still_executing"] = (execPollCount > 0);
! 825 ! 826// --- SQLFetch polling loop ----------------------------
! 827if (SQL_SUCCEEDED(execRet)) {
! 828auto fetchStart = Clock::now();
! 829unsignedint fetchPollCount = 0;
! 830SQLRETURN fetchRet;
! 831 {
! 832 py::gil_scoped_release release;
! 833 fetchRet = SQLFetch_ptr(hStmt);
! 834while (fetchRet == SQL_STILL_EXECUTING) {
! 835 ++fetchPollCount;
! 836std::this_thread::sleep_for(std::chrono::milliseconds(5));
! 837 fetchRet = SQLFetch_ptr(hStmt);
! 838 }
! 839 }
! 840auto fetchMs = std::chrono::duration_cast<std::chrono::milliseconds>(
841Clock::now() - fetchStart)
! 842 .count();
843 smoke["fetch_sqlreturn"] = static_cast<int>(fetchRet);
844 smoke["fetch_ok"] = SQL_SUCCEEDED(fetchRet);
845 smoke["fetch_poll_count"] = fetchPollCount;
! 846 smoke["fetch_elapsed_ms"] = static_cast<longlong>(fetchMs);
! 847 smoke["fetch_observed_still_executing"] = (fetchPollCount > 0);
! 848 } else {
! 849 smoke["fetch_sqlreturn"] = py::none();
! 850 smoke["fetch_ok"] = false;
! 851 smoke["fetch_poll_count"] = py::none();
! 852 smoke["fetch_elapsed_ms"] = py::none();
853 smoke["fetch_observed_still_executing"] = py::none();
854 }
855856 smoke["ran"] = true;Lines 861-930 861SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
862 result["polling_smoke_test"] = smoke;
863 }
864 }
! 865 }
! 866 }
! 867868// --- Fetch-stream smoke test ------------------------------------------
! 869// The polling smoke test above uses a query that returns a single tiny
! 870// row — SQLFetch completes instantly from the TCP receive buffer with
! 871// zero polling, so it does NOT exercise async fetch. This second test
! 872// deliberately streams a multi-megabyte result set so that SQLFetch has
! 873// to wait on TDS packet arrivals from the network. That is where
! 874// SQL_STILL_EXECUTING on SQLFetch actually shows up in practice.
! 875//
! 876// Query is deliberately server-heavy: TOP 50000 rows from a cross-join
! 877// of sys.all_objects, with three moderately wide columns per row. On a
! 878// stock master DB this yields several MB and spans ~1000 TDS packets.
! 879 {
! 880 py::dict fetchStream;
! 881 fetchStream["ran"] = false;
! 882 ! 883if (!SQLExecDirect_ptr || !SQLFetch_ptr || !SQLFreeStmt_ptr) {
! 884 fetchStream["error"] = std::string(
! 885"fetch-stream test skipped: required driver function pointers "
! 886"not initialized");
! 887 result["polling_fetch_stream_test"] = fetchStream;
! 888 } else {
! 889SQLHANDLE hStmt = nullptr;
! 890SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, hDbc, &hStmt);
! 891if (!SQL_SUCCEEDED(ret) || hStmt == nullptr) {
! 892 fetchStream["error"] =
893std::string("SQLAllocHandle(SQL_HANDLE_STMT) failed");
894 fetchStream["sqlreturn"] = static_cast<int>(ret);
! 895 result["polling_fetch_stream_test"] = fetchStream;
! 896 } else {
! 897 ret = SQLSetStmtAttr_ptr(
! 898 hStmt, SQL_ATTR_ASYNC_ENABLE,
! 899 (SQLPOINTER)(uintptr_t)SQL_ASYNC_ENABLE_ON, 0);
! 900 fetchStream["set_async_enable_sqlreturn"] = static_cast<int>(ret);
901if (!SQL_SUCCEEDED(ret)) {
902 fetchStream["error"] =
903std::string("SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE) failed");
! 904SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
! 905 result["polling_fetch_stream_test"] = fetchStream;
! 906 } else {
! 907// ~50000 rows × 3 columns from a cross-join. Uses only
! 908// system catalog views so it runs on any SQL Server DB.
! 909 std::u16string queryStr = u"SELECT TOP 50000 "
! 910u"CAST(a.object_id AS BIGINT) AS id, "
! 911u"CAST(a.name AS NVARCHAR(128)) AS n, "
! 912u"CAST(a.create_date AS DATETIME2) AS cd "
! 913u"FROM sys.all_objects a CROSS JOIN sys.all_objects b";
! 914SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(queryStr);
! 915 ! 916// Poll the execute first — this may or may not observe
! 917// STILL_EXECUTING depending on server plan-cache state.
! 918using Clock = std::chrono::steady_clock;
! 919auto execStart = Clock::now();
! 920unsignedint execPollCount = 0;
! 921SQLRETURN execRet;
! 922 {
! 923 py::gil_scoped_release release;
! 924 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
925while (execRet == SQL_STILL_EXECUTING) {
! 926 ++execPollCount;
927std::this_thread::sleep_for(std::chrono::milliseconds(10));
928 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
929 }
930 }Lines 928-1015 928 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
929 }
930 }
931auto execMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 932Clock::now() - execStart)
! 933 .count();
! 934 fetchStream["execute_sqlreturn"] = static_cast<int>(execRet);
! 935 fetchStream["execute_ok"] = SQL_SUCCEEDED(execRet);
! 936 fetchStream["execute_poll_count"] = execPollCount;
! 937 fetchStream["execute_elapsed_ms"] =
! 938static_cast<longlong>(execMs);
! 939 ! 940if (SQL_SUCCEEDED(execRet)) {
! 941// Stream all rows. For each SQLFetch call, count how
! 942// many STILL_EXECUTING returns we saw before it
! 943// completed. Rows are not bound/materialized — we're944// only measuring whether the fetch phase EVER blocks945// on the network and how often.
! 946auto fetchStart = Clock::now();
! 947unsignedint rowsRead = 0;
! 948unsignedint totalPollCount = 0;
! 949unsignedint rowsThatPolled = 0;
! 950SQLRETURN fetchRet = SQL_SUCCESS;
! 951 {
! 952 py::gil_scoped_release release;
! 953while (true) {
! 954unsignedint rowPolls = 0;
! 955 fetchRet = SQLFetch_ptr(hStmt);
! 956while (fetchRet == SQL_STILL_EXECUTING) {
! 957 ++rowPolls;
! 958// 500us — tight enough to catch short
! 959// network waits without spinning.
! 960std::this_thread::sleep_for(
! 961std::chrono::microseconds(500));
! 962 fetchRet = SQLFetch_ptr(hStmt);
! 963 }
! 964if (fetchRet == SQL_NO_DATA) {
! 965break;
! 966 }
! 967if (!SQL_SUCCEEDED(fetchRet)) {
! 968break;
! 969 }
! 970 ++rowsRead;
! 971 totalPollCount += rowPolls;
! 972if (rowPolls > 0) {
! 973 ++rowsThatPolled;
! 974 }
! 975 }
! 976 }
! 977auto fetchMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 978Clock::now() - fetchStart)
! 979 .count();
! 980 fetchStream["fetch_final_sqlreturn"] =
! 981static_cast<int>(fetchRet);
! 982 fetchStream["fetch_ok"] =
! 983 (fetchRet == SQL_NO_DATA) ||
! 984SQL_SUCCEEDED(fetchRet);
! 985 fetchStream["fetch_rows_read"] = rowsRead;
! 986 fetchStream["fetch_total_poll_count"] = totalPollCount;
987 fetchStream["fetch_rows_that_polled"] = rowsThatPolled;
! 988 fetchStream["fetch_elapsed_ms"] =
989static_cast<longlong>(fetchMs);
! 990 fetchStream["fetch_observed_still_executing"] =
! 991 (totalPollCount > 0);
! 992 } else {
! 993 fetchStream["fetch_final_sqlreturn"] = py::none();
! 994 fetchStream["fetch_ok"] = false;
! 995 fetchStream["fetch_rows_read"] = py::none();
! 996 fetchStream["fetch_total_poll_count"] = py::none();
997 fetchStream["fetch_rows_that_polled"] = py::none();
! 998 fetchStream["fetch_elapsed_ms"] = py::none();
! 999 fetchStream["fetch_observed_still_executing"] = py::none();
1000 }
! 1001 ! 1002 fetchStream["ran"] = true;
! 1003 ! 1004SQLFreeStmt_ptr(hStmt, SQL_CLOSE);
! 1005SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
! 1006 result["polling_fetch_stream_test"] = fetchStream;
1007 }
1008 }
1009 }
! 1010 }
! 10111012return result;
1013 }
10141015 py::dict ConnectionHandle::getAsyncCapability() const {📋 Files Needing Attention📉 Files with overall lowest coverage (click to expand)mssql_python.pybind.connection.connection.cpp: 46.5%
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.ddbc_bindings.cpp: 76.2%
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
|
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
2 participants
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.
Work Item / Issue Reference
Summary