diff --git a/docs/api-reference.md b/docs/api-reference.md index b0194ca..b28436b 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -60,9 +60,11 @@ dmPython.connect( - `host` 与 `server` 互斥（只允许设置一个）。 - `user` 支持 `user/password@server:port[/schema][?catalog=...]` 形式。 -- `login_timeout` 以秒为单位，限制首次建连握手；`connection_timeout` 以秒为单位，传给底层 TCP 拨号超时。两者均不限制 SQL 执行时间。 +- `login_timeout` 以毫秒为单位，默认 5000，限制首次建连握手；设为 0 表示不限制。`connection_timeout` 以秒为单位，默认 0 不限制，限制 SQL 执行时间。 - 常量参数建议使用模块常量（如 `DSQL_AUTOCOMMIT_ON`、`ISO_LEVEL_READ_COMMITTED`）。 +超时单位与默认值依据[达梦 dmPython 接口文档](https://eco.dameng.com/document/dm/zh-cn/pm/dmpython-interface)和 [DPI 编程指南](https://eco.dameng.com/document/dm/zh-cn/pm/dpi-rogramming-guide)。 + ### 模块函数 - `DateFromTicks(ticks)` diff --git a/docs/test-results/2026-09-26-type-connection-matrix.md b/docs/test-results/2026-09-26-type-connection-matrix.md index f638862..173d23d 100644 --- a/docs/test-results/2026-09-26-type-connection-matrix.md +++ b/docs/test-results/2026-09-26-type-connection-matrix.md @@ -36,7 +36,7 @@ `INTERVAL DAY TO SECOND` 现可与 Python `datetime.timedelta` 往返，包括负数、微秒、零值和负 10 万天；读取时描述类型为 `dmPython.INTERVAL`。`INTERVAL YEAR TO MONTH` 的文本参数与读取类型 `dmPython.YEAR_MONTH_INTERVAL` 也已验证，两个区间类型的 `NULL` 往返通过。旧实现对大负区间读取发生 32 位整数溢出，本轮已修复。 -`login_timeout` 现限制完整建连握手：本机假服务端接受 TCP 后不回复，设为 1 秒会在约 1 秒内报错。`connection_timeout` 以秒传递给底层 TCP 拨号选项，属性读回与输入一致；`app_name` 会传给底层驱动并可读回。本机官方 DM8 上，Python 3.9、3.10、3.11、3.12、3.13 的完整真实库回归均为 **121 passed、0 failed**，另有 2 个非真实库用例未选入。 +当时的实现将 `login_timeout=1` 解释为 1 秒，并把 `connection_timeout` 用于 TCP 拨号；这与达梦官方接口定义不符，已在后续修复中更正。本机官方 DM8 上，当时 Python 3.9、3.10、3.11、3.12、3.13 的完整真实库回归均为 **121 passed、0 failed**，另有 2 个非真实库用例未选入。 ## 下一轮边界 diff --git a/dpi_bridge/dpi_conn.go b/dpi_bridge/dpi_conn.go index 34ad431..9170060 100644 --- a/dpi_bridge/dpi_conn.go +++ b/dpi_bridge/dpi_conn.go @@ -71,10 +71,11 @@ type connHandle struct { func newConnHandle(env *envHandle) *connHandle { return &connHandle{ - env: env, - port: DSQL_DEAFAULT_TCPIP_PORT, - autocommit: false, - serverCode: PG_UTF8, + env: env, + port: DSQL_DEAFAULT_TCPIP_PORT, + autocommit: false, + loginTimeout: 5000, + serverCode: PG_UTF8, } } @@ -358,9 +359,6 @@ func dpi_login(hcon C.dhcon, svr *C.sdbyte, user *C.sdbyte, pwd *C.sdbyte) C.DPI } else { params = append(params, "autoCommit=false") } - if conn.connTimeout > 0 { - params = append(params, fmt.Sprintf("socketTimeout=%d", conn.connTimeout)) - } if conn.appName != "" { params = append(params, "appName="+url.QueryEscape(conn.appName)) } @@ -381,7 +379,7 @@ func dpi_login(hcon C.dhcon, svr *C.sdbyte, user *C.sdbyte, pwd *C.sdbyte) C.DPI ctx := context.Background() if conn.loginTimeout > 0 { var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, time.Duration(conn.loginTimeout)*time.Second) + ctx, cancel = context.WithTimeout(ctx, time.Duration(conn.loginTimeout)*time.Millisecond) defer cancel() } rawConn, dbErr := db.Conn(ctx) diff --git a/dpi_bridge/dpi_stmt.go b/dpi_bridge/dpi_stmt.go index ed0f15d..909f0e3 100644 --- a/dpi_bridge/dpi_stmt.go +++ b/dpi_bridge/dpi_stmt.go @@ -23,10 +23,12 @@ typedef dhandle dhdesc; */ import "C" import ( + "context" "database/sql" "errors" "fmt" "sync" + "time" "unsafe" dm "gitee.com/chunanyong/dm" @@ -422,10 +424,12 @@ func dpi_exec(hstmt C.dhstmt) C.DPIRETURN { // Build args from parameter bindings args := buildExecArgs(stmt) + ctx, cancel := statementContext(stmt) + defer cancel() // Determine if this is a query or exec if isQuery(stmt.sql) { - rows, dbErr := stmt.prepared.Query(args...) + rows, dbErr := stmt.prepared.QueryContext(ctx, args...) if dbErr != nil { stmt.lastErr = diagFromError(dbErr) return DSQL_ERROR @@ -438,7 +442,7 @@ func dpi_exec(hstmt C.dhstmt) C.DPIRETURN { return DSQL_ERROR } } else { - result, dbErr := stmt.prepared.Exec(args...) + result, dbErr := stmt.prepared.ExecContext(ctx, args...) if dbErr != nil { stmt.lastErr = diagFromError(dbErr) return DSQL_ERROR @@ -483,9 +487,11 @@ func dpi_exec_direct(hstmt C.dhstmt, sqlTxt *C.sdbyte) C.DPIRETURN { stmt.cachedRows = nil stmt.fetchPos = 0 + ctx, cancel := statementContext(stmt) + defer cancel() if isQuery(sqlStr) { - rows, dbErr := stmt.conn.db.Query(sqlStr) + rows, dbErr := stmt.conn.db.QueryContext(ctx, sqlStr) if dbErr != nil { stmt.lastErr = diagFromError(dbErr) return DSQL_ERROR @@ -497,7 +503,7 @@ func dpi_exec_direct(hstmt C.dhstmt, sqlTxt *C.sdbyte) C.DPIRETURN { return DSQL_ERROR } } else { - result, dbErr := stmt.conn.db.Exec(sqlStr) + result, dbErr := stmt.conn.db.ExecContext(ctx, sqlStr) if dbErr != nil { stmt.lastErr = diagFromError(dbErr) return DSQL_ERROR @@ -511,6 +517,16 @@ func dpi_exec_direct(hstmt C.dhstmt, sqlTxt *C.sdbyte) C.DPIRETURN { return DSQL_SUCCESS } +func statementContext(stmt *stmtHandle) (context.Context, context.CancelFunc) { + stmt.conn.mu.Lock() + timeout := stmt.conn.connTimeout + stmt.conn.mu.Unlock() + if timeout > 0 { + return context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second) + } + return context.WithCancel(context.Background()) +} + //export dpi_exec_directW func dpi_exec_directW(hstmt C.dhstmt, sqlTxt *C.sdbyte, sqlLen C.sdint4) C.DPIRETURN { return dpi_exec_direct(hstmt, sqlTxt) @@ -628,6 +644,11 @@ func cacheAllRows(stmt *stmtHandle) error { } stmt.cachedRows = append(stmt.cachedRows, row) } + if err := stmt.rows.Err(); err != nil { + stmt.rows.Close() + stmt.rows = nil + return err + } // Close the rows now that we've cached everything stmt.rows.Close() diff --git a/dpi_bridge/third_party/chunanyong_dm/PATCHES.md b/dpi_bridge/third_party/chunanyong_dm/PATCHES.md index 58ad78e..fbc6ab3 100644 --- a/dpi_bridge/third_party/chunanyong_dm/PATCHES.md +++ b/dpi_bridge/third_party/chunanyong_dm/PATCHES.md @@ -44,6 +44,15 @@ - Regression: `test_login_timeout_interrupts_unresponsive_handshake` uses a local TCP listener that accepts a connection but never replies. +## Patch: interrupt blocked statements on context timeout + +- File: `m.go` +- Problem: context cancellation called cleanup, which tried to roll back on a + socket still blocked in a statement. The caller waited indefinitely. +- Fix: close the socket before cleanup so the blocked read and rollback return. +- Regression: `test_connection_timeout_limits_sql_execution` holds a row lock + while another session executes an update with a one-second timeout. + ## Patch: application name query value - File: `n.go` diff --git a/dpi_bridge/third_party/chunanyong_dm/m.go b/dpi_bridge/third_party/chunanyong_dm/m.go index 8bc00be..9036e7d 100644 --- a/dpi_bridge/third_party/chunanyong_dm/m.go +++ b/dpi_bridge/third_party/chunanyong_dm/m.go @@ -764,6 +764,11 @@ func (conn *DmConnection) CompatibleMysql() bool { func (conn *DmConnection) cancel(err error) { conn.canceled.Set(err) + // Close the socket before cleanup. Cleanup sends a rollback, which cannot + // complete while the server is blocked on the canceled statement. + if conn.Access != nil && conn.Access.dm_build_415 != nil { + conn.Access.dm_build_415.Close() + } conn.close() } diff --git a/tests/integration/test_p1_connection_matrix.py b/tests/integration/test_p1_connection_matrix.py index 8b8a754..8061767 100644 --- a/tests/integration/test_p1_connection_matrix.py +++ b/tests/integration/test_p1_connection_matrix.py @@ -96,7 +96,7 @@ def test_port_rejects_non_numeric_value(conn_params): "option", [ {"connection_timeout": 5}, - {"login_timeout": 5}, + {"login_timeout": 5000}, {"compress_msg": 0}, {"use_stmt_pool": 1}, ], @@ -111,11 +111,11 @@ def test_optional_setting_connects_and_queries(conn_params, option): def test_connection_timeout_options_are_reported(conn_params): with dmPython.connect( **conn_params, - login_timeout=2, + login_timeout=2000, connection_timeout=2, app_name="dmpython & matrix+1", ) as conn: - assert conn.login_timeout == 2 + assert conn.login_timeout == 2000 assert conn.connection_timeout == 2 assert conn.app_name == "dmpython & matrix+1" with conn.cursor() as cur: @@ -123,6 +123,12 @@ def test_connection_timeout_options_are_reported(conn_params): assert cur.fetchone() == (1,) +def test_timeout_defaults_match_dm_interface(conn_params): + with dmPython.connect(**conn_params) as conn: + assert conn.login_timeout == 5000 + assert conn.connection_timeout == 0 + + def test_login_timeout_interrupts_unresponsive_handshake(): code = """ import dmPython @@ -132,7 +138,7 @@ def test_login_timeout_interrupts_unresponsive_handshake(): start = time.monotonic() try: dmPython.connect(user="probe", password="probe", server="127.0.0.1", - port=int(sys.argv[1]), login_timeout=1) + port=int(sys.argv[1]), login_timeout=1000) except dmPython.Error: print(time.monotonic() - start) else: @@ -160,3 +166,67 @@ def test_login_timeout_interrupts_unresponsive_handshake(): assert child.returncode == 0, stderr assert float(stdout.strip()) < 3 + + +@pytest.mark.parametrize("statement", ["direct", "prepared", "select_for_update"]) +def test_connection_timeout_limits_sql_execution( + conn, table_name_factory, drop_table, statement +): + table = table_name_factory("DMPY_TIMEOUT") + cur = conn.cursor() + code = """ +import dmPython +import os +import sys +import time + +conn = dmPython.connect( + user=os.environ["DM_TEST_USER"], + password=os.environ["DM_TEST_PASSWORD"], + server=os.environ["DM_TEST_HOST"], + port=int(os.environ["DM_TEST_PORT"]), + connection_timeout=1, +) +cur = conn.cursor() +start = time.monotonic() +try: + if sys.argv[2] == "select_for_update": + cur.execute(f"SELECT v FROM {sys.argv[1]} WHERE id=1 FOR UPDATE") + cur.fetchone() + elif sys.argv[2] == "prepared": + cur.execute(f"UPDATE {sys.argv[1]} SET v=? WHERE id=1", (3,)) + else: + cur.execute(f"UPDATE {sys.argv[1]} SET v=3 WHERE id=1") +except dmPython.Error: + print(time.monotonic() - start) +else: + raise AssertionError("blocked update unexpectedly completed") +cur.execute("SELECT 1") +assert cur.fetchone() == (1,) +""" + try: + cur.execute(f"CREATE TABLE {table} (id INTEGER PRIMARY KEY, v INTEGER)") + cur.execute(f"INSERT INTO {table} VALUES (1, 1)") + conn.commit() + cur.execute(f"UPDATE {table} SET v=2 WHERE id=1") + + child = subprocess.Popen( + [sys.executable, "-c", code, table, statement], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + env=os.environ.copy(), + ) + try: + stdout, stderr = child.communicate(timeout=6) + finally: + if child.poll() is None: + child.kill() + child.communicate() + assert child.returncode == 0, stderr + assert 0.5 < float(stdout.strip()) < 4 + finally: + conn.rollback() + drop_table(cur, table) + conn.commit() + cur.close()