erigon-pulse/swarm/storage/netstore.go

// Copyright 2016 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package storage

import (
	"context"
	"time"

	"github.com/ethereum/go-ethereum/swarm/log"
	"github.com/ethereum/go-ethereum/swarm/spancontext"
	opentracing "github.com/opentracing/opentracing-go"
)

var (
	// NetStore.Get timeout for get and get retries
	// This is the maximum period that the Get will block.
	// If it is reached, Get will return ErrChunkNotFound.
	netStoreRetryTimeout = 30 * time.Second
	// Minimal period between calling get method on NetStore
	// on retry. It protects calling get very frequently if
	// it returns ErrChunkNotFound very fast.
	netStoreMinRetryDelay = 3 * time.Second
	// Timeout interval before retrieval is timed out.
	// It is used in NetStore.get on waiting for ReqC to be
	// closed on a single retrieve request.
	searchTimeout = 10 * time.Second
)

// NetStore implements the ChunkStore interface,
// this chunk access layer assumed 2 chunk stores
// local storage eg. LocalStore and network storage eg., NetStore
// access by calling network is blocking with a timeout
type NetStore struct {
	localStore *LocalStore
	retrieve   func(ctx context.Context, chunk *Chunk) error
}

func NewNetStore(localStore *LocalStore, retrieve func(ctx context.Context, chunk *Chunk) error) *NetStore {
	return &NetStore{localStore, retrieve}
}

// Get is the entrypoint for local retrieve requests
// waits for response or times out
//
// Get uses get method to retrieve request, but retries if the
// ErrChunkNotFound is returned by get, until the netStoreRetryTimeout
// is reached.
func (ns *NetStore) Get(ctx context.Context, addr Address) (chunk *Chunk, err error) {

	var sp opentracing.Span
	ctx, sp = spancontext.StartSpan(
		ctx,
		"netstore.get.global")
	defer sp.Finish()

	timer := time.NewTimer(netStoreRetryTimeout)
	defer timer.Stop()

	// result and resultC provide results from the goroutine
	// where NetStore.get is called.
	type result struct {
		chunk *Chunk
		err   error
	}
	resultC := make(chan result)

	// quitC ensures that retring goroutine is terminated
	// when this function returns.
	quitC := make(chan struct{})
	defer close(quitC)

	// do retries in a goroutine so that the timer can
	// force this method to return after the netStoreRetryTimeout.
	go func() {
		// limiter ensures that NetStore.get is not called more frequently
		// then netStoreMinRetryDelay. If NetStore.get takes longer
		// then netStoreMinRetryDelay, the next retry call will be
		// without a delay.
		limiter := time.NewTimer(netStoreMinRetryDelay)
		defer limiter.Stop()

		for {
			chunk, err := ns.get(ctx, addr, 0)
			if err != ErrChunkNotFound {
				// break retry only if the error is nil
				// or other error then ErrChunkNotFound
				select {
				case <-quitC:
					// Maybe NetStore.Get function has returned
					// by the timer.C while we were waiting for the
					// results. Terminate this goroutine.
				case resultC <- result{chunk: chunk, err: err}:
					// Send the result to the parrent goroutine.
				}
				return

			}
			select {
			case <-quitC:
				// NetStore.Get function has returned, possibly
				// by the timer.C, which makes this goroutine
				// not needed.
				return
			case <-limiter.C:
			}
			// Reset the limiter for the next iteration.
			limiter.Reset(netStoreMinRetryDelay)
			log.Debug("NetStore.Get retry chunk", "key", addr)
		}
	}()

	select {
	case r := <-resultC:
		return r.chunk, r.err
	case <-timer.C:
		return nil, ErrChunkNotFound
	}
}

// GetWithTimeout makes a single retrieval attempt for a chunk with a explicit timeout parameter
func (ns *NetStore) GetWithTimeout(ctx context.Context, addr Address, timeout time.Duration) (chunk *Chunk, err error) {
	return ns.get(ctx, addr, timeout)
}

func (ns *NetStore) get(ctx context.Context, addr Address, timeout time.Duration) (chunk *Chunk, err error) {
	if timeout == 0 {
		timeout = searchTimeout
	}

	var sp opentracing.Span
	ctx, sp = spancontext.StartSpan(
		ctx,
		"netstore.get")
	defer sp.Finish()

	if ns.retrieve == nil {
		chunk, err = ns.localStore.Get(ctx, addr)
		if err == nil {
			return chunk, nil
		}
		if err != ErrFetching {
			return nil, err
		}
	} else {
		var created bool
		chunk, created = ns.localStore.GetOrCreateRequest(ctx, addr)

		if chunk.ReqC == nil {
			return chunk, nil
		}

		if created {
			err := ns.retrieve(ctx, chunk)
			if err != nil {
				// mark chunk request as failed so that we can retry it later
				chunk.SetErrored(ErrChunkUnavailable)
				return nil, err
			}
		}
	}

	t := time.NewTicker(timeout)
	defer t.Stop()

	select {
	case <-t.C:
		// mark chunk request as failed so that we can retry
		chunk.SetErrored(ErrChunkNotFound)
		return nil, ErrChunkNotFound
	case <-chunk.ReqC:
	}
	chunk.SetErrored(nil)
	return chunk, nil
}

// Put is the entrypoint for local store requests coming from storeLoop
func (ns *NetStore) Put(ctx context.Context, chunk *Chunk) {
	ns.localStore.Put(ctx, chunk)
}

// Close chunk store
func (ns *NetStore) Close() {
	ns.localStore.Close()
}