dagstore

README

DAG store

This README will be populated soon. In the meantime, please refer to the design document.

License

Dual-licensed: MIT, Apache Software License v2, by way of the Permissive License Stack.

Documentation

Index

• Variables
• type AcquireOpts
• type AllShardsInfo
• type Config
• type DAGStore
  • func NewDAGStore(cfg Config) (*DAGStore, error)
  • func (d *DAGStore) AcquireShard(ctx context.Context, key shard.Key, out chan ShardResult, _ AcquireOpts) error
  • func (d *DAGStore) AllShardsInfo() AllShardsInfo
  • func (d *DAGStore) Close() error
  • func (d *DAGStore) DestroyShard(ctx context.Context, key shard.Key, out chan ShardResult, _ DestroyOpts) error
  • func (d *DAGStore) GC(ctx context.Context) (map[shard.Key]error, error)
  • func (d *DAGStore) GetShardInfo(k shard.Key) (ShardInfo, error)
  • func (d *DAGStore) RecoverShard(ctx context.Context, key shard.Key, out chan ShardResult, _ RecoverOpts) error
  • func (d *DAGStore) RegisterShard(ctx context.Context, key shard.Key, mnt mount.Mount, out chan ShardResult, ...) error
• type DestroyOpts
• type OpType
  • func (o OpType) String() string
• type PersistedShard
• type ReadBlockstore
• type RecoverOpts
• type RegisterOpts
• type Shard
  • func (s *Shard) MarshalJSON() ([]byte, error)
  • func (s *Shard) UnmarshalJSON(b []byte) error
• type ShardAccessor
  • func NewShardAccessor(data mount.Reader, idx index.Index, s *Shard) (*ShardAccessor, error)
  • func (sa *ShardAccessor) Blockstore() (ReadBlockstore, error)
  • func (sa *ShardAccessor) Close() error
  • func (sa *ShardAccessor) Shard() shard.Key
• type ShardInfo
• type ShardResult
• type ShardState
  • func (ss ShardState) String() string
• type Throttler
  • func NewThrottler(maxConcurrency int) Throttler
• type Trace

Variables

var (
	// ErrShardUnknown is the error returned when the requested shard is
	// not known to the DAG store.
	ErrShardUnknown = errors.New("shard not found")

	// ErrShardExists is the error returned upon registering a duplicate shard.
	ErrShardExists = errors.New("shard already exists")

	// ErrShardInitializationFailed is returned when shard initialization fails.
	ErrShardInitializationFailed = errors.New("shard initialization failed")

	// ErrShardInUse is returned when the user attempts to destroy a shard that
	// is in use.
	ErrShardInUse = errors.New("shard in use")
)

var (
	// StoreNamespace is the namespace under which shard state will be persisted.
	StoreNamespace = ds.NewKey("dagstore")
)

Types

type AcquireOpts

type AcquireOpts struct {
}

type AllShardsInfo

type AllShardsInfo map[shard.Key]ShardInfo

type Config

type Config struct {
	// TransientsDir is the path to directory where local transient files will
	// be created for remote mounts.
	TransientsDir string

	// IndexDir is the path where indices are stored.
	IndexDir string

	// Datastore is the datastore where shard state will be persisted.
	Datastore ds.Datastore

	// MountRegistry contains the set of recognized mount types.
	MountRegistry *mount.Registry

	// TraceCh is a channel where the caller desires to be notified of every
	// shard operation. Publishing to this channel blocks the event loop, so the
	// caller must ensure the channel is serviced appropriately.
	TraceCh chan<- Trace

	// FailureCh is a channel to be notified every time that a shard moves to
	// ShardStateErrored. A nil value will send no failure notifications.
	// Failure events can be used to evaluate the error and call
	// DAGStore.RecoverShard if deemed recoverable.
	FailureCh chan<- ShardResult

	// MaxConcurrentIndex is the maximum indexing jobs that can
	// run concurrently. 0 (default) disables throttling.
	MaxConcurrentIndex int

	// MaxConcurrentFetch is the maximum fetching jobs that can
	// run concurrently. 0 (default) disables throttling.
	MaxConcurrentFetch int
}

type DAGStore

type DAGStore struct {
	// contains filtered or unexported fields
}

DAGStore is the central object of the DAG store.

func NewDAGStore

func NewDAGStore(cfg Config) (*DAGStore, error)

NewDAGStore constructs a new DAG store with the supplied configuration.

func (*DAGStore) AcquireShard

func (d *DAGStore) AcquireShard(ctx context.Context, key shard.Key, out chan ShardResult, _ AcquireOpts) error

AcquireShard acquires access to the specified shard, and returns a ShardAccessor, an object that enables various patterns of access to the data contained within the shard.

This operation may resolve near-instantaneously if the shard is available locally. If not, the shard data may be fetched from its mount.

This method returns an error synchronously if preliminary validation fails. Otherwise, it queues the shard for acquisition. The caller should monitor supplied channel for a result.

func (*DAGStore) AllShardsInfo

func (d *DAGStore) AllShardsInfo() AllShardsInfo

AllShardsInfo returns the current state of all registered shards, as well as any errors.

func (*DAGStore) Close

func (d *DAGStore) Close() error

func (*DAGStore) DestroyShard

func (d *DAGStore) DestroyShard(ctx context.Context, key shard.Key, out chan ShardResult, _ DestroyOpts) error

func (*DAGStore) GC

func (d *DAGStore) GC(ctx context.Context) (map[shard.Key]error, error)

GC attempts to reclaim the transient files of shards that are currently available but inactive.

It is not strictly atomic for now, as it determines which shards to reclaim first, sends operations to the event loop, and waits for them to execute. In the meantime, there could be state transitions that change reclaimability of shards (some shards deemed reclaimable are no longer so, and vice versa).

However, the event loop checks for safety prior to deletion, so it will skip over shards that are no longer safe to delete.

func (*DAGStore) GetShardInfo

func (d *DAGStore) GetShardInfo(k shard.Key) (ShardInfo, error)

GetShardInfo returns the current state of shard with key k.

If the shard is not known, ErrShardUnknown is returned.

func (*DAGStore) RecoverShard

func (d *DAGStore) RecoverShard(ctx context.Context, key shard.Key, out chan ShardResult, _ RecoverOpts) error

RecoverShard recovers a shard in ShardStateErrored state.

If the shard referenced by the key doesn't exist, an error is returned immediately and no result is delivered on the supplied channel.

If the shard is not in the ShardStateErrored state, the operation is accepted but an error will be returned quickly on the supplied channel.

Otherwise, the recovery operation will be queued and the supplied channel will be notified when it completes.

TODO add an operation identifier to ShardResult -- starts to look like

a Trace event?

func (*DAGStore) RegisterShard

func (d *DAGStore) RegisterShard(ctx context.Context, key shard.Key, mnt mount.Mount, out chan ShardResult, opts RegisterOpts) error

RegisterShard initiates the registration of a new shard.

This method returns an error synchronously if preliminary validation fails. Otherwise, it queues the shard for registration. The caller should monitor supplied channel for a result.

type DestroyOpts

type DestroyOpts struct {
}

type OpType

type OpType int

const (
	OpShardRegister OpType = iota
	OpShardInitialize
	OpShardMakeAvailable
	OpShardDestroy
	OpShardAcquire
	OpShardFail
	OpShardRelease
	OpShardGC
	OpShardRecover
)

func (OpType) String

func (o OpType) String() string

type PersistedShard

type PersistedShard struct {
	Key   string     `json:"k"`
	URL   string     `json:"u"`
	State ShardState `json:"s"`
	Lazy  bool       `json:"l"`
	Error string     `json:"e"`
}

PersistedShard is the persistent representation of the Shard.

type ReadBlockstore

type ReadBlockstore interface {
	Has(cid.Cid) (bool, error)
	Get(cid.Cid) (blocks.Block, error)
	GetSize(cid.Cid) (int, error)
	AllKeysChan(ctx context.Context) (<-chan cid.Cid, error)
	HashOnRead(enabled bool)
}

ReadBlockstore is a read-only view of Blockstores. This will be implemented by the CARv2 indexed blockstore.

type RecoverOpts

type RecoverOpts struct {
}

type RegisterOpts

type RegisterOpts struct {
	// ExistingTransient can be supplied when registering a shard to indicate
	// that there's already an existing local transient copy that can be used
	// for indexing.
	ExistingTransient string

	// LazyInitialization defers shard indexing to the first access instead of
	// performing it at registration time. Use this option when fetching the
	// asset is expensive.
	//
	// When true, the registration channel will fire as soon as the DAG store
	// has acknowledged the inclusion of the shard, without waiting for any
	// indexing to happen.
	LazyInitialization bool
}

type Shard

type Shard struct {
	// contains filtered or unexported fields
}

Shard encapsulates the state of a shard within the DAG store.

func (*Shard) MarshalJSON

func (s *Shard) MarshalJSON() ([]byte, error)

MarshalJSON returns a serialized representation of the state. It must be called from inside the event loop, as it accesses mutable state, or under a shard read lock.

func (*Shard) UnmarshalJSON

func (s *Shard) UnmarshalJSON(b []byte) error

type ShardAccessor

type ShardAccessor struct {
	// contains filtered or unexported fields
}

ShardAccessor provides various means to access the data contained in a shard.

func NewShardAccessor

func NewShardAccessor(data mount.Reader, idx index.Index, s *Shard) (*ShardAccessor, error)

func (*ShardAccessor) Blockstore

func (sa *ShardAccessor) Blockstore() (ReadBlockstore, error)

func (*ShardAccessor) Close

func (sa *ShardAccessor) Close() error

Close terminates this shard accessor, releasing any resources associated with it, and decrementing internal refcounts.

func (*ShardAccessor) Shard

func (sa *ShardAccessor) Shard() shard.Key

type ShardInfo

type ShardInfo struct {
	ShardState
	Error error
	// contains filtered or unexported fields
}

type ShardResult

type ShardResult struct {
	Key      shard.Key
	Error    error
	Accessor *ShardAccessor
}

ShardResult encapsulates a result from an asynchronous operation.

type ShardState

type ShardState byte

const (
	// ShardStateNew indicates that a shard has just been registered and is
	// about to be processed for activation.
	ShardStateNew ShardState = iota

	// ShardStateInitializing indicates that the shard is being initialized
	// by being fetched from the mount and being indexed.
	ShardStateInitializing

	// ShardStateAvailable indicates that the shard has been initialized and is
	// active for serving queries. There are no active shard readers.
	ShardStateAvailable

	// ShardStateServing indicates the shard has active readers and thus is
	// currently actively serving requests.
	ShardStateServing

	// ShardStateRecovering indicates that the shard is recovering from an
	// errored state. Such recoveries are always initiated by the user through
	// DAGStore.RecoverShard().
	ShardStateRecovering = 0x80

	// ShardStateErrored indicates that an unexpected error was encountered
	// during a shard operation, and therefore the shard needs to be recovered.
	ShardStateErrored ShardState = 0xf0

	// ShardStateUnknown indicates that it's not possible to determine the state
	// of the shard. This state is currently unused, but it's reserved.
	ShardStateUnknown ShardState = 0xff
)

func (ShardState) String

func (ss ShardState) String() string

type Throttler

type Throttler interface {
	// Do performs the supplied action under the guard of the throttler.
	//
	// The supplied context is obeyed when parking to claim a throttler spot, and is
	// passed to the action. Errors from the action are propagated to the caller,
	// as are context deadline errors.
	//
	// Do blocks until the action has executed.
	Do(context.Context, func(ctx context.Context) error) error
}

Throttler is a component to perform throttling of concurrent requests.

func NewThrottler

func NewThrottler(maxConcurrency int) Throttler

NewThrottler creates a new throttler that allows the specified concurrency at most.

type Trace

type Trace struct {
	Key   shard.Key
	Op    OpType
	After ShardInfo
}