Merge pull request #10934 from filecoin-project/sbansal/backport-v1.23.1-rc3

feat: chainstore: sharded mutex for filling chain height index
Appease the linter
2023-05-31 13:38:21 -04:00 · 2023-05-31 12:48:34 -04:00 · 2023-05-31 12:48:25 -04:00 · 2023-05-25 19:28:43 -04:00 · 2023-05-24 12:38:44 -04:00 · 2023-05-10 17:38:27 -04:00
28 changed files with 573 additions and 1132 deletions
@@ -63,7 +63,7 @@ commands:
                name: Install Rust
                command: |
                  curl https://sh.rustup.rs -sSf | sh -s -- -y
-      - run: make deps lotus
+      - run: make deps
  download-params:
    steps:
      - restore_cache:
@@ -304,9 +304,7 @@ jobs:
          darwin: true
          darwin-architecture: arm64
      - run: |
-          export CPATH=$(brew --prefix)/include
-          export LIBRARY_PATH=$(brew --prefix)/lib
-          make lotus lotus-miner lotus-worker
+          export CPATH=$(brew --prefix)/include && export LIBRARY_PATH=$(brew --prefix)/lib && make lotus lotus-miner lotus-worker
      - run: otool -hv lotus
      - run:
          name: check tag and version output match
@@ -884,6 +882,12 @@ workflows:
            - build
          suite: itest-sdr_upgrade
          target: "./itests/sdr_upgrade_test.go"
+      - test:
+          name: test-itest-sealing_resources
+          requires:
+            - build
+          suite: itest-sealing_resources
+          target: "./itests/sealing_resources_test.go"
      - test:
          name: test-itest-sector_finalize_early
          requires:
@@ -1069,6 +1073,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -1078,6 +1083,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -1087,6 +1093,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -1099,7 +1106,7 @@ workflows:
          filters:
            branches:
              ignore:
-                - /.*/
+                - /^.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -1114,6 +1121,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
      - build-docker:
          name: "Docker push (lotus-all-in-one / stable / mainnet)"
          image: lotus-all-in-one
@@ -63,7 +63,7 @@ commands:
                name: Install Rust
                command: |
                  curl https://sh.rustup.rs -sSf | sh -s -- -y
-      - run: make deps lotus
+      - run: make deps
  download-params:
    steps:
      - restore_cache:
@@ -304,9 +304,7 @@ jobs:
          darwin: true
          darwin-architecture: arm64
      - run: |
-          export CPATH=$(brew --prefix)/include
-          export LIBRARY_PATH=$(brew --prefix)/lib
-          make lotus lotus-miner lotus-worker
+          export CPATH=$(brew --prefix)/include && export LIBRARY_PATH=$(brew --prefix)/lib && make lotus lotus-miner lotus-worker
      - run: otool -hv lotus
      - run:
          name: check tag and version output match
@@ -583,6 +581,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -592,6 +591,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -601,6 +601,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -613,7 +614,7 @@ workflows:
          filters:
            branches:
              ignore:
-                - /.*/
+                - /^.*$/
            tags:
              only:
                - /^v\d+\.\d+\.\d+(-rc\d+)?$/
@@ -628,6 +629,7 @@ workflows:
            branches:
              only:
                - /^release\/v\d+\.\d+\.\d+(-rc\d+)?$/
+                - /^ci\/.*$/
      [[- range .Networks]]
      - build-docker:
          name: "Docker push (lotus-all-in-one / stable / [[.]])"
@@ -1,5 +1,132 @@
 # Lotus changelog

+# v1.23.1-rc2 / 2023-05-10
+
+This is the first release candidate of the upcoming optional feature release of Lotus v1.23.1.
+
+**☢️ Upgrade Warnings ☢️**
+
+- If you are upgrading to this release candidate from Lotus v1.22.1, please make sure to read the upgrade warnings section in the [v1.23.0 release first.](https://github.com/filecoin-project/lotus/releases/tag/v1.23.0)
+
+## New features
+
+- feat: daemon: Auto-resume interrupted snapshot imports ([filecoin-project/lotus#10636](https://github.com/filecoin-project/lotus/pull/10636))
+- feat: VM Execution Lanes ([filecoin-project/lotus#10551](https://github.com/filecoin-project/lotus/pull/10551))
+- feat: chainstore: batch writes of tipsets ([filecoin-project/lotus#10800](https://github.com/filecoin-project/lotus/pull/10800))
+- Add API and CLI to unseal sector (#10626) ([filecoin-project/lotus#10626](https://github.com/filecoin-project/lotus/pull/10626))
+- feat: sealing: Split PCA/PCB batches if gas used exceeds block limit ([filecoin-project/lotus#10647](https://github.com/filecoin-project/lotus/pull/10647))
+- Add feature to stagger sector prove commit submission (#10543) ([filecoin-project/lotus#10543](https://github.com/filecoin-project/lotus/pull/10543))
+- Set default for MaxSectorProveCommitsSubmittedPerEpoch ([filecoin-project/lotus#10728](https://github.com/filecoin-project/lotus/pull/10728))
+- fix: storage: Remove temp fetching files after failed fetch ([filecoin-project/lotus#10661](https://github.com/filecoin-project/lotus/pull/10661))
+- feat: worker: Ensure tempdir exists (#10433) ([filecoin-project/lotus#10433](https://github.com/filecoin-project/lotus/pull/10433))
+- feat: sync: harden chain sync (#10756) ([filecoin-project/lotus#10756](https://github.com/filecoin-project/lotus/pull/10756))
+- feat: populate the index on snapshot import ([filecoin-project/lotus#10556](https://github.com/filecoin-project/lotus/pull/10556))
+- feat:chain: Message Index ([filecoin-project/lotus#10452](https://github.com/filecoin-project/lotus/pull/10452))
+- feat: Add small cache to execution traces ([filecoin-project/lotus#10517](https://github.com/filecoin-project/lotus/pull/10517))
+- feat: shed: incoming block-sub chainwatch tool ([filecoin-project/lotus#10513](https://github.com/filecoin-project/lotus/pull/10513))
+
+## Improvements
+- fix: sched: Address GET_32G_MAX_CONCURRENT regression
+- fix: cli: Hide legacy markets cmds
+- fix: ci: Debugging m1 build
+- Disable lotus markets by default (#10809) ([filecoin-project/lotus#10809](https://github.com/filecoin-project/lotus/pull/10809))
+- perf: mempool: lower priority optimizations (#10693) ([filecoin-project/lotus#10693](https://github.com/filecoin-project/lotus/pull/10693))
+- perf: message pool: change locks to RWMutexes for performance ([filecoin-project/lotus#10561](https://github.com/filecoin-project/lotus/pull/10561))
+- perf: eth: gas estimate set applyTsMessages false (#10546) ([filecoin-project/lotus#10546](https://github.com/filecoin-project/lotus/pull/10546))
+- Change args check ([filecoin-project/lotus#10812](https://github.com/filecoin-project/lotus/pull/10812))
+- fix: sealing: Make lotus-worker report GPU usage to miner during ReplicaUpdate task (#10806) ([filecoin-project/lotus#10806](https://github.com/filecoin-project/lotus/pull/10806))
+- fix:splitstore:Don't block when potentially holding txnLk as writer ([filecoin-project/lotus#10811](https://github.com/filecoin-project/lotus/pull/10811))
+- fix: prover: Propagate skipped sectors in local PoSt
+- fix: unseal: check if sealed/update sector exists ([filecoin-project/lotus#10639](https://github.com/filecoin-project/lotus/pull/10639))
+- fix: sealing pipeline: Allow nil message in TerminateWait ([filecoin-project/lotus#10696](https://github.com/filecoin-project/lotus/pull/10696))
+- fix: cli: Check if the sectorID exists before removing ([filecoin-project/lotus#10611](https://github.com/filecoin-project/lotus/pull/10611))
+- feat:splitstore:limit moving gc threads ([filecoin-project/lotus#10621](https://github.com/filecoin-project/lotus/pull/10621))
+- fix: cli: Make `net connect` to miner address work ([filecoin-project/lotus#10599](https://github.com/filecoin-project/lotus/pull/10599))
+- fix: log: Stop logging `file does not exists` ([filecoin-project/lotus#10588](https://github.com/filecoin-project/lotus/pull/10588))
+- Update config default value (#10605) ([filecoin-project/lotus#10605](https://github.com/filecoin-project/lotus/pull/10605))
+- fix: cap the message gas limit at the block gas limit (#10637) ([filecoin-project/lotus#10637](https://github.com/filecoin-project/lotus/pull/10637))
+- fix: miner: correctly count sector extensions ([filecoin-project/lotus#10544](https://github.com/filecoin-project/lotus/pull/10544))
+- fix:mpool: prune excess messages before selection ([filecoin-project/lotus#10648](https://github.com/filecoin-project/lotus/pull/10648))
+- fix: proving: Initialize slice with with same length as partition ([filecoin-project/lotus#10569](https://github.com/filecoin-project/lotus/pull/10569))
+- perf: Address performance of EthGetTransactionCount  ([filecoin-project/lotus#10700](https://github.com/filecoin-project/lotus/pull/10700))
+- fix: sync: reduce log from error to info ([filecoin-project/lotus#10759](https://github.com/filecoin-project/lotus/pull/10759))
+- fix: state: lotus-miner info should show deals info without admin permission ([filecoin-project/lotus#10323](https://github.com/filecoin-project/lotus/pull/10323))
+- fix: tvx: make extract-multiple support the FVM ([filecoin-project/lotus#10714](https://github.com/filecoin-project/lotus/pull/10714))
+- feat: badger: add a has check before writing to reduce duplicates ([filecoin-project/lotus#10680](https://github.com/filecoin-project/lotus/pull/10680))
+- fix: chain: record heaviest tipset before notifying (#10694) ([filecoin-project/lotus#10694](https://github.com/filecoin-project/lotus/pull/10694))
+- fix: Eth JSON-RPC api: handle messages with gasFeeCap less than baseFee (#10614) ([filecoin-project/lotus#10614](https://github.com/filecoin-project/lotus/pull/10614))
+- feat: chainstore: optimize BlockMsgsForTipset ([filecoin-project/lotus#10552](https://github.com/filecoin-project/lotus/pull/10552))
+- refactor: stop using deprecated io/ioutil ([filecoin-project/lotus#10596](https://github.com/filecoin-project/lotus/pull/10596))
+- feat: shed: refactor market cron-state command ([filecoin-project/lotus#10746](https://github.com/filecoin-project/lotus/pull/10746))
+- fix: events: don't set GC confidence to 1 ([filecoin-project/lotus#10713](https://github.com/filecoin-project/lotus/pull/10713))
+- feat: sync: validate (early) that blocks fall within range (#10691) ([filecoin-project/lotus#10691](https://github.com/filecoin-project/lotus/pull/10691))
+- chainstore: Fix raw blocks getting scanned for links during snapshots (#10684) ([filecoin-project/lotus#10684](https://github.com/filecoin-project/lotus/pull/10684))
+- fix: remove pointless panic ([filecoin-project/lotus#10690](https://github.com/filecoin-project/lotus/pull/10690))
+- fix: check for nil bcastDict (#10646) ([filecoin-project/lotus#10646](https://github.com/filecoin-project/lotus/pull/10646))
+- fix: make state compute --html work with unknown methods ([filecoin-project/lotus#10619](https://github.com/filecoin-project/lotus/pull/10619))
+- shed: get balances of evm accounts  ([filecoin-project/lotus#10489](https://github.com/filecoin-project/lotus/pull/10489))
+- feat: Use MessageIndex in WaitForMessage ([filecoin-project/lotus#10587](https://github.com/filecoin-project/lotus/pull/10587))
+- fix: searchForIndexedMsg always returns an error ([filecoin-project/lotus#10586](https://github.com/filecoin-project/lotus/pull/10586))
+- Fix: export-range: Ignore ipld Blocks not found in Receipts. ([filecoin-project/lotus#10535](https://github.com/filecoin-project/lotus/pull/10535))
+- feat: stmgr: speed up calculation of genesis circ supply ([filecoin-project/lotus#10553](https://github.com/filecoin-project/lotus/pull/10553))
+- fix: gas estimation: don't special case paych collects ([filecoin-project/lotus#10549](https://github.com/filecoin-project/lotus/pull/10549))
+- fix: tracer: emit raw peer ids for compatibility with libp2p tracer ([filecoin-project/lotus#10271](https://github.com/filecoin-project/lotus/pull/10271))
+- Merge branch 'feat/new-gw-methods'
+
+## Dependencies
+- chore: deps: update to FVM 3.3.1 ([filecoin-project/lotus#10786](https://github.com/filecoin-project/lotus/pull/10786))
+-  chore: boxo: migrate from go-libipfs to boxo ([filecoin-project/lotus#10562](https://github.com/filecoin-project/lotus/pull/10562))
+- chore: deps: update to go-state-types v0.11.0-alpha-3 ([filecoin-project/lotus#10606](https://github.com/filecoin-project/lotus/pull/10606))
+-  chore: bump go-libipfs ([filecoin-project/lotus#10531](https://github.com/filecoin-project/lotus/pull/10531))
+
+## Others
+- feat:networking:  (Synchronous) Consistent Broadcast for Filecoin EC ([filecoin-project/lotus#9858](https://github.com/filecoin-project/lotus/pull/9858))
+- Revert #9858 (consistent broadcast changes) ([filecoin-project/lotus#10777](https://github.com/filecoin-project/lotus/pull/10777))
+- Update build version for release/v1.23.1
+- chore: drop flaky TestBatchDealInput subcase ([filecoin-project/lotus#10810](https://github.com/filecoin-project/lotus/pull/10810))
+- chore: changelog clean up ([filecoin-project/lotus#10744](https://github.com/filecoin-project/lotus/pull/10744))
+- chore: refactor: drop unused IsTicketWinner (#10801) ([filecoin-project/lotus#10801](https://github.com/filecoin-project/lotus/pull/10801))
+- chore: build: bump matser version to v1.23.1-dev ([filecoin-project/lotus#10709](https://github.com/filecoin-project/lotus/pull/10709))
+- fix: deflake: use 2 miners for flaky tests ([filecoin-project/lotus#10764](https://github.com/filecoin-project/lotus/pull/10764))
+- test: eth: deflake multiblock lookup test (#10769) ([filecoin-project/lotus#10769](https://github.com/filecoin-project/lotus/pull/10769))
+- shed: migrations: add reminder about continuity testing tool ([filecoin-project/lotus#10762](https://github.com/filecoin-project/lotus/pull/10762))
+- chore: merge releases into master ([filecoin-project/lotus#10742](https://github.com/filecoin-project/lotus/pull/10742))
+- test: events: fix race when recording tipsets (#10665) ([filecoin-project/lotus#10665](https://github.com/filecoin-project/lotus/pull/10665))
+- fix: build: add CBDeliveryDelay to testground ([filecoin-project/lotus#10613](https://github.com/filecoin-project/lotus/pull/10613))
+- fix: build: Fixed incorrect words that could not be compiled ([filecoin-project/lotus#10610](https://github.com/filecoin-project/lotus/pull/10610))
+- build: docker: Update GO-version ([filecoin-project/lotus#10581](https://github.com/filecoin-project/lotus/pull/10581))
+- fix: itests: Don't call t.Error in MineBlocks goroutine ([filecoin-project/lotus#10572](https://github.com/filecoin-project/lotus/pull/10572))
+- docs: api: clarify MpoolClear params ([filecoin-project/lotus#10550](https://github.com/filecoin-project/lotus/pull/10550))
+
+Contributors
+
+| Contributor | Commits | Lines ± | Files Changed |
+|-------------|---------|---------|---------------|
+| vyzo | 70 | +1990/-429 | 135 |
+| Alfonso de la Rocha | 25 | +814/-299 | 56 |
+| Steven Allen | 14 | +125/-539 | 28 |
+| Shrenuj Bansal | 13 | +482/-138 | 52 |
+| Aayush | 17 | +317/-301 | 90 |
+| Łukasz Magiera | 13 | +564/-26 | 16 |
+| Jennifer Wang | 7 | +401/-140 | 10 |
+| Fridrik Asmundsson | 14 | +315/-84 | 20 |
+| Jorropo | 2 | +139/-137 | 74 |
+| Mikers | 6 | +114/-43 | 14 |
+| Hector Sanjuan | 5 | +92/-44 | 5 |
+| Ales Dumikau | 1 | +117/-0 | 10 |
+| Mike Seiler | 4 | +51/-51 | 6 |
+| zenground0 | 6 | +33/-25 | 8 |
+| Phi | 8 | +32/-10 | 10 |
+| Aayush Rajasekaran | 1 | +1/-32 | 2 |
+| Ian Davis | 2 | +7/-10 | 3 |
+| Marcel Telka | 1 | +5/-7 | 1 |
+| ychiao | 1 | +8/-3 | 2 |
+| jennijuju | 1 | +4/-4 | 8 |
+| adlrocha | 2 | +2/-2 | 2 |
+| Jiaying Wang | 1 | +0/-4 | 1 |
+| ZenGround0 | 1 | +2/-1 | 2 |
+| Zeng Li | 1 | +1/-1 | 1 |
+
 # v1.23.0 / 2023-04-21

 This is the stable feature release for the upcoming MANDATORY network upgrade at `2023-04-27T13:00:00Z`, epoch `2809800`. This feature release delivers the nv19 Lighting and nv20 Thunder network upgrade for mainnet, and includes numerous improvements and enhancements for node operators, ETH RPC-providers and storage providers.
@@ -37,7 +37,7 @@ func BuildTypeString() string {
 }

 // BuildVersion is the local build version
-const BuildVersion = "1.23.1-dev"
+const BuildVersion = "1.23.1-rc2"

 func UserVersion() string {
 	if os.Getenv("LOTUS_VERSION_IGNORE_COMMIT") == "1" {
@@ -2,18 +2,21 @@ package store

 import (
 	"context"
+	"hash/maphash"
 	"os"
 	"strconv"
-	"sync"

+	"github.com/puzpuzpuz/xsync/v2"
 	"golang.org/x/xerrors"

 	"github.com/filecoin-project/go-state-types/abi"

 	"github.com/filecoin-project/lotus/chain/types"
+	"github.com/filecoin-project/lotus/lib/shardedmutex"
 )

-var DefaultChainIndexCacheSize = 32 << 15
+// DefaultChainIndexCacheSize no longer sets the maximum size, just the initial size of the map.
+var DefaultChainIndexCacheSize = 1 << 15

 func init() {
 	if s := os.Getenv("LOTUS_CHAIN_INDEX_CACHE"); s != "" {
@@ -27,8 +30,9 @@ func init() {
 }

 type ChainIndex struct {
-	indexCacheLk sync.Mutex
-	indexCache   map[types.TipSetKey]*lbEntry
+	indexCache *xsync.MapOf[types.TipSetKey, *lbEntry]
+
+	fillCacheLock shardedmutex.ShardedMutexFor[types.TipSetKey]

 	loadTipSet loadTipSetFunc

@@ -36,11 +40,16 @@ type ChainIndex struct {
 }
 type loadTipSetFunc func(context.Context, types.TipSetKey) (*types.TipSet, error)

+func maphashTSK(s maphash.Seed, tsk types.TipSetKey) uint64 {
+	return maphash.Bytes(s, tsk.Bytes())
+}
+
 func NewChainIndex(lts loadTipSetFunc) *ChainIndex {
 	return &ChainIndex{
-		indexCache: make(map[types.TipSetKey]*lbEntry, DefaultChainIndexCacheSize),
-		loadTipSet: lts,
-		skipLength: 20,
+		indexCache:    xsync.NewTypedMapOfPresized[types.TipSetKey, *lbEntry](maphashTSK, DefaultChainIndexCacheSize),
+		fillCacheLock: shardedmutex.NewFor(maphashTSK, 32),
+		loadTipSet:    lts,
+		skipLength:    20,
 	}
 }

@@ -59,17 +68,23 @@ func (ci *ChainIndex) GetTipsetByHeight(ctx context.Context, from *types.TipSet,
 		return nil, xerrors.Errorf("failed to round down: %w", err)
 	}

-	ci.indexCacheLk.Lock()
-	defer ci.indexCacheLk.Unlock()
 	cur := rounded.Key()
 	for {
-		lbe, ok := ci.indexCache[cur]
+		lbe, ok := ci.indexCache.Load(cur) // check the cache
 		if !ok {
-			fc, err := ci.fillCache(ctx, cur)
-			if err != nil {
-				return nil, xerrors.Errorf("failed to fill cache: %w", err)
+			lk := ci.fillCacheLock.GetLock(cur)
+			lk.Lock()                         // if entry is missing, take the lock
+			lbe, ok = ci.indexCache.Load(cur) // check if someone else added it while we waited for lock
+			if !ok {
+				fc, err := ci.fillCache(ctx, cur)
+				if err != nil {
+					lk.Unlock()
+					return nil, xerrors.Errorf("failed to fill cache: %w", err)
+				}
+				lbe = fc
+				ci.indexCache.Store(cur, lbe)
 			}
-			lbe = fc
+			lk.Unlock()
 		}

 		if to == lbe.targetHeight {
@@ -137,7 +152,6 @@ func (ci *ChainIndex) fillCache(ctx context.Context, tsk types.TipSetKey) (*lbEn
 		targetHeight: skipTarget.Height(),
 		target:       skipTarget.Key(),
 	}
-	ci.indexCache[tsk] = lbe

 	return lbe, nil
 }
@@ -1156,6 +1156,10 @@ func (cs *ChainStore) TryFillTipSet(ctx context.Context, ts *types.TipSet) (*Ful
 // selects the tipset before the null round if true, and the tipset following
 // the null round if false.
 func (cs *ChainStore) GetTipsetByHeight(ctx context.Context, h abi.ChainEpoch, ts *types.TipSet, prev bool) (*types.TipSet, error) {
+	if h < 0 {
+		return nil, xerrors.Errorf("height %d is negative", h)
+	}
+
 	if ts == nil {
 		ts = cs.GetHeaviestTipSet()
 	}
@@ -43,16 +43,16 @@ func main() {
 		backupCmd,
 		lcli.WithCategory("chain", actorCmd),
 		lcli.WithCategory("chain", infoCmd),
-		lcli.WithCategory("market", storageDealsCmd),
-		lcli.WithCategory("market", retrievalDealsCmd),
-		lcli.WithCategory("market", dataTransfersCmd),
-		lcli.WithCategory("market", dagstoreCmd),
-		lcli.WithCategory("market", indexProvCmd),
+		lcli.WithCategory("market", setHidden(storageDealsCmd)),
+		lcli.WithCategory("market", setHidden(retrievalDealsCmd)),
+		lcli.WithCategory("market", setHidden(dataTransfersCmd)),
+		lcli.WithCategory("market", setHidden(dagstoreCmd)),
+		lcli.WithCategory("market", setHidden(indexProvCmd)),
 		lcli.WithCategory("storage", sectorsCmd),
 		lcli.WithCategory("storage", provingCmd),
 		lcli.WithCategory("storage", storageCmd),
 		lcli.WithCategory("storage", sealingCmd),
-		lcli.WithCategory("retrieval", piecesCmd),
+		lcli.WithCategory("retrieval", setHidden(piecesCmd)),
 	}

 	jaeger := tracing.SetupJaegerTracing("lotus")
@@ -86,6 +86,7 @@ func main() {
 	// adapt the Net* commands to always hit the node running the markets
 	// subsystem, as that is the only one that runs a libp2p node.
 	netCmd := *lcli.NetCmd // make a copy.
+	netCmd.Hidden = true
 	prev := netCmd.Before
 	netCmd.Before = func(c *cli.Context) error {
 		if prev != nil {
@@ -137,11 +138,12 @@ func main() {
 			&cli.StringFlag{
 				Name:    FlagMarketsRepo,
 				EnvVars: []string{"LOTUS_MARKETS_PATH"},
-				Usage:   fmt.Sprintf("Markets repo path"),
+				Hidden:  true,
 			},
 			&cli.BoolFlag{
-				Name:  "call-on-markets",
-				Usage: "(experimental; may be removed) call this command against a markets node; use only with common commands like net, auth, pprof, etc. whose target may be ambiguous",
+				Name:   "call-on-markets",
+				Usage:  "(experimental; may be removed) call this command against a markets node; use only with common commands like net, auth, pprof, etc. whose target may be ambiguous",
+				Hidden: true,
 			},
 			cliutil.FlagVeryVerbose,
 		},
@@ -190,3 +192,8 @@ func getActorAddress(ctx context.Context, cctx *cli.Context) (maddr address.Addr

 	return maddr, nil
 }
+
+func setHidden(cmd *cli.Command) *cli.Command {
+	cmd.Hidden = true
+	return cmd
+}
@@ -7,7 +7,7 @@ USAGE:
   lotus-worker [global options] command [command options] [arguments...]

 VERSION:
-   1.23.1-dev
+   1.23.1-rc2

 COMMANDS:
   run        Start lotus worker
@@ -7,7 +7,7 @@ USAGE:
   lotus [global options] command [command options] [arguments...]

 VERSION:
-   1.23.1-dev
+   1.23.1-rc2

 COMMANDS:
   daemon   Start a lotus daemon process
@@ -139,6 +139,7 @@ require (
 	github.com/open-rpc/meta-schema v0.0.0-20201029221707-1b72ef2ea333
 	github.com/polydawn/refmt v0.89.0
 	github.com/prometheus/client_golang v1.14.0
+	github.com/puzpuzpuz/xsync/v2 v2.4.0
 	github.com/raulk/clock v1.1.0
 	github.com/raulk/go-watchdog v1.3.0
 	github.com/stretchr/testify v1.8.2
@@ -1484,6 +1484,8 @@ github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5
 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
 github.com/prometheus/statsd_exporter v0.21.0 h1:hA05Q5RFeIjgwKIYEdFd59xu5Wwaznf33yKI+pyX6T8=
 github.com/prometheus/statsd_exporter v0.21.0/go.mod h1:rbT83sZq2V+p73lHhPZfMc3MLCHmSHelCh9hSGYNLTQ=
+github.com/puzpuzpuz/xsync/v2 v2.4.0 h1:5sXAMHrtx1bg9nbRZTOn8T4MkWe5V+o8yKRH02Eznag=
+github.com/puzpuzpuz/xsync/v2 v2.4.0/go.mod h1:gD2H2krq/w52MfPLE+Uy64TzJDVY7lP2znR9qmR35kU=
 github.com/quic-go/qpack v0.4.0 h1:Cr9BXA1sQS2SmDUWjSofMPNKmvF6IiIfDRmgU0w1ZCo=
 github.com/quic-go/qpack v0.4.0/go.mod h1:UZVnYIfi5GRk+zI9UMaCPsmZ2xKJP7XBUvVyT1Knj9A=
 github.com/quic-go/qtls-go1-19 v0.2.1 h1:aJcKNMkH5ASEJB9FXNeZCyTEIHU1J7MmHyz1Q1TSG1A=
@@ -0,0 +1,64 @@
+package itests
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/filecoin-project/lotus/itests/kit"
+	"github.com/filecoin-project/lotus/storage/sealer/storiface"
+)
+
+// Regression check for a fix introduced in https://github.com/filecoin-project/lotus/pull/10633
+func TestPledgeMaxConcurrentGet(t *testing.T) {
+	require.NoError(t, os.Setenv("GET_2K_MAX_CONCURRENT", "1"))
+	t.Cleanup(func() {
+		require.NoError(t, os.Unsetenv("GET_2K_MAX_CONCURRENT"))
+	})
+
+	kit.QuietMiningLogs()
+
+	blockTime := 50 * time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	_, miner, ens := kit.EnsembleMinimal(t, kit.NoStorage()) // no mock proofs
+	ens.InterconnectAll().BeginMiningMustPost(blockTime)
+
+	// separate sealed and storage paths so that finalize move needs to happen
+	miner.AddStorage(ctx, t, func(meta *storiface.LocalStorageMeta) {
+		meta.CanSeal = true
+	})
+	miner.AddStorage(ctx, t, func(meta *storiface.LocalStorageMeta) {
+		meta.CanStore = true
+	})
+
+	// NOTE: This test only repros the issue when Fetch tasks take ~10s, there's
+	// no great way to do that in a non-horribly-hacky way
+
+	/* The horribly hacky way:
+
+	diff --git a/storage/sealer/sched_worker.go b/storage/sealer/sched_worker.go
+	index 35acd755d..76faec859 100644
+	--- a/storage/sealer/sched_worker.go
+	+++ b/storage/sealer/sched_worker.go
+	@@ -513,6 +513,10 @@ func (sw *schedWorker) startProcessingTask(req *WorkerRequest) error {
+	                        tw.start()
+	                        err = <-werr
+
+	+                       if req.TaskType == sealtasks.TTFetch {
+	+                               time.Sleep(10 * time.Second)
+	+                       }
+	+
+	                        select {
+	                        case req.ret <- workerResponse{err: err}:
+	                        case <-req.Ctx.Done():
+
+	*/
+
+	miner.PledgeSectors(ctx, 3, 0, nil)
+}
@@ -0,0 +1,75 @@
+package shardedmutex
+
+import (
+	"hash/maphash"
+	"sync"
+)
+
+const cacheline = 64
+
+// padding a mutex to a cacheline improves performance as the cachelines are not contested
+// name     old time/op  new time/op  delta
+// Locks-8  74.6ns ± 7%  12.3ns ± 2%  -83.54%  (p=0.000 n=20+18)
+type paddedMutex struct {
+	mt  sync.Mutex
+	pad [cacheline - 8]uint8
+}
+
+type ShardedMutex struct {
+	shards []paddedMutex
+}
+
+// New creates a new ShardedMutex with N shards
+func New(nShards int) ShardedMutex {
+	if nShards < 1 {
+		panic("n_shards cannot be less than 1")
+	}
+	return ShardedMutex{
+		shards: make([]paddedMutex, nShards),
+	}
+}
+
+func (sm ShardedMutex) Shards() int {
+	return len(sm.shards)
+}
+
+func (sm ShardedMutex) Lock(shard int) {
+	sm.shards[shard].mt.Lock()
+}
+
+func (sm ShardedMutex) Unlock(shard int) {
+	sm.shards[shard].mt.Unlock()
+}
+
+func (sm ShardedMutex) GetLock(shard int) sync.Locker {
+	return &sm.shards[shard].mt
+}
+
+type ShardedMutexFor[K any] struct {
+	inner ShardedMutex
+
+	hasher func(maphash.Seed, K) uint64
+	seed   maphash.Seed
+}
+
+func NewFor[K any](hasher func(maphash.Seed, K) uint64, nShards int) ShardedMutexFor[K] {
+	return ShardedMutexFor[K]{
+		inner:  New(nShards),
+		hasher: hasher,
+		seed:   maphash.MakeSeed(),
+	}
+}
+
+func (sm ShardedMutexFor[K]) shardFor(key K) int {
+	return int(sm.hasher(sm.seed, key) % uint64(len(sm.inner.shards)))
+}
+
+func (sm ShardedMutexFor[K]) Lock(key K) {
+	sm.inner.Lock(sm.shardFor(key))
+}
+func (sm ShardedMutexFor[K]) Unlock(key K) {
+	sm.inner.Unlock(sm.shardFor(key))
+}
+func (sm ShardedMutexFor[K]) GetLock(key K) sync.Locker {
+	return sm.inner.GetLock(sm.shardFor(key))
+}
@@ -0,0 +1,159 @@
+package shardedmutex
+
+import (
+	"fmt"
+	"hash/maphash"
+	"runtime"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestLockingDifferentShardsDoesNotBlock(t *testing.T) {
+	shards := 16
+	sm := New(shards)
+	done := make(chan struct{})
+	go func() {
+		select {
+		case <-done:
+			return
+		case <-time.After(5 * time.Second):
+			panic("test locked up")
+		}
+	}()
+	for i := 0; i < shards; i++ {
+		sm.Lock(i)
+	}
+
+	close(done)
+}
+func TestLockingSameShardsBlocks(t *testing.T) {
+	shards := 16
+	sm := New(shards)
+	wg := sync.WaitGroup{}
+	wg.Add(shards)
+	ch := make(chan int, shards)
+
+	for i := 0; i < shards; i++ {
+		go func(i int) {
+			if i != 15 {
+				sm.Lock(i)
+			}
+			wg.Done()
+			wg.Wait()
+			sm.Lock((15 + i) % shards)
+			ch <- i
+			sm.Unlock(i)
+		}(i)
+	}
+
+	wg.Wait()
+	for i := 0; i < 2*shards; i++ {
+		runtime.Gosched()
+	}
+	for i := 0; i < shards; i++ {
+		if a := <-ch; a != i {
+			t.Errorf("got %d instead of %d", a, i)
+		}
+	}
+}
+
+func TestShardedByString(t *testing.T) {
+	shards := 16
+	sm := NewFor(maphash.String, shards)
+
+	wg1 := sync.WaitGroup{}
+	wg1.Add(shards * 20)
+	wg2 := sync.WaitGroup{}
+	wg2.Add(shards * 20)
+
+	active := atomic.Int32{}
+	max := atomic.Int32{}
+
+	for i := 0; i < shards*20; i++ {
+		go func(i int) {
+			wg1.Done()
+			wg1.Wait()
+			sm.Lock(fmt.Sprintf("goroutine %d", i))
+			activeNew := active.Add(1)
+			for {
+				curMax := max.Load()
+				if curMax >= activeNew {
+					break
+				}
+				if max.CompareAndSwap(curMax, activeNew) {
+					break
+				}
+			}
+			for j := 0; j < 100; j++ {
+				runtime.Gosched()
+			}
+			active.Add(-1)
+			sm.Unlock(fmt.Sprintf("goroutine %d", i))
+			wg2.Done()
+		}(i)
+	}
+
+	wg2.Wait()
+
+	if max.Load() != 16 {
+		t.Fatal("max load not achieved", max.Load())
+	}
+
+}
+
+func BenchmarkShardedMutex(b *testing.B) {
+	shards := 16
+	sm := New(shards)
+
+	done := atomic.Int32{}
+	go func() {
+		for {
+			sm.Lock(0)
+			sm.Unlock(0)
+			if done.Load() != 0 {
+				return
+			}
+		}
+	}()
+	for i := 0; i < 100; i++ {
+		runtime.Gosched()
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		sm.Lock(1)
+		sm.Unlock(1)
+	}
+	done.Add(1)
+}
+
+func BenchmarkShardedMutexOf(b *testing.B) {
+	shards := 16
+	sm := NewFor(maphash.String, shards)
+
+	str1 := "string1"
+	str2 := "string2"
+
+	done := atomic.Int32{}
+	go func() {
+		for {
+			sm.Lock(str1)
+			sm.Unlock(str1)
+			if done.Load() != 0 {
+				return
+			}
+		}
+	}()
+	for i := 0; i < 100; i++ {
+		runtime.Gosched()
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		sm.Lock(str2)
+		sm.Unlock(str2)
+	}
+	done.Add(1)
+}
@@ -103,7 +103,7 @@ func (a *AssignerCommon) TrySched(sh *Scheduler) {
 				needRes := worker.Info.Resources.ResourceSpec(task.Sector.ProofType, task.TaskType)

 				// TODO: allow bigger windows
-				if !windows[wnd].Allocated.CanHandleRequest(task.SealTask(), needRes, windowRequest.Worker, "schedAcceptable", worker.Info) {
+				if !windows[wnd].Allocated.CanHandleRequest(task.SchedId, task.SealTask(), needRes, windowRequest.Worker, "schedAcceptable", worker.Info) {
 					continue
 				}

@@ -37,7 +37,7 @@ func RandomWS(sh *Scheduler, queueLen int, acceptableWindows [][]int, windows []

 			log.Debugf("SCHED try assign sqi:%d sector %d to window %d (awi:%d)", sqi, task.Sector.ID.Number, wnd, i)

-			if !windows[wnd].Allocated.CanHandleRequest(task.SealTask(), res, wid, "schedAssign", w.Info) {
+			if !windows[wnd].Allocated.CanHandleRequest(task.SchedId, task.SealTask(), res, wid, "schedAssign", w.Info) {
 				continue
 			}

@@ -71,7 +71,7 @@ func RandomWS(sh *Scheduler, queueLen int, acceptableWindows [][]int, windows []
 			"worker", bestWid,
 			"choices", len(choices))

-		windows[selectedWindow].Allocated.Add(task.SealTask(), info.Resources, needRes)
+		windows[selectedWindow].Allocated.Add(task.SchedId, task.SealTask(), info.Resources, needRes)
 		windows[selectedWindow].Todo = append(windows[selectedWindow].Todo, task)

 		rmQueue = append(rmQueue, sqi)
@@ -35,7 +35,7 @@ func SpreadWS(queued bool) func(sh *Scheduler, queueLen int, acceptableWindows [

 				log.Debugf("SCHED try assign sqi:%d sector %d to window %d (awi:%d)", sqi, task.Sector.ID.Number, wnd, i)

-				if !windows[wnd].Allocated.CanHandleRequest(task.SealTask(), res, wid, "schedAssign", w.Info) {
+				if !windows[wnd].Allocated.CanHandleRequest(task.SchedId, task.SealTask(), res, wid, "schedAssign", w.Info) {
 					continue
 				}

@@ -71,7 +71,7 @@ func SpreadWS(queued bool) func(sh *Scheduler, queueLen int, acceptableWindows [
 				"assigned", bestAssigned)

 			workerAssigned[bestWid]++
-			windows[selectedWindow].Allocated.Add(task.SealTask(), info.Resources, needRes)
+			windows[selectedWindow].Allocated.Add(task.SchedId, task.SealTask(), info.Resources, needRes)
 			windows[selectedWindow].Todo = append(windows[selectedWindow].Todo, task)

 			rmQueue = append(rmQueue, sqi)
@@ -41,7 +41,7 @@ func SpreadTasksWS(queued bool) func(sh *Scheduler, queueLen int, acceptableWind

 				log.Debugf("SCHED try assign sqi:%d sector %d to window %d (awi:%d)", sqi, task.Sector.ID.Number, wnd, i)

-				if !windows[wnd].Allocated.CanHandleRequest(task.SealTask(), res, wid, "schedAssign", w.Info) {
+				if !windows[wnd].Allocated.CanHandleRequest(task.SchedId, task.SealTask(), res, wid, "schedAssign", w.Info) {
 					continue
 				}

@@ -80,7 +80,7 @@ func SpreadTasksWS(queued bool) func(sh *Scheduler, queueLen int, acceptableWind
 				"assigned", bestAssigned)

 			workerAssigned[bestWid]++
-			windows[selectedWindow].Allocated.Add(task.SealTask(), info.Resources, needRes)
+			windows[selectedWindow].Allocated.Add(task.SchedId, task.SealTask(), info.Resources, needRes)
 			windows[selectedWindow].Todo = append(windows[selectedWindow].Todo, task)

 			rmQueue = append(rmQueue, sqi)
@@ -35,7 +35,7 @@ func LowestUtilizationWS(sh *Scheduler, queueLen int, acceptableWindows [][]int,
 			log.Debugf("SCHED try assign sqi:%d sector %d to window %d (awi:%d)", sqi, task.Sector.ID.Number, wnd, i)

 			// TODO: allow bigger windows
-			if !windows[wnd].Allocated.CanHandleRequest(task.SealTask(), res, wid, "schedAssign", w.Info) {
+			if !windows[wnd].Allocated.CanHandleRequest(task.SchedId, task.SealTask(), res, wid, "schedAssign", w.Info) {
 				continue
 			}

@@ -82,7 +82,7 @@ func LowestUtilizationWS(sh *Scheduler, queueLen int, acceptableWindows [][]int,
 			"worker", bestWid,
 			"utilization", bestUtilization)

-		workerUtil[bestWid] += windows[selectedWindow].Allocated.Add(task.SealTask(), info.Resources, needRes)
+		workerUtil[bestWid] += windows[selectedWindow].Allocated.Add(task.SchedId, task.SealTask(), info.Resources, needRes)
 		windows[selectedWindow].Todo = append(windows[selectedWindow].Todo, task)

 		rmQueue = append(rmQueue, sqi)
@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"

+	"github.com/google/uuid"
 	"github.com/hashicorp/go-multierror"
 	"golang.org/x/xerrors"

@@ -110,7 +111,7 @@ func (ps *poStScheduler) Schedule(ctx context.Context, primary bool, spt abi.Reg
 	for i, selected := range candidates {
 		worker := ps.workers[selected.id]

-		err := worker.active.withResources(selected.id, worker.Info, ps.postType.SealTask(spt), selected.res, &ps.lk, func() error {
+		err := worker.active.withResources(uuid.UUID{}, selected.id, worker.Info, ps.postType.SealTask(spt), selected.res, &ps.lk, func() error {
 			ps.lk.Unlock()
 			defer ps.lk.Lock()

@@ -148,7 +149,7 @@ func (ps *poStScheduler) readyWorkers(spt abi.RegisteredSealProof) (bool, []cand
 			continue
 		}

-		if !wr.active.CanHandleRequest(ps.postType.SealTask(spt), needRes, wid, "post-readyWorkers", wr.Info) {
+		if !wr.active.CanHandleRequest(uuid.UUID{}, ps.postType.SealTask(spt), needRes, wid, "post-readyWorkers", wr.Info) {
 			continue
 		}

@@ -3,6 +3,8 @@ package sealer
 import (
 	"sync"

+	"github.com/google/uuid"
+
 	"github.com/filecoin-project/lotus/storage/sealer/sealtasks"
 	"github.com/filecoin-project/lotus/storage/sealer/storiface"
 )
@@ -20,7 +22,7 @@ type ActiveResources struct {
 }

 type taskCounter struct {
-	taskCounters map[sealtasks.SealTaskType]int
+	taskCounters map[sealtasks.SealTaskType]map[uuid.UUID]int

 	// this lock is technically redundant, as ActiveResources is always accessed
 	// with the worker lock, but let's not panic if we ever change that
@@ -29,34 +31,48 @@ type taskCounter struct {

 func newTaskCounter() *taskCounter {
 	return &taskCounter{
-		taskCounters: map[sealtasks.SealTaskType]int{},
+		taskCounters: make(map[sealtasks.SealTaskType]map[uuid.UUID]int),
 	}
 }

-func (tc *taskCounter) Add(tt sealtasks.SealTaskType) {
+func (tc *taskCounter) Add(tt sealtasks.SealTaskType, schedID uuid.UUID) {
 	tc.lk.Lock()
 	defer tc.lk.Unlock()
-	tc.taskCounters[tt]++
+	tc.getUnlocked(tt)[schedID]++
 }

-func (tc *taskCounter) Free(tt sealtasks.SealTaskType) {
+func (tc *taskCounter) Free(tt sealtasks.SealTaskType, schedID uuid.UUID) {
 	tc.lk.Lock()
 	defer tc.lk.Unlock()
-	tc.taskCounters[tt]--
+	m := tc.getUnlocked(tt)
+	if m[schedID] <= 1 {
+		delete(m, schedID)
+	} else {
+		m[schedID]--
+	}
 }

-func (tc *taskCounter) Get(tt sealtasks.SealTaskType) int {
-	tc.lk.Lock()
-	defer tc.lk.Unlock()
+func (tc *taskCounter) getUnlocked(tt sealtasks.SealTaskType) map[uuid.UUID]int {
+	if tc.taskCounters[tt] == nil {
+		tc.taskCounters[tt] = make(map[uuid.UUID]int)
+	}
+
 	return tc.taskCounters[tt]
 }

+func (tc *taskCounter) Get(tt sealtasks.SealTaskType) map[uuid.UUID]int {
+	tc.lk.Lock()
+	defer tc.lk.Unlock()
+
+	return tc.getUnlocked(tt)
+}
+
 func (tc *taskCounter) Sum() int {
 	tc.lk.Lock()
 	defer tc.lk.Unlock()
 	sum := 0
 	for _, v := range tc.taskCounters {
-		sum += v
+		sum += len(v)
 	}
 	return sum
 }
@@ -64,8 +80,8 @@ func (tc *taskCounter) Sum() int {
 func (tc *taskCounter) ForEach(cb func(tt sealtasks.SealTaskType, count int)) {
 	tc.lk.Lock()
 	defer tc.lk.Unlock()
-	for tt, count := range tc.taskCounters {
-		cb(tt, count)
+	for tt, v := range tc.taskCounters {
+		cb(tt, len(v))
 	}
 }

@@ -75,8 +91,8 @@ func NewActiveResources(tc *taskCounter) *ActiveResources {
 	}
 }

-func (a *ActiveResources) withResources(id storiface.WorkerID, wr storiface.WorkerInfo, tt sealtasks.SealTaskType, r storiface.Resources, locker sync.Locker, cb func() error) error {
-	for !a.CanHandleRequest(tt, r, id, "withResources", wr) {
+func (a *ActiveResources) withResources(schedID uuid.UUID, id storiface.WorkerID, wr storiface.WorkerInfo, tt sealtasks.SealTaskType, r storiface.Resources, locker sync.Locker, cb func() error) error {
+	for !a.CanHandleRequest(schedID, tt, r, id, "withResources", wr) {
 		if a.cond == nil {
 			a.cond = sync.NewCond(locker)
 		}
@@ -85,11 +101,11 @@ func (a *ActiveResources) withResources(id storiface.WorkerID, wr storiface.Work
 		a.waiting--
 	}

-	a.Add(tt, wr.Resources, r)
+	a.Add(schedID, tt, wr.Resources, r)

 	err := cb()

-	a.Free(tt, wr.Resources, r)
+	a.Free(schedID, tt, wr.Resources, r)

 	return err
 }
@@ -100,7 +116,7 @@ func (a *ActiveResources) hasWorkWaiting() bool {
 }

 // add task resources to ActiveResources and return utilization difference
-func (a *ActiveResources) Add(tt sealtasks.SealTaskType, wr storiface.WorkerResources, r storiface.Resources) float64 {
+func (a *ActiveResources) Add(schedID uuid.UUID, tt sealtasks.SealTaskType, wr storiface.WorkerResources, r storiface.Resources) float64 {
 	startUtil := a.utilization(wr)

 	if r.GPUUtilization > 0 {
@@ -109,19 +125,19 @@ func (a *ActiveResources) Add(tt sealtasks.SealTaskType, wr storiface.WorkerReso
 	a.cpuUse += r.Threads(wr.CPUs, len(wr.GPUs))
 	a.memUsedMin += r.MinMemory
 	a.memUsedMax += r.MaxMemory
-	a.taskCounters.Add(tt)
+	a.taskCounters.Add(tt, schedID)

 	return a.utilization(wr) - startUtil
 }

-func (a *ActiveResources) Free(tt sealtasks.SealTaskType, wr storiface.WorkerResources, r storiface.Resources) {
+func (a *ActiveResources) Free(schedID uuid.UUID, tt sealtasks.SealTaskType, wr storiface.WorkerResources, r storiface.Resources) {
 	if r.GPUUtilization > 0 {
 		a.gpuUsed -= r.GPUUtilization
 	}
 	a.cpuUse -= r.Threads(wr.CPUs, len(wr.GPUs))
 	a.memUsedMin -= r.MinMemory
 	a.memUsedMax -= r.MaxMemory
-	a.taskCounters.Free(tt)
+	a.taskCounters.Free(tt, schedID)

 	if a.cond != nil {
 		a.cond.Broadcast()
@@ -130,9 +146,10 @@ func (a *ActiveResources) Free(tt sealtasks.SealTaskType, wr storiface.WorkerRes

 // CanHandleRequest evaluates if the worker has enough available resources to
 // handle the request.
-func (a *ActiveResources) CanHandleRequest(tt sealtasks.SealTaskType, needRes storiface.Resources, wid storiface.WorkerID, caller string, info storiface.WorkerInfo) bool {
+func (a *ActiveResources) CanHandleRequest(schedID uuid.UUID, tt sealtasks.SealTaskType, needRes storiface.Resources, wid storiface.WorkerID, caller string, info storiface.WorkerInfo) bool {
 	if needRes.MaxConcurrent > 0 {
-		if a.taskCounters.Get(tt) >= needRes.MaxConcurrent {
+		tasks := a.taskCounters.Get(tt)
+		if len(tasks) >= needRes.MaxConcurrent && (schedID == uuid.UUID{} || tasks[schedID] == 0) {
 			log.Debugf("sched: not scheduling on worker %s for %s; at task limit tt=%s, curcount=%d", wid, caller, tt, a.taskCounters.Get(tt))
 			return false
 		}
@@ -226,7 +243,7 @@ func (a *ActiveResources) taskCount(tt *sealtasks.SealTaskType) int {
 		return a.taskCounters.Sum()
 	}

-	return a.taskCounters.Get(*tt)
+	return len(a.taskCounters.Get(*tt))
 }

 func (wh *WorkerHandle) Utilization() float64 {
@@ -698,7 +698,7 @@ func TestWindowCompact(t *testing.T) {
 						TaskType: task,
 						Sector:   storiface.SectorRef{ProofType: spt},
 					})
-					window.Allocated.Add(task.SealTask(spt), wh.Info.Resources, storiface.ResourceTable[task][spt])
+					window.Allocated.Add(uuid.UUID{}, task.SealTask(spt), wh.Info.Resources, storiface.ResourceTable[task][spt])
 				}

 				wh.activeWindows = append(wh.activeWindows, window)
@@ -717,7 +717,7 @@ func TestWindowCompact(t *testing.T) {

 				for ti, task := range tasks {
 					require.Equal(t, task, wh.activeWindows[wi].Todo[ti].TaskType, "%d, %d", wi, ti)
-					expectRes.Add(task.SealTask(spt), wh.Info.Resources, storiface.ResourceTable[task][spt])
+					expectRes.Add(uuid.UUID{}, task.SealTask(spt), wh.Info.Resources, storiface.ResourceTable[task][spt])
 				}

 				require.Equal(t, expectRes.cpuUse, wh.activeWindows[wi].Allocated.cpuUse, "%d", wi)
@@ -294,14 +294,14 @@ func (sw *schedWorker) workerCompactWindows() {

 			for ti, todo := range window.Todo {
 				needRes := worker.Info.Resources.ResourceSpec(todo.Sector.ProofType, todo.TaskType)
-				if !lower.Allocated.CanHandleRequest(todo.SealTask(), needRes, sw.wid, "compactWindows", worker.Info) {
+				if !lower.Allocated.CanHandleRequest(todo.SchedId, todo.SealTask(), needRes, sw.wid, "compactWindows", worker.Info) {
 					continue
 				}

 				moved = append(moved, ti)
 				lower.Todo = append(lower.Todo, todo)
-				lower.Allocated.Add(todo.SealTask(), worker.Info.Resources, needRes)
-				window.Allocated.Free(todo.SealTask(), worker.Info.Resources, needRes)
+				lower.Allocated.Add(todo.SchedId, todo.SealTask(), worker.Info.Resources, needRes)
+				window.Allocated.Free(todo.SchedId, todo.SealTask(), worker.Info.Resources, needRes)
 			}

 			if len(moved) > 0 {
@@ -355,7 +355,7 @@ assignLoop:
 			worker.lk.Lock()
 			for t, todo := range firstWindow.Todo {
 				needResPrep := worker.Info.Resources.PrepResourceSpec(todo.Sector.ProofType, todo.TaskType, todo.prepare.PrepType)
-				if worker.preparing.CanHandleRequest(todo.PrepSealTask(), needResPrep, sw.wid, "startPreparing", worker.Info) {
+				if worker.preparing.CanHandleRequest(todo.SchedId, todo.PrepSealTask(), needResPrep, sw.wid, "startPreparing", worker.Info) {
 					tidx = t
 					break
 				}
@@ -416,7 +416,7 @@ assignLoop:
 				}

 				needRes := worker.Info.Resources.ResourceSpec(todo.Sector.ProofType, todo.TaskType)
-				if worker.active.CanHandleRequest(todo.SealTask(), needRes, sw.wid, "startPreparing", worker.Info) {
+				if worker.active.CanHandleRequest(todo.SchedId, todo.SealTask(), needRes, sw.wid, "startPreparing", worker.Info) {
 					tidx = t
 					break
 				}
@@ -457,7 +457,7 @@ func (sw *schedWorker) startProcessingTask(req *WorkerRequest) error {
 	needResPrep := w.Info.Resources.PrepResourceSpec(req.Sector.ProofType, req.TaskType, req.prepare.PrepType)

 	w.lk.Lock()
-	w.preparing.Add(req.PrepSealTask(), w.Info.Resources, needResPrep)
+	w.preparing.Add(req.SchedId, req.PrepSealTask(), w.Info.Resources, needResPrep)
 	w.lk.Unlock()

 	go func() {
@@ -468,7 +468,7 @@ func (sw *schedWorker) startProcessingTask(req *WorkerRequest) error {
 		w.lk.Lock()

 		if err != nil {
-			w.preparing.Free(req.PrepSealTask(), w.Info.Resources, needResPrep)
+			w.preparing.Free(req.SchedId, req.PrepSealTask(), w.Info.Resources, needResPrep)
 			w.lk.Unlock()

 			select {
@@ -497,11 +497,12 @@ func (sw *schedWorker) startProcessingTask(req *WorkerRequest) error {
 		}()

 		// wait (if needed) for resources in the 'active' window
-		err = w.active.withResources(sw.wid, w.Info, req.SealTask(), needRes, &w.lk, func() error {
-			w.preparing.Free(req.PrepSealTask(), w.Info.Resources, needResPrep)
+		err = w.active.withResources(req.SchedId, sw.wid, w.Info, req.SealTask(), needRes, &w.lk, func() error {
+			w.preparing.Free(req.SchedId, req.PrepSealTask(), w.Info.Resources, needResPrep)
 			w.lk.Unlock()
 			defer w.lk.Lock() // we MUST return locked from this function

+			// make sure the worker loop sees that the prepare task has finished
 			select {
 			case sw.taskDone <- struct{}{}:
 			case <-sh.closing:
@@ -525,6 +526,12 @@ func (sw *schedWorker) startProcessingTask(req *WorkerRequest) error {

 		w.lk.Unlock()

+		// make sure the worker loop sees that the task has finished
+		select {
+		case sw.taskDone <- struct{}{}:
+		default: // there is a notification pending already
+		}
+
 		// This error should always be nil, since nothing is setting it, but just to be safe:
 		if err != nil {
 			log.Errorf("error executing worker (withResources): %+v", err)
@@ -539,7 +546,7 @@ func (sw *schedWorker) startProcessingReadyTask(req *WorkerRequest) error {

 	needRes := w.Info.Resources.ResourceSpec(req.Sector.ProofType, req.TaskType)

-	w.active.Add(req.SealTask(), w.Info.Resources, needRes)
+	w.active.Add(req.SchedId, req.SealTask(), w.Info.Resources, needRes)

 	go func() {
 		// Do the work!
@@ -557,7 +564,7 @@ func (sw *schedWorker) startProcessingReadyTask(req *WorkerRequest) error {

 		w.lk.Lock()

-		w.active.Free(req.SealTask(), w.Info.Resources, needRes)
+		w.active.Free(req.SchedId, req.SealTask(), w.Info.Resources, needRes)

 		select {
 		case sw.taskDone <- struct{}{}:
Author	SHA1	Message	Date
Shrenuj Bansal	05d3dc811a	Merge pull request #10934 from filecoin-project/sbansal/backport-v1.23.1-rc3 feat: chainstore: sharded mutex for filling chain height index	2023-05-31 13:38:21 -04:00
Jakub Sztandera	80aa96ec25	Appease the linter Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>	2023-05-31 12:48:34 -04:00
Jakub Sztandera	b65c93b2b4	feat: chainstore: sharded mutex for filling chain height index This PR introduces as sharded mutex within the ChainIndex#GetTipsetByHeight. It also replaces a go map with xsync.Map which doesn't require locking. The lock is taken when it appears that ChainIndex filling work should be started. After claiming the lock, the status of the cache is rechecked, if the entry is still missing, the fillCache is started. Thanks to @snissn and @arajasek for debugging and taking initial stabs at this. Supersedes #10866 and 10885 Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>	2023-05-31 12:48:25 -04:00
Jiaying Wang	9900dbe1de	Merge pull request #10918 from filecoin-project/asr/backport Check if epoch is negative in GetTipsetByHeight	2023-05-25 19:28:43 -04:00
Shrenuj Bansal	a509ca23ec	Check if epoch is negative in GetTipsetByHeight	2023-05-24 12:38:44 -04:00
Shrenuj Bansal	5a1738c9ba	Merge pull request #10853 from filecoin-project/sbansal/1.23.1-rc2 Changelog and build version for 1.23.1-rc2	2023-05-10 17:38:27 -04:00
Shrenuj Bansal	5aea90776c	update changelog	2023-05-10 16:25:05 -04:00
Shrenuj Bansal	67df60f26f	Changelog and build version for 1.23.1-rc2	2023-05-10 16:14:42 -04:00
Shrenuj Bansal	86f32d1c1f	Merge pull request #10851 from filecoin-project/sbansal/backport-1.23.1-rc2 Backport changes for 1.23.1-rc2	2023-05-10 16:04:48 -04:00
Łukasz MagieraandSteffen Butzer	60d576241c	fix: sched: Address GET_32G_MAX_CONCURRENT regression (#10850 ) * Fix 1.21 regression: GET_32G_MAX_CONCURRENT + mixed prepared/executing leads to stuck scheduler If you have 12 GET tasks and GET_32G_MAX_CONCURRENT=1, sealing jobs will only show assigned tasks for GET of the miner and is stuck. I believe this to be a regression of 1.21 unifying the counters, in the case of GETs where PrepType and TaskType both being seal/v0/fetch leading to a state where tasks are blocked since already counted towards the limit. * itests: Repro issue from PR #10633 * make counters int (non-working) * fix: worker sched: Send taskDone notifs after tasks are done * itests: Make TestPledgeMaxConcurrentGet actually reproduce the issue * make the linter happy --------- Co-authored-by: Steffen Butzer <steffen.butzer@outlook.com>	2023-05-10 15:45:50 -04:00
Phi	f03c06964f	Make docsgen-cli Make docsgen-cli	2023-05-10 15:27:45 -04:00
Phi	4e49b8b671	Hide lotus-miner legacy markets cmds Hide lotus-miner legacy markets cmds	2023-05-10 15:27:39 -04:00
Łukasz Magiera	0cfb69b95a	ci: run release on ci/ branches, include m1	2023-05-10 15:21:34 -04:00
Łukasz Magiera	9558df7cda	temp drop m1 from releases	2023-05-10 15:21:27 -04:00
Łukasz Magiera	779975c3a9	ci: Debugging m1 build	2023-05-10 15:21:19 -04:00
Shrenuj Bansal	e8dcf9f838	update cl (#10819 )	2023-05-03 19:07:15 -04:00
Shrenuj Bansal	a03535ae20	Update changelog.md for release/v1.23.1	2023-05-03 18:32:10 -04:00
Shrenuj Bansal	fadc94f253	Update build version for release/v1.23.1	2023-05-03 17:20:57 -04:00