thunguo commented on code in PR #1137:
URL:
https://github.com/apache/incubator-seata-go/pull/1137#discussion_r3543972615
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -156,27 +161,31 @@ func (c *XAConn) BeginTx(ctx context.Context, opts
driver.TxOptions) (driver.Tx,
return nil, err
}
- if !c.autoCommit {
- if c.xaActive {
- return nil, errors.New("should NEVER happen:
setAutoCommit from true to false while xa branch is active")
- }
+ // Create XA branch for both explicit transactions and autoCommit mode
(branch reuse)
+ // In autoCommit mode, we register the branch but keep it open for
multiple SQL statements
+ if c.xaActive {
+ return nil, errors.New("should NEVER happen: setAutoCommit from
true to false while xa branch is active")
+ }
- baseTx, ok := tx.(*Tx)
- if !ok {
- return nil, fmt.Errorf("start xa %s transaction failure
for the tx is a wrong type", c.txCtx.XID)
- }
+ baseTx, ok := tx.(*Tx)
+ if !ok {
+ return nil, fmt.Errorf("start xa %s transaction failure for the
tx is a wrong type", c.txCtx.XID)
+ }
- baseTx.xaConn = c
+ baseTx.xaConn = c
- c.branchRegisterTime = time.Now()
- if err := baseTx.register(c.txCtx); err != nil {
- c.cleanXABranchContext()
- return nil, fmt.Errorf("failed to register xa branch
%s, err:%w", c.txCtx.XID, err)
- }
+ c.branchRegisterTime = time.Now()
+ if err := baseTx.register(c.txCtx); err != nil {
+ c.cleanXABranchContext()
+ return nil, fmt.Errorf("failed to register xa branch %s,
err:%w", c.txCtx.XID, err)
+ }
- c.xaBranchXid = XaIdBuild(c.txCtx.XID, c.txCtx.BranchID)
- c.keepIfNecessary()
+ c.xaBranchXid = XaIdBuild(c.txCtx.XID, c.txCtx.BranchID)
+ c.keepIfNecessary()
+ // For autoCommit mode (branch reuse), skip XA START here
+ // It will be done when needed for the actual execution
+ if !wasAutoCommit {
if err = c.start(ctx); err != nil {
Review Comment:
这里语义有问题？autoCommit下跳过了 XA_START，但是看起来后续的时机没有补充调用。autoCommit下完全没有拉起 XA Branch。
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -358,6 +407,10 @@ func (c *XAConn) Rollback(ctx context.Context) error {
c.rollBacked = true
}
c.cleanXABranchContext()
+
+ // Clean up resource holder on rollback
+ c.res.UnregisterXABranch(c.txCtx.XID)
Review Comment:
是不是应该先Unregister
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn)
(bool, *xaEntry) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ now := time.Now()
+
+ // Check if entry already exists
+ if entry, ok := r.entries[xid]; ok {
+ entry.lastAccessTime = now
+ entry.statementCount++
+ log.Infof("Reusing existing XA branch for xid: %s, branchID:
%s, statementCount: %d, state: %v",
+ xid, entry.branchID, entry.statementCount, entry.state)
+ return false, entry
+ }
+
+ // Create new entry
+ entry := &xaEntry{
+ conn: conn,
+ xid: xid,
+ branchID: branchID,
+ resourceID: resourceID,
+ state: xaStateStarted,
+ createTime: now,
+ lastAccessTime: now,
+ statementCount: 1,
+ }
+
+ r.entries[xid] = entry
+ log.Infof("Registered new XA branch, xid: %s, branchID: %s, resourceID:
%s", xid, branchID, resourceID)
+
+ return true, entry
+}
+
+// get retrieves an XA entry by xid
+func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ entry, ok := r.entries[xid]
+ if ok {
+ entry.lastAccessTime = time.Now()
+ }
+ return entry, ok
+}
+
+// canReuse checks if an XA connection can be reused for the given xid
+func (r *xaRegistry) canReuse(xid string) bool {
+ entry, ok := r.get(xid)
+ if !ok {
+ return false
+ }
+
+ // Can reuse if: entry exists, connection is active, and not yet
ended/prepared
+ // Also check that the connection is the same (for connection pooling)
+ if entry.conn == nil {
+ return false
+ }
+ return entry.state == xaStateStarted && entry.conn.xaActive &&
entry.branchID != "pending"
+}
+
+// setState updates the state of an XA entry
+func (r *xaRegistry) setState(xid string, state xaState) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ if entry, ok := r.entries[xid]; ok {
+ oldState := entry.state
+ entry.state = state
+ log.Infof("XA state changed, xid: %s, branchID: %s, %v -> %v",
+ xid, entry.branchID, oldState, state)
+ }
+}
+
+// unregister removes an XA entry from the registry
+func (r *xaRegistry) unregister(xid string) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ if entry, ok := r.entries[xid]; ok {
+ log.Infof("Unregistered XA branch, xid: %s, branchID: %s,
statementCount: %d",
+ xid, entry.branchID, entry.statementCount)
+ delete(r.entries, xid)
+ }
+}
+
+// cleanup removes stale entries (called periodically)
+func (r *xaRegistry) cleanup() {
Review Comment:
同上，没有使用
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -289,12 +331,19 @@ func (c *XAConn) start(ctx context.Context) error {
return fmt.Errorf("xa xid %s resource connection start err:%w",
c.txCtx.XID, err)
}
+ // For multi-statement XA transactions (originally in autoCommit mode),
+ // skip the termination check. The check will be done during Phase 2
commit/rollback.
+ if c.txCtx.IsAutoCommitXABranch {
Review Comment:
这里是不不太对，`c.start(ctx)` 调用前已经 check `wasAutoCommit` 了
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn)
(bool, *xaEntry) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ now := time.Now()
+
+ // Check if entry already exists
+ if entry, ok := r.entries[xid]; ok {
+ entry.lastAccessTime = now
+ entry.statementCount++
+ log.Infof("Reusing existing XA branch for xid: %s, branchID:
%s, statementCount: %d, state: %v",
+ xid, entry.branchID, entry.statementCount, entry.state)
+ return false, entry
+ }
+
+ // Create new entry
+ entry := &xaEntry{
+ conn: conn,
+ xid: xid,
+ branchID: branchID,
+ resourceID: resourceID,
+ state: xaStateStarted,
+ createTime: now,
+ lastAccessTime: now,
+ statementCount: 1,
+ }
+
+ r.entries[xid] = entry
+ log.Infof("Registered new XA branch, xid: %s, branchID: %s, resourceID:
%s", xid, branchID, resourceID)
+
+ return true, entry
+}
+
+// get retrieves an XA entry by xid
+func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
+ r.mu.RLock()
+ defer r.mu.RUnlock()
+
+ entry, ok := r.entries[xid]
+ if ok {
+ entry.lastAccessTime = time.Now()
Review Comment:
为什么在读锁下做写操作？这里会出现竞态，可以调整下：
```go
type xaEntry struct {
// ...
lastAccessTime atomic.Int64
}
func (r *xaRegistry) get(xid string) (*xaEntry, bool) {
// ...
if ok {
entry.lastAccessTime.Store(time.Now().UnixNano())
}
// ...
}
```
##########
pkg/datasource/sql/xa_registry.go:
##########
@@ -0,0 +1,192 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package sql
+
+import (
+ "sync"
+ "time"
+
+ "seata.apache.org/seata-go/v2/pkg/util/log"
+)
+
+// xaState represents the state of an XA transaction branch
+type xaState int
+
+const (
+ xaStateIdle xaState = iota // No XA transaction
+ xaStateStarted // XA START executed
+ xaStateEnded // XA END executed
+ xaStatePrepared // XA PREPARE executed
+)
+
+// xaEntry represents an active XA transaction branch
+type xaEntry struct {
+ conn *XAConn
+ xid string
+ branchID string
+ resourceID string
+ state xaState
+ createTime time.Time
+ lastAccessTime time.Time
+ statementCount int
+}
+
+// xaRegistry manages XA connections for global transactions
+// It ensures multiple SQL operations in the same global transaction
+// reuse the same XA branch, preventing "busy buffer" errors
+type xaRegistry struct {
+ mu sync.RWMutex
+ entries map[string]*xaEntry // key: xid
+}
+
+var (
+ globalRegistry *xaRegistry
+ registryOnce sync.Once
+)
+
+// getXARegistry returns the global XA registry (singleton)
+func getXARegistry() *xaRegistry {
+ registryOnce.Do(func() {
+ globalRegistry = &xaRegistry{
+ entries: make(map[string]*xaEntry),
+ }
+ })
+ return globalRegistry
+}
+
+// register registers or retrieves an XA connection for the given xid
+// Returns (isNew, entry, error)
+func (r *xaRegistry) register(xid, branchID, resourceID string, conn *XAConn)
(bool, *xaEntry) {
Review Comment:
xa_registry感觉是非必要的，这里完全没有使用
##########
pkg/datasource/sql/conn_xa.go:
##########
@@ -196,63 +205,96 @@ func (c *XAConn) createOnceTxContext(ctx context.Context)
bool {
c.txCtx.ResourceID = c.res.resourceID
c.txCtx.XID = tm.GetXID(ctx)
c.txCtx.TransactionMode = types.XAMode
- c.txCtx.GlobalLockRequire = true
Review Comment:
这里为什么删除？
##########
pkg/datasource/sql/db.go:
##########
@@ -117,7 +117,8 @@ type DBResource struct {
// for xa
metaCache datasource.TableMetaCache
shouldBeHeld bool
- keeper sync.Map
+ keeper sync.Map // xaBranchID -> *XAConn
+ xaConnsByXID sync.Map // xid -> *XAConn (for XA branch reuse in
autoCommit mode)
Review Comment:
如果异常导致Unregistry行为没有执行这里是不是会出现泄漏的情况？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
