Simplify Snapshots (#2475)

We had a complicated mechanism to keep track of which transactions have been concluded, and which Raft proposals corresponded to those transactions. This required us to have a txnMark watermark, which kept track of all applied mutations and their Raft indices, which is quite confusing.

Instead, now we just make use of the StartTs stored in proposal.Mutations and the fact that we have a Oracle which already knows the status of transactions. I make use of those givens to remove txnMarks and simplify how we do snapshots.

Further more, I now propose snapshot indices so that the entire Raft group is synchronized on the same snapshot index, calculated by the leader. This also makes it clear that only leader does the snapshot calculation, and the corresponding action of aborting old transactions, with no health signal for some time.

Also, rename MaxPending to MaxAssigned, which is a better term for what the variable holds, i.e. MaxAssigned timestamp by Zero so far.

Author: manishrjain

contrib/scripts/functions.sh

@@ -10,7 +10,7 @@ function runCluster {
     go build . && go install . && md5sum dgraph $GOPATH/bin/dgraph
     DATA="/tmp/dg" docker-compose up --force-recreate --remove-orphans --detach
   popd
-  $basedir/contrib/wait-for-it.sh localhost:6080
-  $basedir/contrib/wait-for-it.sh localhost:9180
+  $basedir/contrib/wait-for-it.sh -t 60 localhost:6080
+  $basedir/contrib/wait-for-it.sh -t 60 localhost:9180
   sleep 10 # Sleep 10 seconds to get things ready.
 }

dgraph/cmd/zero/oracle.go

@@ -28,9 +28,9 @@ type Oracle struct {
 	x.SafeMutex
 	commits map[uint64]uint64 // startTs -> commitTs
 	// TODO: Check if we need LRU.
-	rowCommit  map[string]uint64   // fp(key) -> commitTs. Used to detect conflict.
-	aborts     map[uint64]struct{} // key is startTs
-	maxPending uint64              // max transaction startTs given out by us.
+	rowCommit   map[string]uint64   // fp(key) -> commitTs. Used to detect conflict.
+	aborts      map[uint64]struct{} // key is startTs
+	maxAssigned uint64              // max transaction assigned by us.
 
 	// timestamp at the time of start of server or when it became leader. Used to detect conflicts.
 	tmax uint64
@@ -131,7 +131,7 @@ func (o *Oracle) currentState() *intern.OracleDelta {
 	for abort := range o.aborts {
 		resp.Aborts = append(resp.Aborts, abort)
 	}
-	resp.MaxPending = o.maxPending
+	resp.MaxAssigned = o.maxAssigned
 	return resp
 }
 
@@ -169,8 +169,8 @@ func (o *Oracle) sendDeltasToSubscribers() {
 	slurp_loop:
 		for {
 			// Consume tctx.
-			if update.MaxPending > delta.MaxPending {
-				delta.MaxPending = update.MaxPending
+			if update.MaxAssigned > delta.MaxAssigned {
+				delta.MaxAssigned = update.MaxAssigned
 			}
 			for _, startTs := range update.Aborts {
 				delta.Aborts = append(delta.Aborts, startTs)
@@ -244,19 +244,19 @@ func (o *Oracle) storePending(ids *api.AssignedIds) {
 	// Wait to finish up processing everything before start id.
 	o.doneUntil.WaitForMark(context.Background(), ids.EndId)
 	// Now send it out to updates.
-	o.updates <- &intern.OracleDelta{MaxPending: ids.EndId}
+	o.updates <- &intern.OracleDelta{MaxAssigned: ids.EndId}
 	o.Lock()
 	defer o.Unlock()
 	max := ids.EndId
-	if o.maxPending < max {
-		o.maxPending = max
+	if o.maxAssigned < max {
+		o.maxAssigned = max
 	}
 }
 
 func (o *Oracle) MaxPending() uint64 {
 	o.RLock()
 	defer o.RUnlock()
-	return o.maxPending
+	return o.maxAssigned
 }
 
 var errConflict = errors.New("Transaction conflict")
@@ -317,8 +317,7 @@ func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
 	// map to be keyed by uint64, which would be cheaper. But, unsure about the repurcussions of
 	// that. It would save some memory. So, worth a try.
 
-	var num intern.Num
-	num.Val = 1
+	num := intern.Num{Val: 1}
 	assigned, err := s.lease(ctx, &num, true)
 	if err != nil {
 		return err

dgraph/docker-compose.yml

@@ -20,42 +20,42 @@ services:
       # - type: bind
       #   source: "${DATA}"
       #   target: /data
-  zero2:
-    image: debian:latest
-    container_name: bank-dg0.2
-    command: /gobin/dgraph zero --my=zero2:5080 --replicas 3 --peer=zero1:5080 --idx 2
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
-  zero3:
-    image: debian:latest
-    container_name: bank-dg0.3
-    command: /gobin/dgraph zero --my=zero3:5080 --replicas 3 --peer=zero1:5080 --idx 3
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
-  zero4:
-    image: debian:latest
-    container_name: bank-dg0.4
-    command: /gobin/dgraph zero --my=zero4:5080 --replicas 3 --peer=zero1:5080 --idx 4
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
-  zero5:
-    image: debian:latest
-    container_name: bank-dg0.5
-    command: /gobin/dgraph zero --my=zero5:5080 --replicas 3 --peer=zero1:5080 --idx 5
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
+  # zero2:
+  #   image: debian:latest
+  #   container_name: bank-dg0.2
+  #   command: /gobin/dgraph zero --my=zero2:5080 --replicas 3 --peer=zero1:5080 --idx 2
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
+  # zero3:
+  #   image: debian:latest
+  #   container_name: bank-dg0.3
+  #   command: /gobin/dgraph zero --my=zero3:5080 --replicas 3 --peer=zero1:5080 --idx 3
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
+  # zero4:
+  #   image: debian:latest
+  #   container_name: bank-dg0.4
+  #   command: /gobin/dgraph zero --my=zero4:5080 --replicas 3 --peer=zero1:5080 --idx 4
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
+  # zero5:
+  #   image: debian:latest
+  #   container_name: bank-dg0.5
+  #   command: /gobin/dgraph zero --my=zero5:5080 --replicas 3 --peer=zero1:5080 --idx 5
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
 
   dg1:
     image: debian:latest
@@ -73,6 +73,7 @@ services:
       - 8180:8180
       - 9180:9180
     command: /gobin/dgraph server --my=dg1:7180 --lru_mb=1024 --zero=zero1:5080 -o 100 --expose_trace  --trace 1.0 --profile_mode block --block_rate 10
+
   dg2:
     image: debian:latest
     container_name: bank-dg2
@@ -108,37 +109,37 @@ services:
     command: /gobin/dgraph server --my=dg3:7183 --lru_mb=1024 --zero=zero1:5080 -o 103 --expose_trace   --trace 1.0 --profile_mode block --block_rate 10
 
 
-  dg4:
-    image: debian:latest
-    container_name: bank-dg4
-    working_dir: /data/dg4
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
-      # - type: bind
-      #   source: "${DATA}"
-      #   target: /data
-    ports:
-      - 8184:8184
-      - 9184:9184
-    command: /gobin/dgraph server --my=dg4:7184 --lru_mb=1024 --zero=zero1:5080 -o 104 --expose_trace   --trace 1.0 --profile_mode block --block_rate 10
+  # dg4:
+  #   image: debian:latest
+  #   container_name: bank-dg4
+  #   working_dir: /data/dg4
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
+  #     # - type: bind
+  #     #   source: "${DATA}"
+  #     #   target: /data
+  #   ports:
+  #     - 8184:8184
+  #     - 9184:9184
+  #   command: /gobin/dgraph server --my=dg4:7184 --lru_mb=1024 --zero=zero1:5080 -o 104 --expose_trace   --trace 1.0 --profile_mode block --block_rate 10
 
-  dg5:
-    image: debian:latest
-    container_name: bank-dg5
-    working_dir: /data/dg5
-    volumes:
-      - type: bind
-        source: $GOPATH/bin
-        target: /gobin
-        read_only: true
-      # - type: bind
-      #   source: "${DATA}"
-      #   target: /data
-    ports:
-      - 8185:8185
-      - 9185:9185
-    command: /gobin/dgraph server --my=dg5:7185 --lru_mb=1024 --zero=zero1:5080 -o 105 --expose_trace   --trace 1.0 --profile_mode block --block_rate 10
+  # dg5:
+  #   image: debian:latest
+  #   container_name: bank-dg5
+  #   working_dir: /data/dg5
+  #   volumes:
+  #     - type: bind
+  #       source: $GOPATH/bin
+  #       target: /gobin
+  #       read_only: true
+  #     # - type: bind
+  #     #   source: "${DATA}"
+  #     #   target: /data
+  #   ports:
+  #     - 8185:8185
+  #     - 9185:9185
+  #   command: /gobin/dgraph server --my=dg5:7185 --lru_mb=1024 --zero=zero1:5080 -o 105 --expose_trace   --trace 1.0 --profile_mode block --block_rate 10
 

posting/mvcc.go

@@ -29,17 +29,15 @@ var (
 	txnMarks    *x.WaterMark // Used to find out till what RAFT index we can snapshot entries.
 )
 
+type transactions struct {
+	x.SafeMutex
+	m map[uint64]*Txn
+}
+
 func init() {
 	txns = new(transactions)
 	txns.m = make(map[uint64]*Txn)
-	txnMarks = &x.WaterMark{Name: "Transaction watermark"}
-	txnMarks.Init()
-}
-
-func TxnMarks() *x.WaterMark {
-	return txnMarks
 }
-
 func Txns() *transactions {
 	return txns
 }
@@ -52,6 +50,8 @@ type delta struct {
 	posting       *intern.Posting
 	checkConflict bool // Check conflict detection.
 }
+
+// TODO: This structure can be merged back into Oracle.
 type Txn struct {
 	StartTs uint64
 
@@ -66,11 +66,7 @@ type Txn struct {
 	nextKeyIdx int
 }
 
-type transactions struct {
-	x.SafeMutex
-	m map[uint64]*Txn
-}
-
+// TODO: What is this for?
 func (t *transactions) MinTs() uint64 {
 	t.Lock()
 	var minTs uint64
@@ -94,28 +90,9 @@ func (t *transactions) MinTs() uint64 {
 	return minTs
 }
 
-func (t *transactions) TxnsSinceSnapshot(pending uint64) []uint64 {
-	lastSnapshotIdx := TxnMarks().DoneUntil()
-	var timestamps []uint64
-	t.Lock()
-	defer t.Unlock()
-	var oldest float64 = 0.2 * float64(pending)
-	for _, txn := range t.m {
-		index := txn.startIdx()
-		// We abort oldest 20% of the transactions.
-		if index-lastSnapshotIdx <= uint64(oldest) {
-			timestamps = append(timestamps, txn.StartTs)
-		}
-	}
-	return timestamps
-}
-
 func (t *transactions) Reset() {
 	t.Lock()
 	defer t.Unlock()
-	for _, txn := range t.m {
-		txn.done()
-	}
 	t.m = make(map[uint64]*Txn)
 }
 
@@ -158,21 +135,9 @@ func (t *transactions) Get(startTs uint64) *Txn {
 func (t *transactions) Done(startTs uint64) {
 	t.Lock()
 	defer t.Unlock()
-	txn, ok := t.m[startTs]
-	if !ok {
-		return
-	}
-	txn.done()
 	delete(t.m, startTs)
 }
 
-func (t *Txn) done() {
-	t.Lock()
-	defer t.Unlock()
-	// All indices should have been added by now.
-	TxnMarks().DoneMany(t.Indices)
-}
-
 // LastIndex returns the index of last prewrite proposal associated with
 // the transaction.
 func (t *Txn) LastIndex() uint64 {