diff --git a/transport/internet/xdrive/slots.go b/transport/internet/xdrive/slots.go index 280f19988..b9f3f7d3f 100644 --- a/transport/internet/xdrive/slots.go +++ b/transport/internet/xdrive/slots.go @@ -22,8 +22,12 @@ import ( // reuses a slot only once that says what was in it before was taken const ( - slotMagic = "XDRVSL01" + slotMagic = "XDRVSL02" + // slotMagicV1 frames have no checksum, they are still read + slotMagicV1 = "XDRVSL01" slotHeaderLen = len(slotMagic) + 8 + 1 + 4 + 4 + // slotHeaderLenV1 is the header without the checksum + slotHeaderLenV1 = len(slotMagicV1) + 8 + 1 + 4 // slotInlineLimit is how large a segment is at most to go in the description // of its slot, which takes one request less than a body slotInlineLimit = 14 << 10 @@ -52,6 +56,13 @@ var ( slotStallTimeout = 60 * time.Second slotCloseGrace = 10 * time.Second slotGoneTimeout = 60 * time.Second + + // a failed write of a segment is tried slotWriteAttempts times + slotWriteAttempts = 3 + // a read of the head that has not answered for slotHedgeTick is asked a + // second time, for slotHedgeBudget at most + slotHedgeTick = 250 * time.Millisecond + slotHedgeBudget = time.Second ) // slotStore is a storage of objects that are written and read again by id, @@ -72,6 +83,8 @@ type slotFrame struct { size int crc uint32 data []byte + // legacy frames carry no checksum + legacy bool } func newSlotFrame(seq int64, mode byte, data []byte) slotFrame { @@ -94,7 +107,20 @@ func (f slotFrame) encode() string { func decodeSlot(description string) (slotFrame, bool) { b, err := base64.StdEncoding.DecodeString(description) - if err != nil || len(b) < slotHeaderLen || string(b[:len(slotMagic)]) != slotMagic { + if err != nil || len(b) < slotHeaderLenV1 { + return slotFrame{}, false + } + switch string(b[:len(slotMagic)]) { + case slotMagic: + return decodeSlotV2(b) + case slotMagicV1: + return decodeSlotV1(b) + } + return slotFrame{}, false +} + +func decodeSlotV2(b []byte) (slotFrame, bool) { + if len(b) < slotHeaderLen { return slotFrame{}, false } f := slotFrame{ @@ -112,10 +138,28 @@ func decodeSlot(description string) (slotFrame, bool) { return f, true } +func decodeSlotV1(b []byte) (slotFrame, bool) { + f := slotFrame{ + seq: int64(binary.BigEndian.Uint64(b[8:])), + mode: b[16], + size: int(binary.BigEndian.Uint32(b[17:])), + + legacy: true, + } + if f.mode == slotInline { + f.data = b[slotHeaderLenV1:] + if len(f.data) != f.size { + return slotFrame{}, false + } + f.crc = crc32.ChecksumIEEE(f.data) + } + return f, true +} + // holds reports whether body is the one that f is about, which a body read // right after its description need not be yet func (f slotFrame) holds(body []byte) bool { - return len(body) == f.size && crc32.ChecksumIEEE(body) == f.crc + return len(body) == f.size && (f.legacy || crc32.ChecksumIEEE(body) == f.crc) } // slotTable is the ring of one direction: the ids of its slots and of the @@ -347,11 +391,31 @@ func (w *slotWriter) upload(seq int64, chunk []byte) { w.fail(errors.New("failed to store segment ", seq).Base(err)) return } - if err := w.write(seq, mode, chunk); err != nil { - w.fail(errors.New("failed to store segment ", seq).Base(err)) - // the slot is free, the reader learns there what happened - w.write(seq, slotErr, nil) + var lastErr error + for attempt := range slotWriteAttempts { + if attempt > 0 { + select { + case <-w.ctx.Done(): + return + case <-time.After(time.Duration(attempt) * 200 * time.Millisecond): + } + // the failed write may have gone through + if d, err := w.store.readDescription(w.ctx, w.slotID(seq)); err == nil { + if f, ok := decodeSlot(d); ok && f.seq >= seq { + return + } + } + } + if lastErr = w.write(seq, mode, chunk); lastErr == nil { + return + } + if w.ctx.Err() != nil { + return + } } + w.fail(errors.New("failed to store segment ", seq).Base(lastErr)) + // the slot is free, the reader learns there what happened + w.write(seq, slotErr, nil) } func (w *slotWriter) write(seq int64, mode byte, data []byte) error { @@ -609,7 +673,13 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe backoff := slotHuntFloor var goneSince, warnedAt time.Time for { - description, err := r.store.readDescription(ctx, id) + var description string + var err error + if r.isHead(seq) { + description, err = r.readHedged(ctx, id) + } else { + description, err = r.store.readDescription(ctx, id) + } frame, ok := decodeSlot(description) switch { case err == nil && ok && frame.seq == seq: @@ -673,6 +743,55 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe } } +// readHedged reads the description of the head, and asks a second time if the +// first request is slow to answer, as one that is stuck costs more than one +// too many. The first good answer is taken +func (r *slotReader) readHedged(ctx context.Context, id string) (string, error) { + if slotHedgeTick <= 0 { + return r.store.readDescription(ctx, id) + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + type result struct { + description string + err error + } + results := make(chan result, 2) + launch := func() { + go func() { + d, err := r.store.readDescription(ctx, id) + results <- result{d, err} + }() + } + launch() + outstanding := 1 + deadline := time.Now().Add(slotHedgeBudget) + ticker := time.NewTicker(slotHedgeTick) + defer ticker.Stop() + var last result + for { + select { + case res := <-results: + outstanding-- + if res.err == nil { + return res.description, nil + } + last = res + if outstanding == 0 { + return last.description, last.err + } + case <-ticker.C: + if outstanding < 2 && time.Now().Before(deadline) { + launch() + outstanding++ + } + case <-ctx.Done(): + return "", ctx.Err() + } + } +} + // pace is about how long segments take to come one after the other func (r *slotReader) pace() time.Duration { r.mu.Lock()