mirror of
https://github.com/restic/restic.git
synced 2026-09-21 11:38:31 +00:00
Update dependencies
This commit is contained in:
Alexander Neumann
+63
-312
@@ -15,201 +15,21 @@
|
||||
package pubsub
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
"google.golang.org/api/iterator"
|
||||
"google.golang.org/api/support/bundler"
|
||||
pb "google.golang.org/genproto/googleapis/pubsub/v1"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
)
|
||||
|
||||
type messageIterator struct {
|
||||
impl interface {
|
||||
next() (*Message, error)
|
||||
stop()
|
||||
}
|
||||
}
|
||||
|
||||
type pollingMessageIterator struct {
|
||||
// kaTicker controls how often we send an ack deadline extension request.
|
||||
kaTicker *time.Ticker
|
||||
// ackTicker controls how often we acknowledge a batch of messages.
|
||||
ackTicker *time.Ticker
|
||||
|
||||
ka *keepAlive
|
||||
acker *acker
|
||||
nacker *bundler.Bundler
|
||||
puller *puller
|
||||
|
||||
// mu ensures that cleanup only happens once, and concurrent Stop
|
||||
// invocations block until cleanup completes.
|
||||
mu sync.Mutex
|
||||
|
||||
// closed is used to signal that Stop has been called.
|
||||
closed chan struct{}
|
||||
}
|
||||
|
||||
var useStreamingPull = false
|
||||
|
||||
// newMessageIterator starts a new messageIterator. Stop must be called on the messageIterator
|
||||
// newMessageIterator starts a new streamingMessageIterator. Stop must be called on the messageIterator
|
||||
// when it is no longer needed.
|
||||
// subName is the full name of the subscription to pull messages from.
|
||||
// ctx is the context to use for acking messages and extending message deadlines.
|
||||
func newMessageIterator(ctx context.Context, s service, subName string, po *pullOptions) *messageIterator {
|
||||
if !useStreamingPull {
|
||||
return &messageIterator{
|
||||
impl: newPollingMessageIterator(ctx, s, subName, po),
|
||||
}
|
||||
}
|
||||
func newMessageIterator(ctx context.Context, s service, subName string, po *pullOptions) *streamingMessageIterator {
|
||||
sp := s.newStreamingPuller(ctx, subName, int32(po.ackDeadline.Seconds()))
|
||||
err := sp.open()
|
||||
if grpc.Code(err) == codes.Unimplemented {
|
||||
log.Println("pubsub: streaming pull unimplemented; falling back to legacy pull")
|
||||
return &messageIterator{
|
||||
impl: newPollingMessageIterator(ctx, s, subName, po),
|
||||
}
|
||||
}
|
||||
// TODO(jba): handle other non-nil error?
|
||||
log.Println("using streaming pull")
|
||||
return &messageIterator{
|
||||
impl: newStreamingMessageIterator(ctx, sp, po),
|
||||
}
|
||||
}
|
||||
|
||||
func newPollingMessageIterator(ctx context.Context, s service, subName string, po *pullOptions) *pollingMessageIterator {
|
||||
// TODO: make kaTicker frequency more configurable.
|
||||
// (ackDeadline - 5s) is a reasonable default for now, because the minimum ack period is 10s. This gives us 5s grace.
|
||||
keepAlivePeriod := po.ackDeadline - 5*time.Second
|
||||
kaTicker := time.NewTicker(keepAlivePeriod) // Stopped in it.Stop
|
||||
|
||||
// Ack promptly so users don't lose work if client crashes.
|
||||
ackTicker := time.NewTicker(100 * time.Millisecond) // Stopped in it.Stop
|
||||
ka := &keepAlive{
|
||||
s: s,
|
||||
Ctx: ctx,
|
||||
Sub: subName,
|
||||
ExtensionTick: kaTicker.C,
|
||||
Deadline: po.ackDeadline,
|
||||
MaxExtension: po.maxExtension,
|
||||
}
|
||||
|
||||
ack := &acker{
|
||||
s: s,
|
||||
Ctx: ctx,
|
||||
Sub: subName,
|
||||
AckTick: ackTicker.C,
|
||||
Notify: ka.Remove,
|
||||
}
|
||||
|
||||
nacker := bundler.NewBundler("", func(ackIDs interface{}) {
|
||||
// NACK by setting the ack deadline to zero, to make the message
|
||||
// immediately available for redelivery.
|
||||
//
|
||||
// If the RPC fails, nothing we can do about it. In the worst case, the
|
||||
// deadline for these messages will expire and they will still get
|
||||
// redelivered.
|
||||
_ = s.modifyAckDeadline(ctx, subName, 0, ackIDs.([]string))
|
||||
})
|
||||
nacker.DelayThreshold = 100 * time.Millisecond // nack promptly
|
||||
nacker.BundleCountThreshold = 10
|
||||
|
||||
pull := newPuller(s, subName, ctx, po.maxPrefetch, ka.Add, ka.Remove)
|
||||
|
||||
ka.Start()
|
||||
ack.Start()
|
||||
return &pollingMessageIterator{
|
||||
kaTicker: kaTicker,
|
||||
ackTicker: ackTicker,
|
||||
ka: ka,
|
||||
acker: ack,
|
||||
nacker: nacker,
|
||||
puller: pull,
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Next returns the next Message to be processed. The caller must call
|
||||
// Message.Done when finished with it.
|
||||
// Once Stop has been called, calls to Next will return iterator.Done.
|
||||
func (it *messageIterator) Next() (*Message, error) {
|
||||
return it.impl.next()
|
||||
}
|
||||
|
||||
func (it *pollingMessageIterator) next() (*Message, error) {
|
||||
m, err := it.puller.Next()
|
||||
if err == nil {
|
||||
m.doneFunc = it.done
|
||||
return m, nil
|
||||
}
|
||||
|
||||
select {
|
||||
// If Stop has been called, we return Done regardless the value of err.
|
||||
case <-it.closed:
|
||||
return nil, iterator.Done
|
||||
default:
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Client code must call Stop on a messageIterator when finished with it.
|
||||
// Stop will block until Done has been called on all Messages that have been
|
||||
// returned by Next, or until the context with which the messageIterator was created
|
||||
// is cancelled or exceeds its deadline.
|
||||
// Stop need only be called once, but may be called multiple times from
|
||||
// multiple goroutines.
|
||||
func (it *messageIterator) Stop() {
|
||||
it.impl.stop()
|
||||
}
|
||||
|
||||
func (it *pollingMessageIterator) stop() {
|
||||
it.mu.Lock()
|
||||
defer it.mu.Unlock()
|
||||
|
||||
select {
|
||||
case <-it.closed:
|
||||
// Cleanup has already been performed.
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
// We close this channel before calling it.puller.Stop to ensure that we
|
||||
// reliably return iterator.Done from Next.
|
||||
close(it.closed)
|
||||
|
||||
// Stop the puller. Once this completes, no more messages will be added
|
||||
// to it.ka.
|
||||
it.puller.Stop()
|
||||
|
||||
// Start acking messages as they arrive, ignoring ackTicker. This will
|
||||
// result in it.ka.Stop, below, returning as soon as possible.
|
||||
it.acker.FastMode()
|
||||
|
||||
// This will block until
|
||||
// (a) it.ka.Ctx is done, or
|
||||
// (b) all messages have been removed from keepAlive.
|
||||
// (b) will happen once all outstanding messages have been either ACKed or NACKed.
|
||||
it.ka.Stop()
|
||||
|
||||
// There are no more live messages, so kill off the acker.
|
||||
it.acker.Stop()
|
||||
it.nacker.Flush()
|
||||
it.kaTicker.Stop()
|
||||
it.ackTicker.Stop()
|
||||
}
|
||||
|
||||
func (it *pollingMessageIterator) done(ackID string, ack bool) {
|
||||
if ack {
|
||||
it.acker.Ack(ackID)
|
||||
// There's no need to call it.ka.Remove here, as acker will
|
||||
// call it via its Notify function.
|
||||
} else {
|
||||
it.ka.Remove(ackID)
|
||||
_ = it.nacker.Add(ackID, len(ackID)) // ignore error; this is just an optimization
|
||||
}
|
||||
_ = sp.open() // error stored in sp
|
||||
return newStreamingMessageIterator(ctx, sp, po)
|
||||
}
|
||||
|
||||
type streamingMessageIterator struct {
|
||||
@@ -222,7 +42,6 @@ type streamingMessageIterator struct {
|
||||
failed chan struct{} // closed on stream error
|
||||
stopped chan struct{} // closed when Stop is called
|
||||
drained chan struct{} // closed when stopped && no more pending messages
|
||||
msgc chan *Message
|
||||
wg sync.WaitGroup
|
||||
|
||||
mu sync.Mutex
|
||||
@@ -242,104 +61,36 @@ func newStreamingMessageIterator(ctx context.Context, sp *streamingPuller, po *p
|
||||
ackTicker := time.NewTicker(100 * time.Millisecond)
|
||||
nackTicker := time.NewTicker(100 * time.Millisecond)
|
||||
it := &streamingMessageIterator{
|
||||
ctx: ctx,
|
||||
sp: sp,
|
||||
po: po,
|
||||
kaTicker: kaTicker,
|
||||
ackTicker: ackTicker,
|
||||
nackTicker: nackTicker,
|
||||
failed: make(chan struct{}),
|
||||
stopped: make(chan struct{}),
|
||||
drained: make(chan struct{}),
|
||||
// use maxPrefetch as the channel's buffer size.
|
||||
msgc: make(chan *Message, po.maxPrefetch),
|
||||
ctx: ctx,
|
||||
sp: sp,
|
||||
po: po,
|
||||
kaTicker: kaTicker,
|
||||
ackTicker: ackTicker,
|
||||
nackTicker: nackTicker,
|
||||
failed: make(chan struct{}),
|
||||
stopped: make(chan struct{}),
|
||||
drained: make(chan struct{}),
|
||||
keepAliveDeadlines: map[string]time.Time{},
|
||||
pendingReq: &pb.StreamingPullRequest{},
|
||||
}
|
||||
it.wg.Add(2)
|
||||
go it.receiver()
|
||||
it.wg.Add(1)
|
||||
go it.sender()
|
||||
return it
|
||||
}
|
||||
|
||||
func (it *streamingMessageIterator) next() (*Message, error) {
|
||||
// If ctx has been cancelled or the iterator is done, return straight
|
||||
// away (even if there are buffered messages available).
|
||||
select {
|
||||
case <-it.ctx.Done():
|
||||
return nil, it.ctx.Err()
|
||||
|
||||
case <-it.failed:
|
||||
break
|
||||
|
||||
case <-it.stopped:
|
||||
break
|
||||
|
||||
default:
|
||||
// Wait for a message, but also for one of the above conditions.
|
||||
select {
|
||||
case msg := <-it.msgc:
|
||||
// Since active select cases are chosen at random, this can return
|
||||
// nil (from the channel close) even if it.failed or it.stopped is
|
||||
// closed.
|
||||
if msg == nil {
|
||||
break
|
||||
}
|
||||
msg.doneFunc = it.done
|
||||
return msg, nil
|
||||
|
||||
case <-it.ctx.Done():
|
||||
return nil, it.ctx.Err()
|
||||
|
||||
case <-it.failed:
|
||||
break
|
||||
|
||||
case <-it.stopped:
|
||||
break
|
||||
}
|
||||
}
|
||||
// Here if the iterator is done.
|
||||
it.mu.Lock()
|
||||
defer it.mu.Unlock()
|
||||
return nil, it.err
|
||||
}
|
||||
|
||||
// Subscription.receive will call stop on its messageIterator when finished with it.
|
||||
// Stop will block until Done has been called on all Messages that have been
|
||||
// returned by Next, or until the context with which the messageIterator was created
|
||||
// is cancelled or exceeds its deadline.
|
||||
func (it *streamingMessageIterator) stop() {
|
||||
it.mu.Lock()
|
||||
select {
|
||||
case <-it.stopped:
|
||||
it.mu.Unlock()
|
||||
it.wg.Wait()
|
||||
return
|
||||
default:
|
||||
close(it.stopped)
|
||||
}
|
||||
if it.err == nil {
|
||||
it.err = iterator.Done
|
||||
}
|
||||
// Before reading from the channel, see if we're already drained.
|
||||
it.checkDrained()
|
||||
it.mu.Unlock()
|
||||
// Nack all the pending messages.
|
||||
// Grab the lock separately for each message to allow the receiver
|
||||
// and sender goroutines to make progress.
|
||||
// Why this will eventually terminate:
|
||||
// - If the receiver is not blocked on a stream Recv, then
|
||||
// it will write all the messages it has received to the channel,
|
||||
// then exit, closing the channel.
|
||||
// - If the receiver is blocked, then this loop will eventually
|
||||
// nack all the messages in the channel. Once done is called
|
||||
// on the remaining messages, the iterator will be marked as drained,
|
||||
// which will trigger the sender to terminate. When it does, it
|
||||
// performs a CloseSend on the stream, which will result in the blocked
|
||||
// stream Recv returning.
|
||||
for m := range it.msgc {
|
||||
it.mu.Lock()
|
||||
delete(it.keepAliveDeadlines, m.ackID)
|
||||
it.addDeadlineMod(m.ackID, 0)
|
||||
it.checkDrained()
|
||||
it.mu.Unlock()
|
||||
}
|
||||
it.wg.Wait()
|
||||
}
|
||||
|
||||
@@ -394,52 +145,40 @@ func (it *streamingMessageIterator) fail(err error) {
|
||||
it.mu.Unlock()
|
||||
}
|
||||
|
||||
// receiver runs in a goroutine and handles all receives from the stream.
|
||||
func (it *streamingMessageIterator) receiver() {
|
||||
defer it.wg.Done()
|
||||
defer close(it.msgc)
|
||||
for {
|
||||
// Stop retrieving messages if the context is done, the stream
|
||||
// failed, or the iterator's Stop method was called.
|
||||
select {
|
||||
case <-it.ctx.Done():
|
||||
return
|
||||
case <-it.failed:
|
||||
return
|
||||
case <-it.stopped:
|
||||
return
|
||||
default:
|
||||
}
|
||||
// Receive messages from stream. This may block indefinitely.
|
||||
msgs, err := it.sp.fetchMessages()
|
||||
|
||||
// The streamingPuller handles retries, so any error here
|
||||
// is fatal to the iterator.
|
||||
if err != nil {
|
||||
it.fail(err)
|
||||
return
|
||||
}
|
||||
// We received some messages. Remember them so we can
|
||||
// keep them alive.
|
||||
deadline := time.Now().Add(it.po.maxExtension)
|
||||
it.mu.Lock()
|
||||
for _, m := range msgs {
|
||||
it.keepAliveDeadlines[m.ackID] = deadline
|
||||
}
|
||||
it.mu.Unlock()
|
||||
// Deliver the messages to the channel.
|
||||
for _, m := range msgs {
|
||||
select {
|
||||
case <-it.ctx.Done():
|
||||
return
|
||||
case <-it.failed:
|
||||
return
|
||||
// Don't return if stopped. We want to send the remaining
|
||||
// messages on the channel, where they will be nacked.
|
||||
case it.msgc <- m:
|
||||
}
|
||||
}
|
||||
// receive makes a call to the stream's Recv method and returns
|
||||
// its messages.
|
||||
func (it *streamingMessageIterator) receive() ([]*Message, error) {
|
||||
// Stop retrieving messages if the context is done, the stream
|
||||
// failed, or the iterator's Stop method was called.
|
||||
select {
|
||||
case <-it.ctx.Done():
|
||||
return nil, it.ctx.Err()
|
||||
default:
|
||||
}
|
||||
it.mu.Lock()
|
||||
err := it.err
|
||||
it.mu.Unlock()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// Receive messages from stream. This may block indefinitely.
|
||||
msgs, err := it.sp.fetchMessages()
|
||||
// The streamingPuller handles retries, so any error here
|
||||
// is fatal.
|
||||
if err != nil {
|
||||
it.fail(err)
|
||||
return nil, err
|
||||
}
|
||||
// We received some messages. Remember them so we can
|
||||
// keep them alive.
|
||||
deadline := time.Now().Add(it.po.maxExtension)
|
||||
it.mu.Lock()
|
||||
for _, m := range msgs {
|
||||
m.doneFunc = it.done
|
||||
it.keepAliveDeadlines[m.ackID] = deadline
|
||||
}
|
||||
it.mu.Unlock()
|
||||
return msgs, nil
|
||||
}
|
||||
|
||||
// sender runs in a goroutine and handles all sends to the stream.
|
||||
@@ -518,3 +257,15 @@ func (it *streamingMessageIterator) handleKeepAlives() bool {
|
||||
it.checkDrained()
|
||||
return len(live) > 0
|
||||
}
|
||||
|
||||
func getKeepAliveAckIDs(items map[string]time.Time) (live, expired []string) {
|
||||
now := time.Now()
|
||||
for id, expiry := range items {
|
||||
if expiry.Before(now) {
|
||||
expired = append(expired, id)
|
||||
} else {
|
||||
live = append(live, id)
|
||||
}
|
||||
}
|
||||
return live, expired
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user