Make watermark not use timer

[srh]

This avoids the question of fixing issue #1345 -- it keeps watermark behavior the same.

---

This change is 

[srh]

Review status: 0 of 13 files reviewed at latest revision, 4 unresolved discussions.

---

client/mutations.go, line 375 at r1 (raw file):

	if req.line != 0 && req.mark != nil {
		atomic.AddUint64(&d.rdfs, uint64(req.size()))
		req.mark.Done(req.line)

Wound up encapsulating the channel.

---

cmd/dgraph/main_test.go, line 1374 at r1 (raw file):

	// we need watermarks for reindexing
	x.AssertTrue(!x.IsTestRun())

I removed the SetTestRun/IsTestRun stuff. I don't know what exactly its reason for existing was -- tests still work.

---

worker/groups.go, line 664 at r1 (raw file):

				return
			}
			n.syncAllMarks(ctx, lastIndex)

syncAllMarks never returned an error before.

---

worker/index.go, line 90 at r1 (raw file):

}

func (n *node) waitForAppliedMark(ctx context.Context, lastIndex uint64) {

The downside of centralizing WaitForMark is that we do lose the specific message.

---

Comments from Reviewable

[manishrjain]

Nice refactoring. Have some comments to simplify the waiter code.

Also, I think the reason for those test run cases was that query package was failing randomly. You might want to run the tests multiple times to ensure that the failures don't reoccur. We might have modified the tests in some way, which might have fixed the issue; but hard to say.

---

Reviewed 12 of 13 files at r1, 1 of 1 files at r2.
Review status: all files reviewed at latest revision, 6 unresolved discussions.

---

worker/index.go, line 91 at r2 (raw file):

func (n *node) waitForAppliedMark(ctx context.Context, lastIndex uint64) {
	n.applied.WaitForMark(lastIndex)

Can remove this single line function.

---

x/watermark.go, line 179 at r2 (raw file):

				// Some partly duplicated code with what's below.
				heap.Pop(&waiterIndices)
				toNotify := waiters[min]

This logic seems weird. We might have someone waiting for 101, but the heap only sees 10, 50 and 150. This map lookup would then not trigger; or so it seems.

A simpler solution might be to keep a heap of waiters and compare the waiter index against the min done index. Not sure if a heap allows duplication, but if it does, then we'd avoid having a map of list of chan; which is another benefit to readability.

P.S. Another thought was that you can attach a waiter to an index in the heap itself, thus making the wait done trigger even simpler. Just add the index as a done index when attaching waiter to the heap element.

---

Comments from Reviewable

[srh]

I'll run that test in a loop overnight.

---

Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed.

---

worker/index.go, line 91 at r2 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

Can remove this single line function.

Done.

---

x/watermark.go, line 179 at r2 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

This logic seems weird. We might have someone waiting for 101, but the heap only sees 10, 50 and 150. This map lookup would then not trigger; or so it seems.

A simpler solution might be to keep a heap of waiters and compare the waiter index against the min done index. Not sure if a heap allows duplication, but if it does, then we'd avoid having a map of list of chan; which is another benefit to readability.

P.S. Another thought was that you can attach a waiter to an index in the heap itself, thus making the wait done trigger even simpler. Just add the index as a done index when attaching waiter to the heap element.

If we have someone waiting for 101, then the heap will contain 101. waiterIndices is a different heap from indices.

We could keep a heap of waiters -- it can surely have duplicates -- the main reason I didn't do that was because I just mirrored the existing implementation with pending/indices.

If we put waiters on the pending/indices heap I think it would make the 10/50/150 while somebody waits on 101 case much more difficult.

---

Comments from Reviewable

[srh]

I have not seen test failures.

---

Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed.

---

Comments from Reviewable

[manishrjain]

Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed.

---

x/watermark.go, line 179 at r2 (raw file):

If we put waiters on the pending/indices heap I think it would make the 10/50/150 while somebody waits on 101 case much more difficult.

Not really, in fact it should be simplified. Say, the heap has 10, 50 and 150. It's waiting on 10 right now. Then we ask to wait for 101. We add 101 in the heap (as done), and attach the waiting channel to it. Once 10 is done, and then 50 is done, heap moves to 101 -- sees that it's already done, and writes to the channel, and moves onto 150. This is just simpler logic to think through and hence code.

---

Comments from Reviewable

[srh]

Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed.

---

x/watermark.go, line 179 at r2 (raw file):

Previously, manishrjain (Manish R Jain) wrote…

If we put waiters on the pending/indices heap I think it would make the 10/50/150 while somebody waits on 101 case much more difficult.

Not really, in fact it should be simplified. Say, the heap has 10, 50 and 150. It's waiting on 10 right now. Then we ask to wait for 101. We add 101 in the heap (as done), and attach the waiting channel to it. Once 10 is done, and then 50 is done, heap moves to 101 -- sees that it's already done, and writes to the channel, and moves onto 150. This is just simpler logic to think through and hence code.

What if after we consider 101 to be done, 101 begins?

---

Comments from Reviewable

[manishrjain]

This particular change is a NOOP from that perspective. We already put things in place so that the watermark doesn't miss marks. In fact, watermark code has an asset failure for this case. Sent from Nexus 6P

…

On Aug 29, 2017 3:18 PM, "Sam Hughes" ***@***.***> wrote: Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed. ------------------------------ *x/watermark.go, line 179 at r2 <https://reviewable.io:443/reviews/dgraph-io/dgraph/1366#-KsbkZqlYXERnmTGDpJx:-KsgPi7h_R5JMefYAcUY:b-ieypzu> (raw file <https://github.com/dgraph-io/dgraph/blob/52ec7f97c60b8fa7031d057c80457229a03ab2e3/x/watermark.go#L179>):* *Previously, manishrjain (Manish R Jain) wrote…* If we put waiters on the pending/indices heap I think it would make the 10/50/150 while somebody waits on 101 case much more difficult. Not really, in fact it should be simplified. Say, the heap has 10, 50 and 150. It's waiting on 10 right now. Then we ask to wait for 101. We add 101 in the heap (as done), and attach the waiting channel to it. Once 10 is done, and then 50 is done, heap moves to 101 -- sees that it's already done, and writes to the channel, and moves onto 150. This is just simpler logic to think through and hence code. What if after we consider 101 to be done, 101 begins? ------------------------------ *Comments from Reviewable <https://reviewable.io:443/reviews/dgraph-io/dgraph/1366>* — You are receiving this because you were assigned. Reply to this email directly, view it on GitHub <#1366 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ABsyNLmabDanjLo106ZmQYs5JRZiTKblks5sc59BgaJpZM4PEGXx> .

[srh]

The watermark code has an assert that we don't create entries below doneUntil. It does not assert that a waiter for level L can be added after a begin Mark for L has been made, but never before.

---

Review status: 12 of 13 files reviewed at latest revision, 6 unresolved discussions, some commit checks failed.

---

Comments from Reviewable

[manishrjain]

Review status: 12 of 13 files reviewed at latest revision, 8 unresolved discussions, some commit checks failed.

---

x/watermark.go, line 125 at r3 (raw file):

	// pending maps raft proposal index to the number of pending mutations for this proposal.
	pending := make(map[uint64]int)
	waiters := make(map[uint64][]chan struct{})

Just use one chan, and close it to represent done for all waiters.

---

x/watermark.go, line 208 at r3 (raw file):

	for mark := range w.markCh {
		if mark.waiter != nil {

Maybe better to do this:

if mark.index > 0 {
  ...
  continue
}
if len(mark.indices) > 0 {
  ...
  continue
}
if mark.waiter != nil {
   ...
  continue
}

Reads better. Also, mark.index is the most frequent code path, followed by indices, followed by waiter. This puts them in that order.

---

Comments from Reviewable

[manishrjain]

---

Reviewed 1 of 1 files at r3.
Review status: all files reviewed at latest revision, 7 unresolved discussions, some commit checks failed.

---

Comments from Reviewable

[srh]

In master as of 5e1709e.