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

ensure test messages avoid the chain

Browse Source
This commit is contained in:
Andrew Jackson (Ajax) 2023-11-21 22:43:13 -06:00
parent deae3ad05f
commit 9f9af40d3b
5 changed files with 31 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cmd/lotus-provider/proving.go
View File

@ -37,6 +37,11 @@ var wdPostCmd = &cli.Command{
},
}
// wdPostTaskCmd writes to harmony_task and wdpost_partition_tasks, then waits for the result.
// It is intended to be used to test the windowpost scheduler.
// The end of the compute task puts the task_id onto wdpost_proofs, which is read by the submit task.
// The submit task will not send test tasks to the chain, and instead will write the result to harmony_test.
// The result is read by this command, and printed to stdout.
var wdPostTaskCmd = &cli.Command{
Name: "task",
Aliases: []string{"scheduled", "schedule", "async", "asynchronous"},
@ -113,6 +118,9 @@ var wdPostTaskCmd = &cli.Command{
},
}
// This command is intended to be used to verify PoSt compute performance.
// It will not send any messages to the chain. Since it can compute any deadline, output may be incorrectly timed for the chain.
// The entire processing happens in this process while you wait. It does not use the scheduler.
var wdPostHereCmd = &cli.Command{
Name: "here",
Aliases: []string{"cli"},

3
lib/harmony/harmonydb/sql/20231120-testing1.sql
View File

@ -4,4 +4,5 @@ CREATE TABLE harmony_test (
primary key,
options text,
result text
);
);
ALTER TABLE wdpost_proofs ADD COLUMN test_task_id bigint;

6
provider/lpmessage/sender.go
View File

@ -21,7 +21,7 @@ var log = logging.Logger("lpmessage")
type str string // makes ctx value collissions impossible
var CtxTaskID str = "task_id"
var CtxTestTaskID str = "task_id"
type SenderAPI interface {
StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error)
@ -106,7 +106,7 @@ func (s *Sender) Send(ctx context.Context, msg *types.Message, mss *api.MessageS
var idCount int
err = s.db.QueryRow(ctx, `SELECT COUNT(*) FROM harmony_test WHERE task_id=$1`,
ctx.Value(CtxTaskID)).Scan(&idCount)
ctx.Value(CtxTestTaskID)).Scan(&idCount)
if err != nil {
return cid.Undef, xerrors.Errorf("reading harmony_test: %w", err)
}
@ -163,7 +163,7 @@ func (s *Sender) Send(ctx context.Context, msg *types.Message, mss *api.MessageS
if err != nil {
return false, xerrors.Errorf("marshaling message: %w", err)
}
id := ctx.Value(CtxTaskID)
id := ctx.Value(CtxTestTaskID)
tx.Exec(`UPDATE harmony_test SET result=$1 WHERE task_id=$2`, string(data), id)
log.Infof("SKIPPED sending test message to chain. Query harmony_test WHERE task_id= %v", id)
return true, nil // nothing committed

17
provider/lpwindow/compute_task.go
View File

@ -161,6 +161,14 @@ func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
return false, xerrors.Errorf("marshaling PoSt: %w", err)
}
testTaskID := 0
testTaskIDCt := 0
if err = t.db.QueryRow(context.Background(), `SELECT COUNT(*) FROM harmony_test WHERE task_id = $1`, taskID).Scan(&testTaskIDCt); err != nil {
return false, xerrors.Errorf("querying for test task: %w", err)
}
if testTaskIDCt == 1 {
testTaskID = int(taskID)
}
// Insert into wdpost_proofs table
n, err := t.db.Exec(context.Background(),
`INSERT INTO wdpost_proofs (
@ -170,15 +178,18 @@ func (t *WdPostTask) Do(taskID harmonytask.TaskID, stillOwned func() bool) (done
partition,
submit_at_epoch,
submit_by_epoch,
proof_params)
VALUES ($1, $2, $3, $4, $5, $6, $7)`,
proof_params,
test_task_id)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
spID,
pps,
deadline.Index,
partIdx,
deadline.Open,
deadline.Close,
msgbuf.Bytes())
msgbuf.Bytes(),
testTaskID,
)
if err != nil {
log.Errorf("WdPostTask.Do() failed to insert into wdpost_proofs: %v", err)

7
provider/lpwindow/submit_task.go
View File

@ -78,11 +78,12 @@ func (w *WdPostSubmitTask) Do(taskID harmonytask.TaskID, stillOwned func() bool)
var pps, submitAtEpoch, submitByEpoch abi.ChainEpoch
var earlyParamBytes []byte
var dbTask uint64
var testTaskID uint64
err = w.db.QueryRow(
context.Background(), `SELECT sp_id, proving_period_start, deadline, partition, submit_at_epoch, submit_by_epoch, proof_params, submit_task_id
context.Background(), `SELECT sp_id, proving_period_start, deadline, partition, submit_at_epoch, submit_by_epoch, proof_params, submit_task_id, test_task_id
FROM wdpost_proofs WHERE submit_task_id = $1`, taskID,
).Scan(&spID, &pps, &deadline, &partition, &submitAtEpoch, &submitByEpoch, &earlyParamBytes, &dbTask)
).Scan(&spID, &pps, &deadline, &partition, &submitAtEpoch, &submitByEpoch, &earlyParamBytes, &dbTask, &testTaskID)
if err != nil {
return false, xerrors.Errorf("query post proof: %w", err)
}
@ -149,7 +150,7 @@ func (w *WdPostSubmitTask) Do(taskID harmonytask.TaskID, stillOwned func() bool)
return false, xerrors.Errorf("preparing proof message: %w", err)
}
ctx := context.WithValue(context.Background(), lpmessage.CtxTaskID, taskID)
ctx := context.WithValue(context.Background(), lpmessage.CtxTestTaskID, testTaskID)
smsg, err := w.sender.Send(ctx, msg, mss, "wdpost")
if err != nil {
return false, xerrors.Errorf("sending proof message: %w", err)