xk6-frostfs/internal/registry/obj_selector.go

package registry

import (
	"context"
	"fmt"
	"time"

	"github.com/nspcc-dev/neo-go/pkg/io"
	"go.etcd.io/bbolt"
)

type ObjFilter struct {
	Status string
	Age    int
}

type ObjSelector struct {
	ctx       context.Context
	objChan   chan *ObjectInfo
	boltDB    *bbolt.DB
	filter    *ObjFilter
	cacheSize int
	looped    bool
}

// objectSelectCache is the default maximum size of a batch to select from DB.
const objectSelectCache = 1000

// NewObjSelector creates a new instance of object selector that can iterate over
// objects in the specified registry.
func NewObjSelector(registry *ObjRegistry, selectionSize int, looped bool, filter *ObjFilter) *ObjSelector {
	if selectionSize <= 0 {
		selectionSize = objectSelectCache
	}
	if filter == nil || filter.Status == "" {
		panic("filtering without status is not supported")
	}
	objSelector := &ObjSelector{
		ctx:       registry.ctx,
		boltDB:    registry.boltDB,
		filter:    filter,
		objChan:   make(chan *ObjectInfo, selectionSize*2),
		cacheSize: selectionSize,
		looped:    looped,
	}

	go objSelector.selectLoop()

	return objSelector
}

// NextObject returns the next object from the registry that matches filter of
// the selector. NextObject only roams forward from the current position of the
// selector. If there are no objects that match the filter, blocks until one of
// the following happens:
//   - a "new" next object is available;
//   - underlying registry context is done, nil objects will be returned on the
//     currently blocked and every further NextObject calls.
func (o *ObjSelector) NextObject() *ObjectInfo {
	return <-o.objChan
}

// Count returns total number of objects that match filter of the selector.
func (o *ObjSelector) Count() (int, error) {
	count := 0
	err := o.boltDB.View(func(tx *bbolt.Tx) error {
		b := tx.Bucket([]byte(o.filter.Status))
		if b == nil {
			return nil
		}

		if o.filter.Age == 0 {
			count = b.Stats().KeyN
			return nil
		}

		return b.ForEach(func(_, objBytes []byte) error {
			if objBytes != nil {
				r := io.NewBinReaderFromBuf(objBytes)

				var obj ObjectInfo
				obj.decodeFilterableFields(r)
				if r.Err != nil {
					// Ignore malformed objects
					return nil
				}
				if o.filter.match(obj) {
					count++
				}
			}
			return nil
		})
	})
	return count, err
}

func (o *ObjSelector) selectLoop() {
	cache := make([]*ObjectInfo, 0, o.cacheSize)
	var lastID uint64
	defer close(o.objChan)

	for {
		select {
		case <-o.ctx.Done():
			return
		default:
		}

		// cache the objects
		err := o.boltDB.View(func(tx *bbolt.Tx) error {
			b := tx.Bucket([]byte(o.filter.Status))
			if b == nil {
				return nil
			}

			c := b.Cursor()

			// Establish the start position for searching the next object:
			// If we should go from the beginning (lastID=0), then we start
			// from the first element. Otherwise, we start from the last
			// handled ID + 1.
			var keyBytes, objBytes []byte
			if lastID == 0 {
				keyBytes, objBytes = c.First()
			} else {
				keyBytes, objBytes = c.Seek(encodeId(lastID))
				if keyBytes != nil && decodeId(keyBytes) == lastID {
					keyBytes, objBytes = c.Next()
				}
			}

			// Iterate over objects to find the next object matching the filter.
			for ; keyBytes != nil && len(cache) != o.cacheSize; keyBytes, objBytes = c.Next() {
				if objBytes != nil {
					var obj ObjectInfo
					if err := obj.Unmarshal(objBytes); err != nil {
						// Ignore malformed objects for now. Maybe it should be panic?
						continue
					}

					if o.filter.match(obj) {
						cache = append(cache, &obj)
					}
				}
			}

			if len(cache) > 0 {
				lastID = cache[len(cache)-1].Id
			}

			return nil
		})
		if err != nil {
			panic(fmt.Errorf("fetching objects failed: %w", err))
		}

		for _, obj := range cache {
			select {
			case <-o.ctx.Done():
				return
			case o.objChan <- obj:
			}
		}

		if !o.looped && len(cache) != o.cacheSize {
			// no more objects, wait a little; the logic could be improved.
			select {
			case <-time.After(time.Second):
			case <-o.ctx.Done():
				return
			}
		}

		if o.looped && len(cache) != o.cacheSize {
			lastID = 0
		}

		// clean handled objects
		cache = cache[:0]
	}
}

func (f *ObjFilter) match(o ObjectInfo) bool {
	if f.Status != "" && f.Status != o.Status {
		return false
	}
	if f.Age != 0 {
		objAge := time.Now().UTC().Unix() - o.CreatedAt
		if objAge < int64(f.Age) {
			return false
		}
	}
	return true
}