ASYNC Probe#676
Draft
subrata-ms wants to merge 1 commit into
Draft
Conversation
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| """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 these
655 // numeric infotypes we want the value as an int, decoded on the C++ side.
! 656 // -----------------------------------------------------------------------------
657
658 // 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_FUNCTIONS 10005
! 663 #endif
! 664
! 665 py::dict Connection::getAsyncCapability() const {
! 666 if (!_dbcHandle) {
! 667 ThrowStdException("Connection handle not allocated");
668 }
! 669 if (!SQLGetInfo_ptr || !SQLAllocHandle_ptr || !SQLSetStmtAttr_ptr ||
! 670 !SQLFreeHandle_ptr) {
671 LOG("getAsyncCapability: driver function pointers not initialized, loading driver");
672 DriverLoader::getInstance().loadDriver();
! 673 }
! 674
! 675 SQLHDBC hDbc = _dbcHandle->get();
! 676 py::dict result;
! 677
! 678 // --- SQL_ASYNC_MODE (SQLUSMALLINT) -----------------------------------
! 679 {
! 680 SQLUSMALLINT asyncMode = 0;
! 681 SQLSMALLINT outLen = 0;
! 682 SQLRETURN ret =
! 683 SQLGetInfo_ptr(hDbc, SQL_ASYNC_MODE, &asyncMode, sizeof(asyncMode), &outLen);
! 684 if (SQL_SUCCEEDED(ret)) {
! 685 result["async_mode"] = static_cast<unsigned int>(asyncMode);
! 686 const char* name = "SQL_AM_NONE";
! 687 if (asyncMode == SQL_AM_CONNECTION) {
! 688 name = "SQL_AM_CONNECTION";
! 689 } else if (asyncMode == SQL_AM_STATEMENT) {
! 690 name = "SQL_AM_STATEMENT";
! 691 }
! 692 result["async_mode_name"] = std::string(name);
693 } else {
694 LOG("getAsyncCapability: SQLGetInfo(SQL_ASYNC_MODE) failed - "
695 "SQLRETURN=%d",
696 ret);Lines 694-727 694 LOG("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 {
! 707 SQLUINTEGER stmtFns = 0;
! 708 SQLSMALLINT outLen = 0;
! 709 SQLRETURN ret = SQLGetInfo_ptr(hDbc, SQL_ASYNC_STMT_FUNCTIONS, &stmtFns,
710 sizeof(stmtFns), &outLen);
711 if (SQL_SUCCEEDED(ret)) {
! 712 result["async_stmt_functions_bitmask"] = static_cast<unsigned int>(stmtFns);
! 713 } else {
! 714 LOG("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 {
! 723 SQLUINTEGER dbcFns = 0;
724 SQLSMALLINT outLen = 0;
725 SQLRETURN ret =
726 SQLGetInfo_ptr(hDbc, SQL_ASYNC_DBC_FUNCTIONS, &dbcFns, sizeof(dbcFns), &outLen);
727 if (SQL_SUCCEEDED(ret)) {Lines 726-747 726 SQLGetInfo_ptr(hDbc, SQL_ASYNC_DBC_FUNCTIONS, &dbcFns, sizeof(dbcFns), &outLen);
727 if (SQL_SUCCEEDED(ret)) {
728 result["async_dbc_functions_bitmask"] = static_cast<unsigned int>(dbcFns);
729 } else {
! 730 LOG("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 {
! 743 SQLUINTEGER notify = 0;
744 SQLSMALLINT outLen = 0;
745 SQLRETURN ret =
746 SQLGetInfo_ptr(hDbc, SQL_ASYNC_NOTIFICATION, ¬ify, sizeof(notify), &outLen);
747 if (SQL_SUCCEEDED(ret)) {Lines 751-856 751 LOG("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
! 773 if (!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 {
! 778 SQLHANDLE hStmt = nullptr;
! 779 SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, hDbc, &hStmt);
780 if (!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);
! 788 if (!SQL_SUCCEEDED(ret)) {
! 789 smoke["error"] =
! 790 std::string("SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE) failed");
! 791 SQLFreeHandle_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 =
! 797 u"WAITFOR DELAY '00:00:01'; SELECT 1 AS async_probe";
! 798 SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(queryStr);
! 799
! 800 // --- SQLExecDirect polling loop -----------------------
! 801 using Clock = std::chrono::steady_clock;
! 802 auto execStart = Clock::now();
! 803 unsigned int execPollCount = 0;
! 804 SQLRETURN execRet;
! 805 {
! 806 py::gil_scoped_release release;
807 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
808 while (execRet == SQL_STILL_EXECUTING) {
809 ++execPollCount;
! 810 std::this_thread::sleep_for(std::chrono::milliseconds(10));
811 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
812 }
! 813 }
! 814 auto execMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 815 Clock::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<long long>(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 ----------------------------
! 827 if (SQL_SUCCEEDED(execRet)) {
! 828 auto fetchStart = Clock::now();
! 829 unsigned int fetchPollCount = 0;
! 830 SQLRETURN fetchRet;
! 831 {
! 832 py::gil_scoped_release release;
! 833 fetchRet = SQLFetch_ptr(hStmt);
! 834 while (fetchRet == SQL_STILL_EXECUTING) {
! 835 ++fetchPollCount;
! 836 std::this_thread::sleep_for(std::chrono::milliseconds(5));
! 837 fetchRet = SQLFetch_ptr(hStmt);
! 838 }
! 839 }
! 840 auto fetchMs = std::chrono::duration_cast<std::chrono::milliseconds>(
841 Clock::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<long long>(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 }
855
856 smoke["ran"] = true;Lines 861-930 861 SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
862 result["polling_smoke_test"] = smoke;
863 }
864 }
! 865 }
! 866 }
! 867
868 // --- 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
! 883 if (!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 {
! 889 SQLHANDLE hStmt = nullptr;
! 890 SQLRETURN ret = SQLAllocHandle_ptr(SQL_HANDLE_STMT, hDbc, &hStmt);
! 891 if (!SQL_SUCCEEDED(ret) || hStmt == nullptr) {
! 892 fetchStream["error"] =
893 std::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);
901 if (!SQL_SUCCEEDED(ret)) {
902 fetchStream["error"] =
903 std::string("SQLSetStmtAttr(SQL_ATTR_ASYNC_ENABLE) failed");
! 904 SQLFreeHandle_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 "
! 910 u"CAST(a.object_id AS BIGINT) AS id, "
! 911 u"CAST(a.name AS NVARCHAR(128)) AS n, "
! 912 u"CAST(a.create_date AS DATETIME2) AS cd "
! 913 u"FROM sys.all_objects a CROSS JOIN sys.all_objects b";
! 914 SQLWCHAR* queryPtr = reinterpretU16stringAsSqlWChar(queryStr);
! 915
! 916 // Poll the execute first — this may or may not observe
! 917 // STILL_EXECUTING depending on server plan-cache state.
! 918 using Clock = std::chrono::steady_clock;
! 919 auto execStart = Clock::now();
! 920 unsigned int execPollCount = 0;
! 921 SQLRETURN execRet;
! 922 {
! 923 py::gil_scoped_release release;
! 924 execRet = SQLExecDirect_ptr(hStmt, queryPtr, SQL_NTS);
925 while (execRet == SQL_STILL_EXECUTING) {
! 926 ++execPollCount;
927 std::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 }
931 auto execMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 932 Clock::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"] =
! 938 static_cast<long long>(execMs);
! 939
! 940 if (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're
944 // only measuring whether the fetch phase EVER blocks
945 // on the network and how often.
! 946 auto fetchStart = Clock::now();
! 947 unsigned int rowsRead = 0;
! 948 unsigned int totalPollCount = 0;
! 949 unsigned int rowsThatPolled = 0;
! 950 SQLRETURN fetchRet = SQL_SUCCESS;
! 951 {
! 952 py::gil_scoped_release release;
! 953 while (true) {
! 954 unsigned int rowPolls = 0;
! 955 fetchRet = SQLFetch_ptr(hStmt);
! 956 while (fetchRet == SQL_STILL_EXECUTING) {
! 957 ++rowPolls;
! 958 // 500us — tight enough to catch short
! 959 // network waits without spinning.
! 960 std::this_thread::sleep_for(
! 961 std::chrono::microseconds(500));
! 962 fetchRet = SQLFetch_ptr(hStmt);
! 963 }
! 964 if (fetchRet == SQL_NO_DATA) {
! 965 break;
! 966 }
! 967 if (!SQL_SUCCEEDED(fetchRet)) {
! 968 break;
! 969 }
! 970 ++rowsRead;
! 971 totalPollCount += rowPolls;
! 972 if (rowPolls > 0) {
! 973 ++rowsThatPolled;
! 974 }
! 975 }
! 976 }
! 977 auto fetchMs = std::chrono::duration_cast<std::chrono::milliseconds>(
! 978 Clock::now() - fetchStart)
! 979 .count();
! 980 fetchStream["fetch_final_sqlreturn"] =
! 981 static_cast<int>(fetchRet);
! 982 fetchStream["fetch_ok"] =
! 983 (fetchRet == SQL_NO_DATA) ||
! 984 SQL_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"] =
989 static_cast<long long>(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
! 1004 SQLFreeStmt_ptr(hStmt, SQL_CLOSE);
! 1005 SQLFreeHandle_ptr(SQL_HANDLE_STMT, hStmt);
! 1006 result["polling_fetch_stream_test"] = fetchStream;
1007 }
1008 }
1009 }
! 1010 }
! 1011
1012 return result;
1013 }
1014
1015 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 free
to 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