executor

package
v0.2.3-alpha.4 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 17, 2023 License: GPL-3.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// DefaultExecutorMaxExecuteNum defines the default max parallel execute task number.
	DefaultExecutorMaxExecuteNum int64 = 16
	// DefaultExecutorAskTaskInterval defines the default ask task interval from manager.
	DefaultExecutorAskTaskInterval int = 1
	// DefaultExecutorAskReplicateApprovalTimeout defines the ask replicate piece approval
	// timeout that send the request to the p2p node,
	DefaultExecutorAskReplicateApprovalTimeout int64 = 4
	// DefaultExecutorAskReplicateApprovalExFactor defines the expanded factor for asking
	// secondary SP.
	// Example: need data chunk + data parity chunk numbers SPs as secondary, consider fault
	// 	tolerance, should collect (data chunk + data parity chunk) * factor numbers SPs as
	//	backup secondary, if some of these are failed to replicate can pick up again from
	//  backups. So it is always bigger than 1.0.
	DefaultExecutorAskReplicateApprovalExFactor float64 = 1.0
	// DefaultExecutorListenSealTimeoutHeight defines the default listen seal object on
	// greenfield timeout height, if after current block height + timeout height, the object
	// is not sealed, it is judged failed to seal object on greenfield.
	DefaultExecutorListenSealTimeoutHeight int = 10
	// DefaultExecutorListenSealRetryTimeout defines the sleep time when listen seal object
	// fail, until retry ExecutorMaxListenSealRetry times, the task is set error.
	DefaultExecutorListenSealRetryTimeout int = 2
	// DefaultExecutorMaxListenSealRetry defines the default max retry number for listening
	// object.
	DefaultExecutorMaxListenSealRetry int = 3
	// DefaultStatisticsOutputInterval defines the default interval for output statistics info,
	// it is used to log and debug.
	DefaultStatisticsOutputInterval int = 60
	// DefaultSleepInterval defines the sleep interval when failed to ask task
	// it is millisecond level
	DefaultSleepInterval = 100
)
View Source 
const (
	ExeutorSuccessAskTask   = "executor_ask_task_success"
	ExeutorRunTask          = "executor_run_task"
	ExeutorFailureAskTask   = "executor_ask_task_failure"
	ExeutorFailureAskNoTask = "executor_ask_no_task_failure"

	ExeutorSuccessReplicateTask  = "executor_replicate_task_success"
	ExeutorFailureReplicateTask  = "executor_replicate_task_failure"
	ExeutorSuccessSealObjectTask = "executor_seal_object_task_success"
	ExeutorFailureSealObjectTask = "executor_seal_object_task_failure"
	ExeutorSuccessReceiveTask    = "executor_receive_task_success"
	ExeutorFailureReceiveTask    = "executor_receive_task_failure"
	ExeutorSuccessRecoveryTask   = "executor_recovery_task_success"
	ExeutorFailureRecoveryTask   = "executor_recovery_task_failure"

	ExeutorSuccessReportTask = "executor_report_task_to_manager_success"
	ExeutorFailureReportTask = "executor_report_task_to_manager_failure"

	ExeutorSuccessP2P                = "executor_p2p_success"
	ExeutorFailureP2P                = "executor_p2p_failure"
	ExeutorSuccessReplicateAllPiece  = "executor_replicate_all_piece_success"
	ExeutorFailureReplicateAllPiece  = "executor_replicate_all_piece_failure"
	ExeutorSuccessReplicateOnePiece  = "executor_replicate_one_piece_success"
	ExeutorFailureReplicateOnePiece  = "executor_replicate_one_piece_failure"
	ExeutorSuccessDoneReplicatePiece = "executor_done_replicate_piece_success"
	ExeutorFailureDoneReplicatePiece = "executor_done_replicate_piece_failure"
	ExeutorSuccessSealObject         = "executor_seal_object_success"
	ExeutorFailureSealObject         = "executor_seal_object_failure"
)

Variables

View Source 
var (
	ErrDanglingPointer         = gfsperrors.Register(module.ExecuteModularName, http.StatusBadRequest, 40001, "OoooH.... request lost")
	ErrInsufficientApproval    = gfsperrors.Register(module.ExecuteModularName, http.StatusNotFound, 40002, "insufficient approvals from p2p")
	ErrUnsealed                = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 40003, "seal object on chain failed")
	ErrExhaustedApproval       = gfsperrors.Register(module.ExecuteModularName, http.StatusNotFound, 40004, "approvals exhausted")
	ErrInvalidIntegrity        = gfsperrors.Register(module.ExecuteModularName, http.StatusNotAcceptable, 40005, "secondary integrity hash verification failed")
	ErrSecondaryMismatch       = gfsperrors.Register(module.ExecuteModularName, http.StatusNotAcceptable, 40006, "secondary sp mismatch")
	ErrReplicateIdsOutOfBounds = gfsperrors.Register(module.ExecuteModularName, http.StatusNotAcceptable, 40007, "replicate idx out of bounds")
	ErrGfSpDB                  = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45201, "server slipped away, try again later")
	ErrRecoveryRedundancyType  = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45202, "recovery only support EC redundancy type")
	ErrRecoveryPieceNotEnough  = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45203, "fail to get enough piece data to recovery")
	ErrRecoveryDecode          = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45204, "EC decode error")
	ErrPieceStore              = gfsperrors.Register(module.ReceiveModularName, http.StatusInternalServerError, 45205, "server slipped away, try again later")
	ErrRecoveryPieceChecksum   = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45206, "recovery checksum not correct")
	ErrRecoveryPieceLength     = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45207, "get secondary piece data length error")
	ErrPrimaryNotFound         = gfsperrors.Register(module.ExecuteModularName, http.StatusNotAcceptable, 45208, "primary sp endpoint not found when recovering")
	ErrRecoveryPieceIndex      = gfsperrors.Register(module.ExecuteModularName, http.StatusNotAcceptable, 45209, "recovery piece index invalid")
	ErrMigratedPieceChecksum   = gfsperrors.Register(module.ExecuteModularName, http.StatusInternalServerError, 45210, "migrate piece checksum is not correct")
	ErrConsensus               = gfsperrors.Register(module.ExecuteModularName, http.StatusBadRequest, 45211, "server slipped away, try again later")
)

Functions

func DefaultExecutorOptions 

func DefaultExecutorOptions(executor *ExecuteModular, cfg *gfspconfig.GfSpConfig) error

func NewExecuteModular 

func NewExecuteModular(app *gfspapp.GfSpBaseApp, cfg *gfspconfig.GfSpConfig) (coremodule.Modular, error)

Types

type ExecuteModular 

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

func (*ExecuteModular) AskTask 

func (e *ExecuteModular) AskTask(ctx context.Context) error

func (*ExecuteModular) HandleGCMetaTask 

func (e *ExecuteModular) HandleGCMetaTask(ctx context.Context, task coretask.GCMetaTask)

func (*ExecuteModular) HandleGCObjectTask 

func (e *ExecuteModular) HandleGCObjectTask(ctx context.Context, task coretask.GCObjectTask)

func (*ExecuteModular) HandleGCZombiePieceTask 

func (e *ExecuteModular) HandleGCZombiePieceTask(ctx context.Context, task coretask.GCZombiePieceTask)

func (*ExecuteModular) HandleMigrateGVGTask added in v0.2.3 

func (e *ExecuteModular) HandleMigrateGVGTask(ctx context.Context, task coretask.MigrateGVGTask) error

HandleMigrateGVGTask handles the migrate gvg task. There are two cases: sp exit and bucket migration srcSP is a sp who wants to exit or need to migrate bucket, destSP is used to accept data from srcSP

func (*ExecuteModular) HandleMigratePieceTask added in v0.2.3 

func (e *ExecuteModular) HandleMigratePieceTask(ctx context.Context, task *gfsptask.GfSpMigratePieceTask) error

HandleMigratePieceTask handles the migrate piece task, it will send requests to the exiting SP to get piece data integrity hash and piece checksum to migrate the receiving SP. PieceData would be written to PieceStore, integrity hash and piece checksum will be written sql db. currently get and handle data one by one; in the future, use concurrency storagetypes.ObjectInfo struct contains LocalVirtualGroupId field which we can use it to get a GVG consisting of one PrimarySP and six ordered secondarySP(the order cannot be changed). Therefore, we can know what kinds of object we want to migrate: primary or secondary. Now we cannot use objectInfo operator address or secondaryAddress straightly. We should encapsulate a new method to get. objectInfo->lvg->gvg->(1 primarySP, 6 secondarySPs)

func (*ExecuteModular) HandleReceivePieceTask 

func (e *ExecuteModular) HandleReceivePieceTask(ctx context.Context, task coretask.ReceivePieceTask)

func (*ExecuteModular) HandleRecoverPieceTask added in v0.2.3 

func (e *ExecuteModular) HandleRecoverPieceTask(ctx context.Context, task coretask.RecoveryPieceTask)

HandleRecoverPieceTask handle the recovery piece task, it will send request to other SPs to get piece data to recovery, recovery the original data, and write the recovered data to piece store

func (*ExecuteModular) HandleReplicatePieceTask 

func (e *ExecuteModular) HandleReplicatePieceTask(ctx context.Context, task coretask.ReplicatePieceTask)

func (*ExecuteModular) HandleSealObjectTask 

func (e *ExecuteModular) HandleSealObjectTask(ctx context.Context, task coretask.SealObjectTask)

func (*ExecuteModular) Name 

func (e *ExecuteModular) Name() string

func (*ExecuteModular) ReleaseResource 

func (e *ExecuteModular) ReleaseResource(
	ctx context.Context,
	span corercmgr.ResourceScopeSpan)

func (*ExecuteModular) ReportTask 

func (e *ExecuteModular) ReportTask(
	ctx context.Context,
	task coretask.Task) (err error)

func (*ExecuteModular) ReserveResource 

func (e *ExecuteModular) ReserveResource(
	ctx context.Context,
	st *corercmgr.ScopeStat) (
	corercmgr.ResourceScopeSpan, error)

func (*ExecuteModular) Start 

func (e *ExecuteModular) Start(ctx context.Context) error

func (*ExecuteModular) Statistics 

func (e *ExecuteModular) Statistics() string

func (*ExecuteModular) Stop 

func (e *ExecuteModular) Stop(ctx context.Context) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.