cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

lpwindow: Make tasks go into the database

Browse Source
This commit is contained in:
Łukasz Magiera 2023-10-30 19:40:06 +01:00
parent ebca1d0142
commit 080128c6b6
2 changed files with 17 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
provider/builder.go
View File

@ -30,5 +30,12 @@ func WindowPostScheduler(ctx context.Context, fc config.LotusProviderFees, pc co
// todo config // todo config
ft := lpwindow.NewSimpleFaultTracker(stor, idx, 32, 5*time.Second, 300*time.Second) ft := lpwindow.NewSimpleFaultTracker(stor, idx, 32, 5*time.Second, 300*time.Second)
return lpwindow.NewWdPostTask(db, api, ft, lw, verif, chainSched, maddr, max) task, err := lpwindow.NewWdPostTask(db, api, ft, lw, verif, chainSched, maddr, max)
if err != nil {
return nil, err
}
go chainSched.Run(ctx)
return task, nil
} }

13
provider/lpwindow/task.go
View File

@ -82,6 +82,8 @@ type wdTaskIdentity struct {
func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) { func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done bool, err error) {
log.Errorw("WDPOST DO", "taskID", taskID)
time.Sleep(5 * time.Second) time.Sleep(5 * time.Second)
log.Errorf("WdPostTask.Do() called with taskID: %v", taskID) log.Errorf("WdPostTask.Do() called with taskID: %v", taskID)
@ -89,7 +91,7 @@ func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
err = t.db.QueryRow(context.Background(), err = t.db.QueryRow(context.Background(),
`Select sp_id, proving_period_start, deadline_index, partition_index `Select sp_id, proving_period_start, deadline_index, partition_index
from wdpost_tasks from wdpost_partition_tasks
where task_id = $1`, taskID).Scan( where task_id = $1`, taskID).Scan(
&spID, &pps, &dlIdx, &partIdx, &spID, &pps, &dlIdx, &partIdx,
) )
@ -174,6 +176,9 @@ func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
} }
func (t *WdPostTask) CanAccept(ids []harmonytask.TaskID, te *harmonytask.TaskEngine) (*harmonytask.TaskID, error) { func (t *WdPostTask) CanAccept(ids []harmonytask.TaskID, te *harmonytask.TaskEngine) (*harmonytask.TaskID, error) {
log.Errorw("WDPOST CANACCEPT", "ids", ids)
// GetEpoch // GetEpoch
ts, err := t.api.ChainHead(context.Background()) ts, err := t.api.ChainHead(context.Background())
@ -201,7 +206,7 @@ func (t *WdPostTask) CanAccept(ids []harmonytask.TaskID, te *harmonytask.TaskEng
proving_period_start, proving_period_start,
deadline_index, deadline_index,
partition_index partition_index
from wdpost_tasks from wdpost_partition_tasks
where task_id IN $1`, ids) where task_id IN $1`, ids)
if err != nil { if err != nil {
return nil, err return nil, err
@ -374,13 +379,13 @@ func NewWdPostTask(db *harmonydb.DB,
func (t *WdPostTask) addTaskToDB(taskId harmonytask.TaskID, taskIdent wdTaskIdentity, tx *harmonydb.Tx) (bool, error) { func (t *WdPostTask) addTaskToDB(taskId harmonytask.TaskID, taskIdent wdTaskIdentity, tx *harmonydb.Tx) (bool, error) {
_, err := tx.Exec( _, err := tx.Exec(
`INSERT INTO wdpost_tasks ( `INSERT INTO wdpost_partition_tasks (
task_id, task_id,
sp_id, sp_id,
proving_period_start, proving_period_start,
deadline_index, deadline_index,
partition_index partition_index
) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10 , $11, $12, $13, $14)`, ) VALUES ($1, $2, $3, $4, $5)`,
taskId, taskId,
taskIdent.Sp_id, taskIdent.Sp_id,
taskIdent.Proving_period_start, taskIdent.Proving_period_start,