Merge pull request #271 from CovenantSQL/bugfix/isolation_level

laodouya · web-flow · commit 6d9e58447777 · 2019-03-01T17:54:23.000+08:00
Fix isolation level auto rollback bug
diff --git a/xenomint/state.go b/xenomint/state.go
@@ -404,6 +404,11 @@ func (s *State) write(
 			}
 			defer s.executer.Exec(`ROLLBACK TO "?"`, lastSeq)
 		}
+		if s.level != sql.LevelReadUncommitted {
+			// NOTE(leventeliu): this will cancel any uncommitted transaction, and do not harm to
+			// committed ones.
+			defer s.executer.Exec(`ROLLBACK`)
+		}
 		for i, v := range req.Payload.Queries {
 			var res sql.Result
 			if res, ierr = s.writeSingle(ctx, &v); ierr != nil {
@@ -426,10 +431,6 @@ func (s *State) write(
 					return
 				}
 			}
-		} else {
-			// NOTE(leventeliu): this will cancel any uncommitted transaction, and do not harm to
-			// committed ones.
-			s.executer.Exec(`ROLLBACK`)
 		}
 		// Try to commit if the ongoing tx is too large or schema is changed
 		if s.getSeq()-s.getLastCommitPoint() > s.maxTx ||
diff --git a/xenomint/state_test.go b/xenomint/state_test.go
@@ -818,6 +818,82 @@ func TestSerializableState(t *testing.T) {
 					})
 				}
 			})
+			Convey("The state should see changes", FailureContinues, func(c C) {
+				// Build transaction query
+				var (
+					count   = 1000
+					queries = make([]types.Query, count+2)
+					req     *types.Request
+				)
+				queries[0] = buildQuery(`BEGIN`)
+				for i := 0; i < count; i++ {
+					queries[i+1] = buildQuery(
+						`INSERT INTO t1(k, v) VALUES (?, ?)`, i, fmt.Sprintf("v%d", i),
+					)
+				}
+				queries[count+1] = buildQuery(`COMMIT`)
+				req = buildRequest(types.WriteQuery, queries)
+				// Send uncommitted transaction on background
+				var _, resp, err = state.Query(req, true)
+				c.So(err, ShouldBeNil)
+				c.So(resp.Header.RowCount, ShouldEqual, 0)
+
+				// Test isolation level
+				for i := 0; i < count; i++ {
+					_, resp, err = state.Query(buildRequest(types.ReadQuery, []types.Query{
+						buildQuery(`SELECT COUNT(1) AS cnt FROM t1`),
+					}), true)
+					So(resp.Payload, ShouldResemble, types.ResponsePayload{
+						Columns:   []string{"cnt"},
+						DeclTypes: []string{""},
+						Rows:      []types.ResponseRow{{Values: []interface{}{int64(count)}}},
+					})
+				}
+
+				req = buildRequest(types.WriteQuery, []types.Query{
+					buildQuery("DELETE FROM t1"),
+				})
+				_, resp, err = state.Query(req, true)
+				c.So(err, ShouldBeNil)
+			})
+			Convey("The state should not see changes because of failure query content", FailureContinues, func(c C) {
+				// Build transaction query
+				var (
+					count   = 1000
+					queries = make([]types.Query, count+3)
+					req     *types.Request
+				)
+				queries[0] = buildQuery(`BEGIN`)
+				for i := 0; i < count; i++ {
+					queries[i+1] = buildQuery(
+						`INSERT INTO t1(k, v) VALUES (?, ?)`, i, fmt.Sprintf("v%d", i),
+					)
+				}
+				queries[count+1] = buildQuery(`HAHA`)
+				queries[count+2] = buildQuery(`COMMIT`)
+				req = buildRequest(types.WriteQuery, queries)
+				// Send uncommitted transaction on background
+				var _, resp, err = state.Query(req, true)
+				c.So(err, ShouldNotBeNil)
+
+				// Test isolation level
+				for i := 0; i < count; i++ {
+					_, resp, err = state.Query(buildRequest(types.ReadQuery, []types.Query{
+						buildQuery(`SELECT COUNT(1) AS cnt FROM t1`),
+					}), true)
+					So(resp.Payload, ShouldResemble, types.ResponsePayload{
+						Columns:   []string{"cnt"},
+						DeclTypes: []string{""},
+						Rows:      []types.ResponseRow{{Values: []interface{}{int64(0)}}},
+					})
+				}
+
+				req = buildRequest(types.WriteQuery, []types.Query{
+					buildQuery("DELETE FROM t1"),
+				})
+				_, resp, err = state.Query(req, true)
+				c.So(err, ShouldBeNil)
+			})
 		})
 	})
 }
diff --git a/xenomint/xxx_test.go b/xenomint/xxx_test.go
@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"os"
 	"path"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"syscall"
@@ -198,16 +199,18 @@ func setup() {
 
 	rand.Seed(time.Now().UnixNano())
 
-	// Set NOFILE limit
-	if err = syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lmt); err != nil {
-		panic(err)
-	}
-	if lmt.Max < minNoFile {
-		panic("insufficient max RLIMIT_NOFILE")
-	}
-	lmt.Cur = lmt.Max
-	if err = syscall.Setrlimit(syscall.RLIMIT_NOFILE, &lmt); err != nil {
-		panic(err)
+	if runtime.GOOS == "linux" {
+		// Set NOFILE limit
+		if err = syscall.Getrlimit(syscall.RLIMIT_NOFILE, &lmt); err != nil {
+			panic(err)
+		}
+		if lmt.Max < minNoFile {
+			panic("insufficient max RLIMIT_NOFILE")
+		}
+		lmt.Cur = lmt.Max
+		if err = syscall.Setrlimit(syscall.RLIMIT_NOFILE, &lmt); err != nil {
+			panic(err)
+		}
 	}
 
 	// Initialze kms

Original file line number	Diff line number	Diff line change
`@@ -404,6 +404,11 @@ func (s *State) write(`
`404`	`404`	`}`
`405`	`405`	defer s.executer.Exec(`ROLLBACK TO "?"`, lastSeq)
`406`	`406`	`}`
	`407`	`+ if s.level != sql.LevelReadUncommitted {`
	`408`	`+ // NOTE(leventeliu): this will cancel any uncommitted transaction, and do not harm to`
	`409`	`+ // committed ones.`
	`410`	+ defer s.executer.Exec(`ROLLBACK`)
	`411`	`+ }`
`407`	`412`	`for i, v := range req.Payload.Queries {`
`408`	`413`	`var res sql.Result`
`409`	`414`	`if res, ierr = s.writeSingle(ctx, &v); ierr != nil {`
`@@ -426,10 +431,6 @@ func (s *State) write(`
`426`	`431`	`return`
`427`	`432`	`}`
`428`	`433`	`}`
`429`		`- } else {`
`430`		`- // NOTE(leventeliu): this will cancel any uncommitted transaction, and do not harm to`
`431`		`- // committed ones.`
`432`		- s.executer.Exec(`ROLLBACK`)
`433`	`434`	`}`
`434`	`435`	`// Try to commit if the ongoing tx is too large or schema is changed`
`435`	`436`	`if s.getSeq()-s.getLastCommitPoint() > s.maxTx \|\|`