[all] rename & refactor (#419)

huangzhiran · web-flow · commit 4a67fd5d018f · 2024-04-11T17:35:12.000+08:00
diff --git a/cmd/coordinator/main.go b/cmd/coordinator/main.go
@@ -30,13 +30,13 @@ func main() {
 		log.Fatal(errors.Wrap(err, "failed to new postgres persistence"))
 	}
 
-	projectConfigManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,
+	projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,
 		conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)
 	if err != nil {
 		log.Fatal(errors.Wrap(err, "failed to new project config manager"))
 	}
 
-	if err := task.RunDispatcher(persistence, datasource.NewPostgres, projectConfigManager.GetAllCacheProjectIDs, projectConfigManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519, conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {
+	if err := task.RunDispatcher(persistence, datasource.NewPostgres, projectManager.GetCachedProjectIDs, projectManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519, conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {
 		log.Fatal(errors.Wrap(err, "failed to run dispatcher"))
 	}
 
diff --git a/cmd/prover/main.go b/cmd/prover/main.go
@@ -40,7 +40,7 @@ func main() {
 		},
 	)
 
-	projectConfigManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,
+	projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,
 		conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)
 	if err != nil {
 		log.Fatal(err)
@@ -59,7 +59,7 @@ func main() {
 		log.Fatal(errors.Wrap(err, "failed to decode sequencer pubkey"))
 	}
 
-	taskProcessor := task.NewProcessor(vmHandler, projectConfigManager, sk, sequencerPubKey, proverID)
+	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID)
 
 	pubSubs, err := p2p.NewPubSubs(taskProcessor.HandleP2PData, conf.BootNodeMultiAddr, conf.IoTeXChainID)
 	if err != nil {
@@ -68,7 +68,7 @@ func main() {
 
 	if err := scheduler.Run(conf.SchedulerEpoch, conf.ChainEndpoint, conf.ProverContractAddress,
 		conf.ProjectContractAddress, conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers,
-		projectConfigManager.GetAllCacheProjectIDs); err != nil {
+		projectManager.GetCachedProjectIDs); err != nil {
 		log.Fatal(err)
 	}
 
diff --git a/cmd/tests/init.go b/cmd/tests/init.go
@@ -115,7 +115,7 @@ func runProver(conf *proverconfig.Config) {
 		},
 	)
 
-	projectConfigManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)
+	projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -131,15 +131,15 @@ func runProver(conf *proverconfig.Config) {
 		log.Fatal(errors.Wrap(err, "failed to decode sequencer pubkey"))
 	}
 
-	taskProcessor := task.NewProcessor(vmHandler, projectConfigManager, sk, sequencerPubKey, proverID)
+	taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID)
 
 	pubSubs, err := p2p.NewPubSubs(taskProcessor.HandleP2PData, conf.BootNodeMultiAddr, conf.IoTeXChainID)
 	if err != nil {
 		log.Fatal(err)
 	}
 
 	if err := scheduler.Run(conf.SchedulerEpoch, conf.ChainEndpoint, conf.ProverContractAddress, conf.ProjectContractAddress,
-		conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers, projectConfigManager.GetAllCacheProjectIDs); err != nil {
+		conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers, projectManager.GetCachedProjectIDs); err != nil {
 		log.Fatal(err)
 	}
 
@@ -159,7 +159,7 @@ func runCoordinator(conf *coordinatorconfig.Config) {
 		log.Fatal(err)
 	}
 
-	if err := task.RunDispatcher(pg, datasource.NewPostgres, projectConfigManager.GetAllCacheProjectIDs,
+	if err := task.RunDispatcher(pg, datasource.NewPostgres, projectConfigManager.GetCachedProjectIDs,
 		projectConfigManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519,
 		conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {
 		log.Fatal(errors.Wrap(err, "failed to run dispatcher"))
diff --git a/project/manager.go b/project/manager.go
@@ -23,7 +23,7 @@ type Manager struct {
 	cache        *cache   // optional
 }
 
-func (m *Manager) GetAllCacheProjectIDs() []uint64 {
+func (m *Manager) GetCachedProjectIDs() []uint64 {
 	var ids []uint64
 	m.projects.Range(func(key, value any) bool {
 		ids = append(ids, key.(uint64))
diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go
@@ -18,7 +18,7 @@ import (
 
 type HandleProjectProvers func(projectID uint64, provers []string)
 
-type GetCacheProjectIDs func() []uint64
+type GetCachedProjectIDs func() []uint64
 
 type scheduler struct {
 	provers              *sync.Map // proverID(string) -> Prover(*contract.Prover)
@@ -101,7 +101,7 @@ func watchChainHead(head chan<- uint64, chainEndpoint string) error {
 }
 
 func Run(epoch uint64, chainEndpoint, proverContractAddress, projectContractAddress, projectFileDirectory, proverID string,
-	pubSubs *p2p.PubSubs, handleProjectProvers HandleProjectProvers, getProjectIDs GetCacheProjectIDs) error {
+	pubSubs *p2p.PubSubs, handleProjectProvers HandleProjectProvers, getProjectIDs GetCachedProjectIDs) error {
 
 	if projectFileDirectory != "" {
 		dummySchedule(proverID, pubSubs, handleProjectProvers, getProjectIDs)
@@ -165,7 +165,7 @@ func Run(epoch uint64, chainEndpoint, proverContractAddress, projectContractAddr
 	return nil
 }
 
-func dummySchedule(proverID string, pubSubs *p2p.PubSubs, handleProjectProvers HandleProjectProvers, getProjectIDs GetCacheProjectIDs) {
+func dummySchedule(proverID string, pubSubs *p2p.PubSubs, handleProjectProvers HandleProjectProvers, getProjectIDs GetCachedProjectIDs) {
 	s := &scheduler{
 		pubSubs:              pubSubs,
 		handleProjectProvers: handleProjectProvers,
diff --git a/task/dispatcher.go b/task/dispatcher.go
@@ -28,9 +28,7 @@ type Datasource interface {
 	Retrieve(projectID, nextTaskID uint64) (*types.Task, error)
 }
 
-type ProjectManager interface {
-	Get(projectID uint64) (*project.Project, error)
-}
+type GetCachedProjectIDs func() []uint64
 
 type dispatcher struct {
 	projectDispatchers *sync.Map // projectID(uint64) -> *ProjectDispatcher
@@ -51,7 +49,7 @@ func (d *dispatcher) handleP2PData(data *p2p.Data, topic *pubsub.Topic) {
 }
 
 func RunDispatcher(persistence Persistence, newDatasource internaldispatcher.NewDatasource,
-	getProjectIDs handler.GetCacheProjectIDs, getProject handler.GetProject,
+	getProjectIDs GetCachedProjectIDs, getProject handler.GetProject,
 	bootNodeMultiaddr, operatorPrivateKey, operatorPrivateKeyED25519, chainEndpoint, projectContractAddress, projectFileDirectory string,
 	iotexChainID int) error {
 	projectDispatchers := &sync.Map{}
@@ -79,7 +77,7 @@ func RunDispatcher(persistence Persistence, newDatasource internaldispatcher.New
 }
 
 func dummyDispatch(persistence Persistence, newDatasource internaldispatcher.NewDatasource,
-	getProjectIDs handler.GetCacheProjectIDs, getProject handler.GetProject,
+	getProjectIDs GetCachedProjectIDs, getProject handler.GetProject,
 	projectDispatchers *sync.Map, ps *p2p.PubSubs, handler *handler.TaskStateHandler) error {
 	projectIDs := getProjectIDs()
 	for _, id := range projectIDs {
diff --git a/task/internal/handler/task_state_handler.go b/task/internal/handler/task_state_handler.go
@@ -13,8 +13,6 @@ type SaveTaskStateLog func(s *types.TaskStateLog, t *types.Task) error
 
 type GetProject func(projectID uint64) (*project.Project, error)
 
-type GetCacheProjectIDs func() []uint64
-
 type TaskStateHandler struct {
 	saveTaskStateLog          SaveTaskStateLog
 	getProject                GetProject
diff --git a/task/processor.go b/task/processor.go
@@ -16,6 +16,7 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/machinefi/sprout/p2p"
+	"github.com/machinefi/sprout/project"
 	"github.com/machinefi/sprout/types"
 	"github.com/machinefi/sprout/utils/distance"
 	"github.com/machinefi/sprout/vm"
@@ -25,9 +26,11 @@ type VMHandler interface {
 	Handle(task *types.Task, vmtype vm.Type, code string, expParam string) ([]byte, error)
 }
 
+type GetProject func(projectID uint64) (*project.Project, error)
+
 type Processor struct {
 	vmHandler        VMHandler
-	projectManager   ProjectManager
+	getProject       GetProject
 	proverPrivateKey *ecdsa.PrivateKey
 	sequencerPubKey  []byte
 	proverID         string
@@ -44,7 +47,7 @@ func (r *Processor) HandleP2PData(d *p2p.Data, topic *pubsub.Topic) {
 	}
 	t := d.Task
 
-	p, err := r.projectManager.Get(t.ProjectID)
+	p, err := r.getProject(t.ProjectID)
 	if err != nil {
 		slog.Error("failed to get project", "error", err, "project_id", t.ProjectID)
 		r.reportFail(t, err, topic)
@@ -161,10 +164,10 @@ func (r *Processor) reportSuccess(t *types.Task, state types.TaskState, result [
 	}
 }
 
-func NewProcessor(vmHandler VMHandler, projectManager ProjectManager, proverPrivateKey *ecdsa.PrivateKey, seqPubkey []byte, proverID string) *Processor {
+func NewProcessor(vmHandler VMHandler, getProject GetProject, proverPrivateKey *ecdsa.PrivateKey, seqPubkey []byte, proverID string) *Processor {
 	return &Processor{
 		vmHandler:        vmHandler,
-		projectManager:   projectManager,
+		getProject:       getProject,
 		proverPrivateKey: proverPrivateKey,
 		sequencerPubKey:  seqPubkey,
 		proverID:         proverID,
diff --git a/task/processor_test.go b/task/processor_test.go
@@ -59,9 +59,11 @@ func TestProcessor_ReportSuccess(t *testing.T) {
 
 func TestProcessor_HandleP2PData(t *testing.T) {
 	r := require.New(t)
+
+	m := &project.Manager{}
 	processor := &Processor{
-		vmHandler:      &vm.Handler{},
-		projectManager: &project.Manager{},
+		vmHandler:  &vm.Handler{},
+		getProject: m.Get,
 	}
 
 	t.Run("TaskNil", func(t *testing.T) {

Original file line number	Diff line number	Diff line change
`@@ -30,13 +30,13 @@ func main() {`
`30`	`30`	`log.Fatal(errors.Wrap(err, "failed to new postgres persistence"))`
`31`	`31`	`}`
`32`	`32`
`33`		`- projectConfigManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,`
	`33`	`+ projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress,`
`34`	`34`	`conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)`
`35`	`35`	`if err != nil {`
`36`	`36`	`log.Fatal(errors.Wrap(err, "failed to new project config manager"))`
`37`	`37`	`}`
`38`	`38`
`39`		`- if err := task.RunDispatcher(persistence, datasource.NewPostgres, projectConfigManager.GetAllCacheProjectIDs, projectConfigManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519, conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {`
	`39`	`+ if err := task.RunDispatcher(persistence, datasource.NewPostgres, projectManager.GetCachedProjectIDs, projectManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519, conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {`
`40`	`40`	`log.Fatal(errors.Wrap(err, "failed to run dispatcher"))`
`41`	`41`	`}`
`42`	`42`
Original file line number	Diff line number	Diff line change
`@@ -115,7 +115,7 @@ func runProver(conf *proverconfig.Config) {`
`115`	`115`	`},`
`116`	`116`	`)`
`117`	`117`
`118`		`- projectConfigManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)`
	`118`	`+ projectManager, err := project.NewManager(conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectCacheDirectory, conf.IPFSEndpoint, conf.ProjectFileDirectory)`
`119`	`119`	`if err != nil {`
`120`	`120`	`log.Fatal(err)`
`121`	`121`	`}`
`@@ -131,15 +131,15 @@ func runProver(conf *proverconfig.Config) {`
`131`	`131`	`log.Fatal(errors.Wrap(err, "failed to decode sequencer pubkey"))`
`132`	`132`	`}`
`133`	`133`
`134`		`- taskProcessor := task.NewProcessor(vmHandler, projectConfigManager, sk, sequencerPubKey, proverID)`
	`134`	`+ taskProcessor := task.NewProcessor(vmHandler, projectManager.Get, sk, sequencerPubKey, proverID)`
`135`	`135`
`136`	`136`	`pubSubs, err := p2p.NewPubSubs(taskProcessor.HandleP2PData, conf.BootNodeMultiAddr, conf.IoTeXChainID)`
`137`	`137`	`if err != nil {`
`138`	`138`	`log.Fatal(err)`
`139`	`139`	`}`
`140`	`140`
`141`	`141`	`if err := scheduler.Run(conf.SchedulerEpoch, conf.ChainEndpoint, conf.ProverContractAddress, conf.ProjectContractAddress,`
`142`		`- conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers, projectConfigManager.GetAllCacheProjectIDs); err != nil {`
	`142`	`+ conf.ProjectFileDirectory, proverID, pubSubs, taskProcessor.HandleProjectProvers, projectManager.GetCachedProjectIDs); err != nil {`
`143`	`143`	`log.Fatal(err)`
`144`	`144`	`}`
`145`	`145`
`@@ -159,7 +159,7 @@ func runCoordinator(conf *coordinatorconfig.Config) {`
`159`	`159`	`log.Fatal(err)`
`160`	`160`	`}`
`161`	`161`
`162`		`- if err := task.RunDispatcher(pg, datasource.NewPostgres, projectConfigManager.GetAllCacheProjectIDs,`
	`162`	`+ if err := task.RunDispatcher(pg, datasource.NewPostgres, projectConfigManager.GetCachedProjectIDs,`
`163`	`163`	`projectConfigManager.Get, conf.BootNodeMultiAddr, conf.OperatorPrivateKey, conf.OperatorPrivateKeyED25519,`
`164`	`164`	`conf.ChainEndpoint, conf.ProjectContractAddress, conf.ProjectFileDirectory, conf.IoTeXChainID); err != nil {`
`165`	`165`	`log.Fatal(errors.Wrap(err, "failed to run dispatcher"))`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@ type Manager struct {`
`23`	`23`	`cache *cache // optional`
`24`	`24`	`}`
`25`	`25`
`26`		`-func (m *Manager) GetAllCacheProjectIDs() []uint64 {`
	`26`	`+func (m *Manager) GetCachedProjectIDs() []uint64 {`
`27`	`27`	`var ids []uint64`
`28`	`28`	`m.projects.Range(func(key, value any) bool {`
`29`	`29`	`ids = append(ids, key.(uint64))`