Merge pull request #11664 from filecoin-project/feat/lpdeal-cache

feat: curio deal cache (Piece Park)
2024-03-18 21:13:35 +01:00
parent 6179e40b24 154cf09f52
commit 356ea2d774
23 changed files with 852 additions and 45 deletions
@@ -13,9 +13,12 @@ import (
 	"github.com/filecoin-project/lotus/curiosrc/chainsched"
 	"github.com/filecoin-project/lotus/curiosrc/ffi"
 	"github.com/filecoin-project/lotus/curiosrc/message"
+	"github.com/filecoin-project/lotus/curiosrc/piece"
 	"github.com/filecoin-project/lotus/curiosrc/seal"
 	"github.com/filecoin-project/lotus/curiosrc/winning"
 	"github.com/filecoin-project/lotus/lib/harmony/harmonytask"
+	"github.com/filecoin-project/lotus/lib/lazy"
+	"github.com/filecoin-project/lotus/lib/must"
 	"github.com/filecoin-project/lotus/node/modules"
 )

@@ -66,6 +69,19 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 		}
 	}

+	slrLazy := lazy.MakeLazy(func() (*ffi.SealCalls, error) {
+		return ffi.NewSealCalls(stor, lstor, si), nil
+	})
+
+	{
+		// Piece handling
+		if cfg.Subsystems.EnableParkPiece {
+			parkPieceTask := piece.NewParkPieceTask(db, must.One(slrLazy.Val()), cfg.Subsystems.ParkPieceMaxTasks)
+			cleanupPieceTask := piece.NewCleanupPieceTask(db, must.One(slrLazy.Val()), 0)
+			activeTasks = append(activeTasks, parkPieceTask, cleanupPieceTask)
+		}
+	}
+
 	hasAnySealingTask := cfg.Subsystems.EnableSealSDR ||
 		cfg.Subsystems.EnableSealSDRTrees ||
 		cfg.Subsystems.EnableSendPrecommitMsg ||
@@ -81,7 +97,7 @@ func StartTasks(ctx context.Context, dependencies *deps.Deps) (*harmonytask.Task
 			sp = seal.NewPoller(db, full)
 			go sp.RunPoller(ctx)

-			slr = ffi.NewSealCalls(stor, lstor, si)
+			slr = must.One(slrLazy.Val())
 		}

 		// NOTE: Tasks with the LEAST priority are at the top
@@ -14,7 +14,8 @@ import (
 	"time"

 	"github.com/fatih/color"
-	"github.com/ipfs/go-cid"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
 	"github.com/mitchellh/go-homedir"
 	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/urfave/cli/v2"
@@ -37,6 +38,7 @@ import (
 	"github.com/filecoin-project/lotus/cmd/curio/deps"
 	cumarket "github.com/filecoin-project/lotus/curiosrc/market"
 	"github.com/filecoin-project/lotus/curiosrc/market/fakelm"
+	"github.com/filecoin-project/lotus/lib/harmony/harmonydb"
 	"github.com/filecoin-project/lotus/lib/must"
 	"github.com/filecoin-project/lotus/lib/nullreader"
 	"github.com/filecoin-project/lotus/metrics/proxy"
@@ -427,7 +429,7 @@ var lpBoostProxyCmd = &cli.Command{
 		}

 		pieceInfoLk := new(sync.Mutex)
-		pieceInfos := map[cid.Cid][]pieceInfo{}
+		pieceInfos := map[uuid.UUID][]pieceInfo{}

 		ast.Internal.SectorAddPieceToAny = func(ctx context.Context, pieceSize abi.UnpaddedPieceSize, pieceData storiface.Data, deal api.PieceDealInfo) (api.SectorOffset, error) {
 			origPieceData := pieceData
@@ -449,25 +451,103 @@ var lpBoostProxyCmd = &cli.Command{
 				done: make(chan struct{}),
 			}

+			pieceUUID := uuid.New()
+
+			color.Blue("%s %s piece assign request with id %s", deal.DealProposal.PieceCID, deal.DealProposal.Provider, pieceUUID)
+
 			pieceInfoLk.Lock()
-			pieceInfos[deal.DealProposal.PieceCID] = append(pieceInfos[deal.DealProposal.PieceCID], pi)
+			pieceInfos[pieceUUID] = append(pieceInfos[pieceUUID], pi)
 			pieceInfoLk.Unlock()

 			// /piece?piece_cid=xxxx
 			dataUrl := rootUrl
 			dataUrl.Path = "/piece"
-			dataUrl.RawQuery = "piece_cid=" + deal.DealProposal.PieceCID.String()
+			dataUrl.RawQuery = "piece_id=" + pieceUUID.String()
+
+			// add piece entry
+
+			var refID int64
+			var pieceWasCreated bool
+
+			comm, err := db.BeginTransaction(ctx, func(tx *harmonydb.Tx) (commit bool, err error) {
+				var pieceID int64
+				// Attempt to select the piece ID first
+				err = tx.QueryRow(`SELECT id FROM parked_pieces WHERE piece_cid = $1`, deal.DealProposal.PieceCID.String()).Scan(&pieceID)
+
+				if err != nil {
+					if err == pgx.ErrNoRows {
+						// Piece does not exist, attempt to insert
+						err = tx.QueryRow(`
+							INSERT INTO parked_pieces (piece_cid, piece_padded_size, piece_raw_size)
+							VALUES ($1, $2, $3)
+							ON CONFLICT (piece_cid) DO NOTHING
+							RETURNING id`, deal.DealProposal.PieceCID.String(), int64(pieceSize.Padded()), int64(pieceSize)).Scan(&pieceID)
+						if err != nil {
+							return false, xerrors.Errorf("inserting new parked piece and getting id: %w", err)
+						}
+						pieceWasCreated = true // New piece was created
+					} else {
+						// Some other error occurred during select
+						return false, xerrors.Errorf("checking existing parked piece: %w", err)
+					}
+				} else {
+					pieceWasCreated = false // Piece already exists, no new piece was created
+				}
+
+				// Add parked_piece_ref
+				err = tx.QueryRow(`INSERT INTO parked_piece_refs (piece_id, data_url)
+        			VALUES ($1, $2) RETURNING ref_id`, pieceID, dataUrl.String()).Scan(&refID)
+				if err != nil {
+					return false, xerrors.Errorf("inserting parked piece ref: %w", err)
+				}
+
+				// If everything went well, commit the transaction
+				return true, nil // This will commit the transaction
+			}, harmonydb.OptionRetry())
+			if err != nil {
+				return api.SectorOffset{}, xerrors.Errorf("inserting parked piece: %w", err)
+			}
+			if !comm {
+				return api.SectorOffset{}, xerrors.Errorf("piece tx didn't commit")
+			}
+
+			// wait for piece to be parked
+			if pieceWasCreated {
+				<-pi.done
+			} else {
+				// If the piece was not created, we need to close the done channel
+				close(pi.done)
+
+				go func() {
+					// close the data reader (drain to eof if it's not a closer)
+					if closer, ok := pieceData.(io.Closer); ok {
+						if err := closer.Close(); err != nil {
+							log.Warnw("closing pieceData in DataCid", "error", err)
+						}
+					} else {
+						log.Warnw("pieceData is not an io.Closer", "type", fmt.Sprintf("%T", pieceData))
+
+						_, err := io.Copy(io.Discard, pieceData)
+						if err != nil {
+							log.Warnw("draining pieceData in DataCid", "error", err)
+						}
+					}
+				}()
+			}
+
+			pieceIDUrl := url.URL{
+				Scheme: "pieceref",
+				Opaque: fmt.Sprintf("%d", refID),
+			}

 			// make a sector
-			so, err := pin.AllocatePieceToSector(ctx, maddr, deal, int64(pieceSize), dataUrl, nil)
+			so, err := pin.AllocatePieceToSector(ctx, maddr, deal, int64(pieceSize), pieceIDUrl, nil)
 			if err != nil {
 				return api.SectorOffset{}, err
 			}

 			color.Blue("%s piece assigned to sector f0%d:%d @ %d", deal.DealProposal.PieceCID, mid, so.Sector, so.Offset)

-			<-pi.done
-
 			return so, nil
 		}

@@ -484,10 +564,12 @@ var lpBoostProxyCmd = &cli.Command{
 		ast.Internal.StorageGetLocks = si.StorageGetLocks

 		var pieceHandler http.HandlerFunc = func(w http.ResponseWriter, r *http.Request) {
-			// /piece?piece_cid=xxxx
-			pieceCid, err := cid.Decode(r.URL.Query().Get("piece_cid"))
+			// /piece?piece_id=xxxx
+			pieceUUID := r.URL.Query().Get("piece_id")
+
+			pu, err := uuid.Parse(pieceUUID)
 			if err != nil {
-				http.Error(w, "bad piece_cid", http.StatusBadRequest)
+				http.Error(w, "bad piece id", http.StatusBadRequest)
 				return
 			}

@@ -496,13 +578,13 @@ var lpBoostProxyCmd = &cli.Command{
 				return
 			}

-			fmt.Printf("%s request for piece from %s\n", pieceCid, r.RemoteAddr)
+			fmt.Printf("%s request for piece from %s\n", pieceUUID, r.RemoteAddr)

 			pieceInfoLk.Lock()
-			pis, ok := pieceInfos[pieceCid]
+			pis, ok := pieceInfos[pu]
 			if !ok {
 				http.Error(w, "piece not found", http.StatusNotFound)
-				color.Red("%s not found", pieceCid)
+				color.Red("%s not found", pu)
 				pieceInfoLk.Unlock()
 				return
 			}
@@ -511,7 +593,10 @@ var lpBoostProxyCmd = &cli.Command{
 			pi := pis[0]
 			pis = pis[1:]

-			pieceInfos[pieceCid] = pis
+			pieceInfos[pu] = pis
+			if len(pis) == 0 {
+				delete(pieceInfos, pu)
+			}

 			pieceInfoLk.Unlock()

@@ -533,7 +618,7 @@ var lpBoostProxyCmd = &cli.Command{
 				return
 			}

-			color.Green("%s served %.3f MiB in %s (%.2f MiB/s)", pieceCid, float64(n)/(1024*1024), took, mbps)
+			color.Green("%s served %.3f MiB in %s (%.2f MiB/s)", pu, float64(n)/(1024*1024), took, mbps)
 		}

 		finalApi := proxy.LoggingAPI[api.StorageMiner, api.StorageMinerStruct](&ast)