mirror of https://github.com/dtm-labs/dtm.git
yedf2
4 years ago
39 changed files with 691 additions and 219 deletions
@ -0,0 +1,29 @@ |
|||
package dtmcli |
|||
|
|||
import "fmt" |
|||
|
|||
// ConcurrentSaga struct of concurrent saga
|
|||
type ConcurrentSaga struct { |
|||
Saga |
|||
orders map[int][]int |
|||
} |
|||
|
|||
// NewConcurrentSaga create a concurrent saga
|
|||
func NewConcurrentSaga(server string, gid string) *ConcurrentSaga { |
|||
return &ConcurrentSaga{Saga: Saga{TransBase: *NewTransBase(gid, "csaga", server, "")}, orders: map[int][]int{}} |
|||
} |
|||
|
|||
// AddStepOrder specify that step should be after preSteps. Step is larger than all the element in preSteps
|
|||
func (s *ConcurrentSaga) AddStepOrder(step int, preSteps []int) *ConcurrentSaga { |
|||
PanicIf(step > len(s.Steps), fmt.Errorf("step value: %d is invalid. which cannot be larger than total steps: %d", step, len(s.Steps))) |
|||
s.orders[step] = preSteps |
|||
return s |
|||
} |
|||
|
|||
// Submit submit the saga trans
|
|||
func (s *ConcurrentSaga) Submit() error { |
|||
if len(s.orders) > 0 { |
|||
s.CustomData = MustMarshalString(M{"orders": s.orders}) |
|||
} |
|||
return s.callDtm(s, "submit") |
|||
} |
|||
@ -0,0 +1,175 @@ |
|||
package dtmsvr |
|||
|
|||
import ( |
|||
"time" |
|||
|
|||
"github.com/yedf/dtm/common" |
|||
"github.com/yedf/dtm/dtmcli" |
|||
"gorm.io/gorm/clause" |
|||
) |
|||
|
|||
type transCSagaProcessor struct { |
|||
*TransGlobal |
|||
} |
|||
|
|||
func init() { |
|||
registorProcessorCreator("csaga", func(trans *TransGlobal) transProcessor { return &transCSagaProcessor{TransGlobal: trans} }) |
|||
} |
|||
|
|||
func (t *transCSagaProcessor) GenBranches() []TransBranch { |
|||
return genSagaBranches(t.TransGlobal) |
|||
} |
|||
|
|||
type cSagaCustom struct { |
|||
Orders map[int][]int `json:"orders"` |
|||
} |
|||
|
|||
func isPreconditionsSucceed(branches []TransBranch, pres []int) bool { |
|||
for _, pre := range pres { |
|||
if branches[pre].Status != dtmcli.StatusSucceed { |
|||
return false |
|||
} |
|||
} |
|||
return true |
|||
} |
|||
|
|||
type branchResult struct { |
|||
index int |
|||
status string |
|||
started bool |
|||
branchType string |
|||
} |
|||
|
|||
func (t *transCSagaProcessor) ProcessOnce(db *common.DB, branches []TransBranch) error { |
|||
if t.Status == dtmcli.StatusFailed || t.Status == dtmcli.StatusSucceed { |
|||
return nil |
|||
} |
|||
n := len(branches) |
|||
|
|||
orders := map[int][]int{} |
|||
if t.CustomData != "" { |
|||
csc := cSagaCustom{Orders: map[int][]int{}} |
|||
dtmcli.MustUnmarshalString(t.CustomData, &csc) |
|||
for k, v := range csc.Orders { // new branches is doubled, so change the order value
|
|||
orders[2*k+1] = []int{} |
|||
for j := 0; j < len(v); j++ { |
|||
orders[2*k+1] = append(orders[2*k+1], csc.Orders[k][j]*2+1) |
|||
} |
|||
} |
|||
} |
|||
// resultStats
|
|||
var rsActionToStart, rsActionDone, rsActionFailed, rsActionSucceed, rsCompensateToStart, rsCompensateDone, rsCompensateSucceed int |
|||
branchResults := make([]branchResult, n) // save the branch result
|
|||
for i := 0; i < n; i++ { |
|||
b := branches[i] |
|||
if b.BranchType == dtmcli.BranchAction { |
|||
if b.Status == dtmcli.StatusPrepared || b.Status == dtmcli.StatusDoing { |
|||
rsActionToStart++ |
|||
} else if b.Status == dtmcli.StatusFailed { |
|||
rsActionFailed++ |
|||
} |
|||
} |
|||
branchResults[i] = branchResult{status: branches[i].Status, branchType: branches[i].BranchType} |
|||
} |
|||
stopChan := make(chan branchResult, n) |
|||
asyncExecBranch := func(i int) { |
|||
var err error |
|||
defer func() { |
|||
if x := recover(); x != nil { |
|||
err = dtmcli.AsError(x) |
|||
} |
|||
stopChan <- branchResult{index: i, status: branches[i].Status, branchType: branches[i].BranchType} |
|||
if err != nil { |
|||
dtmcli.LogRedf("exec branch error: %v", err) |
|||
} |
|||
}() |
|||
err = t.execBranch(db, &branches[i]) |
|||
} |
|||
needRollback := func(i int) bool { |
|||
br := &branchResults[i] |
|||
return !br.started && br.branchType == dtmcli.BranchCompensate && br.status != dtmcli.StatusSucceed && branchResults[i+1].branchType == dtmcli.BranchAction && branchResults[i+1].status != dtmcli.StatusPrepared |
|||
} |
|||
pickAndRun := func(branchType string) { |
|||
toRun := []int{} |
|||
for current := 0; current < n; current++ { |
|||
br := &branchResults[current] |
|||
if br.branchType == branchType && branchType == dtmcli.BranchAction { |
|||
if (br.status == dtmcli.StatusPrepared || br.status == dtmcli.StatusDoing) && |
|||
!br.started && isPreconditionsSucceed(branches, orders[current]) { |
|||
br.status = dtmcli.StatusDoing |
|||
toRun = append(toRun, current) |
|||
} |
|||
} else if br.branchType == branchType && branchType == dtmcli.BranchCompensate { |
|||
if needRollback(current) { |
|||
toRun = append(toRun, current) |
|||
} |
|||
} |
|||
} |
|||
if branchType == dtmcli.BranchAction && len(toRun) > 0 { |
|||
updates := make([]TransBranch, len(toRun)) |
|||
for i, b := range toRun { |
|||
updates[i].ID = branches[b].ID |
|||
branches[b].Status = dtmcli.StatusDoing |
|||
updates[i].Status = dtmcli.StatusDoing |
|||
} |
|||
dbGet().Must().Clauses(clause.OnConflict{ |
|||
OnConstraint: "trans_branch_pkey", |
|||
DoUpdates: clause.AssignmentColumns([]string{"status"}), |
|||
}).Create(updates) |
|||
} else if branchType == dtmcli.BranchCompensate { |
|||
rsCompensateToStart = len(toRun) |
|||
} |
|||
for _, b := range toRun { |
|||
branchResults[b].started = true |
|||
go asyncExecBranch(b) |
|||
} |
|||
} |
|||
processorTimeout := func() bool { |
|||
return time.Since(t.processStarted)+NowForwardDuration > time.Duration(t.getRetryInterval()-3)*time.Second |
|||
} |
|||
waitOnceForDone := func() { |
|||
select { |
|||
case r := <-stopChan: |
|||
br := &branchResults[r.index] |
|||
br.status = r.status |
|||
if r.branchType == dtmcli.BranchAction { |
|||
rsActionDone++ |
|||
if r.status == dtmcli.StatusFailed { |
|||
rsActionFailed++ |
|||
} else if r.status == dtmcli.StatusSucceed { |
|||
rsActionSucceed++ |
|||
} |
|||
} else { |
|||
rsCompensateDone++ |
|||
if r.status == dtmcli.StatusSucceed { |
|||
rsCompensateSucceed++ |
|||
} |
|||
} |
|||
dtmcli.Logf("branch done: %v", r) |
|||
case <-time.After(time.Duration(time.Second * 3)): |
|||
dtmcli.Logf("wait once for done") |
|||
} |
|||
} |
|||
|
|||
for t.Status == dtmcli.StatusSubmitted && !t.isTimeout() && rsActionFailed == 0 && rsActionDone != rsActionToStart && !processorTimeout() { |
|||
pickAndRun(dtmcli.BranchAction) |
|||
waitOnceForDone() |
|||
} |
|||
if t.Status == dtmcli.StatusSubmitted && rsActionFailed == 0 && rsActionToStart == rsActionSucceed { |
|||
t.changeStatus(db, dtmcli.StatusSucceed) |
|||
return nil |
|||
} |
|||
if t.Status == dtmcli.StatusSubmitted && (rsActionFailed > 0 || t.isTimeout()) { |
|||
t.changeStatus(db, dtmcli.StatusAborting) |
|||
} |
|||
if t.Status == dtmcli.StatusAborting { |
|||
pickAndRun(dtmcli.BranchCompensate) |
|||
for rsCompensateDone != rsCompensateToStart && !processorTimeout() { |
|||
waitOnceForDone() |
|||
} |
|||
} |
|||
if (t.Status == dtmcli.StatusSubmitted || t.Status == dtmcli.StatusAborting) && rsActionFailed > 0 && rsCompensateToStart == rsCompensateSucceed { |
|||
t.changeStatus(db, dtmcli.StatusFailed) |
|||
} |
|||
return nil |
|||
} |
|||
@ -0,0 +1,53 @@ |
|||
package test |
|||
|
|||
import ( |
|||
"fmt" |
|||
"testing" |
|||
|
|||
"github.com/stretchr/testify/assert" |
|||
"github.com/yedf/dtm/common" |
|||
"github.com/yedf/dtm/dtmcli" |
|||
"github.com/yedf/dtm/examples" |
|||
) |
|||
|
|||
func TestSqlDB(t *testing.T) { |
|||
asserts := assert.New(t) |
|||
db := common.DbGet(config.DB) |
|||
barrier := &dtmcli.BranchBarrier{ |
|||
TransType: "saga", |
|||
Gid: "gid2", |
|||
BranchID: "branch_id2", |
|||
BranchType: dtmcli.BranchAction, |
|||
} |
|||
db.Must().Exec("insert into dtm_barrier.barrier(trans_type, gid, branch_id, branch_type, reason) values('saga', 'gid1', 'branch_id1', 'action', 'saga')") |
|||
tx, err := db.ToSQLDB().Begin() |
|||
asserts.Nil(err) |
|||
err = barrier.Call(tx, func(db dtmcli.DB) error { |
|||
dtmcli.Logf("rollback gid2") |
|||
return fmt.Errorf("gid2 error") |
|||
}) |
|||
asserts.Error(err, fmt.Errorf("gid2 error")) |
|||
dbr := db.Model(&BarrierModel{}).Where("gid=?", "gid1").Find(&[]BarrierModel{}) |
|||
asserts.Equal(dbr.RowsAffected, int64(1)) |
|||
dbr = db.Model(&BarrierModel{}).Where("gid=?", "gid2").Find(&[]BarrierModel{}) |
|||
asserts.Equal(dbr.RowsAffected, int64(0)) |
|||
barrier.BarrierID = 0 |
|||
tx2, err := db.ToSQLDB().Begin() |
|||
asserts.Nil(err) |
|||
err = barrier.Call(tx2, func(db dtmcli.DB) error { |
|||
dtmcli.Logf("submit gid2") |
|||
return nil |
|||
}) |
|||
asserts.Nil(err) |
|||
dbr = db.Model(&BarrierModel{}).Where("gid=?", "gid2").Find(&[]BarrierModel{}) |
|||
asserts.Equal(dbr.RowsAffected, int64(1)) |
|||
} |
|||
|
|||
func TestHttp(t *testing.T) { |
|||
resp, err := dtmcli.RestyClient.R().SetQueryParam("panic_string", "1").Post(examples.Busi + "/TestPanic") |
|||
assert.Nil(t, err) |
|||
assert.Contains(t, resp.String(), "panic_string") |
|||
resp, err = dtmcli.RestyClient.R().SetQueryParam("panic_error", "1").Post(examples.Busi + "/TestPanic") |
|||
assert.Nil(t, err) |
|||
assert.Contains(t, resp.String(), "panic_error") |
|||
} |
|||
@ -0,0 +1,72 @@ |
|||
package test |
|||
|
|||
import ( |
|||
"testing" |
|||
|
|||
"github.com/stretchr/testify/assert" |
|||
"github.com/yedf/dtm/dtmcli" |
|||
"github.com/yedf/dtm/examples" |
|||
) |
|||
|
|||
func TestCSaga(t *testing.T) { |
|||
csagaNormal(t) |
|||
csagaRollback(t) |
|||
csagaRollback2(t) |
|||
csagaCommittedOngoing(t) |
|||
} |
|||
|
|||
func genCSaga(gid string, outFailed bool, inFailed bool) *dtmcli.ConcurrentSaga { |
|||
dtmcli.Logf("beginning a concurrent saga test ---------------- %s", gid) |
|||
csaga := dtmcli.NewConcurrentSaga(examples.DtmServer, gid) |
|||
req := examples.GenTransReq(30, outFailed, inFailed) |
|||
csaga.Add(examples.Busi+"/TransOut", examples.Busi+"/TransOutRevert", &req) |
|||
csaga.Add(examples.Busi+"/TransIn", examples.Busi+"/TransInRevert", &req) |
|||
return csaga |
|||
} |
|||
|
|||
func csagaNormal(t *testing.T) { |
|||
csaga := genCSaga("gid-noraml-csaga", false, false) |
|||
csaga.Submit() |
|||
WaitTransProcessed(csaga.Gid) |
|||
assert.Equal(t, []string{dtmcli.StatusPrepared, dtmcli.StatusSucceed, dtmcli.StatusPrepared, dtmcli.StatusSucceed}, getBranchesStatus(csaga.Gid)) |
|||
assert.Equal(t, dtmcli.StatusSucceed, getTransStatus(csaga.Gid)) |
|||
} |
|||
|
|||
func csagaRollback(t *testing.T) { |
|||
csaga := genCSaga("gid-rollback-csaga", true, false) |
|||
examples.MainSwitch.TransOutRevertResult.SetOnce(dtmcli.ResultOngoing) |
|||
err := csaga.Submit() |
|||
assert.Nil(t, err) |
|||
WaitTransProcessed(csaga.Gid) |
|||
assert.Equal(t, dtmcli.StatusAborting, getTransStatus(csaga.Gid)) |
|||
CronTransOnce() |
|||
assert.Equal(t, dtmcli.StatusFailed, getTransStatus(csaga.Gid)) |
|||
assert.Equal(t, []string{dtmcli.StatusSucceed, dtmcli.StatusFailed, dtmcli.StatusSucceed, dtmcli.StatusSucceed}, getBranchesStatus(csaga.Gid)) |
|||
err = csaga.Submit() |
|||
assert.Error(t, err) |
|||
} |
|||
|
|||
func csagaRollback2(t *testing.T) { |
|||
csaga := genCSaga("gid-rollback-csaga2", true, false) |
|||
csaga.AddStepOrder(1, []int{0}) |
|||
err := csaga.Submit() |
|||
assert.Nil(t, err) |
|||
WaitTransProcessed(csaga.Gid) |
|||
assert.Equal(t, dtmcli.StatusFailed, getTransStatus(csaga.Gid)) |
|||
assert.Equal(t, []string{dtmcli.StatusSucceed, dtmcli.StatusFailed, dtmcli.StatusPrepared, dtmcli.StatusPrepared}, getBranchesStatus(csaga.Gid)) |
|||
err = csaga.Submit() |
|||
assert.Error(t, err) |
|||
} |
|||
|
|||
func csagaCommittedOngoing(t *testing.T) { |
|||
csaga := genCSaga("gid-committed-ongoing-csaga", false, false) |
|||
examples.MainSwitch.TransOutResult.SetOnce(dtmcli.ResultOngoing) |
|||
csaga.Submit() |
|||
WaitTransProcessed(csaga.Gid) |
|||
assert.Equal(t, []string{dtmcli.StatusPrepared, dtmcli.StatusDoing, dtmcli.StatusPrepared, dtmcli.StatusSucceed}, getBranchesStatus(csaga.Gid)) |
|||
assert.Equal(t, dtmcli.StatusSubmitted, getTransStatus(csaga.Gid)) |
|||
|
|||
CronTransOnce() |
|||
assert.Equal(t, []string{dtmcli.StatusPrepared, dtmcli.StatusSucceed, dtmcli.StatusPrepared, dtmcli.StatusSucceed}, getBranchesStatus(csaga.Gid)) |
|||
assert.Equal(t, dtmcli.StatusSucceed, getTransStatus(csaga.Gid)) |
|||
} |
|||
Loading…
Reference in new issue