Method WaitLinearizableRead

conn/node.go:643–674

WaitLinearizableRead waits until a linearizable read can be performed.

(ctx context.Context)

// WaitLinearizableRead waits until a linearizable read can be performed.
func (n *Node) WaitLinearizableRead(ctx context.Context) error {
	span := trace.SpanFromContext(ctx)
	span.AddEvent("WaitLinearizableRead")

	if num := atomic.AddUint64(&readIndexTotal, 1); num%1000 == 0 {
		glog.V(2).Infof("ReadIndex Total: %d\n", num)
	}
	indexCh := make(chan uint64, 1)
	select {
	case n.requestCh <- linReadReq{indexCh: indexCh}:
		span.AddEvent("Pushed to requestCh")
	case <-ctx.Done():
		span.AddEvent("Context expired")
		return ctx.Err()
	}

	select {
	case index := <-indexCh:
		span.AddEvent(fmt.Sprintf("Received index %d", index))
		if index == 0 {
			return errReadIndex
		} else if num := atomic.AddUint64(&readIndexOk, 1); num%1000 == 0 {
			glog.V(2).Infof("ReadIndex OK: %d\n", num)
		}
		err := n.Applied.WaitForMark(ctx, index)
		span.AddEvent(fmt.Sprintf("Error from Applied.WaitForMark: %v", err))
		return err
	case <-ctx.Done():
		span.AddEvent("Context expired")
		return ctx.Err()
	}
}

// RunReadIndexLoop runs the RAFT index in a loop.
func (n *Node) RunReadIndexLoop(closer *z.Closer, readStateCh <-chan raft.ReadState) {