add timeout

conf.sample.yml

@@ -42,13 +42,15 @@
 #   Target: 'etcd://localhost:2379/dtmservice' # register dtm server to this url
 #   EndPoint: 'localhost:36790'
 
-# the unit of following configurations is second
-
+### the unit of following configurations is second
 # TransCronInterval: 3 # the interval to poll unfinished global transaction for every dtm process
 # TimeoutToFail: 35 # timeout for XA, TCC to fail. saga's timeout default to infinite, which can be overwritten in saga options
 # RetryInterval: 10 # the subtrans branch will be retried after this interval
+# RequestTimeout: 3 # the timeout of HTTP/gRPC request in dtm
 
-# UpdateBranchAsyncGoroutineNum: 1 # num of async goroutine to update branch status
 # LogLevel: 'info' # default: info. can be debug|info|warn|error
 # HttpPort: 36789
 # GrpcPort: 36790
+
+### advanced options
+# UpdateBranchAsyncGoroutineNum: 1 # num of async goroutine to update branch status

dtmcli/dtmimp/vars.go

@@ -38,9 +38,6 @@ var PassthroughHeaders = []string{}
 var BarrierTableName = "dtm_barrier.barrier"
 
 func init() {
-	// RestyClient.SetTimeout(3 * time.Second)
-	// RestyClient.SetRetryCount(2)
-	// RestyClient.SetRetryWaitTime(1 * time.Second)
 	RestyClient.OnBeforeRequest(func(c *resty.Client, r *resty.Request) error {
 		r.URL = MayReplaceLocalhost(r.URL)
 		logger.Debugf("requesting: %s %s %s", r.Method, r.URL, MustMarshalString(r.Body))

dtmsvr/config/config.go

@@ -66,6 +66,7 @@ type configType struct {
 	TransCronInterval             int64        `yaml:"TransCronInterval" default:"3"`
 	TimeoutToFail                 int64        `yaml:"TimeoutToFail" default:"35"`
 	RetryInterval                 int64        `yaml:"RetryInterval" default:"10"`
+	RequestTimeout                int64        `yaml:"RequestTimeout" default:"3"`
 	HTTPPort                      int64        `yaml:"HttpPort" default:"36789"`
 	GrpcPort                      int64        `yaml:"GrpcPort" default:"36790"`
 	MicroService                  MicroService `yaml:"MicroService"`

dtmsvr/svr.go

@@ -7,11 +7,14 @@
 package dtmsvr
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"time"
 
+	"github.com/dtm-labs/dtm/dtmcli"
 	"github.com/dtm-labs/dtm/dtmcli/logger"
+	"github.com/dtm-labs/dtm/dtmgrpc"
 	"github.com/dtm-labs/dtm/dtmgrpc/dtmgimp"
 	"github.com/dtm-labs/dtm/dtmgrpc/dtmgpb"
 	"github.com/dtm-labs/dtm/dtmutil"
@@ -23,6 +26,12 @@ import (
 // StartSvr StartSvr
 func StartSvr() {
 	logger.Infof("start dtmsvr")
+	dtmcli.GetRestyClient().SetTimeout(time.Duration(conf.RequestTimeout) * time.Second)
+	dtmgrpc.AddUnaryInterceptor(func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		ctx2, cancel := context.WithTimeout(ctx, time.Duration(conf.RequestTimeout)*time.Second)
+		defer cancel()
+		return invoker(ctx2, method, req, reply, cc, opts...)
+	})
 	app := dtmutil.GetGinApp()
 	app = httpMetrics(app)
 	addRoute(app)