-
Notifications
You must be signed in to change notification settings - Fork 1.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improve efficiency of readonly transactions by reusing the same read …
…ts (#2604) Currently, every transaction gets a new timestamp from Zero, which then has to propagate the allocated timestamp to every Alpha, so that it knows when to service that transaction. It would apply all the updates up until that ts, so the txn always returns consistent results. If a user is only doing read-only transactions, then this system causes a lot of unnecessary work, because each txn gets a new ts from Zero. This PR changes that by allocating a new read-only timestamp, which can be reused across many read-only transactions, if no RW txns are going on in the system. This speeds up reads, because it avoids the wait for ts propagation from Zero to Alpha leader to followers. Commits: * Logic to retrieve read only timestamps from Dgraph cluster. This would make readonly queries more efficient by utilizing the same readonly timestamp, instead of having to lease and propagate a new MaxAssigned each time. * Working code for ReadOnly Txns. * Add a test to ensure that all read-only txns get the same timestamp. Also, avoid streaming MaxAssigned if nothing new was assigned.
- Loading branch information
1 parent
8f24674
commit fe7b749
Showing
17 changed files
with
1,015 additions
and
664 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,105 @@ | ||
package main | ||
|
||
import ( | ||
"context" | ||
"log" | ||
"strings" | ||
"sync" | ||
"testing" | ||
|
||
"github.com/dgraph-io/dgo" | ||
"github.com/dgraph-io/dgo/protos/api" | ||
"github.com/dgraph-io/dgraph/x" | ||
"github.com/stretchr/testify/require" | ||
"google.golang.org/grpc" | ||
) | ||
|
||
const N = 10 | ||
|
||
func increment(t *testing.T, dg *dgo.Dgraph) int { | ||
var max int | ||
var mu sync.Mutex | ||
storeMax := func(a int) { | ||
mu.Lock() | ||
if max < a { | ||
max = a | ||
} | ||
mu.Unlock() | ||
} | ||
|
||
var wg sync.WaitGroup | ||
// N goroutines, process N times each goroutine. | ||
for i := 0; i < N; i++ { | ||
wg.Add(1) | ||
go func() { | ||
defer wg.Done() | ||
for i := 0; i < N; i++ { | ||
cnt, err := process(dg, false) | ||
if err != nil { | ||
if strings.Index(err.Error(), "Transaction has been aborted") >= 0 { | ||
// pass | ||
} else { | ||
t.Logf("Error while incrementing: %v\n", err) | ||
} | ||
} else { | ||
storeMax(cnt.Val) | ||
} | ||
} | ||
}() | ||
} | ||
wg.Wait() | ||
return max | ||
} | ||
|
||
func read(t *testing.T, dg *dgo.Dgraph, expected int) { | ||
cnt, err := process(dg, true) | ||
require.NoError(t, err) | ||
ts := cnt.startTs | ||
t.Logf("Readonly stage counter: %+v\n", cnt) | ||
|
||
var wg sync.WaitGroup | ||
for i := 0; i < N; i++ { | ||
wg.Add(1) | ||
go func() { | ||
defer wg.Done() | ||
for i := 0; i < N; i++ { | ||
cnt, err := process(dg, true) | ||
if err != nil { | ||
t.Logf("Error while reading: %v\n", err) | ||
} else { | ||
require.Equal(t, expected, cnt.Val) | ||
require.Equal(t, ts, cnt.startTs) | ||
} | ||
} | ||
}() | ||
} | ||
wg.Wait() | ||
} | ||
|
||
func TestIncrement(t *testing.T) { | ||
conn, err := grpc.Dial("localhost:9180", grpc.WithInsecure()) | ||
if err != nil { | ||
log.Fatal(err) | ||
} | ||
dc := api.NewDgraphClient(conn) | ||
dg := dgo.NewDgraphClient(dc) | ||
|
||
op := api.Operation{DropAll: true} | ||
x.Check(dg.Alter(context.Background(), &op)) | ||
|
||
cnt, err := process(dg, false) | ||
if err != nil { | ||
t.Logf("Error while reading: %v\n", err) | ||
} else { | ||
t.Logf("Initial value: %d\n", cnt.Val) | ||
} | ||
|
||
val := increment(t, dg) | ||
t.Logf("Increment stage done. Got value: %d\n", val) | ||
read(t, dg, val) | ||
t.Logf("Read stage done with value: %d\n", val) | ||
val = increment(t, dg) | ||
t.Logf("Increment stage done. Got value: %d\n", val) | ||
read(t, dg, val) | ||
t.Logf("Read stage done with value: %d\n", val) | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.