Mainlining of Fast Snap logic

.github/workflows/ci.yml

@@ -259,7 +259,7 @@ jobs:
 
       - name: Install golangci-lint
         run: |
-          curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v2.4.0
+          curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(go env GOPATH)/bin v2.6.2
         shell: bash
 
       - name: Lint

api/api_chain_all.go

@@ -1,5 +1,4 @@
 //go:build !forest
-// +build !forest
 
 package api
 

api/api_chain_limited.go

@@ -1,5 +1,4 @@
 //go:build forest
-// +build forest
 
 package api
 

build/params_2k.go

@@ -1,5 +1,4 @@
 //go:build 2k
-// +build 2k
 
 package build
 

build/params_calibnet.go

@@ -1,5 +1,4 @@
 //go:build calibnet
-// +build calibnet
 
 package build
 

build/params_mainnet.go

@@ -1,5 +1,4 @@
 //go:build !calibnet && !debug && !2k
-// +build !calibnet,!debug,!2k
 
 package build
 

build/params_testnet.go

@@ -1,5 +1,4 @@
 //go:build debug
-// +build debug
 
 package build
 

cmd/curio/internal/translations/locales/en/out.gotext.json

@@ -1170,6 +1170,13 @@
             "translatorComment": "Copied from source.",
             "fuzzy": true
         },
+        {
+            "id": "Supra consensus testing utilities",
+            "message": "Supra consensus testing utilities",
+            "translation": "Supra consensus testing utilities",
+            "translatorComment": "Copied from source.",
+            "fuzzy": true
+        },
         {
             "id": "Tool Box for Curio",
             "message": "Tool Box for Curio",

cmd/curio/internal/translations/locales/ko/messages.gotext.json

@@ -2931,6 +2931,12 @@
       "translation": "PDP 지갑이 가져와졌습니다",
       "message": "PDP wallet imported",
       "placeholder": null
+    },
+    {
+      "id": "Supra consensus testing utilities",
+      "translation": "수프라 합의 테스트 유틸리티",
+      "message": "Supra consensus testing utilities",
+      "placeholder": null
     }
   ]
 }

cmd/curio/internal/translations/locales/zh/messages.gotext.json

@@ -2901,6 +2901,12 @@
       "translation": "1. 使用以下命令测试您的 PDP 服务：pdptool ping --service-url https://your-domain.com --service-name public",
       "message": "1. Test your PDP service with: pdptool ping --service-url https://your-domain.com --service-name public",
       "placeholder": null
+    },
+    {
+      "id": "Supra consensus testing utilities",
+      "translation": "超越共识测试工具",
+      "message": "Supra consensus testing utilities",
+      "placeholder": null
     }
   ]
 }

cmd/curio/tasks/tasks.go

@@ -222,18 +222,20 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 		cfg.Subsystems.EnableProofShare ||
 		cfg.Subsystems.EnableRemoteProofs
 
+	var p2Active sealsupra.P2Active
 	if hasAnySealingTask {
-		sealingTasks, err := addSealingTasks(ctx, hasAnySealingTask, db, full, sender, as, cfg, slrLazy, asyncParams, si, stor, bstore, machine, prover)
+		sealingTasks, p2a, err := addSealingTasks(ctx, hasAnySealingTask, db, full, sender, as, cfg, slrLazy, asyncParams, si, stor, bstore, machine, prover)
 		if err != nil {
 			return nil, err
 		}
 		activeTasks = append(activeTasks, sealingTasks...)
+		p2Active = p2a
 	}
 
 	{
 		// Piece handling
 		if cfg.Subsystems.EnableParkPiece {
-			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), stor, cfg.Subsystems.ParkPieceMaxTasks)
+			parkPieceTask, err := piece2.NewParkPieceTask(db, must.One(slrLazy.Val()), stor, cfg.Subsystems.ParkPieceMaxTasks, cfg.Subsystems.ParkPieceMaxInPark, p2Active, cfg.Subsystems.ParkPieceMinFreeStoragePercent)
 			if err != nil {
 				return nil, err
 			}
@@ -268,7 +270,7 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps, shutdownChan chan
 			sdeps.DealMarket = dm
 
 			if cfg.Subsystems.EnableCommP {
-				commpTask := storage_market.NewCommpTask(dm, db, must.One(slrLazy.Val()), full, cfg.Subsystems.CommPMaxTasks)
+				commpTask := storage_market.NewCommpTask(dm, db, must.One(slrLazy.Val()), full, cfg.Subsystems.CommPMaxTasks, cfg.Subsystems.BindCommPToData)
 				activeTasks = append(activeTasks, commpTask)
 			}
 
@@ -385,7 +387,7 @@ func addSealingTasks(
 	ctx context.Context, hasAnySealingTask bool, db *harmonydb.DB, full api.Chain, sender *message.Sender,
 	as *multictladdr.MultiAddressSelector, cfg *config.CurioConfig, slrLazy *lazy.Lazy[*ffi.SealCalls],
 	asyncParams func() func() (bool, error), si paths.SectorIndex, stor *paths.Remote,
-	bstore curiochain.CurioBlockstore, machineHostPort string, prover storiface.Prover) ([]harmonytask.TaskInterface, error) {
+	bstore curiochain.CurioBlockstore, machineHostPort string, prover storiface.Prover) ([]harmonytask.TaskInterface, sealsupra.P2Active, error) {
 	var activeTasks []harmonytask.TaskInterface
 	// Sealing / Snap
 
@@ -407,18 +409,20 @@ func addSealingTasks(
 		activeTasks = append(activeTasks, scrubUnsealedTask)
 	}
 
+	var p2Active sealsupra.P2Active
 	if cfg.Subsystems.EnableBatchSeal {
-		batchSealTask, sm, err := sealsupra.NewSupraSeal(
+		batchSealTask, sm, p2a, err := sealsupra.NewSupraSeal(
 			cfg.Seal.BatchSealSectorSize,
 			cfg.Seal.BatchSealBatchSize,
 			cfg.Seal.BatchSealPipelines,
 			!cfg.Seal.SingleHasherPerThread,
 			cfg.Seal.LayerNVMEDevices,
 			machineHostPort, db, full, stor, si, slr)
 		if err != nil {
-			return nil, xerrors.Errorf("setting up batch sealer: %w", err)
+			return nil, nil, xerrors.Errorf("setting up batch sealer: %w", err)
 		}
 		slotMgr = sm
+		p2Active = p2a
 		activeTasks = append(activeTasks, batchSealTask)
 		addFinalize = true
 	}
@@ -457,7 +461,7 @@ func addSealingTasks(
 
 		storePieceTask, err := piece2.NewStorePieceTask(db, must.One(slrLazy.Val()), stor, cfg.Subsystems.MoveStorageMaxTasks)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 
 		activeTasks = append(activeTasks, moveStorageTask, moveStorageSnapTask, storePieceTask)
@@ -478,7 +482,7 @@ func addSealingTasks(
 	}
 
 	if cfg.Subsystems.EnableUpdateEncode {
-		encodeTask := snap.NewEncodeTask(slr, db, cfg.Subsystems.UpdateEncodeMaxTasks)
+		encodeTask := snap.NewEncodeTask(slr, db, cfg.Subsystems.UpdateEncodeMaxTasks, cfg.Subsystems.BindEncodeToData, cfg.Subsystems.AllowEncodeGPUOverprovision)
 		activeTasks = append(activeTasks, encodeTask)
 	}
 	if cfg.Subsystems.EnableUpdateProve || cfg.Subsystems.EnableRemoteProofs {
@@ -523,7 +527,7 @@ func addSealingTasks(
 		activeTasks = append(activeTasks, storageEndpointGcTask, pipelineGcTask, storageGcMarkTask, storageGcSweepTask, sectorMetadataTask)
 	}
 
-	return activeTasks, nil
+	return activeTasks, p2Active, nil
 }
 
 func machineDetails(deps *deps.Deps, activeTasks []harmonytask.TaskInterface, machineID int, machineName string) {

cmd/curio/test-cli.go

@@ -41,6 +41,7 @@ var testCmd = &cli.Command{
 		//provingInfoCmd,
 		wdPostCmd,
 		testDebugCmd,
+		testSupraCmd,
 	},
 	Before: func(cctx *cli.Context) error {
 		return nil

cmd/curio/test-supra.go

@@ -0,0 +1,205 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"os"
+	"time"
+
+	"github.com/ipfs/go-cid"
+	"github.com/urfave/cli/v2"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/go-state-types/abi"
+
+	"github.com/filecoin-project/curio/cmd/curio/internal/translations"
+	"github.com/filecoin-project/curio/lib/ffi/cunative"
+	"github.com/filecoin-project/curio/lib/supraffi"
+)
+
+var testSupraCmd = &cli.Command{
+	Name:  "supra",
+	Usage: translations.T("Supra consensus testing utilities"),
+	Subcommands: []*cli.Command{
+		testSupraTreeRFileCmd,
+		testSnapEncodeCmd,
+	},
+}
+
+var testSupraTreeRFileCmd = &cli.Command{
+	Name:  "tree-r-file",
+	Usage: "Test tree-r-file",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "last-layer-filename",
+			Usage:    "Last layer filename",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "data-filename",
+			Usage:    "Data filename",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "output-dir",
+			Usage:    "Output directory",
+			Required: true,
+		},
+		&cli.Uint64Flag{
+			Name:     "sector-size",
+			Usage:    "Sector size",
+			Required: true,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		res := supraffi.TreeRFile(cctx.String("last-layer-filename"), cctx.String("data-filename"), cctx.String("output-dir"), cctx.Uint64("sector-size"))
+		if res != 0 {
+			return xerrors.Errorf("tree-r-file failed: %d", res)
+		}
+		return nil
+	},
+}
+
+var testSnapEncodeCmd = &cli.Command{
+	Name:  "snap-encode",
+	Usage: "Test snap-encode",
+	Flags: []cli.Flag{
+		&cli.StringFlag{
+			Name:     "sealed-filename",
+			Usage:    "Sealed filename",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "unsealed-filename",
+			Usage:    "Unsealed filename",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "update-filename",
+			Usage:    "Update filename",
+			Required: true,
+		},
+		&cli.Uint64Flag{
+			Name:     "sector-size",
+			Usage:    "Sector size (bytes). Supported: 2048, 8388608, 549755813888, 34359738368, 68719476736",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "commd",
+			Usage:    "Unsealed CommD CID (v1)",
+			Required: true,
+		},
+		&cli.StringFlag{
+			Name:     "commk",
+			Usage:    "SectorKey CommR (commK) CID (v1)",
+			Required: true,
+		},
+		&cli.BoolFlag{
+			Name:  "membuffer",
+			Usage: "Use memory buffer instead of disk (load and store)",
+			Value: false,
+		},
+	},
+	Action: func(cctx *cli.Context) error {
+		sealedPath := cctx.String("sealed-filename")
+		unsealedPath := cctx.String("unsealed-filename")
+		updatePath := cctx.String("update-filename")
+		useMem := cctx.Bool("membuffer")
+
+		commD, err := cid.Parse(cctx.String("commd"))
+		if err != nil {
+			return xerrors.Errorf("parse commD: %w", err)
+		}
+		commK, err := cid.Parse(cctx.String("commk"))
+		if err != nil {
+			return xerrors.Errorf("parse commK: %w", err)
+		}
+
+		spt, err := proofFromSectorSize(cctx.Uint64("sector-size"))
+		if err != nil {
+			return err
+		}
+		ssize, err := spt.SectorSize()
+		if err != nil {
+			return err
+		}
+
+		start := time.Now()
+		if useMem {
+			sealedBytes, err := os.ReadFile(sealedPath)
+			if err != nil {
+				return xerrors.Errorf("read sealed: %w", err)
+			}
+			unsealedBytes, err := os.ReadFile(unsealedPath)
+			if err != nil {
+				return xerrors.Errorf("read unsealed: %w", err)
+			}
+
+			elapsed := time.Since(start)
+			mbps := float64(ssize) / elapsed.Seconds() / 1024.0 / 1024.0
+			fmt.Printf("Load time: %s\n", elapsed)
+			fmt.Printf("Load throughput: %.2f MB/s\n", mbps)
+
+			var outBuf bytes.Buffer
+			outBuf.Grow(int(ssize))
+			start = time.Now() //nolint:staticcheck // false positive: used on line 181
+			if err := cunative.EncodeSnap(spt, commD, commK, bytes.NewReader(sealedBytes), bytes.NewReader(unsealedBytes), &outBuf); err != nil {
+				return xerrors.Errorf("EncodeSnap: %w", err)
+			}
+		} else {
+			keyF, err := os.Open(sealedPath)
+			if err != nil {
+				return xerrors.Errorf("open sealed: %w", err)
+			}
+			defer func() { _ = keyF.Close() }()
+
+			dataF, err := os.Open(unsealedPath)
+			if err != nil {
+				return xerrors.Errorf("open unsealed: %w", err)
+			}
+			defer func() { _ = dataF.Close() }()
+
+			outF, err := os.OpenFile(updatePath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644)
+			if err != nil {
+				return xerrors.Errorf("create update: %w", err)
+			}
+			defer func() { _ = outF.Close() }()
+
+			if err := cunative.EncodeSnap(spt, commD, commK, keyF, dataF, outF); err != nil {
+				return xerrors.Errorf("EncodeSnap: %w", err)
+			}
+
+			if err := outF.Sync(); err != nil {
+				return xerrors.Errorf("sync update: %w", err)
+			}
+
+			_, _ = io.Copy(io.Discard, keyF)
+			_, _ = io.Copy(io.Discard, dataF)
+			start = time.Now()
+		}
+		elapsed := time.Since(start)
+		mbps := float64(ssize) / elapsed.Seconds() / 1024.0 / 1024.0
+		fmt.Printf("EncodeSnap time: %s\n", elapsed)
+		fmt.Printf("EncodeSnap throughput: %.2f MB/s\n", mbps)
+
+		return nil
+	},
+}
+
+func proofFromSectorSize(size uint64) (abi.RegisteredSealProof, error) {
+	switch size {
+	case 2 << 10:
+		return abi.RegisteredSealProof_StackedDrg2KiBV1_1, nil
+	case 8 << 20:
+		return abi.RegisteredSealProof_StackedDrg8MiBV1_1, nil
+	case 512 << 20:
+		return abi.RegisteredSealProof_StackedDrg512MiBV1_1, nil
+	case 32 << 30:
+		return abi.RegisteredSealProof_StackedDrg32GiBV1_1, nil
+	case 64 << 30:
+		return abi.RegisteredSealProof_StackedDrg64GiBV1_1, nil
+	default:
+		return 0, xerrors.Errorf("unsupported sector size: %d", size)
+	}
+}