Author | Liu Xiaomin Yu Yu

I. Introduction​

In the Java world, netty is a widely used high-performance network communication framework, and many RPC frameworks are based on netty. In the golang world, getty is also a high-performance network communication library similar to netty. getty was originally developed by Yu Yu, the leader of the dubbogo project, and is available in dubbo-go as an underlying communication library. github.com/apache/dubbo-go). With the donation of dubbo-go to the apache foundation, getty eventually made its way into the apache family and was renamed dubbo-getty, thanks to the efforts of the community.

In '18, I was practicing microservices in my company, and the biggest problem I encountered at that time was distributed transactions. In the same year, Ali open-sourced their distributed transaction solution in the community, and I quickly noticed this project, which was initially called fescar, but later renamed seata. Since I was very interested in open source technology, I added a lot of community groups, and at that time, I also paid attention to the dubbo-go project, and silently dived in it. As I learnt more about seata, the idea of making a go version of a distributed transaction framework gradually emerged.

To make a golang version of distributed transaction framework, one of the first problems is how to achieve RPC communication. dubbo-go is a very good example in front of us, so we started to study the underlying getty of dubbo-go.

How to implement RPC communication based on getty?​

The overall model of the getty framework is as follows:

! [image.png]( https://img.alicdn.com/imgextra/i1/O1CN011TIcL01jY4JaweOfV_! !6000000004559-2-tps-954-853.png)

The following is a detailed description of the RPC communication process of seata-golang with related code.

1. Establish Connection​

To implement RPC communication, we need to establish a network connection first, let's start from client.go.

func (c *client) connect() {
 var (
  err error
  ss Session
 )

 for {
       // Create a session connection
  ss = c.dial()
  if ss == nil {
   if ss == nil { // client has been closed
   if ss == nil { // client has been closed
  }
  err = c.newSession(ss)
  if err == nil {
           // send and receive messages
   ss.(*session).run()
   // Omit some code here

   break
  }
  // don't distinguish between tcp connection and websocket connection. because
  // gorilla/websocket/conn.go:(Conn)Close also invoke net.Conn.Close()
  ss.Conn().Close()
 Close()
}

The connect() method gets a session connection via the dial() method into the dial() method:

func (c *client) dial() Session {
switch c.endPointType {
case TCP_CLIENT.
return c.dialTCP()
case UDP_CLIENT: return c.dialUDP()
return c.dialUDP()
case WS_CLIENT: return c.dialWS()
return c.dialWS()
case WSS_CLIENT: return c.dialWSS()
return c.dialWSS()
}

return nil
}

We're concerned with TCP connections, so we continue into the c.dialTCP() method:

func (c *client) dialTCP() Session {
 var (
  err error
  conn net.
 )

 for {
  if c.IsClosed() {
   return nil
  }
  if c.sslEnabled {
   if sslConfig, err := c.tlsConfigBuilder.BuildTlsConfig(); err == nil && sslConfig ! = nil {
    d := &net.Dialer{Timeout: connectTimeout}
    // Establish an encrypted connection
    conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig)
   }
  } else {
           // Establish a tcp connection
   conn, err = net.DialTimeout("tcp", c.addr, connectTimeout)
  }
  if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) {
   conn.Close()
   err = errSelfConnect
  }
  if err == nil {
           // Return a TCPSession
   return newTCPSession(conn, c)
  }

  log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err))
  <-wheel.After(connectInterval)
 }
}

At this point, we know how getty establishes a TCP connection and returns a TCPSession.

2. Sending and Receiving Messages​

How does it send and receive messages? Let's go back to the connection method and look at the next line, which is ss.(*session).run(). After this line of code, the code is a very simple operation, so we guess that the logic of this line of code must include sending and receiving messages, and then go to the run() method:

func (s *session) run() {
// Omit some of the code

go s.handleLoop()
go s.handlePackage()
}

There are two goroutines up here, handleLoop and handlePackage, which literally match our guesses into the handleLoop() method:

func (s *session) handleLoop() {
   // Omit some of the code

 for {
  // A select blocks until one of its cases is ready to run.
  // It choose one at random if multiple are ready. Otherwise it choose default branch if none is ready.
  It choose one at random if multiple are ready.
  // Otherwise it choose default branch if none is ready.

  case outPkg, ok = <-s.wQ.
   // Omit some of the code

   iovec = iovec[:0]
   for idx := 0; idx < maxIovecNum; idx++ {
       // Encode interface{} type outPkg into binary bits via s.writer
    pkgBytes, err = s.writer.Write(s, outPkg)
    // Omit some of the code

    iovec = append(iovec, pkgBytes)

               // omit some code
   }
           // Send these binary bits out
   err = s.WriteBytesArray(iovec[:]...)
   if err ! = nil {
    log.Errorf("%s, [session.handleLoop]s.WriteBytesArray(iovec len:%d) = error:%+v",
     s.sessionToken(), len(iovec), perrors.WithStack(err))
    s.stop()
    // break LOOP
    flag = false
   }

  case <-wheel.After(s.period).
   if flag {
    if wsFlag {
     err := wsConn.writePing()
     if err ! = nil {
      log.Warnf("wsConn.writePing() = error:%+v", perrors.WithStack(err))
     }
    }
               // Logic for timed execution, heartbeat, etc.
    s.listener.OnCron(s)
   }
  }
 }
}

With the above code, it is easy to see that the handleLoop() method handles the logic of sending the message, which is encoded into binary bits by s.writer and then sent over the established TCP connection. This s.writer corresponds to the Writer interface, which is an interface that must be implemented by the RPC framework.

Moving on to the handlePackage() method:

func (s *session) handlePackage() {
// Omit some of the code

if _, ok := s.Connection.(*gettyTCPConn); ok {
if s.reader == nil {
errStr := fmt.Sprintf("session{name:%s, conn:%#v, reader:%#v}", s.name, s.Connection, s.reader)
log.Error(errStr)
panic(errStr)
}

err = s.handleTCPPackage()
} else if _, ok := s.Connection.(*gettyWSConn); ok {
err = s.handleWSPackage()
} else if _, ok := s.Connection.(*gettyUDPConn); ok {
err = s.handleUDPPackage()
} else {
panic(fmt.Sprintf("unknown type session{%#v}", s))
}
}

Go to the handleTCPPackage() method:

func (s *session) handleTCPPackage() error {
   // Omit some of the code

 conn = s.Connection.(*gettyTCPConn)
 for {
  // omit some code

  bufLen = 0
  for {
   // for clause for the network timeout condition check
   // s.conn.SetReadTimeout(time.Now().Add(s.rTimeout))
           // Receive a message from the TCP connection
   bufLen, err = conn.recv(buf)
   // Omit some of the code

   break
  }
  // Omit part of the code

       // Write the binary bits of the received message to pkgBuf
  pktBuf.Write(buf[:bufLen])
  for {
   if pktBuf.Len() <= 0 {
    Write(buf[:bufLen]) for { if pktBuf.
   }
           // Decode the received message into an RPC message via s.reader
   pkg, pkgLen, err = s.reader.Read(s, pktBuf.Bytes())
   // Omit some of the code

     s.UpdateActive()
           // Put the received message into a TaskQueue for consumption by the RPC consumer.
   s.addTask(pkg)
   pktBuf.Next(pkgLen)
   // continue to handle case 5
  If exit { pktBuf.Next(pkgLen) // continue to handle case 5
  if exit {
   pktBuf.Next(pkgLen) // continue to handle case 5 } if exit {
  }
 }

 return perrors.WithStack(err)
}

From the above code logic, we analyse that the RPC consumer needs to decode the binary bits received from the TCP connection into messages that can be consumed by RPC, and this work is implemented by s.reader, so we need to implement the Reader interface corresponding to s.reader in order to build the RPC communication layer.

3. How to decouple the underlying network message processing logic from the business logic​

We all know that netty decouples the underlying network logic from the business logic through the boss thread and the worker thread. So how does getty do it?

At the end of the handlePackage() method, we see that the incoming message is put into the s.addTask(pkg) method, so let's move on:

func (s *session) addTask(pkg interface{}) {
f := func() {
s.listener.OnMessage(s, pkg)
s.incReadPkgNum()
}
if taskPool := s.EndPoint().GetTaskPool(); taskPool ! = nil {
taskPool.AddTaskAlways(f)
return
}
f()
}

The pkg argument is passed to an anonymous method that ends up in taskPool. This method is critical, and I ran into a pitfall later on when I wrote the seata-golang code, which is analysed later.

Next we look at the definition of taskPool:

// NewTaskPoolSimple builds a simple task pool.
func NewTaskPoolSimple(size int) GenericTaskPool {
 if size < 1 {
  size = runtime.NumCPU() * 100
 NumCPU() * 100 }
 return &taskPoolSimple{
  work: make(chan task), sem: make(chan struct{task
  sem:  make(chan struct{}, size),
  done: make(chan struct{}),
 }
}

Builds a channel sem with a buffer size of size (defaults to runtime.NumCPU() * 100). Then look at the method AddTaskAlways(t task):

func (p *taskPoolSimple) AddTaskAlways(t task) {
select {
case <-p.done.
return
default.
}

select {
case p.work <- t.
return
default: }
}
select {
case p.work <- t: return default: }
case p.sem <- struct{}{}.
p.wg.Add(1)
go p.worker(t)
default.
goSafely(t)
}
}

When a task is added, it is consumed by len(p.sem) goroutines, and if no goroutine is free, a temporary goroutine is started to run t(). This is equivalent to having len(p.sem) goroutines to form a goroutine pool, and the goroutines in the pool process business logic instead of the goroutines that process network messages to run business logic, thus achieving decoupling. One of the pitfalls I encountered when writing seata-golang was that I forgot to set the taskPool, which resulted in the same goroutine handling the business logic and the underlying network message logic. When I blocked the business logic and waited for a task to complete, I blocked the entire goroutine, and I couldn't receive any messages during the blocking period.

4. Implementation​

The following code is available at getty.go:

// Reader is used to unmarshal a complete pkg from buffer
type Reader interface {
 Read(Session, []byte) (interface{}, int, error)
}

// Writer is used to marshal a pkg and write to session.
type Writer interface {
 // If @Session is udpGettySession, the second parameter is UDPContext.
 Write(Session, interface{}) ([]byte, error)
Write(Session, interface{}) ([]byte, error) }

// ReadWriter interface use for handle application packages.
type ReadWriter interface {
 Writer
 Writer
}

// EventListener is used to process pkg that received from remote session
type EventListener interface {
 // invoked when session opened
 // If the return error is not nil, @Session will be closed.
 OnOpen(Session) error

 OnOpen(Session) error // invoked when session closed.
 EventListener { // invoked when session opened // If the return error is not nil, @Session will be closed.)

 OnOpen(Session) error // invoked when session closed.
 OnError(Session, error)

 // invoked periodically, its period can be set by (Session)SetCronPeriod
 OnCron(Session)

 // invoked when getty received a package. Pls attention that do not handle long time
 // logic processing in this func. You'd better set the package's maximum length.
 // If the message's length is greater than it, u should should return err in
 If the message's length is greater than it, u should should return err in // Reader{Read} and getty will close this connection soon.
 // If ur logic processing in this func
 // If ur logic processing in this func will take a long time, u should start a goroutine
 // If ur logic processing in this func will take a long time, u should start a goroutine pool (like working thread pool in cpp) to handle the processing asynchronously.
 // can do the logic processing in other asynchronous way.
 Or u // can do the logic processing in other asynchronous way. !In short, ur OnMessage callback func should return asap.
 // In short, ur OnMessage callback func should return asap.
 // If this is a udp event listener, the second parameter type is UDPContext.
 OnMessage(Session, interface{})
}

By analysing the entire getty code, we only need to implement ReadWriter to encode and decode RPC messages, and then implement EventListener to handle the corresponding specific logic of RPC messages, and then inject the ReadWriter implementation and the EventLister implementation into the Client and Server sides of RPC, then we can implement RPC communication. Inject the ReadWriter implementation and EventLister implementation into the Client and Server side of RPC to achieve RPC communication.

4.1 Codec Protocol Implementation​

The following is the definition of the seata protocol: ! [image-20201205214556457.png](https://cdn.nlark.com/yuque/0/2020/png/737378/1607180799872-5f96afb6-680d-4e69-8c95-b8fd1ac4c3a7.png #align=left&display=inline&height=209&margin=%5Bobject%20Object%5D&name=image-20201205214556457.png& originHeight=209&originWidth=690&size=18407&status=done&style=none&width=690)

In the ReadWriter interface implementation RpcPackageHandler, call the Codec method to codec the message body in the above format:

// Encode the message into binary bits
func MessageEncoder(codecType byte, in interface{}) []byte {
switch codecType {
case SEATA.
return SeataEncoder(in)
default.
log.Errorf("not support codecType, %s", codecType)
return nil
}
}

// Decode the binary bits into the message body
func MessageDecoder(codecType byte, in []byte) (interface{}, int) {
switch codecType {
case SEATA.
return SeataDecoder(in)
default.
log.Errorf("not support codecType, %s", codecType)
return nil, 0
}
}

4.2 Client Side Implementation​

Let's look at the client-side implementation of EventListener [RpcRemotingClient](https://github.com/opentrx/seata-golang/blob/dev/pkg/client/rpc_remoting_client. go):

func (client *RpcRemoteClient) OnOpen(session getty.Session) error {
 go func()
  request := protocal.RegisterTMRequest{AbstractIdentifyRequest: protocal.
   ApplicationId: client.conf.
   TransactionServiceGroup: client.conf.
  }}
   // Once the connection is established, make a request to the Transaction Coordinator to register the TransactionManager.
  _, err := client.sendAsyncRequestWithResponse(session, request, RPC_REQUEST_TIMEOUT)
  if err == nil {
     // Save the connection to the Transaction Coordinator in the connection pool for future use.
   clientSessionManager.RegisterGettySession(session)
   client.GettySessionOnOpenChannel <- session.RemoteAddr()
  }
 }()

 return nil
}

// OnError ...
func (client *RpcRemoteClient) OnError(session getty.Session, err error) {
 clientSessionManager.ReleaseGettySession(session)
}

// OnClose ...
func (client *RpcRemoteClient) OnClose(session getty.Session) {
 clientSessionManager.ReleaseGettySession(session)
}

// OnMessage ...
func (client *RpcRemoteClient) OnMessage(session getty.Session, pkg interface{}) {
 log.Info("received message:{%v}", pkg)
 rpcMessage, ok := pkg.(clientRpcRemoteClient.Session, pkg interface{}) { log.Info("received message:{%v}", pkg)
 if ok {
  heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
  if isHeartBeat && heartBeat == protocal.HeartBeatMessagePong {
   log.Debugf("received PONG from %s", session.RemoteAddr())
  }
 }

 if rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
  rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
  log.Debugf("msgId:%s, body:%v", rpcMessage.Id, rpcMessage.Body)

  // Process the transaction message, commit or rollback
  client.onMessage(rpcMessage, session.RemoteAddr())
 } else {
  resp, loaded := client.futures.Load(rpcMessage.Id)
  if loaded {
   response := resp.(*getty2.MessageFuture)
   response.Response = rpcMessage.Body
   response.Done <- true
   client.futures.Delete(rpcMessage.Id)
  }
 }
}

// OnCron ...
func (client *RpcRemoteClient) OnCron(session getty.Session) {
 // Send a heartbeat
 client.defaultSendRequest(session, protocal.HeartBeatMessagePing)
}

The logic of clientSessionManager.RegisterGettySession(session) is analysed in subsection 4.4.

4.3 Server-side Transaction Coordinator Implementation​

See DefaultCoordinator for code:

func (coordinator *DefaultCoordinator) OnOpen(session getty.Session) error {
 log.Infof("got getty_session:%s", session.Stat())
 error { log.Infof("got getty_session:%s", session.Stat())
}

func (coordinator *DefaultCoordinator) OnError(session getty.Session, err error) {
 // Release the TCP connection
 SessionManager.ReleaseGettySession(session)
 session.Close()
 log.Errorf("getty_session{%s} got error{%v}, will be closed.", session.Stat(), err)
}

func (coordinator *DefaultCoordinator) OnClose(session getty.Session) {
 log.Info("getty_session{%s} is closing......" , session.Stat())
}

func (coordinator *DefaultCoordinator) OnMessage(session getty.Session, pkg interface{}) {
 log.Debugf("received message:{%v}", pkg)
 rpcMessage, ok := pkg.(protocal.)
 RpcMessage) if ok {
  _, isRegTM := rpcMessage.Body.(protocal.RegisterTMRequest)
  if isRegTM {
     // Map the TransactionManager information to the TCP connection.
   coordinator.OnRegTmMessage(rpcMessage, session)
   OnRegTmMessage(rpcMessage, session)
  }

  heartBeat, isHeartBeat := rpcMessage.Body.(protocal.HeartBeatMessage)
  if isHeartBeat && heartBeat == protocal.HeartBeatMessagePing {
   coordinator.OnCheckMessage(rpcMessage, session)
   OnCheckMessage(rpcMessage, session)
  }

  if rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST ||
   rpcMessage.MessageType == protocal.MSGTYPE_RESQUEST_ONEWAY {
   log.Debugf("msgId:%s, body:%v", rpcMessage.Id, rpcMessage.Body)
   _, isRegRM := rpcMessage.Body.(protocal.RegisterRMRequest)
   if isRegRM {
       // Map the ResourceManager information to the TCP connection.
    coordinator.OnRegRmMessage(rpcMessage, session)
   } else {
    if SessionManager.IsRegistered(session) {
     if err := recover(); } else { if SessionManager.IsRegistered(session) {
      if err := recover(); err ! = nil { log.Errorf(); err !
       log.Errorf("Catch Exception while do RPC, request: %v,err: %w", rpcMessage, err)
      }
     }()
         // Handle transaction messages, global transaction registration, branch transaction registration, branch transaction commit, global transaction rollback, etc.
     coordinator.OnTrxMessage(rpcMessage, session)
    } else {
     session.Close()
     log.Infof("Close an unhandled connection! [%v]", session)
    }
   }
  } else {
   resp, loaded := coordinator.futures.Load(rpcMessage.Id)
   if loaded {
    response := resp.(*getty2.MessageFuture)
    response.Response = rpcMessage.Body
    response.Done <- true
    coordinator.futures.Delete(rpcMessage.Id)
   }
  }
 }
}

func (coordinator *DefaultCoordinator) OnCron(session getty.Session) {

}

coordinator.OnRegTmMessage(rpcMessage, session) registers the Transaction Manager, coordinator.OnRegRmMessage(rpcMessage, session) registers the Resource The logic is analysed in Section 4.4. The message enters the coordinator.OnTrxMessage(rpcMessage, session) method and is routed to the specific logic according to the message type code:

switch msg.GetTypeCode() {
case protocal.TypeGlobalBegin:
req := msg.(protocal.GlobalBeginRequest)
resp := coordinator.doGlobalBegin(req, ctx)
return resp
case protocal.TypeGlobalStatus.
TypeGlobalStatus. req := msg.(protocal.GlobalStatusRequest)
resp := coordinator.doGlobalStatus(req, ctx)
return resp
case protocal.TypeGlobalReport.
req := msg.(protocal.GlobalReportRequest)
resp := coordinator.doGlobalReport(req, ctx)
return resp
case protocal.TypeGlobalCommit.
req := msg.(protocal.GlobalCommitRequest)
resp := coordinator.doGlobalCommit(req, ctx)
return resp
case protocal.TypeGlobalRollback.
req := msg.(protocal.GlobalRollbackRequest)
resp := coordinator.doGlobalRollback(req, ctx)
return resp
case protocal.TypeBranchRegister.
TypeBranchRegister. req := msg.(protocal.BranchRegisterRequest)
resp := coordinator.doBranchRegister(req, ctx)
return resp
case protocal.TypeBranchStatusReport.
TypeBranchStatusReport: req := msg.(protocal.BranchReportRequest)
resp := coordinator.doBranchReport(req, ctx)
return resp
default.
return nil
}

4.4 Session Manager Analysis​

After the Client establishes a connection with the Transaction Coordinator, it saves the connection in the map serverSessions = sync.Map{} by using clientSessionManager.RegisterGettySession(session). The key of the map is the RemoteAddress of the Transaction Coordinator obtained from the session, and the value is the session. This allows the Client to register the Transaction Manager and Resource Manager with the Transaction Coordinator through a session in the map. See [getty_client_session_manager.go]. (https://github.com/opentrx/seata-golang/blob/dev/pkg/client/getty_client_session_manager.go) After the Transaction Manager and Resource Manager are registered with the Transaction Coordinator, a connection can be used to send either TM messages or RM messages. We identify a connection with an RpcContext:

type RpcContext struct {
 Version string
 TransactionServiceGroup string
 ClientRole meta.TransactionRole
 ApplicationId string
 ClientId string
 ResourceSets *model.
 Session getty.
Session }

When a transaction message is received, we need to construct such an RpcContext to be used by the subsequent transaction logic. So, we will construct the following map to cache the mapping relationships:

var (
 // session -> transactionRole
 // TM will register before RM, if a session is not the TM registered, // it will be the RM registered.
 // it will be the RM registered
 session_transactionroles = sync.Map{}

 // session -> applicationId
 identified_sessions = sync.Map{}

 // applicationId -> ip -> port -> session
 client_sessions = sync.Map{}

 // applicationId -> resourceIds
 client_resources = sync.Map{}
)

In this way, the Transaction Manager and Resource Manager are registered to the Transaction Coordinator via coordinator.OnRegTmMessage(rpcMessage, session) and coordinator.OnRegRmMessage(rpcMessage, session) respectively. session) are registered with the Transaction Coordinator, the relationship between applicationId, ip, port and session is cached in the above client_sessions map, and the relationship between applicationId, ip, port and resourceIds (an application may be able to register with the Transaction Coordinator) is cached in the client_resources map. and resourceIds (there may be multiple Resource Managers for an application) in the client_resources map. When needed, we can construct an RpcContext from these mappings, which is very different from the java version of seata, so if you're interested, you can dig a little deeper. See [getty_session_manager.go`]. (https://github.com/opentrx/seata-golang/blob/dev/tc/server/getty_session_manager.go) At this point, we have analysed seata-golang the entire mechanism of the RPC communication model.

III. The Future of seata-golang​

The development of seata-golang started in April this year, and in August it basically realised the interoperability with the java version of seata 1.2 protocol. seata) protocol, implemented AT mode for mysql database (automatically coordinating the commit rollback of distributed transactions), implemented TCC mode, and used mysql to store data on the TC side, which turned TC into a stateless application to support high-availability deployment. The following figure shows the principle of AT mode: ! [image20201205-232516.png]( https://img.alicdn.com/imgextra/i3/O1CN01alqsQS1G2oQecFYIs_! !6000000000565-2-tps-1025-573.png)

There is still a lot of work to be done, such as support for the registry, support for the configuration centre, protocol interoperability with the java version of seata 1.4, support for other databases, implementation of the craft transaction coordinator, etc. We hope that developers interested in the distributed transaction problem can join in to build a perfect golang's distributed transaction framework.

If you have any questions, please feel free to join the group [group number 33069364]:

Author Bio​

Xiaomin Liu (GitHubID dk-lockdown), currently working at h3c Chengdu, is good at using Java/Go language, and has dabbled in cloud-native and microservices related technologies, currently specialising in distributed transactions. Yu Yu (github @AlexStocks), dubbo-go project and community leader, a programmer with more than 10 years of frontline experience in server-side infrastructure R&D, has participated in the improvement of Muduo/Pika/Dubbo/Sentinel-go and other well-known projects, and is currently engaged in container orchestration and service mesh work in the Trusted Native Department of ants. Currently, he is working on container orchestration and service mesh in the Trusted Native Department of AntGold.

References​

seata official: https://seata.apache.org

java version seata：https://github.com/apache/incubator-seata

seata-golang project address: https://github.com/apache/incubator-seata-go

seata-golang go night reading b站分享：https://www.bilibili.com/video/BV1oz411e72T

In Seata version 1.3.0, data source auto-proxy and manual proxy must not be mixed, otherwise it will lead to multi-layer proxy, which will lead to the following problems:

1. single data source case: cause branch transaction commit, undo_log itself is also proxied, i.e. generated undo_log for undo_log, assumed to be undo_log2, at this time, undo_log will be treated as a branch transaction; branch transaction rollback, because of the undo_log2 generated by the faulty in undo_log corresponding transaction branch rollback. When the branch transaction is rolled back, because there is a problem with the generation of undo_log2, when the transaction branch corresponding to the undo_log is rolled back, it will delete the undo_log associated with the business table, which will lead to the discovery that the business table corresponding to the business tableis rolled back and theundo_logdoesn't exist, and thus generate an additional status of 1 for theundo_log.' This time, the overall logic is already messed up, which is a very serious problem!
2. multiple data sources and logical data sources are proxied case: in addition to the problems that will occur in the case of a single data source, may also cause deadlock problems. The reason for the deadlock is that for the undo_log operation, the select for update and delete operations that should have been performed in one transaction are spread out over multiple transactions, resulting in one transaction not committing after executing the select for update, and one transaction waiting for a lock when executing the delete until the timeout expires, and then the lock will not lock until the timeout expires. until it times out.

Proxy description​

This is a layer of DataSource proxying that overrides some methods. For example, the getConnection method does not return a Connection, but a ConnectionProxy, and so on.

// DataSourceProxy

public DataSourceProxy(DataSource targetDataSource) {
this(targetDataSource, DEFAULT_RESOURCE_GROUP_ID);
}

private void init(DataSource dataSource, String resourceGroupId) {
DefaultResourceManager.get().registerResource(this); }
}

public Connection getPlainConnection() throws SQLException {
return targetDataSource.getConnection(); } public Connection getPlainConnection(); return targetDataSource.
}

@Override
public ConnectionProxy getConnection() throws SQLException {
Connection targetConnection = targetDataSource.getConnection(); } @Override public ConnectionProxy getConnection(); }
return new ConnectionProxy(this, targetConnection);
}

Manual Proxy​

That is, manually inject a DataSourceProxy as follows

@Bean
public DataSource druidDataSource() {
return new DruidDataSource()
}

@Primary
@Bean("dataSource")
public DataSourceProxy dataSource(DataSource druidDataSource) {
return new DataSourceProxy(druidDataSource); }
}

AutoProxy​

Create a proxy class for DataSource, get DataSourceProxy (create it if it doesn't exist) based on DataSource inside the proxy class, and then call the relevant methods of DataSourceProxy. The core logic is in SeataAutoDataSourceProxyCreator.

public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {
private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoDataSourceProxyCreator.class);
private final String[] excludes; private final Advisor advisor = new SeataAutoDataSourceProxyCreator.class
private final Advisor advisor = new DefaultIntroductionAdvisor(new SeataAutoDataSourceProxyAdvice());

    public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes) {
        this.excludes = excludes;
        setProxyTargetClass(!useJdkProxy);
    }

    @Override
    protected Object[] getAdvicesAndAdvisorsForBean(Class<? > beanClass, String beanName, TargetSource customTargetSource) throws BeansException {
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("Auto proxy of [{}]", beanName);
        }
        return new Object[]{advisor};
    }

    @Override
    protected boolean shouldSkip(Class<? > beanClass, String beanName) {
        return SeataProxy.class.isAssignableFrom(beanClass) ||
                DataSourceProxy.class.isAssignableFrom(beanClass) ||
                !DataSource.class.isAssignableFrom(beanClass) ||
                Arrays.asList(excludes).contains(beanClass.getName());
    }
}

public class SeataAutoDataSourceProxyAdvice implements MethodInterceptor, IntroductionInfo {
}
public Object invoke(MethodInvocation invocation) throws Throwable {
DataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) invocation.getThis());
Method method = invocation.getMethod();
Object[] args = invocation.getArguments();
Method m = BeanUtils.findDeclaredMethod(DataSourceProxy.class, method.getName(), method.getParameterTypes());
if (m ! = null) {
return m.invoke(dataSourceProxy, args); } else { m.invoke(DataSourceProxy.class, method.getName(), method.getParameterTypes())
} else {
return invocation.proceed();
}
}

    @Override
    public Class<? >[] getInterfaces() {
        return new Class[]{SeataProxy.class};
    }
}

Data Source Multi-Level Proxy​

@Bean.
@DependsOn("strangeAdapter")
public DataSource druidDataSource(StrangeAdapter strangeAdapter) {
druidDataSource(StrangeAdapter strangeAdapter) { doxx
return new DruidDataSource()
}

@Primary
@Bean("dataSource")
public DataSourceProxy dataSource(DataSource druidDataSource) {
return new DataSourceProxy(druidDataSource); }
}

1. First we inject two DataSources into the configuration class: DruidDataSource and DataSourceProxy, where DruidDataSource is used as the targetDataSource attribute of DataSourceProxy and DataSourceProxy is used as the targetDataSource attribute of DruidDataSource. DataSourceProxyis declared using the@Primary` annotation.
2. The application has automatic data source proxying enabled by default, so when calling methods related to DruidDataSource, a corresponding data source proxy DataSourceProxy2 will be created for DruidDataSource.
3. What happens when we want to get a Connection in our application?
4. first get a DataSource, because the DataSourceProxy is Primary, so we get a DataSourceProxy. 2. based on the DataSource, we create a corresponding DataSourceProxy2.
5. get a Connection based on the DataSource, i.e. get the Connection through the DataSourceProxy. At this time, we will first call the getConnection method of targetDataSource, i.e. DruidDataSource, but since the cutover will intercept DruidDataSource, according to the interception logic in step 2, we can know that a DataSourceProxy2will be created automatically, and then call theDataSourceProxy2. Then call DataSourceProxy2#getConnection, and then call DruidDataSource#getConnection. This results in a two-tier proxy, and the returned Connectionis also a two-tierConnectionProxy`.

!

The above is actually the modified proxy logic, Seata's default autoproxy will proxy the DataSourceProxy again, the consequence is that there is one more layer of proxy at this time the corresponding diagram is as follows

!

The two problems that can result from multiple layers of proxies for a data source are summarised at the beginning of the article, with case studies below.

Branching Transaction Commits

What happens when the corresponding method is executed through the ConnectionProxy? Let's take an example of a branching transaction commit involving an update operation:

1. Execute ConnectionProxy#prepareStatement, which returns a PreparedStatementProxy.
2. Execute PreparedStatementProxy#executeUpdate, PreparedStatementProxy#executeUpdate will probably do two things: execute the business SQL and commit the undo_log.

Commit business SQL​

// ExecuteTemplate#execute
if (sqlRecognizers.size() == 1) {
   SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
   switch (sqlRecognizer.getSQLType()) {
       case INSERT.
           executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType, new Class[]{StatementLoader.load(InsertExecutor.class, dbType)) { case INSERT.
                   new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class}, new
                   new Object[]{statementProxy, statementCallback, sqlRecognizer});
           statementProxy, statementCallback, sqlRecognizer}); break;
       case UPDATE: executor = new UpdateExecutor
           executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer); break; case UPDATE.
           break;
       case DELETE.
           executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer); break; case DELETE.
           break; case SELECT_FOR_UPDATE.
       case SELECT_FOR_UPDATE: executor = new SelectForUpdate.
           executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer); break; case SELECT_FOR_UPDATE.
           break; break
       default: executor = new PlainExecutor
           executor = new PlainExecutor<>(statementProxy, statementCallback); break; default.
           break;
   }
} else {
   executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers); } else { executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers); }
}

The main process is: first execute the business SQL, then execute the commit method of the ConnectionProxy, in which the corresponding undo_log SQL will be executed for us, and then commit the transaction.

PreparedStatementProxy#executeUpdate =>
ExecuteTemplate#execute =>
BaseTransactionalExecutor#execute =>
AbstractDMLBaseExecutor#doExecute =>
AbstractDMLBaseExecutor#executeAutoCommitTrue =>
AbstractDMLBaseExecutor#executeAutoCommitFalse => In this step, the statementCallback#execute method will be triggered, i.e. the native PreparedStatement#executeUpdate method will be called.
ConnectionProxy#commit
ConnectionProxy#processGlobalTransactionCommit

UNDO_LOG insert​

// ConnectionProxy#processGlobalTransactionCommit
private void processGlobalTransactionCommit() throws SQLException {
   try {
       // Register for a branch transaction, simply understand that a request is sent to the server, and then the server inserts a record into the branch_table table.
       register();
   } catch (TransactionException e) {
       // If there is no for update sql, it will register directly before commit, then not only insert a branch record, but also lock information for the competition, the following exception is generally thrown in the registration did not get the lock, generally is pure update statement concurrency will trigger the competition lock failure exception @FUNKYE
       recognizeLockKeyConflictException(e, context.buildLockKeys());
   }
   try {
       // undo_log handling, expect targetConnection handling @1
       UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this); // Commit local transaction, expect targetConnection.

       // Commit the local transaction, expecting it to be handled by targetConnection @2
       targetConnection.commit(); } catch (Throwable ex)
   } catch (Throwable ex) {
       LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex); report(false); }
       report(false); } catch (Throwable ex); }
       throw new SQLException(ex);
   }
   if (IS_REPORT_SUCCESS_ENABLE) {
       report(true); }
   }
   context.reset();
}

1. undo_log processing @1, parses the undo_log involved in the current transaction branch and writes it to the database using TargetConnection.

   public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
   ConnectionContext connectionContext = cp.getContext();
   if (!connectionContext.hasUndoLog()) {
   return;
   }

   String xid = connectionContext.getXid(); long branchId = connectionContext.hasUndoLog(); { return; }
   long branchId = connectionContext.getBranchId(); }

   BranchUndoLog branchUndoLog = new BranchUndoLog(); branchUndoLog.setBranchId = connectionContext.getBranchId(); }
   branchUndoLog.setXid(xid); branchUndoLog.
   branchUndoLog.setBranchId(branchId); branchUndoLog.
   branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());

   UndoLogParser parser = UndoLogParserFactory.getInstance();
   byte[] undoLogContent = parser.encode(branchUndoLog);

   if (LOGGER.isDebugEnabled()) {
   LOGGER.debug("Flushing UNDO LOG: {}", new String(undoLogContent, Constants.DEFAULT_CHARSET));
   }

   insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName()), undoLogContent,cp.getTargetConnection());
   }

2. Commit local transaction @2, i.e., commit the transaction via TargetConnection. That is, the same TargetConnection is used for service SQL execution, undo_log write, and i.e. transaction commit.

lcn's built-in database solution, lcn is to write undolog to his embedded h2 (I forget if it's this one) database, at this time it will become two local transactions, one is h2's undolog insertion transaction, one is the transaction of the business database, if the business database is abnormal after the insertion of the h2, lcn's solution will be data redundancy, roll back the data. data is the same, delete undolog and rollback business data is not a local transaction. But the advantage of lcn is the invasion of small, do not need to add another undolog table. Thanks to @FUNKYE for the advice, I don't know much about lcn, I'll look into it when I get a chance!

Branch Transaction Rollback

1. Server sends a rollback request to Client. 2.
2. Client receives the request from Server, and after a series of processing, it ends up in the DataSourceManager#branchRollback method. 3.
3. first according to the resourceId from the DataSourceManager.dataSourceCache to get the corresponding DataSourceProxy, at this time for the masterSlaveProxy (rollback stage we do not test the proxy data source, simple and direct, anyway, the final get all the TragetConnection)
4. According to the xid and branchId sent from the Server side to find the corresponding undo_log and parse its rollback_info attribute, each undo_log may be parsed out of more than one SQLUndoLog, each SQLUndoLog can be interpreted as an operation. For example, if a branch transaction updates table A and then table B, the undo_log generated for the branch transaction contains two SQLUndoLogs: the first SQLUndoLog corresponds to the snapshot before and after the update of table A; the second SQLUndoLog corresponds to the snapshot before and after the update of table B.
5. for each SQLUndoLog execute the corresponding rollback operation, for example, a SQLUndoLog corresponds to the operation INSERT, then its corresponding rollback operation is DELETE.
6. Delete the undo_log based on the xid and branchId.

// AbstractUndoLogManager#undo removes some non-critical code

public void undo(DataSourceProxy dataSourceProxy, String xid, long branchId) throws TransactionException {
   Connection conn = null;
   ResultSet rs = null;
   PreparedStatement selectPST = null;
   boolean originalAutoCommit = true; for (; ; ) {

   for (; ; ) {
       try {
           // Get the connection to the original data source, we don't care about the proxy data source in the rollback phase, we'll end up with the TargetConnection.
           conn = dataSourceProxy.getPlainConnection(); // Get the connection to the native data source.

           // Put the rollback operation in a local transaction and commit it manually, making sure that the final business SQL operation is committed along with the undo_log delete operation.
           if (originalAutoCommit = conn.getAutoCommit()) {
               conn.setAutoCommit(false);
           }

           // Query undo_log based on xid and branchId, note the SQL statement SELECT * FROM undo_log WHERE branch_id = ? AND xid = ? FOR UPDATE
           selectPST = conn.prepareStatement(SELECT_UNDO_LOG_SQL);
           selectPST.setLong(1, branchId); selectPST.setString(1, branchId); selectPST.setString(1, branchId)
           selectPST.setString(2, xid);
           rs = selectPST.executeQuery(); boolean exists = false; rs = selectPST.

           boolean exists = false; while (rs.next())
           while (rs.next()) {
               exists = true; boolean exists = false; while (rs.next()) {
               // status == 1 undo_log is not processed, related to anti-suspension
               if (!canUndo(state)) {
                   return; }
               }

               // Parsing the undo_log
               byte[] rollbackInfo = getRollbackInfo(rs); // Parsing the undo_log.
               BranchUndoLog branchUndoLog = UndoLogParserFactory.getInstance(serialiser).decode(rollbackInfo);
               try {
                   setCurrentSerializer(parser.getName());
                   List<SQLUndoLog> sqlUndoLogs = branchUndoLog.getSqlUndoLogs(); if (sqlUndoLog.getSqlUndoLogs(parser.getName()); } }
                   if (sqlUndoLogs.size() > 1) {
                       Collections.reverse(sqlUndoLogs);
                   }
                   for (SQLUndoLog sqlUndoLog : sqlUndoLogs) {
                       AbstractUndoExecutor undoExecutor = UndoExecutorFactory.getUndoExecutor(dataSourceProxy.getDbType(), sqlUndoLog);
                       // Execute the corresponding rollback operation
                       undoExecutor.executeOn(conn);
                   }
               }
           }

           // If (exists) { undoExecutor.executeOn(conn); }
           if (exists) {
               LOGGER.error("\n delete from undo_log where xid={} AND branchId={} \n", xid, branchId);
               deleteUndoLog(xid, branchId, conn);
               conn.commit();
           // and anti-suspension related If no undo_log is found based on xid and branchId, it means that there is an exception in the branch transaction: for example, the business process timed out, resulting in a global transaction rollback, but the business undo_log was not inserted at that time.
           } else {
               LOGGER.error("\n insert into undo_log xid={},branchId={} \n", xid, branchId);
               insertUndoLogWithGlobalFinished(xid, branchId, UndoLogParserFactory.getInstance(), conn);
               conn.commit();
           }
           return; }
       } catch (Throwable e) {
           throw new BranchTransactionException(BranchRollbackFailed_Retriable, String
               .format("Branch session rollback failed and try again later xid = %s branchId = %s %s", xid,branchId, e.getMessage()), e); }
       }
   }
}

There are several notes:

1. rollback does not take into account data source proxying, and ends up using TargetConnection.
2. set atuoCommit to false, i.e. you need to commit the transaction manually
3. for update is added when querying the undo_log based on xid and branchId, i.e. the transaction will hold the lock for this undo_log until all rollbacks are complete, as it is not until they are done that the

Multi-Tier Proxy Issues

Several issues that can be caused by multi-tier proxying of data sources have been mentioned at the beginning of the article, focusing on analysing why the above issues are caused:

Impact on branch transaction commits​

Let's start by analysing what happens if we use a two-tier proxy. Let's analyse it from two aspects: business SQL and undo_log

1. business SQL

   PreparedStatementProxy1.executeUpdate =>
   statementCallback#executeUpdate(PreparedStatementProxy2#executeUpdate) =>
   PreparedStatement#executeUpdate

It doesn't seem to matter, it's just an extra loop, and it's still executed through PreparedStatement in the end.

2. undo_log

ConnectionProxy1#getTargetConnection ->
ConnectionProxy2#prepareStatement ->
PreparedStatementProxy2#executeUpdate ->
PreparedStatement#executeUpdate (native undo_log write, before generating undo_log2 (the undo_log of undo_log) for that undo_log) ->
ConnectionProxy2#commit ->
ConnectionProxy2#processGlobalTransactionCommit(write undo_log2) ->
ConnectionProxy2#getTargetConnection ->
TargetConnection#prepareStatement ->
PreparedStatement#executeUpdate

Impact on branch transaction rollback​

Why isn't the undo_log deleted after a transaction rollback?

It is not actually not deleted. As I said before, the two-tier proxy causes the undo_log to be treated as a branch transaction, so it generates an undo_log for that undo_log (assuming it's undo_log2), and undo_log2 is generated wrongly (which is fine, it should be generated this way), which results in the business-table-associated undo_log being deleted when rolling back. This leads to a rollback that deletes the undo_log associated with the business table, which ultimately leads to the business table corresponding to the transaction branch rolling back to find that the undo_log does not exist, thus generating one more undo_log with a status of 1.

Before the rollback

// undo_log
84 59734070967644161 172.16.120.59:23004:59734061438185472 serializer=jackson 1.1KB 0
85 59734075254222849 172.16.120.59:23004:59734061438185472 serializer=jackson 4.0KB 0

// branch_table
59734070967644161 172.16.120.59:23004:59734061438185472 jdbc:mysql://172.16.248.10:3306/tuya_middleware
59734075254222849 172.16.120.59:23004:59734061438185472 jdbc:mysql://172.16.248.10:3306/tuya_middleware

// lock_table
jdbc:mysql://xx^^^seata_storage^^^1 59734070967644161 jdbc:mysql://172.16.248.10:3306/tuya_middleware seata_storage 1
jdbc:mysql://xx^^^^undo_log^^^^84 59734075254222849 jdbc:mysql://172.16.248.10:3306/tuya_middleware undo_log 84

After the rollback

// An undo_log with status 1 was generated, corresponding to the log: undo_log added with GlobalFinished
86 59734070967644161 172.16.120.59:23004:59734061438185472 serializer=jackson 1.0Byte 1

Problem analysis​

1. find the corresponding undo_log log based on xid and branchId
2. parse the undo_log, mainly its rollback_info field, rollback_info is a SQLUndoLog collection, each SQLUndoLog corresponds to an operation, which contains a snapshot before and after the operation, and then perform a corresponding rollback
3. Delete undo_log logs based on xid and branchId.

Because of the two-tier proxy problem, an undo_log becomes a branch transaction, so when a rollback occurs, we also need to rollback the undo_log branch transaction: 1, first according to the xid and branchId to find the corresponding undo_log and parse its rollback_info attribute, here the parsed rollback_info contains two SQLUndoLog. Why are there two?

If you think about it, you can understand that the first layer of proxy operations on seata_storage are put into the cache, which should be cleared after execution, but because of the two-tier proxy, the process is not finished at this time. When it's the second tier proxy's turn to operate on undo_log, it puts that operation into the cache, and at that point there are two operations in the cache, UPDATE for seata_storage and INSERT for undo_log. So it's easy to see why the undo_log operation is extra large (4KB) because it has two operations in its rollback_info.

One thing to note is that the first SQLUndoLog corresponds to the after snapshot, which has branchId=59734070967644161 pk=84, i.e., branchIdcorresponding to theseata_storage branch and undo_log corresponding to the seata_storage PK. In other words, the undo_log rollback deletes the seata_storage corresponding undo_log`. How to delete the undo_log itself? In the next logic, it will be deleted according to xid and branchId.

2. Parsing the first SQLUndoLog, it corresponds to the INSERToperation ofundo_log, so its corresponding rollback operation is DELETE. Because undo_logis treated as a business table at this point. So this step will delete the59734075254222849corresponding to theundo_log, **but this is actually the corresponding business table corresponding to the corresponding undo_log`**.

3, parse the second SQLUndoLog, at this time corresponds to the seata_storage UPDATE operation, this time will be through the snapshot of the seata_storage corresponding to the recovery of records

4、Delete the undo_log log according to xid and branchId, here the deletion is the undo_log of undo_log , i.e. undo_log2. So, by this point, both undo_logs have been deleted.

5. Next, roll back seata_storage, because at this time its corresponding undo_log has been deleted in step 2, so at this time can not check the undo_log, and then regenerate a status == 1 undo_log.

Case Study

Background​

1. Three data sources are configured: two physical data sources and one logical data source, but the corresponding connection addresses of the two physical data sources are the same. Is this interesting?

@Bean("dsMaster")
DynamicDataSource dsMaster() {
   return new DynamicDataSource(masterDsRoute);
}

@Bean("dsSlave")
DynamicDataSource dsSlave() {
   return new DynamicDataSource(slaveDsRoute); }
}

@Primary
@Bean("masterSlave")
DataSource masterSlave(@Qualifier("dsMaster") DataSource dataSourceMaster,
                       @Qualifier("dsSlave") DataSource dataSourceSlave) throws SQLException {
   Map<String, DataSource> dataSourceMap = new HashMap<>(2);
   // Master database
   dataSourceMap.put("dsMaster", dataSourceMaster);
   //slave
   dataSourceMap.put("dsSlave", dataSourceSlave); // Configure read/write separation rules.
   // Configure read/write separation rules
   MasterSlaveRuleConfiguration masterSlaveRuleConfig = new MasterSlaveRuleConfiguration(
           "masterSlave", "dsMaster", Lists.newArrayList("dsSlave")
   );
   Properties shardingProperties = new Properties();
   shardingProperties.setProperty("sql.show", "true");
   shardingProperties.setProperty("sql.simple", "true");
   // Get the data source object
   DataSource dataSource = MasterSlaveDataSourceFactory.createDataSource(dataSourceMap, masterSlaveRuleConfig, shardingProperties);
   log.info("datasource initialised!");
   return dataSource;˚
}

!

2, open seata's data source dynamic proxy, according to seata's data source proxy logic can be known, will eventually generate three proxy data sources, the relationship between the native data source and the proxy data source is cached in the DataSourceProxyHolder.dataSourceProxyMap, if the native data source and the proxy data source corresponds to the relationship is as follows:

dsMaster(DynamicDataSource) => dsMasterProxy(DataSourceProxy)
dsSlave(DynamicDataSource) => dsSlaveProxy(DataSourceProxy)
masterSlave(MasterSlaveDataSource) => masterSlaveProxy(DataSourceProxy)

So, ultimately, the three data sources that exist in the IOC container are: dsMasterProxy, dsSlaveProxy, masterSlaveProxy. According to the @Primary feature, when we get a DataSource from the container, the default data source returned is the proxy masterSlaveProxy.

I haven't studied shardingjdbc specifically, but just guessed its working mechanism based on the code I saw during the debug.

masterSlaveProxy can be seen as MasterSlaveDataSource wrapped by DataSourceProxy. We can venture to guess that MasterSlaveDataSource is not a physical data source, but a logical data source, which can simply be thought of as containing routing logic. When we get a connection, we will use the routing rules inside to select a specific physical data source, and then get a real connection through that physical data source. The routing rules should be able to be defined by yourself. According to the phenomenon observed when debugging, the default routing rules should be:

1. for select read operations, will be routed to the slave library, that is, our dsSlave

2. for update write operations, will be routed to the master library, that is, our dsMaster

3. When each DataSourceProxy is initialised, it will parse the connection address of that real DataSource, and then maintain that connection address and the DataSourceProxy itself in DataSourceManager.dataSourceCache. The DataSourceManager.dataSourceCache is used for rollback: when rolling back, it finds the corresponding DataSourceProxy based on the connection address, and then does the rollback operation based on that DataSourceProxy. But we can find this problem, these three data sources are resolved to the same connection address, that is, the key is duplicated, so in the DataSourceManager.dataSourceCache, when the connection place is the same, after the registration of the data source will overwrite the existing one. That is: DataSourceManager.dataSourceCache ultimately exists masterSlaveProxy, that is to say, will ultimately be rolled back through the masterSlaveProxy, this point is very important.

4, the table involved: very simple, we expect a business table seata_account, but because of the duplicate proxy problem, resulting in seata will undo_log also as a business table

1. seata_account
2. undo_log

OK, here's a brief background, moving on to the Seata session

Requirements​

We have a simple requirement to perform a simple update operation inside a branch transaction to update the count value of seata_account. After the update, manually throw an exception that triggers a rollback of the global transaction. To make it easier to troubleshoot and reduce interference, we use one branch transaction in the global transaction and no other branch transactions.SQL is as follows.

update seata_account set count = count - 1 where id = 100;

Problems​

Client: In the console log, the following logs are printed over and over again

1. the above logs are printed at 20s intervals, and I checked the value of the innodb_lock_wait_timeout property of the database, and it happens to be 20, which means that every time a rollback request comes through, the rollback fails because of the timeout for acquiring the lock (20).
2. Why is it not printed once after 20s? Because the server side will have a timer to process the rollback request.

// Branch rollback starts
Branch rollback start: 172.16.120.59:23004:59991911632711680 59991915571163137 jdbc:mysql://172.16.248.10:3306/tuya_middleware

// undo_log transaction branch The original action corresponds to insert, so it rolls back to delete.
undoSQL undoSQL=DELETE FROM undo_log WHERE id = ?  and PK=[[id,139]]
// Since the corresponding operation of the first-level agent is also in the context, when the undo_log branch transaction commits, the corresponding undo_log contains two actions
undoSQL undoSQL=UPDATE seata_account SET money = ? WHERE id = ?  and PK=[[id,1]].

// After the branch transaction has been rolled back, delete the corresponding undo_log for that branch transaction
delete from undo_log where xid=172.16.120.59:23004:59991911632711680 AND branchId=59991915571163137

// Threw an exception indicating that the rollback failed because `Lock wait timeout exceeded`, and failed when deleting the undo_log based on the xid and branchId because a lock acquisition timeout occurred, indicating that there was another operation that held a lock on the record that was not released.
branchRollback failed. branchType:[AT], xid:[172.16.120.59:23004:59991911632711680], branchId:[59991915571163137], resourceId:[jdbc. mysql://172.16.248.10:3306/tuya_middleware], applicationData:[null]. reason:[Branch session rollback failed and try again later xid = 172.16.120.59:23004:59991911632711680 branchId = 59991915571163137 Lock wait timeout exceeded; try restarting transaction]

Server: the following log is printed every 20s, indicating that the server is constantly retrying to send a rollback request

Rollback branch transaction failed and will retry, xid = 172.16.120.59:23004:59991911632711680 branchId = 59991915571163137

The SQL involved in the process is roughly as follows:

1. SELECT * FROM undo_log WHERE branch_id = ? AND xid = ? FOR UPDATE slaveDS
2. SELECT * FROM undo_log WHERE (id ) in ( (?)  ) slaveDS
3. DELETE FROM undo_log WHERE id = ?                                masterDS
4. SELECT * FROM seata_account WHERE (id ) in ( (?)  ) masterDS
5. UPDATE seata_account SET money = ? WHERE id = ?                      masterDS
6. DELETE FROM undo_log WHERE branch_id = ? AND xid = ?                 masterDS

At this point, check the database transaction status, lock status, lock wait relationship 1, check the current transaction being executed

SELECT * FROM information_schema.INNODB_TRX.

!

2. Check the current lock status

SELECT * FROM information_schema.INNODB_LOCKs;

!

3. Check the current lock wait relationship

SELECT * FROM information_schema.INNODB_LOCK_waits;

SELECT
 block_trx.trx_mysql_thread_id AS sessionID that already holds a lock, request_trx.
 request_trx.trx_mysql_thread_id AS the sessionID that is requesting the lock,
 block_trx.trx_query AS the SQL statement that already holds the lock, request_trx.
 request_trx.trx_query AS the SQL statement for which the lock is being requested,
 waits.blocking_trx_id AS Transaction ID that already holds the lock, waits.requesting_trx.trx_query
 waits.requesting_trx_id AS 正在申请锁的事务ID,
 waits.requested_lock_id AS the ID of the lock object, waits.
 locks.lock_table AS lock_table, -- table locked by the lock object
 locks.lock_type AS lock_type, -- lock type
 locks.lock_mode AS lock_mode -- lock mode
FROM
 information_schema.innodb_lock_waits AS waits
 INNER JOIN information_schema.innodb_trx AS block_trx ON waits.blocking_trx_id = block_trx.trx_id
 INNER JOIN information_schema.innodb_trx AS request_trx ON waits.requesting_trx_id = request_trx.trx_id
 INNER JOIN information_schema.innodb_locks AS locks ON waits.requested_lock_id = locks.lock_id;