Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Improve reporting of aborts and retries during live load. #3313

Merged
merged 3 commits into from
Apr 25, 2019
Merged

Improve reporting of aborts and retries during live load. #3313

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6eadd10
Improve reporting of aborts and retries during live load.
Apr 23, 2019
2e0fd0c
Rename variable.
Apr 23, 2019
4001f12
Rename variable.
Apr 25, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 20 additions & 9 deletions dgraph/cmd/live/batch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import (
"github.com/dgraph-io/dgraph/protos/pb"
"github.com/dgraph-io/dgraph/x"
"github.com/dgraph-io/dgraph/xidmap"
"github.com/dustin/go-humanize/english"
)

var (
Expand Down Expand Up @@ -78,6 +79,7 @@ type loader struct {
// To get time elapsed
start time.Time

nreqs uint64
reqs chan api.Mutation
zeroconn *grpc.ClientConn
}
Expand Down Expand Up @@ -125,34 +127,42 @@ type Counter struct {
// server expects TLS and our certificate does not match or the host name is not verified. When
// the node certificate is created the name much match the request host name. e.g., localhost not
// 127.0.0.1.
func handleError(err error) {
func handleError(err error, nreq uint64, isRetry bool) {
s := status.Convert(err)
switch {
case s.Code() == codes.Internal, s.Code() == codes.Unavailable:
x.Fatalf(s.Message())
case strings.Contains(s.Message(), "x509"):
x.Fatalf(s.Message())
case s.Code() == codes.Aborted:
if !isRetry {
fmt.Printf("Transaction #%d aborted. Will retry in background.\n", nreq)
}
case strings.Contains(s.Message(), "Server overloaded."):
dur := time.Duration(1+rand.Intn(10)) * time.Minute
fmt.Printf("Server is overloaded. Will retry after %s.", dur.Round(time.Minute))
fmt.Printf("Server is overloaded. Will retry after %s.\n", dur.Round(time.Minute))
time.Sleep(dur)
case err != y.ErrAborted && err != y.ErrConflict:
case err != y.ErrConflict:
fmt.Printf("Error while mutating: %v\n", s.Message())
}
}

func (l *loader) infinitelyRetry(req api.Mutation) {
func (l *loader) infinitelyRetry(req api.Mutation, nreq uint64) {
defer l.retryRequestsWg.Done()
nretries := 1
for i := time.Millisecond; ; i *= 2 {
txn := l.dc.NewTxn()
req.CommitNow = true
_, err := txn.Mutate(l.opts.Ctx, &req)
if err == nil {
fmt.Printf("Transaction #%d succeeded after %s.\n",
nreq, english.Plural(nretries, "retry", "retries"))
atomic.AddUint64(&l.nquads, uint64(len(req.Set)))
atomic.AddUint64(&l.txns, 1)
return
}
handleError(err)
nretries++
handleError(err, nreq, true)
atomic.AddUint64(&l.aborts, 1)
if i >= 10*time.Second {
i = 10 * time.Second
Expand All @@ -161,7 +171,7 @@ func (l *loader) infinitelyRetry(req api.Mutation) {
}
}

func (l *loader) request(req api.Mutation) {
func (l *loader) request(req api.Mutation, nreq uint64) {
txn := l.dc.NewTxn()
req.CommitNow = true
_, err := txn.Mutate(l.opts.Ctx, &req)
Expand All @@ -171,10 +181,10 @@ func (l *loader) request(req api.Mutation) {
atomic.AddUint64(&l.txns, 1)
return
}
handleError(err)
handleError(err, nreq, false)
atomic.AddUint64(&l.aborts, 1)
l.retryRequestsWg.Add(1)
go l.infinitelyRetry(req)
go l.infinitelyRetry(req, nreq)
}

// makeRequests can receive requests from batchNquads or directly from BatchSetWithMark.
Expand All @@ -183,7 +193,8 @@ func (l *loader) request(req api.Mutation) {
func (l *loader) makeRequests() {
defer l.requestsWg.Done()
for req := range l.reqs {
l.request(req)
nreq := atomic.AddUint64(&l.nreqs, 1)
l.request(req, nreq)
}
}

Expand Down