registry

Documentation

Index

• Constants
• Variables
• func IsActorDoesNotExistErr(err error) bool
• func IsNoopModuleStore(m ModuleStore) bool
• func TestAllCommon(t *testing.T, registryCtor func() Registry)
• type Address
• type CreateActorResult
• type EnsureActivationRequest
• type EnsureActivationResult
  • func NewEnsureActivationResult(references []types.ActorReference, versionStamp int64, registryServerID string) EnsureActivationResult
• type HeartbeatResult
• type HeartbeatState
• type KVRegistryOptions
• type ModuleOptions
• type ModuleStore
  • func NewNoopModuleStore() ModuleStore
• type NoOpTransaction
  • func (tr NoOpTransaction) Cancel(ctx context.Context) error
  • func (tr NoOpTransaction) Commit(ctx context.Context) error
  • func (tr NoOpTransaction) Get(ctx context.Context, key []byte) ([]byte, bool, error)
  • func (tr NoOpTransaction) Put(ctx context.Context, key []byte, value []byte) error
• type RegisterModuleResult
• type Registry
  • func NewKVRegistry(serverID string, kv kv.Store, opts KVRegistryOptions) Registry
  • func NewValidatedRegistry(r Registry) Registry

Constants

const (
	// HeartbeatTTL is the maximum amount of time between server heartbeats before
	// the registry will consider a server as dead.
	//
	// TODO: Should be configurable.
	HeartbeatTTL = 5 * time.Second

	// 2GiB, see KVRegistryOptions.RebalanceMemoryThreshold for more details.
	DefaultRebalanceMemoryThreshold = 1 << 31
)

Variables

var ErrWorkerUnimplemented = errors.New("not implemented, tried to use transaction from worker")

Functions

func IsActorDoesNotExistErr

func IsActorDoesNotExistErr(err error) bool

IsActorDoesNotExistErr returns a boolean indicating whether the error is an instance of (or wraps) errActorDoesNotExist.

func IsNoopModuleStore

func IsNoopModuleStore(m ModuleStore) bool

IsNoopModuleStore returns a boolean indicating whether the provided ModuelStore is backed by noopModuleStore. This is only used for producing better error messages.

func TestAllCommon

func TestAllCommon(t *testing.T, registryCtor func() Registry)

This is called from the specific registry implementation subpackages like fdbregistry, localregistry, dnsregistry, etc.

Types

type Address

type Address struct {
	IP   net.IP
	Port int
}

Address is a tuple of net.IP and port so that the implementation can be used without assuming every server is running on the same port.

type CreateActorResult

type CreateActorResult struct{}

CreateActorResult is the result of a call to CreateActor().

type EnsureActivationRequest

type EnsureActivationRequest struct {
	Namespace string `json:"namespace"`
	ModuleID  string `json:"module_id"`
	ActorID   string `json:"actor_id"`

	// ExtraReplicas represents the number of additional replicas requested for an actor.
	// It specifies the desired number of replicas, in addition to the primary replica,
	// that should be created during actor activation.
	// The value of ExtraReplicas should be a non-negative integer.
	ExtraReplicas uint64 `json:"extra_replicas"`
	// BlacklistedServerIDs is set if the caller is calling the EnsureActivation method
	// after receiving an error from the server the actor is *supposed* to be activated
	// on that the server has blacklisted the actor. The server may blacklist the actor
	// temporarily due to excessive resource consumption and/or to accomplish balancing
	// requests initiated by the registry. In those scenarios, the caller will provide
	// the ID of the server that the actor was blacklisted on so the registry can keep
	// track of that information and ensure the actor is activated elsewhere / balanced
	// properly.
	BlacklistedServerIDs      []string `json:"blacklisted_server_ids"`
	CachedActivationServerIDs []string `json:"cached_activation_server_ids"`
}

EnsureActiationRequest contains the arguments for the EnsureActivation method.

type EnsureActivationResult

type EnsureActivationResult struct {
	References       []types.ActorReference `json:"references"`
	VersionStamp     int64                  `json:"versionstamp"`
	RegistryServerID string                 `json:"registry_server_id"`
}

EnsureActivationResult contains the result of invoking the EnsureActivation method.

func NewEnsureActivationResult

func NewEnsureActivationResult(
	references []types.ActorReference,
	versionStamp int64,
	registryServerID string,
) EnsureActivationResult

NewEnsureActivationResult creates a new EnsureActivationResult.

type HeartbeatResult

type HeartbeatResult struct {
	// VersionStamp associated with the successful heartbeat.
	VersionStamp int64 `json:"version_stamp"`
	// TTL of the successful heartbeat in the same unit as the
	// VerisionStamp.
	HeartbeatTTL int64 `json:"heartbeat_ttl"`
	// ServerVersion is incremented every time a server's heartbeat expires and resumes,
	// guaranteeing the server's ability to identify periods of inactivity/death for correctness purposes.
	ServerVersion int64 `json:"server_version"`
	// MemoryBytesToShed is the number of bytes of memory usage that the registry recommends
	// that the server try to shed for balancing purposes. This value will only ever be > 0
	// when the registry things that rebalancing should occur by requesting that the current
	// server shed some of its load.
	MemoryBytesToShed int64
}

HeartbeatResult is the result returned by the Heartbeat() method.

type HeartbeatState

type HeartbeatState struct {
	// NumActivatedActors is the number of actors currently activated on the server.
	NumActivatedActors int `json:"num_activated_actors"`
	// UsedMemory is the amount of memory currently being used by actors on the server.
	UsedMemory int `json:"used_memory"`
	// Address is the address at which the server can be reached.
	Address string `json:"address"`
}

HeartbeatState contains information that accompanies a server's heartbeat. It contains various information about the current state of the server that might be useful to the registry. For example, the number of currently activated actors on the server is useful to the registry so it can load-balance future actor activations around the cluster to achieve uniformity.

TODO: This should include things like how many CPU seconds and memory the actors are using, etc for hotspot detection.

type KVRegistryOptions

type KVRegistryOptions struct {
	// DisableHighConflictOperations disables operations that
	// would lead to high conflict rates when using KV stores
	// like FoundationDB. In general enabling this feature will
	// not break correctness, but it may degrade the efficiency
	// of features like balancing actors across servers.
	DisableHighConflictOperations bool

	// RebalanceMemoryThreshold is the minimum delta between the memory usage of the
	// minimum and maximum servers before the registry will begin making balancing
	// decisions based on memory usage.
	RebalanceMemoryThreshold int

	// DisableMemoryRebalancing will disable rebalancing actors based on memory
	// usage if set.
	DisableMemoryRebalancing bool

	// MinSuccessiveHeartbeatsBeforeAllowActivations is the minimum number of
	// successive heartbeats the registry must receive from any serverID before
	// it will allow EnsureActivation() calls to succeed for any actor. This is
	// used to prevent a newly instantiated registry from making actor placement
	// decisions before every server has had the opportunity to heartbeat at least
	// once. Without this setting, newly intantiated registries may temporarily
	// assign all new actor activations to a small number of servers in the brief
	// window before its received a heartbeat (and thus is aware of the existence)
	// from all the servers in the cluster.
	MinSuccessiveHeartbeatsBeforeAllowActivations int

	// Logger is a logging instance used for logging messages.
	// If no logger is provided, the default logger from the slog
	// package (slog.Default()) will be used.
	Logger *slog.Logger
}

KVRegistryOptions contains the options for the KVRegistry.

type ModuleOptions

type ModuleOptions struct {
}

ModuleOptions contains the options for a given module.

type ModuleStore

type ModuleStore interface {
	// RegisterModule registers the provided module []byte and options with the
	// provided module ID for subsequent calls to CreateActor().
	RegisterModule(
		ctx context.Context,
		namespace,
		moduleID string,
		moduleBytes []byte,
		opts ModuleOptions,
	) (RegisterModuleResult, error)

	// GetModule gets the bytes and options associated with the provided module.
	GetModule(
		ctx context.Context,
		namespace,
		moduleID string,
	) ([]byte, ModuleOptions, error)
}

ModuleStore is the interface that must be implemented by the module store so that the virtual environment can store/retrieve new modules.

func NewNoopModuleStore

func NewNoopModuleStore() ModuleStore

NewNoopModuleStore returns a new ModuleStore that returns an error for every method call.

type NoOpTransaction

type NoOpTransaction struct{}

func (NoOpTransaction) Cancel

func (tr NoOpTransaction) Cancel(ctx context.Context) error

func (NoOpTransaction) Commit

func (tr NoOpTransaction) Commit(ctx context.Context) error

func (NoOpTransaction) Get

func (tr NoOpTransaction) Get(ctx context.Context, key []byte) ([]byte, bool, error)

func (NoOpTransaction) Put

func (tr NoOpTransaction) Put(ctx context.Context, key []byte, value []byte) error

type RegisterModuleResult

type RegisterModuleResult struct{}

RegisterModuleResult is the result of a call to RegisterModule().

type Registry

type Registry interface {
	// Heartbeat updates the "lastHeartbeatedAt" value for the provided server ID. Server's
	// must heartbeat regularly to be considered alive and eligible for hosting actor
	// activations.
	Heartbeat(
		ctx context.Context,
		serverID string,
		state HeartbeatState,
	) (HeartbeatResult, error)

	// EnsureActivation checks the registry to see if the provided actor is already
	// activated, and if so it returns an ActorReference that points to its activated
	// location. Otherwise, the registry will pick a location to activate the actor at
	// and then return an ActorReference that points to the newly selected location.
	//
	// Note that when this method returns it is guaranteed that a location will have
	// been selected for the actor to be activated at, but the actor may not necessarily
	// have been activated. In general, actor activation is handled "lazily" when a
	// location (server) receives its first invocation for an actor ID that it doesn't
	// currently have activated.
	EnsureActivation(
		ctx context.Context,
		req EnsureActivationRequest,
	) (EnsureActivationResult, error)

	// GetVersionStamp() returns a monotonically increasing integer that should increase
	// at a rate of ~ 1 million/s.
	GetVersionStamp(ctx context.Context) (int64, error)

	// Close closes the registry and releases any resources associated (DB connections, etc).
	Close(ctx context.Context) error

	// UnsafeWipeAll wipes the entire registry. Only used for tests. Do not call it anywhere
	// in production code.
	UnsafeWipeAll() error
}

Registry is the interface that is implemented by the virtual actor registry.

func NewKVRegistry

func NewKVRegistry(
	serverID string,
	kv kv.Store,
	opts KVRegistryOptions,
) Registry

NewKVRegistry creates a new KV-backed registry.

func NewValidatedRegistry

func NewValidatedRegistry(r Registry) Registry

NewValidatedRegistry wraps the provided Registry r such that it validates inputs before delegating calls. This makes it easier to write new registry implementations without making all of them re-implement the validation logic.