Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/api-reference.md
Original file line number Diff line number Diff line change
Expand Up @@ -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)`
Expand Down
2 changes: 1 addition & 1 deletion docs/test-results/2026-09-26-type-connection-matrix.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 个非真实库用例未选入。

## 下一轮边界

Expand Down
14 changes: 6 additions & 8 deletions dpi_bridge/dpi_conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}
}

Expand Down Expand Up @@ -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))
}
Expand All @@ -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)
Expand Down
29 changes: 25 additions & 4 deletions dpi_bridge/dpi_stmt.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,12 @@ typedef dhandle dhdesc;
*/
import "C"
import (
"context"
"database/sql"
"errors"
"fmt"
"sync"
"time"
"unsafe"

dm "gitee.com/chunanyong/dm"
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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)
Expand Down Expand Up @@ -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()
Expand Down
9 changes: 9 additions & 0 deletions dpi_bridge/third_party/chunanyong_dm/PATCHES.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`
Expand Down
5 changes: 5 additions & 0 deletions dpi_bridge/third_party/chunanyong_dm/m.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()

}
Expand Down
78 changes: 74 additions & 4 deletions tests/integration/test_p1_connection_matrix.py
Original file line number Diff line number Diff line change
Expand Up @@ -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},
],
Expand All @@ -111,18 +111,24 @@ 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:
cur.execute("SELECT 1")
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
Expand All @@ -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:
Expand Down Expand Up @@ -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()
Loading