5262749833

Committed 14 Jun 2023 04:22AM UTC coverage: 66.818% (-0.2%) from 67.043%

Build Type

push

Committed by

Commit Message

Merge 8b487186b into 2787cfc58

Run Details

58007 of 86814 relevant lines covered (66.82%)

2248264.51 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.13

/dgraph/cmd/zero/oracle.go

/*
 * Copyright 2017-2023 Dgraph Labs, Inc. and Contributors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package zero

import (
        "context"
        "math/rand"
        "strconv"
        "strings"
        "time"

        "github.com/golang/glog"
        "github.com/pkg/errors"
        otrace "go.opencensus.io/trace"

        "github.com/dgraph-io/badger/v4/y"
        "github.com/dgraph-io/dgo/v230/protos/api"
        "github.com/dgraph-io/dgraph/protos/pb"
        "github.com/dgraph-io/dgraph/x"
        "github.com/dgraph-io/ristretto/z"
)

// Oracle stores and manages the transaction state and conflict detection.
type Oracle struct {
        x.SafeMutex
        commits map[uint64]uint64 // startTs -> commitTs
        // TODO: Check if we need LRU.
        keyCommit   *z.Tree // fp(key) -> commitTs. Used to detect conflict.
        maxAssigned uint64  // max transaction assigned by us.

        // All transactions with startTs < startTxnTs return true for hasConflict.
        startTxnTs  uint64
        subscribers map[int]chan pb.OracleDelta
        updates     chan *pb.OracleDelta
        doneUntil   y.WaterMark
}

// Init initializes the oracle.
func (o *Oracle) Init() {
        o.commits = make(map[uint64]uint64)
        // Remove the older btree file, before creating NewTree, as it may contain stale data leading
        // to wrong results.
        o.keyCommit = z.NewTree("oracle")
        o.subscribers = make(map[int]chan pb.OracleDelta)
        o.updates = make(chan *pb.OracleDelta, 100000) // Keeping 1 second worth of updates.
        o.doneUntil.Init(nil)
        go o.sendDeltasToSubscribers()
}

// close releases the memory associated with btree used for keycommit.
func (o *Oracle) close() {
        if err := o.keyCommit.Close(); err != nil {
                glog.Warningf("error while closing tree: %v", err)
        }
}

func (o *Oracle) updateStartTxnTs(ts uint64) {
        o.Lock()
        defer o.Unlock()
        o.startTxnTs = ts
        o.keyCommit.Reset()
}

// TODO: This should be done during proposal application for Txn status.
func (o *Oracle) hasConflict(src *api.TxnContext) bool {
        // This transaction was started before I became leader.
        if src.StartTs < o.startTxnTs {
                return true
        }
        for _, k := range src.Keys {
                ki, err := strconv.ParseUint(k, 36, 64)
                if err != nil {
                        glog.Errorf("Got error while parsing conflict key %q: %v\n", k, err)
                        continue
                }
                if last := o.keyCommit.Get(ki); last > src.StartTs {
                        return true
                }
        }
        return false
}

func (o *Oracle) purgeBelow(minTs uint64) {
        var timer x.Timer
        timer.Start()

        o.Lock()
        defer o.Unlock()

        // Set startTxnTs so that every txn with start ts less than this, would be aborted.
        o.startTxnTs = minTs

        // Dropping would be cheaper if abort/commits map is sharded
        for ts := range o.commits {
                if ts < minTs {
                        delete(o.commits, ts)
                }
        }
        timer.Record("commits")

        // There is no transaction running with startTs less than minTs
        // So we can delete everything from rowCommit whose commitTs < minTs
        stats := o.keyCommit.Stats()
        if stats.Occupancy < 50.0 {
                return
        }
        o.keyCommit.DeleteBelow(minTs)
        timer.Record("deleteBelow")
        glog.V(2).Infof("Purged below ts:%d, len(o.commits):%d, keyCommit: [before: %+v, after: %+v].\n",
                minTs, len(o.commits), stats, o.keyCommit.Stats())
        if timer.Total() > time.Second {
                glog.V(2).Infof("Purge %s\n", timer.String())
        }
}

func (o *Oracle) commit(src *api.TxnContext) error {
        o.Lock()
        defer o.Unlock()

        if o.hasConflict(src) {
                return x.ErrConflict
        }
        // We store src.Keys as string to ensure compatibility with all the various language clients we
        // have. But, really they are just uint64s encoded as strings. We use base 36 during creation of
        // these keys in FillContext in posting/mvcc.go.
        for _, k := range src.Keys {
                ki, err := strconv.ParseUint(k, 36, 64)
                if err != nil {
                        glog.Errorf("Got error while parsing conflict key %q: %v\n", k, err)
                        continue
                }
                o.keyCommit.Set(ki, src.CommitTs) // CommitTs is handed out before calling this func.
        }
        return nil
}

func (o *Oracle) currentState() *pb.OracleDelta {
        o.AssertRLock()
        resp := &pb.OracleDelta{}
        for start, commit := range o.commits {
                resp.Txns = append(resp.Txns,
                        &pb.TxnStatus{StartTs: start, CommitTs: commit})
        }
        resp.MaxAssigned = o.maxAssigned
        return resp
}

func (o *Oracle) newSubscriber() (<-chan pb.OracleDelta, int) {
        o.Lock()
        defer o.Unlock()
        var id int
        for {
                //nolint:gosec // random generator for node id does not require cryptographic precision
                id = rand.Int()
                if _, has := o.subscribers[id]; !has {
                        break
                }
        }

        // The channel takes a delta instead of a pointer as the receiver needs to
        // modify it by setting the group checksums. Passing a pointer previously
        // resulted in a race condition.
        ch := make(chan pb.OracleDelta, 1000)
        ch <- *o.currentState() // Queue up the full state as the first entry.
        o.subscribers[id] = ch
        return ch, id
}

func (o *Oracle) removeSubscriber(id int) {
        o.Lock()
        defer o.Unlock()
        delete(o.subscribers, id)
}

// sendDeltasToSubscribers reads updates from the o.updates
// constructs a delta object containing transactions from one or more updates
// and sends the delta object to each subscriber's channel
func (o *Oracle) sendDeltasToSubscribers() {
        delta := &pb.OracleDelta{}
        ticker := time.NewTicker(time.Second)
        defer ticker.Stop()

        // waitFor calculates the maximum value of delta.MaxAssigned and all the CommitTs of delta.Txns
        waitFor := func() uint64 {
                w := delta.MaxAssigned
                for _, txn := range delta.Txns {
                        w = x.Max(w, txn.CommitTs)
                }
                return w
        }

        for {
        get_update:
                var update *pb.OracleDelta
                select {
                case update = <-o.updates:
                case <-ticker.C:
                        wait := waitFor()
                        if wait == 0 || o.doneUntil.DoneUntil() < wait {
                                goto get_update
                        }
                        // Send empty update.
                        update = &pb.OracleDelta{}
                }
        slurp_loop:
                for {
                        delta.MaxAssigned = x.Max(delta.MaxAssigned, update.MaxAssigned)
                        delta.Txns = append(delta.Txns, update.Txns...)
                        select {
                        case update = <-o.updates:
                        default:
                                break slurp_loop
                        }
                }
                // No need to sort the txn updates here. Alpha would sort them before
                // applying.

                // Let's ensure that we have all the commits up until the max here.
                // Otherwise, we'll be sending commit timestamps out of order, which
                // would cause Alphas to drop some of them, during writes to Badger.
                if o.doneUntil.DoneUntil() < waitFor() {
                        continue // The for loop doing blocking reads from o.updates.
                        // We need at least one entry from the updates channel to pick up a missing update.
                        // Don't goto slurp_loop, because it would break from select immediately.
                }

                if glog.V(3) {
                        glog.Infof("DoneUntil: %d. Sending delta: %+v\n", o.doneUntil.DoneUntil(), delta)
                }
                o.Lock()
                for id, ch := range o.subscribers {
                        select {
                        case ch <- *delta:
                        default:
                                close(ch)
                                delete(o.subscribers, id)
                        }
                }
                o.Unlock()
                delta = &pb.OracleDelta{}
        }
}

func (o *Oracle) updateCommitStatusHelper(index uint64, src *api.TxnContext) bool {
        o.Lock()
        defer o.Unlock()
        if _, ok := o.commits[src.StartTs]; ok {
                return false
        }
        if src.Aborted {
                o.commits[src.StartTs] = 0
        } else {
                o.commits[src.StartTs] = src.CommitTs
        }
        return true
}

func (o *Oracle) updateCommitStatus(index uint64, src *api.TxnContext) {
        // TODO: We should check if the tablet is in read-only status here.
        if o.updateCommitStatusHelper(index, src) {
                delta := new(pb.OracleDelta)
                delta.Txns = append(delta.Txns, &pb.TxnStatus{
                        StartTs:  src.StartTs,
                        CommitTs: o.commitTs(src.StartTs),
                })
                o.updates <- delta
        }
}

func (o *Oracle) commitTs(startTs uint64) uint64 {
        o.RLock()
        defer o.RUnlock()
        return o.commits[startTs]
}

func (o *Oracle) storePending(ids *pb.AssignedIds) {
        // Wait to finish up processing everything before start id.
        max := x.Max(ids.EndId, ids.ReadOnly)
        if err := o.doneUntil.WaitForMark(context.Background(), max); err != nil {
                glog.Errorf("Error while waiting for mark: %+v", err)
        }

        // Now send it out to updates.
        o.updates <- &pb.OracleDelta{MaxAssigned: max}

        o.Lock()
        defer o.Unlock()
        o.maxAssigned = x.Max(o.maxAssigned, max)
}

// MaxPending returns the maximum assigned timestamp.
func (o *Oracle) MaxPending() uint64 {
        o.RLock()
        defer o.RUnlock()
        return o.maxAssigned
}

// proposeTxn proposes a txn update, and then updates src to reflect the state
// of the commit after proposal is run.
func (s *Server) proposeTxn(ctx context.Context, src *api.TxnContext) error {
        var zp pb.ZeroProposal
        zp.Txn = &api.TxnContext{
                StartTs:  src.StartTs,
                CommitTs: src.CommitTs,
                Aborted:  src.Aborted,
        }

        // NOTE: It is important that we continue retrying proposeTxn until we succeed. This should
        // happen, irrespective of what the user context timeout might be. We check for it before
        // reaching this stage, but now that we're here, we have to ensure that the commit proposal goes
        // through. Otherwise, we should block here forever. If we don't do this, we'll see txn
        // violations in Jepsen, because we'll send out a MaxAssigned higher than a commit, which would
        // cause newer txns to see older data.

        // If this node stops being the leader, we want this proposal to not be forwarded to the leader,
        // and get aborted.
        if err := s.Node.proposeAndWait(ctx, &zp); err != nil {
                return err
        }

        // There might be race between this proposal trying to commit and predicate
        // move aborting it. A predicate move, triggered by Zero, would abort all
        // pending transactions.  At the same time, a client which has already done
        // mutations, can proceed to commit it. A race condition can happen here,
        // with both proposing their respective states, only one can succeed after
        // the proposal is done. So, check again to see the fate of the transaction
        // here.
        src.CommitTs = s.orc.commitTs(src.StartTs)
        if src.CommitTs == 0 {
                src.Aborted = true
        }
        return nil
}

func (s *Server) commit(ctx context.Context, src *api.TxnContext) error {
        span := otrace.FromContext(ctx)
        span.Annotate([]otrace.Attribute{otrace.Int64Attribute("startTs", int64(src.StartTs))}, "")
        if src.Aborted {
                return s.proposeTxn(ctx, src)
        }

        // Use the start timestamp to check if we have a conflict, before we need to assign a commit ts.
        s.orc.RLock()
        conflict := s.orc.hasConflict(src)
        s.orc.RUnlock()
        if conflict {
                span.Annotate([]otrace.Attribute{otrace.BoolAttribute("abort", true)},
                        "Oracle found conflict")
                src.Aborted = true
                return s.proposeTxn(ctx, src)
        }

        checkPreds := func() error {
                // Check if any of these tablets is being moved. If so, abort the transaction.
                for _, pkey := range src.Preds {
                        splits := strings.SplitN(pkey, "-", 2)
                        if len(splits) < 2 {
                                return errors.Errorf("Unable to find group id in %s", pkey)
                        }
                        gid, err := strconv.Atoi(splits[0])
                        if err != nil {
                                return errors.Wrapf(err, "unable to parse group id from %s", pkey)
                        }
                        pred := splits[1]
                        tablet := s.ServingTablet(pred)
                        if tablet == nil {
                                return errors.Errorf("Tablet for %s is nil", pred)
                        }
                        if tablet.GroupId != uint32(gid) {
                                return errors.Errorf("Mutation done in group: %d. Predicate %s assigned to %d",
                                        gid, pred, tablet.GroupId)
                        }
                        if s.isBlocked(pred) {
                                return errors.Errorf("Commits on predicate %s are blocked due to predicate move", pred)
                        }
                }
                return nil
        }
        if err := checkPreds(); err != nil {
                span.Annotate([]otrace.Attribute{otrace.BoolAttribute("abort", true)}, err.Error())
                src.Aborted = true
                return s.proposeTxn(ctx, src)
        }

        num := pb.Num{Val: 1, Type: pb.Num_TXN_TS}
        assigned, err := s.lease(ctx, &num)
        if err != nil {
                return err
        }
        src.CommitTs = assigned.StartId
        // Mark the transaction as done, irrespective of whether the proposal succeeded or not.
        defer s.orc.doneUntil.Done(src.CommitTs)
        span.Annotatef([]otrace.Attribute{otrace.Int64Attribute("commitTs", int64(src.CommitTs))},
                "Node Id: %d. Proposing TxnContext: %+v", s.Node.Id, src)

        if err := s.orc.commit(src); err != nil {
                span.Annotatef(nil, "Found a conflict. Aborting.")
                src.Aborted = true
        }
        if err := ctx.Err(); err != nil {
                span.Annotatef(nil, "Aborting txn due to context timing out.")
                src.Aborted = true
        }
        // Propose txn should be used to set watermark as done.
        return s.proposeTxn(ctx, src)
}

// CommitOrAbort either commits a transaction or aborts it.
// The abortion can happen under the following conditions
// 1) the api.TxnContext.Aborted flag is set in the src argument
// 2) if there's an error (e.g server is not the leader or there's a conflicting transaction)
func (s *Server) CommitOrAbort(ctx context.Context, src *api.TxnContext) (*api.TxnContext, error) {
        if ctx.Err() != nil {
                return nil, ctx.Err()
        }
        ctx, span := otrace.StartSpan(ctx, "Zero.CommitOrAbort")
        defer span.End()

        if !s.Node.AmLeader() {
                return nil, errors.Errorf("Only leader can decide to commit or abort")
        }
        err := s.commit(ctx, src)
        if err != nil {
                span.Annotate([]otrace.Attribute{otrace.BoolAttribute("error", true)}, err.Error())
        }
        return src, err
}

var errClosed = errors.New("Streaming closed by oracle")
var errNotLeader = errors.New("Node is no longer leader")

// Oracle streams the oracle state to the alphas.
// The first entry sent by Zero contains the entire state of transactions. Zero periodically
// confirms receipt from the group, and truncates its state. This 2-way acknowledgement is a
// safe way to get the status of all the transactions.
func (s *Server) Oracle(_ *api.Payload, server pb.Zero_OracleServer) error {
        if !s.Node.AmLeader() {
                return errNotLeader
        }
        ch, id := s.orc.newSubscriber()
        defer s.orc.removeSubscriber(id)

        ctx := server.Context()
        leaderChangeCh := s.leaderChangeChannel()
        for {
                select {
                case <-leaderChangeCh:
                        return errNotLeader
                case delta, open := <-ch:
                        if !open {
                                return errClosed
                        }
                        // Pass in the latest group checksum as well, so the Alpha can use that to determine
                        // when not to service a read.
                        delta.GroupChecksums = s.groupChecksums()
                        if err := server.Send(&delta); err != nil {
                                return err
                        }
                case <-ctx.Done():
                        return ctx.Err()
                case <-s.closer.HasBeenClosed():
                        return errServerShutDown
                }
        }
}

// TryAbort attempts to abort the given transactions which are not already committed..
func (s *Server) TryAbort(ctx context.Context,
        txns *pb.TxnTimestamps) (*pb.OracleDelta, error) {
        delta := &pb.OracleDelta{}
        for _, startTs := range txns.Ts {
                // Do via proposals to avoid race
                tctx := &api.TxnContext{StartTs: startTs, Aborted: true}
                if err := s.proposeTxn(ctx, tctx); err != nil {
                        return delta, err
                }
                // Txn should be aborted if not already committed.
                delta.Txns = append(delta.Txns, &pb.TxnStatus{
                        StartTs:  startTs,
                        CommitTs: s.orc.commitTs(startTs)})
        }
        return delta, nil
}

// Timestamps is used to assign startTs for a new transaction
func (s *Server) Timestamps(ctx context.Context, num *pb.Num) (*pb.AssignedIds, error) {
        ctx, span := otrace.StartSpan(ctx, "Zero.Timestamps")
        defer span.End()

        span.Annotatef(nil, "Zero id: %d. Timestamp request: %+v", s.Node.Id, num)
        if ctx.Err() != nil {
                return &emptyAssignedIds, ctx.Err()
        }

        num.Type = pb.Num_TXN_TS
        reply, err := s.lease(ctx, num)
        span.Annotatef(nil, "Response: %+v. Error: %v", reply, err)

        switch err {
        case nil:
                s.orc.doneUntil.Done(x.Max(reply.EndId, reply.ReadOnly))
                go s.orc.storePending(reply)
        case errServedFromMemory:
                // Avoid calling doneUntil.Done, and storePending.
                err = nil
        default:
                glog.Errorf("Got error: %v while leasing timestamps: %+v", err, num)
        }
        return reply, err
}

1	/*
2	* Copyright 2017-2023 Dgraph Labs, Inc. and Contributors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package zero
18
19	import (
20	"context"
21	"math/rand"
22	"strconv"
23	"strings"
24	"time"
25
26	"github.com/golang/glog"
27	"github.com/pkg/errors"
28	otrace "go.opencensus.io/trace"
29
30	"github.com/dgraph-io/badger/v4/y"
31	"github.com/dgraph-io/dgo/v230/protos/api"
32	"github.com/dgraph-io/dgraph/protos/pb"
33	"github.com/dgraph-io/dgraph/x"
34	"github.com/dgraph-io/ristretto/z"
35	)
36
37	// Oracle stores and manages the transaction state and conflict detection.
38	type Oracle struct {
39	x.SafeMutex
40	commits map[uint64]uint64 // startTs -> commitTs
41	// TODO: Check if we need LRU.
42	keyCommit *z.Tree // fp(key) -> commitTs. Used to detect conflict.
43	maxAssigned uint64 // max transaction assigned by us.
44
45	// All transactions with startTs < startTxnTs return true for hasConflict.
46	startTxnTs uint64
47	subscribers map[int]chan pb.OracleDelta
48	updates chan *pb.OracleDelta
49	doneUntil y.WaterMark
50	}
51
52	// Init initializes the oracle.
53	func (o *Oracle) Init() {	65✔
54	o.commits = make(map[uint64]uint64)	65✔
55	// Remove the older btree file, before creating NewTree, as it may contain stale data leading	65✔
56	// to wrong results.	65✔
57	o.keyCommit = z.NewTree("oracle")	65✔
58	o.subscribers = make(map[int]chan pb.OracleDelta)	65✔
59	o.updates = make(chan *pb.OracleDelta, 100000) // Keeping 1 second worth of updates.	65✔
60	o.doneUntil.Init(nil)	65✔
61	go o.sendDeltasToSubscribers()	65✔
62	}	65✔
63
64	// close releases the memory associated with btree used for keycommit.
65	func (o *Oracle) close() {	65✔
66	if err := o.keyCommit.Close(); err != nil {	65✔
67	glog.Warningf("error while closing tree: %v", err)	×
68	}	×
69	}
70
71	func (o *Oracle) updateStartTxnTs(ts uint64) {	52✔
72	o.Lock()	52✔
73	defer o.Unlock()	52✔
74	o.startTxnTs = ts	52✔
75	o.keyCommit.Reset()	52✔
76	}	52✔
77
78	// TODO: This should be done during proposal application for Txn status.
79	func (o Oracle) hasConflict(src api.TxnContext) bool {	20,092✔
80	// This transaction was started before I became leader.	20,092✔
81	if src.StartTs < o.startTxnTs {	20,092✔
82	return true	×
83	}	×
84	for _, k := range src.Keys {	1,075,115✔
85	ki, err := strconv.ParseUint(k, 36, 64)	1,055,023✔
86	if err != nil {	1,055,023✔
87	glog.Errorf("Got error while parsing conflict key %q: %v\n", k, err)	×
88	continue	×
89	}
90	if last := o.keyCommit.Get(ki); last > src.StartTs {	1,056,941✔
91	return true	1,918✔
92	}	1,918✔
93	}
94	return false	18,174✔
95	}
96
97	func (o *Oracle) purgeBelow(minTs uint64) {	40✔
98	var timer x.Timer	40✔
99	timer.Start()	40✔
100		40✔
101	o.Lock()	40✔
102	defer o.Unlock()	40✔
103		40✔
104	// Set startTxnTs so that every txn with start ts less than this, would be aborted.	40✔
105	o.startTxnTs = minTs	40✔
106		40✔
107	// Dropping would be cheaper if abort/commits map is sharded	40✔
108	for ts := range o.commits {	33,898✔
109	if ts < minTs {	51,442✔
110	delete(o.commits, ts)	17,584✔
111	}	17,584✔
112	}
113	timer.Record("commits")	40✔
114		40✔
115	// There is no transaction running with startTs less than minTs	40✔
116	// So we can delete everything from rowCommit whose commitTs < minTs	40✔
117	stats := o.keyCommit.Stats()	40✔
118	if stats.Occupancy < 50.0 {	73✔
119	return	33✔
120	}	33✔
121	o.keyCommit.DeleteBelow(minTs)	7✔
122	timer.Record("deleteBelow")	7✔
123	glog.V(2).Infof("Purged below ts:%d, len(o.commits):%d, keyCommit: [before: %+v, after: %+v].\n",	7✔
124	minTs, len(o.commits), stats, o.keyCommit.Stats())	7✔
125	if timer.Total() > time.Second {	7✔
126	glog.V(2).Infof("Purge %s\n", timer.String())	×
127	}	×
128	}
129
130	func (o Oracle) commit(src api.TxnContext) error {	9,103✔
131	o.Lock()	9,103✔
132	defer o.Unlock()	9,103✔
133		9,103✔
134	if o.hasConflict(src) {	9,135✔
135	return x.ErrConflict	32✔
136	}	32✔
137	// We store src.Keys as string to ensure compatibility with all the various language clients we
138	// have. But, really they are just uint64s encoded as strings. We use base 36 during creation of
139	// these keys in FillContext in posting/mvcc.go.
140	for _, k := range src.Keys {	535,221✔
141	ki, err := strconv.ParseUint(k, 36, 64)	526,150✔
142	if err != nil {	526,150✔
143	glog.Errorf("Got error while parsing conflict key %q: %v\n", k, err)	×
144	continue	×
145	}
146	o.keyCommit.Set(ki, src.CommitTs) // CommitTs is handed out before calling this func.	526,150✔
147	}
148	return nil	9,071✔
149	}
150
151	func (o Oracle) currentState() pb.OracleDelta {	80✔
152	o.AssertRLock()	80✔
153	resp := &pb.OracleDelta{}	80✔
154	for start, commit := range o.commits {	155✔
155	resp.Txns = append(resp.Txns,	75✔
156	&pb.TxnStatus{StartTs: start, CommitTs: commit})	75✔
157	}	75✔
158	resp.MaxAssigned = o.maxAssigned	80✔
159	return resp	80✔
160	}
161
162	func (o *Oracle) newSubscriber() (<-chan pb.OracleDelta, int) {	80✔
163	o.Lock()	80✔
164	defer o.Unlock()	80✔
165	var id int	80✔
166	for {	160✔
167	//nolint:gosec // random generator for node id does not require cryptographic precision	80✔
168	id = rand.Int()	80✔
169	if _, has := o.subscribers[id]; !has {	160✔
170	break	80✔
171	}
172	}
173
174	// The channel takes a delta instead of a pointer as the receiver needs to
175	// modify it by setting the group checksums. Passing a pointer previously
176	// resulted in a race condition.
177	ch := make(chan pb.OracleDelta, 1000)	80✔
178	ch <- *o.currentState() // Queue up the full state as the first entry.	80✔
179	o.subscribers[id] = ch	80✔
180	return ch, id	80✔
181	}
182
183	func (o *Oracle) removeSubscriber(id int) {	80✔
184	o.Lock()	80✔
185	defer o.Unlock()	80✔
186	delete(o.subscribers, id)	80✔
187	}	80✔
188
189	// sendDeltasToSubscribers reads updates from the o.updates
190	// constructs a delta object containing transactions from one or more updates
191	// and sends the delta object to each subscriber's channel
192	func (o *Oracle) sendDeltasToSubscribers() {	65✔
193	delta := &pb.OracleDelta{}	65✔
194	ticker := time.NewTicker(time.Second)	65✔
195	defer ticker.Stop()	65✔
196		65✔
197	// waitFor calculates the maximum value of delta.MaxAssigned and all the CommitTs of delta.Txns	65✔
198	waitFor := func() uint64 {	105,266✔
199	w := delta.MaxAssigned	105,201✔
200	for _, txn := range delta.Txns {	43,644,791✔
201	w = x.Max(w, txn.CommitTs)	43,539,590✔
202	}	43,539,590✔
203	return w	105,201✔
204	}
205
206	for {	93,019✔
207	get_update:	92,954✔
208	var update *pb.OracleDelta	105,215✔
209	select {	105,215✔
210	case update = <-o.updates:	92,838✔
211	case <-ticker.C:	12,312✔
212	wait := waitFor()	12,312✔
213	if wait == 0 \|\| o.doneUntil.DoneUntil() < wait {	24,573✔
214	goto get_update	12,261✔
215	}
216	// Send empty update.
217	update = &pb.OracleDelta{}	51✔
218	}
219	slurp_loop:	92,889✔
220	for {	187,021✔
221	delta.MaxAssigned = x.Max(delta.MaxAssigned, update.MaxAssigned)	94,132✔
222	delta.Txns = append(delta.Txns, update.Txns...)	94,132✔
223	select {	94,132✔
224	case update = <-o.updates:	1,243✔
225	default:	92,889✔
226	break slurp_loop	92,889✔
227	}
228	}
229	// No need to sort the txn updates here. Alpha would sort them before
230	// applying.
231
232	// Let's ensure that we have all the commits up until the max here.
233	// Otherwise, we'll be sending commit timestamps out of order, which
234	// would cause Alphas to drop some of them, during writes to Badger.
235	if o.doneUntil.DoneUntil() < waitFor() {	112,126✔
236	continue // The for loop doing blocking reads from o.updates.	19,237✔
237	// We need at least one entry from the updates channel to pick up a missing update.
238	// Don't goto slurp_loop, because it would break from select immediately.
239	}
240
241	if glog.V(3) {	73,652✔
242	glog.Infof("DoneUntil: %d. Sending delta: %+v\n", o.doneUntil.DoneUntil(), delta)	×
243	}	×
244	o.Lock()	73,652✔
245	for id, ch := range o.subscribers {	222,963✔
246	select {	149,311✔
247	case ch <- *delta:	149,311✔
248	default:	×
249	close(ch)	×
250	delete(o.subscribers, id)	×
251	}
252	}
253	o.Unlock()	73,652✔
254	delta = &pb.OracleDelta{}	73,652✔
255	}
256	}
257
258	func (o Oracle) updateCommitStatusHelper(index uint64, src api.TxnContext) bool {	23,581✔
259	o.Lock()	23,581✔
260	defer o.Unlock()	23,581✔
261	if _, ok := o.commits[src.StartTs]; ok {	24,236✔
262	return false	655✔
263	}	655✔
264	if src.Aborted {	28,317✔
265	o.commits[src.StartTs] = 0	5,391✔
266	} else {	22,926✔
267	o.commits[src.StartTs] = src.CommitTs	17,535✔
268	}	17,535✔
269	return true	22,926✔
270	}
271
272	func (o Oracle) updateCommitStatus(index uint64, src api.TxnContext) {	23,581✔
273	// TODO: We should check if the tablet is in read-only status here.	23,581✔
274	if o.updateCommitStatusHelper(index, src) {	46,507✔
275	delta := new(pb.OracleDelta)	22,926✔
276	delta.Txns = append(delta.Txns, &pb.TxnStatus{	22,926✔
277	StartTs: src.StartTs,	22,926✔
278	CommitTs: o.commitTs(src.StartTs),	22,926✔
279	})	22,926✔
280	o.updates <- delta	22,926✔
281	}	22,926✔
282	}
283
284	func (o *Oracle) commitTs(startTs uint64) uint64 {	34,349✔
285	o.RLock()	34,349✔
286	defer o.RUnlock()	34,349✔
287	return o.commits[startTs]	34,349✔
288	}	34,349✔
289
290	func (o Oracle) storePending(ids pb.AssignedIds) {	63,745✔
291	// Wait to finish up processing everything before start id.	63,745✔
292	max := x.Max(ids.EndId, ids.ReadOnly)	63,745✔
293	if err := o.doneUntil.WaitForMark(context.Background(), max); err != nil {	63,745✔
294	glog.Errorf("Error while waiting for mark: %+v", err)	×
295	}	×
296
297	// Now send it out to updates.
298	o.updates <- &pb.OracleDelta{MaxAssigned: max}	63,745✔
299		63,745✔
300	o.Lock()	63,745✔
301	defer o.Unlock()	63,745✔
302	o.maxAssigned = x.Max(o.maxAssigned, max)	63,745✔
303	}
304
305	// MaxPending returns the maximum assigned timestamp.
306	func (o *Oracle) MaxPending() uint64 {	478✔
307	o.RLock()	478✔
308	defer o.RUnlock()	478✔
309	return o.maxAssigned	478✔
310	}	478✔
311
312	// proposeTxn proposes a txn update, and then updates src to reflect the state
313	// of the commit after proposal is run.
314	func (s Server) proposeTxn(ctx context.Context, src api.TxnContext) error {	11,423✔
315	var zp pb.ZeroProposal	11,423✔
316	zp.Txn = &api.TxnContext{	11,423✔
317	StartTs: src.StartTs,	11,423✔
318	CommitTs: src.CommitTs,	11,423✔
319	Aborted: src.Aborted,	11,423✔
320	}	11,423✔
321		11,423✔
322	// NOTE: It is important that we continue retrying proposeTxn until we succeed. This should	11,423✔
323	// happen, irrespective of what the user context timeout might be. We check for it before	11,423✔
324	// reaching this stage, but now that we're here, we have to ensure that the commit proposal goes	11,423✔
325	// through. Otherwise, we should block here forever. If we don't do this, we'll see txn	11,423✔
326	// violations in Jepsen, because we'll send out a MaxAssigned higher than a commit, which would	11,423✔
327	// cause newer txns to see older data.	11,423✔
328		11,423✔
329	// If this node stops being the leader, we want this proposal to not be forwarded to the leader,	11,423✔
330	// and get aborted.	11,423✔
331	if err := s.Node.proposeAndWait(ctx, &zp); err != nil {	11,423✔
332	return err	×
333	}	×
334
335	// There might be race between this proposal trying to commit and predicate
336	// move aborting it. A predicate move, triggered by Zero, would abort all
337	// pending transactions. At the same time, a client which has already done
338	// mutations, can proceed to commit it. A race condition can happen here,
339	// with both proposing their respective states, only one can succeed after
340	// the proposal is done. So, check again to see the fate of the transaction
341	// here.
342	src.CommitTs = s.orc.commitTs(src.StartTs)	11,423✔
343	if src.CommitTs == 0 {	13,774✔
344	src.Aborted = true	2,351✔
345	}	2,351✔
346	return nil	11,423✔
347	}
348
349	func (s Server) commit(ctx context.Context, src api.TxnContext) error {	11,423✔
350	span := otrace.FromContext(ctx)	11,423✔
351	span.Annotate([]otrace.Attribute{otrace.Int64Attribute("startTs", int64(src.StartTs))}, "")	11,423✔
352	if src.Aborted {	11,857✔
353	return s.proposeTxn(ctx, src)	434✔
354	}	434✔
355
356	// Use the start timestamp to check if we have a conflict, before we need to assign a commit ts.
357	s.orc.RLock()	10,989✔
358	conflict := s.orc.hasConflict(src)	10,989✔
359	s.orc.RUnlock()	10,989✔
360	if conflict {	12,875✔
361	span.Annotate([]otrace.Attribute{otrace.BoolAttribute("abort", true)},	1,886✔
362	"Oracle found conflict")	1,886✔
363	src.Aborted = true	1,886✔
364	return s.proposeTxn(ctx, src)	1,886✔
365	}	1,886✔
366
367	checkPreds := func() error {	18,206✔
368	// Check if any of these tablets is being moved. If so, abort the transaction.	9,103✔
369	for _, pkey := range src.Preds {	28,555✔
370	splits := strings.SplitN(pkey, "-", 2)	19,452✔
371	if len(splits) < 2 {	19,452✔
372	return errors.Errorf("Unable to find group id in %s", pkey)	×
373	}	×
374	gid, err := strconv.Atoi(splits[0])	19,452✔
375	if err != nil {	19,452✔
376	return errors.Wrapf(err, "unable to parse group id from %s", pkey)	×
377	}	×
378	pred := splits[1]	19,452✔
379	tablet := s.ServingTablet(pred)	19,452✔
380	if tablet == nil {	19,452✔
381	return errors.Errorf("Tablet for %s is nil", pred)	×
382	}	×
383	if tablet.GroupId != uint32(gid) {	19,452✔
384	return errors.Errorf("Mutation done in group: %d. Predicate %s assigned to %d",	×
385	gid, pred, tablet.GroupId)	×
386	}	×
387	if s.isBlocked(pred) {	19,452✔
388	return errors.Errorf("Commits on predicate %s are blocked due to predicate move", pred)	×
389	}	×
390	}
391	return nil	9,103✔
392	}
393	if err := checkPreds(); err != nil {	9,103✔
394	span.Annotate([]otrace.Attribute{otrace.BoolAttribute("abort", true)}, err.Error())	×
395	src.Aborted = true	×
396	return s.proposeTxn(ctx, src)	×
397	}	×
398
399	num := pb.Num{Val: 1, Type: pb.Num_TXN_TS}	9,103✔
400	assigned, err := s.lease(ctx, &num)	9,103✔
401	if err != nil {	9,103✔
402	return err	×
403	}	×
404	src.CommitTs = assigned.StartId	9,103✔
405	// Mark the transaction as done, irrespective of whether the proposal succeeded or not.	9,103✔
406	defer s.orc.doneUntil.Done(src.CommitTs)	9,103✔
407	span.Annotatef([]otrace.Attribute{otrace.Int64Attribute("commitTs", int64(src.CommitTs))},	9,103✔
408	"Node Id: %d. Proposing TxnContext: %+v", s.Node.Id, src)	9,103✔
409		9,103✔
410	if err := s.orc.commit(src); err != nil {	9,135✔
411	span.Annotatef(nil, "Found a conflict. Aborting.")	32✔
412	src.Aborted = true	32✔
413	}	32✔
414	if err := ctx.Err(); err != nil {	9,103✔
415	span.Annotatef(nil, "Aborting txn due to context timing out.")	×
416	src.Aborted = true	×
417	}	×
418	// Propose txn should be used to set watermark as done.
419	return s.proposeTxn(ctx, src)	9,103✔
420	}
421
422	// CommitOrAbort either commits a transaction or aborts it.
423	// The abortion can happen under the following conditions
424	// 1) the api.TxnContext.Aborted flag is set in the src argument
425	// 2) if there's an error (e.g server is not the leader or there's a conflicting transaction)
426	func (s Server) CommitOrAbort(ctx context.Context, src api.TxnContext) (*api.TxnContext, error) {	11,423✔
427	if ctx.Err() != nil {	11,423✔
428	return nil, ctx.Err()	×
429	}	×
430	ctx, span := otrace.StartSpan(ctx, "Zero.CommitOrAbort")	11,423✔
431	defer span.End()	11,423✔
432		11,423✔
433	if !s.Node.AmLeader() {	11,423✔
434	return nil, errors.Errorf("Only leader can decide to commit or abort")	×
435	}	×
436	err := s.commit(ctx, src)	11,423✔
437	if err != nil {	11,423✔
438	span.Annotate([]otrace.Attribute{otrace.BoolAttribute("error", true)}, err.Error())	×
439	}	×
440	return src, err	11,423✔
441	}
442
443	var errClosed = errors.New("Streaming closed by oracle")
444	var errNotLeader = errors.New("Node is no longer leader")
445
446	// Oracle streams the oracle state to the alphas.
447	// The first entry sent by Zero contains the entire state of transactions. Zero periodically
448	// confirms receipt from the group, and truncates its state. This 2-way acknowledgement is a
449	// safe way to get the status of all the transactions.
450	func (s Server) Oracle(_ api.Payload, server pb.Zero_OracleServer) error {	80✔
451	if !s.Node.AmLeader() {	80✔
452	return errNotLeader	×
453	}	×
454	ch, id := s.orc.newSubscriber()	80✔
455	defer s.orc.removeSubscriber(id)	80✔
456		80✔
457	ctx := server.Context()	80✔
458	leaderChangeCh := s.leaderChangeChannel()	80✔
459	for {	149,551✔
460	select {	149,471✔
461	case <-leaderChangeCh:	×
462	return errNotLeader	×
463	case delta, open := <-ch:	149,391✔
464	if !open {	149,391✔
465	return errClosed	×
466	}	×
467	// Pass in the latest group checksum as well, so the Alpha can use that to determine
468	// when not to service a read.
469	delta.GroupChecksums = s.groupChecksums()	149,391✔
470	if err := server.Send(&delta); err != nil {	149,391✔
471	return err	×
472	}	×
473	case <-ctx.Done():	44✔
474	return ctx.Err()	44✔
475	case <-s.closer.HasBeenClosed():	36✔
476	return errServerShutDown	36✔
477	}
478	}
479	}
480
481	// TryAbort attempts to abort the given transactions which are not already committed..
482	func (s *Server) TryAbort(ctx context.Context,
483	txns pb.TxnTimestamps) (pb.OracleDelta, error) {	×
484	delta := &pb.OracleDelta{}	×
485	for _, startTs := range txns.Ts {	×
486	// Do via proposals to avoid race	×
487	tctx := &api.TxnContext{StartTs: startTs, Aborted: true}	×
488	if err := s.proposeTxn(ctx, tctx); err != nil {	×
489	return delta, err	×
490	}	×
491	// Txn should be aborted if not already committed.
492	delta.Txns = append(delta.Txns, &pb.TxnStatus{	×
493	StartTs: startTs,	×
494	CommitTs: s.orc.commitTs(startTs)})	×
495	}
496	return delta, nil	×
497	}
498
499	// Timestamps is used to assign startTs for a new transaction
500	func (s Server) Timestamps(ctx context.Context, num pb.Num) (*pb.AssignedIds, error) {	64,409✔
501	ctx, span := otrace.StartSpan(ctx, "Zero.Timestamps")	64,409✔
502	defer span.End()	64,409✔
503		64,409✔
504	span.Annotatef(nil, "Zero id: %d. Timestamp request: %+v", s.Node.Id, num)	64,409✔
505	if ctx.Err() != nil {	64,409✔
506	return &emptyAssignedIds, ctx.Err()	×
507	}	×
508
509	num.Type = pb.Num_TXN_TS	64,409✔
510	reply, err := s.lease(ctx, num)	64,409✔
511	span.Annotatef(nil, "Response: %+v. Error: %v", reply, err)	64,409✔
512		64,409✔
513	switch err {	64,409✔
514	case nil:	63,745✔
515	s.orc.doneUntil.Done(x.Max(reply.EndId, reply.ReadOnly))	63,745✔
516	go s.orc.storePending(reply)	63,745✔
517	case errServedFromMemory:	663✔
518	// Avoid calling doneUntil.Done, and storePending.	663✔
519	err = nil	663✔
520	default:	1✔
521	glog.Errorf("Got error: %v while leasing timestamps: %+v", err, num)	1✔
522	}
523	return reply, err	64,409✔
524	}

dgraph-io / dgraph / 5262749833

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous