Better control excessive traffic to Dgraph

contrib/integration/increment/main.go

@@ -55,7 +55,7 @@ func queryCounter(txn *dgo.Txn) (Counter, error) {
 	query := fmt.Sprintf("{ q(func: has(%s)) { uid, val: %s }}", *pred, *pred)
 	resp, err := txn.Query(ctx, query)
 	if err != nil {
-		return counter, err
+		return counter, fmt.Errorf("Query error: %v", err)
 	}
 	m := make(map[string][]Counter)
 	if err := json.Unmarshal(resp.Json, &m); err != nil {
@@ -92,13 +92,12 @@ func process(dg *dgo.Dgraph, readOnly bool) (Counter, error) {
 	}
 	mu.SetNquads = []byte(fmt.Sprintf(`<%s> <%s> "%d"^^<xs:int> .`, counter.Uid, *pred, counter.Val))
 
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-	_, err = txn.Mutate(ctx, &mu)
+	// Don't put any timeout for mutation.
+	_, err = txn.Mutate(context.Background(), &mu)
 	if err != nil {
 		return Counter{}, err
 	}
-	return counter, txn.Commit(ctx)
+	return counter, txn.Commit(context.Background())
 }
 
 func main() {

contrib/integration/testtxn/main_test.go

@@ -439,8 +439,10 @@ func TestReadIndexKeySameTxn(t *testing.T) {
 
 	txn := s.dg.NewTxn()
 
-	mu := &api.Mutation{}
-	mu.SetJson = []byte(`{"name": "Manish"}`)
+	mu := &api.Mutation{
+		CommitNow: true,
+		SetJson:   []byte(`{"name": "Manish"}`),
+	}
 	assigned, err := txn.Mutate(context.Background(), mu)
 	if err != nil {
 		log.Fatalf("Error while running mutation: %v\n", err)
@@ -453,6 +455,8 @@ func TestReadIndexKeySameTxn(t *testing.T) {
 		uid = u
 	}
 
+	txn = s.dg.NewTxn()
+	defer txn.Discard(context.Background())
 	q := `{ me(func: le(name, "Manish")) { uid }}`
 	resp, err := txn.Query(context.Background(), q)
 	if err != nil {

dgraph/cmd/alpha/run.go

@@ -91,7 +91,7 @@ they form a Raft group and provide synchronous replication.
 			"actions (i.e., --whitelist 127.0.0.1:127.0.0.3,0.0.0.7:0.0.0.9)")
 
 	flag.StringVar(&worker.Config.ExportPath, "export", "export", "Folder in which to store exports.")
-	flag.IntVar(&worker.Config.NumPendingProposals, "pending_proposals", 2000,
+	flag.IntVar(&worker.Config.NumPendingProposals, "pending_proposals", 256,
 		"Number of pending mutation proposals. Useful for rate limiting.")
 	flag.Float64Var(&worker.Config.Tracing, "trace", 0.0, "The ratio of queries to trace.")
 	flag.StringVar(&worker.Config.MyAddr, "my", "",

dgraph/cmd/live/batch.go

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"math"
+	"math/rand"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -139,6 +140,10 @@ func handleError(err error) {
 		x.Fatalf(s.Message())
 	case strings.Contains(s.Message(), "x509"):
 		x.Fatalf(s.Message())
+	case strings.Contains(s.Message(), "Server unavailable."):
+		dur := time.Duration(1+rand.Intn(10)) * time.Minute
+		x.Printf("Server is unavailable. Will retry after %s.", dur.Round(time.Minute))
+		time.Sleep(dur)
 	case err != y.ErrAborted && err != y.ErrConflict:
 		x.Printf("Error while mutating %v\n", s.Message())
 	}
@@ -197,9 +202,9 @@ func (l *loader) printCounters() {
 	for range l.ticker.C {
 		counter := l.Counter()
 		rate := float64(counter.Rdfs) / counter.Elapsed.Seconds()
-		elapsed := ((time.Since(start) / time.Second) * time.Second).String()
-		fmt.Printf("Total Txns done: %8d RDFs per second: %7.0f Time Elapsed: %v, Aborts: %d\n",
-			counter.TxnsDone, rate, elapsed, counter.Aborts)
+		elapsed := time.Since(start).Round(time.Second)
+		fmt.Printf("[%6s] Txns: %d RDFs: %d RDFs/sec: %5.0f Aborts: %d\n",
+			elapsed, counter.TxnsDone, counter.Rdfs, rate, counter.Aborts)
 
 	}
 }

dgraph/cmd/live/run.go

@@ -87,7 +87,7 @@ func init() {
 	flag.StringP("schema", "s", "", "Location of schema file")
 	flag.StringP("dgraph", "d", "127.0.0.1:9080", "Dgraph gRPC server address")
 	flag.StringP("zero", "z", "127.0.0.1:5080", "Dgraphzero gRPC server address")
-	flag.IntP("conc", "c", 100,
+	flag.IntP("conc", "c", 10,
 		"Number of concurrent requests to make to Dgraph")
 	flag.IntP("batch", "b", 1000,
 		"Number of RDF N-Quads to send as part of a mutation.")

posting/index.go

@@ -791,7 +791,6 @@ func RebuildListType(ctx context.Context, attr string, startTs uint64) error {
 }
 
 func DeleteAll() error {
-	btree.DeleteAll()
 	lcache.clear(func([]byte) bool { return true })
 	return deleteEntries(nil, func(key []byte) bool {
 		pk := x.Parse(key)

posting/list.go

@@ -80,7 +80,6 @@ type List struct {
 	deleteMe      int32  // Using atomic for this, to avoid expensive SetForDeletion operation.
 	estimatedSize int32
 	numCommits    int
-	onDisk        bool
 }
 
 // calculateSize would give you the size estimate. This is expensive, so run it carefully.

posting/lists.go

@@ -18,7 +18,6 @@ package posting
 
 import (
 	"crypto/md5"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"math"
@@ -34,7 +33,6 @@ import (
 
 	"github.com/dgraph-io/badger"
 	"github.com/dgraph-io/badger/y"
-	"github.com/golang/glog"
 
 	"github.com/dgraph-io/dgraph/protos/pb"
 	"github.com/dgraph-io/dgraph/x"
@@ -211,96 +209,19 @@ func updateMemoryMetrics(lc *y.Closer) {
 var (
 	pstore *badger.DB
 	lcache *listCache
-	btree  *BTree
 	closer *y.Closer
 )
 
 // Init initializes the posting lists package, the in memory and dirty list hash.
 func Init(ps *badger.DB) {
 	pstore = ps
 	lcache = newListCache(math.MaxUint64)
-	btree = newBTree(2)
 	x.LcacheCapacity.Set(math.MaxInt64)
 
-	closer = y.NewCloser(3)
+	closer = y.NewCloser(2)
 
 	go periodicUpdateStats(closer)
 	go updateMemoryMetrics(closer)
-	go clearBtree(closer)
-}
-
-// clearBtree checks if the keys stored in the btree have reached their conclusion, and if so,
-// removes them from the tree. Conclusion in this case would be that the key got written out to
-// disk, or the txn which introduced the key got aborted.
-func clearBtree(closer *y.Closer) {
-	defer closer.Done()
-	var removeKey = errors.New("Remove key from btree.")
-
-	handleKey := func(txn *badger.Txn, k []byte) error {
-		_, err := txn.Get(k)
-		switch {
-		case err == badger.ErrKeyNotFound:
-			l := GetLru(k) // Retrieve from LRU cache, if it exists.
-			if l == nil {
-				// Posting list no longer in memory. So, it must have been either written to
-				// disk, or removed from memory after a txn abort.
-				return removeKey
-			}
-			l.RLock()
-			defer l.RUnlock()
-			if !l.hasPendingTxn() {
-				// This key's txn was aborted. So, we can remove it from btree.
-				return removeKey
-			}
-			return nil
-		case err != nil:
-			glog.Warningf("Error while checking key: %v\n", err)
-			return err
-		default:
-			// Key was found on disk. Remove from btree.
-			return removeKey
-		}
-	}
-
-	removeKeysOnDisk := func() {
-		var keys []string
-		var count int
-		err := pstore.View(func(txn *badger.Txn) error {
-			var rerr error
-			btree.Ascend(func(k []byte) bool {
-				count++
-				err := handleKey(txn, k)
-				if err == removeKey {
-					keys = append(keys, string(k))
-				} else if err != nil {
-					rerr = err
-					return false
-				}
-				return true
-			})
-			return rerr
-		})
-		if glog.V(2) && count > 0 {
-			glog.Infof("Btree size: %d. Removing=%d. Error=%v\n", count, len(keys), err)
-		}
-		if err == nil {
-			for _, k := range keys {
-				btree.Delete([]byte(k))
-			}
-		}
-	}
-
-	ticker := time.NewTicker(15 * time.Second)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ticker.C:
-			removeKeysOnDisk()
-		case <-closer.HasBeenClosed():
-			return
-		}
-	}
 }
 
 func Cleanup() {
@@ -339,8 +260,6 @@ func Get(key []byte) (rlist *List, err error) {
 	lp = lcache.PutIfMissing(string(key), l)
 	if lp != l {
 		x.CacheRace.Add(1)
-	} else if !lp.onDisk {
-		btree.Insert(lp.key)
 	}
 	return lp, nil
 }

posting/lru.go

@@ -22,7 +22,6 @@ package posting
 import (
 	"container/list"
 	"context"
-	"sync"
 	"sync/atomic"
 	"time"
 
@@ -31,7 +30,7 @@ import (
 
 // listCache is an LRU cache.
 type listCache struct {
-	sync.Mutex
+	x.SafeMutex
 
 	ctx context.Context
 	// MaxSize is the maximum size of cache before an item is evicted.