Pull gossip queues for better throughput

packages/api/src/beacon/routes/lodestar.ts

@@ -81,7 +81,7 @@ export type Api = {
   /** TODO: description */
   getSyncChainsDebugState(): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: SyncChainDebugState[]}}>>;
   /** Dump all items in a gossip queue, by gossipType */
-  getGossipQueueItems(gossipType: string): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: GossipQueueItem[]}}>>;
+  getGossipQueueItems(gossipType: string): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: unknown[]}}>>;
   /** Dump all items in the regen queue */
   getRegenQueueItems(): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: RegenQueueItem[]}}>>;
   /** Dump all items in the block processor queue */

packages/beacon-node/src/api/impl/beacon/pool/index.ts

@@ -9,7 +9,7 @@ import {validateBlsToExecutionChange} from "../../../../chain/validation/blsToEx
 import {validateSyncCommitteeSigOnly} from "../../../../chain/validation/syncCommittee.js";
 import {ApiModules} from "../../types.js";
 import {AttestationError, GossipAction, SyncCommitteeError} from "../../../../chain/errors/index.js";
-import {validateGossipFnRetryUnknownRoot} from "../../../../network/gossip/handlers/index.js";
+import {validateGossipFnRetryUnknownRoot} from "../../../../network/processor/gossipHandlers.js";
 
 export function getBeaconPoolApi({
   chain,
@@ -53,20 +53,20 @@ export function getBeaconPoolApi({
         attestations.map(async (attestation, i) => {
           try {
             // eslint-disable-next-line @typescript-eslint/explicit-function-return-type
-            const validateFn = () => validateGossipAttestation(chain, attestation, null);
+            const validateFn = () => validateGossipAttestation(chain, attestation, null, null);
             const {slot, beaconBlockRoot} = attestation.data;
             // when a validator is configured with multiple beacon node urls, this attestation data may come from another beacon node
             // and the block hasn't been in our forkchoice since we haven't seen / processing that block
             // see https://github.com/ChainSafe/lodestar/issues/5098
-            const {indexedAttestation, subnet} = await validateGossipFnRetryUnknownRoot(
+            const {indexedAttestation, subnet, attDataRootHex} = await validateGossipFnRetryUnknownRoot(
               validateFn,
               chain,
               slot,
               beaconBlockRoot
             );
 
             if (network.attnetsService.shouldProcess(subnet, slot)) {
-              const insertOutcome = chain.attestationPool.add(attestation);
+              const insertOutcome = chain.attestationPool.add(attestation, attDataRootHex);
               metrics?.opPool.attestationPoolInsertOutcome.inc({insertOutcome});
             }
             const sentPeers = await network.gossip.publishBeaconAttestation(attestation, subnet);

packages/beacon-node/src/api/impl/lodestar/index.ts

@@ -60,7 +60,7 @@ export function getLodestarApi({
 
     async getGossipQueueItems(gossipType: GossipType | string) {
       return {
-        data: await network.dumpGossipQueueItems(gossipType),
+        data: await network.dumpGossipQueue(gossipType as GossipType),
       };
     },
 

packages/beacon-node/src/api/impl/validator/index.ts

@@ -25,7 +25,7 @@ import {CommitteeSubscription} from "../../../network/subnets/index.js";
 import {ApiModules} from "../types.js";
 import {RegenCaller} from "../../../chain/regen/index.js";
 import {getValidatorStatus} from "../beacon/state/utils.js";
-import {validateGossipFnRetryUnknownRoot} from "../../../network/gossip/handlers/index.js";
+import {validateGossipFnRetryUnknownRoot} from "../../../network/processor/gossipHandlers.js";
 import {computeSubnetForCommitteesAtSlot, getPubkeysForIndices} from "./utils.js";
 
 /**
@@ -541,7 +541,7 @@ export function getValidatorApi({
             // when a validator is configured with multiple beacon node urls, this attestation may come from another beacon node
             // and the block hasn't been in our forkchoice since we haven't seen / processing that block
             // see https://github.com/ChainSafe/lodestar/issues/5098
-            const {indexedAttestation, committeeIndices} = await validateGossipFnRetryUnknownRoot(
+            const {indexedAttestation, committeeIndices, attDataRootHex} = await validateGossipFnRetryUnknownRoot(
               validateFn,
               chain,
               slot,
@@ -550,6 +550,7 @@ export function getValidatorApi({
 
             chain.aggregatedAttestationPool.add(
               signedAggregateAndProof.message.aggregate,
+              attDataRootHex,
               indexedAttestation.attestingIndices.length,
               committeeIndices
             );

packages/beacon-node/src/chain/bls/interface.ts

@@ -43,4 +43,9 @@ export interface IBlsVerifier {
 
   /** For multithread pool awaits terminating all workers */
   close(): Promise<void>;
+
+  /**
+   * Returns true if BLS worker pool is ready to accept more work jobs.
+   */
+  canAcceptWork(): boolean;
 }

packages/beacon-node/src/chain/bls/multithread/index.ts

@@ -56,6 +56,11 @@ const MAX_BUFFERED_SIGS = 32;
  */
 const MAX_BUFFER_WAIT_MS = 100;
 
+/**
+ * Max concurrent jobs on `canAcceptWork` status
+ */
+const MAX_JOBS_CAN_ACCEPT_WORK = 512;
+
 type WorkerApi = {
   verifyManySignatureSets(workReqArr: BlsWorkReq[]): Promise<BlsWorkResult>;
 };
@@ -110,6 +115,7 @@ export class BlsMultiThreadWorkerPool implements IBlsVerifier {
   } | null = null;
   private blsVerifyAllMultiThread: boolean;
   private closed = false;
+  private workersBusy = 0;
 
   constructor(options: BlsMultiThreadWorkerPoolOptions, modules: BlsMultiThreadWorkerPoolModules) {
     const {logger, metrics} = modules;
@@ -127,10 +133,21 @@ export class BlsMultiThreadWorkerPool implements IBlsVerifier {
     this.workers = this.createWorkers(implementation, defaultPoolSize);
 
     if (metrics) {
-      metrics.blsThreadPool.queueLength.addCollect(() => metrics.blsThreadPool.queueLength.set(this.jobs.length));
+      metrics.blsThreadPool.queueLength.addCollect(() => {
+        metrics.blsThreadPool.queueLength.set(this.jobs.length);
+        metrics.blsThreadPool.workersBusy.set(this.workersBusy);
+      });
     }
   }
 
+  canAcceptWork(): boolean {
+    return (
+      this.workersBusy < defaultPoolSize &&
+      // TODO: Should also bound the jobs queue?
+      this.jobs.length < MAX_JOBS_CAN_ACCEPT_WORK
+    );
+  }
+
   async verifySignatureSets(sets: ISignatureSet[], opts: VerifySignatureOpts = {}): Promise<boolean> {
     // Pubkeys are aggregated in the main thread regardless if verified in workers or in main thread
     this.metrics?.bls.aggregatedPubkeys.inc(getAggregatedPubkeysCount(sets));
@@ -310,6 +327,7 @@ export class BlsMultiThreadWorkerPool implements IBlsVerifier {
 
     const workerApi = worker.status.workerApi;
     worker.status = {code: WorkerStatusCode.running, workerApi};
+    this.workersBusy++;
 
     try {
       let startedSigSets = 0;
@@ -375,6 +393,7 @@ export class BlsMultiThreadWorkerPool implements IBlsVerifier {
     }
 
     worker.status = {code: WorkerStatusCode.idle, workerApi};
+    this.workersBusy--;
 
     // Potentially run a new job
     setTimeout(this.runJob, 0);

packages/beacon-node/src/chain/bls/singleThread.ts

@@ -37,4 +37,9 @@ export class BlsSingleThreadVerifier implements IBlsVerifier {
   async close(): Promise<void> {
     // nothing to do
   }
+
+  canAcceptWork(): boolean {
+    // Since sigs are verified blocking the main thread, there's no mechanism to throttle
+    return true;
+  }
 }

packages/beacon-node/src/chain/chain.ts

@@ -34,7 +34,7 @@ import {BeaconClock, LocalClock} from "./clock/index.js";
 import {ChainEventEmitter, ChainEvent} from "./emitter.js";
 import {IBeaconChain, ProposerPreparationData} from "./interface.js";
 import {IChainOptions} from "./options.js";
-import {IStateRegenerator, QueuedStateRegenerator, RegenCaller} from "./regen/index.js";
+import {QueuedStateRegenerator, RegenCaller} from "./regen/index.js";
 import {initializeForkChoice} from "./forkChoice/index.js";
 import {computeAnchorCheckpoint} from "./initState.js";
 import {IBlsVerifier, BlsSingleThreadVerifier, BlsMultiThreadWorkerPool} from "./bls/index.js";
@@ -64,6 +64,7 @@ import {AssembledBlockType, BlobsResultType, BlockType} from "./produceBlock/ind
 import {BlockAttributes, produceBlockBody} from "./produceBlock/produceBlockBody.js";
 import {computeNewStateRoot} from "./produceBlock/computeNewStateRoot.js";
 import {BlockInput} from "./blocks/types.js";
+import {SeenAttestationDatas} from "./seenCache/seenAttestationData.js";
 
 /**
  * Arbitrary constants, blobs should be consumed immediately in the same slot they are produced.
@@ -91,7 +92,7 @@ export class BeaconChain implements IBeaconChain {
   readonly emitter: ChainEventEmitter;
   readonly stateCache: StateContextCache;
   readonly checkpointStateCache: CheckpointStateCache;
-  readonly regen: IStateRegenerator;
+  readonly regen: QueuedStateRegenerator;
   readonly lightClientServer: LightClientServer;
   readonly reprocessController: ReprocessController;
 
@@ -109,6 +110,7 @@ export class BeaconChain implements IBeaconChain {
   readonly seenBlockProposers = new SeenBlockProposers();
   readonly seenSyncCommitteeMessages = new SeenSyncCommitteeMessages();
   readonly seenContributionAndProof: SeenContributionAndProof;
+  readonly seenAttestationDatas: SeenAttestationDatas;
   // Seen cache for liveness checks
   readonly seenBlockAttesters = new SeenBlockAttesters();
 
@@ -195,6 +197,7 @@ export class BeaconChain implements IBeaconChain {
 
     this.seenAggregatedAttestations = new SeenAggregatedAttestations(metrics);
     this.seenContributionAndProof = new SeenContributionAndProof(metrics);
+    this.seenAttestationDatas = new SeenAttestationDatas(metrics, this.opts?.attDataCacheSlotDistance);
 
     this.beaconProposerCache = new BeaconProposerCache(opts);
     this.checkpointBalancesCache = new CheckpointBalancesCache();
@@ -281,6 +284,14 @@ export class BeaconChain implements IBeaconChain {
     await this.bls.close();
   }
 
+  regenCanAcceptWork(): boolean {
+    return this.regen.canAcceptWork();
+  }
+
+  blsThreadPoolCanAcceptWork(): boolean {
+    return this.bls.canAcceptWork();
+  }
+
   validatorSeenAtEpoch(index: ValidatorIndex, epoch: Epoch): boolean {
     // Caller must check that epoch is not older that current epoch - 1
     // else the caches for that epoch may already be pruned.
@@ -632,6 +643,7 @@ export class BeaconChain implements IBeaconChain {
     this.aggregatedAttestationPool.prune(slot);
     this.syncCommitteeMessagePool.prune(slot);
     this.seenSyncCommitteeMessages.prune(slot);
+    this.seenAttestationDatas.onSlot(slot);
     this.reprocessController.onSlot(slot);
 
     if (isFinite(this.config.BELLATRIX_FORK_EPOCH) && slot % this.exchangeTransitionConfigurationEverySlots === 0) {

packages/beacon-node/src/chain/interface.ts

@@ -1,5 +1,5 @@
 import {allForks, UintNum64, Root, phase0, Slot, RootHex, Epoch, ValidatorIndex, deneb, Wei} from "@lodestar/types";
-import {CachedBeaconStateAllForks} from "@lodestar/state-transition";
+import {CachedBeaconStateAllForks, Index2PubkeyCache, PubkeyIndexMap} from "@lodestar/state-transition";
 import {BeaconConfig} from "@lodestar/config";
 import {CompositeTypeAny, TreeView, Type} from "@chainsafe/ssz";
 import {Logger} from "@lodestar/utils";
@@ -31,6 +31,7 @@ import {SeenBlockAttesters} from "./seenCache/seenBlockAttesters.js";
 import {CheckpointBalancesCache} from "./balancesCache.js";
 import {IChainOptions} from "./options.js";
 import {AssembledBlockType, BlockAttributes, BlockType} from "./produceBlock/produceBlockBody.js";
+import {SeenAttestationDatas} from "./seenCache/seenAttestationData.js";
 
 export type Eth2Context = {
   activeValidatorCount: number;
@@ -68,6 +69,8 @@ export interface IBeaconChain {
   readonly regen: IStateRegenerator;
   readonly lightClientServer: LightClientServer;
   readonly reprocessController: ReprocessController;
+  readonly pubkey2index: PubkeyIndexMap;
+  readonly index2pubkey: Index2PubkeyCache;
 
   // Ops pool
   readonly attestationPool: AttestationPool;
@@ -83,6 +86,7 @@ export interface IBeaconChain {
   readonly seenBlockProposers: SeenBlockProposers;
   readonly seenSyncCommitteeMessages: SeenSyncCommitteeMessages;
   readonly seenContributionAndProof: SeenContributionAndProof;
+  readonly seenAttestationDatas: SeenAttestationDatas;
   // Seen cache for liveness checks
   readonly seenBlockAttesters: SeenBlockAttesters;
 
@@ -133,6 +137,9 @@ export interface IBeaconChain {
   /** Persist bad items to persistInvalidSszObjectsDir dir, for example invalid state, attestations etc. */
   persistInvalidSszView(view: TreeView<CompositeTypeAny>, suffix?: string): void;
   updateBuilderStatus(clockSlot: Slot): void;
+
+  regenCanAcceptWork(): boolean;
+  blsThreadPoolCanAcceptWork(): boolean;
 }
 
 export type SSZObjectType =

packages/beacon-node/src/chain/opPools/aggregatedAttestationPool.ts

@@ -6,7 +6,7 @@ import {
   SLOTS_PER_EPOCH,
   TIMELY_SOURCE_FLAG_INDEX,
 } from "@lodestar/params";
-import {phase0, Epoch, Slot, ssz, ValidatorIndex} from "@lodestar/types";
+import {phase0, Epoch, Slot, ssz, ValidatorIndex, RootHex} from "@lodestar/types";
 import {
   CachedBeaconStateAllForks,
   CachedBeaconStatePhase0,
@@ -70,7 +70,12 @@ export class AggregatedAttestationPool {
     return {attestationCount, attestationDataCount};
   }
 
-  add(attestation: phase0.Attestation, attestingIndicesCount: number, committee: ValidatorIndex[]): InsertOutcome {
+  add(
+    attestation: phase0.Attestation,
+    dataRootHex: RootHex,
+    attestingIndicesCount: number,
+    committee: ValidatorIndex[]
+  ): InsertOutcome {
     const slot = attestation.data.slot;
     const lowestPermissibleSlot = this.lowestPermissibleSlot;
 
@@ -80,9 +85,6 @@ export class AggregatedAttestationPool {
     }
 
     const attestationGroupByDataHash = this.attestationGroupByDataHashBySlot.getOrDefault(slot);
-    const dataRoot = ssz.phase0.AttestationData.hashTreeRoot(attestation.data);
-    const dataRootHex = toHexString(dataRoot);
-
     let attestationGroup = attestationGroupByDataHash.get(dataRootHex);
     if (!attestationGroup) {
       attestationGroup = new MatchingDataAttestationGroup(committee, attestation.data);

packages/beacon-node/src/chain/opPools/attestationPool.ts

@@ -1,4 +1,4 @@
-import {phase0, Slot, Root, ssz} from "@lodestar/types";
+import {phase0, Slot, Root, RootHex} from "@lodestar/types";
 import {PointFormat, Signature} from "@chainsafe/bls/types";
 import bls from "@chainsafe/bls";
 import {BitArray, toHexString} from "@chainsafe/ssz";
@@ -93,7 +93,7 @@ export class AttestationPool {
    * - Valid committeeIndex
    * - Valid data
    */
-  add(attestation: phase0.Attestation): InsertOutcome {
+  add(attestation: phase0.Attestation, attDataRootHex: RootHex): InsertOutcome {
     const slot = attestation.data.slot;
     const lowestPermissibleSlot = this.lowestPermissibleSlot;
 
@@ -113,17 +113,14 @@ export class AttestationPool {
       throw new OpPoolError({code: OpPoolErrorCode.REACHED_MAX_PER_SLOT});
     }
 
-    const dataRoot = ssz.phase0.AttestationData.hashTreeRoot(attestation.data);
-    const dataRootHex = toHexString(dataRoot);
-
     // Pre-aggregate the contribution with existing items
-    const aggregate = aggregateByRoot.get(dataRootHex);
+    const aggregate = aggregateByRoot.get(attDataRootHex);
     if (aggregate) {
       // Aggregate mutating
       return aggregateAttestationInto(aggregate, attestation);
     } else {
       // Create new aggregate
-      aggregateByRoot.set(dataRootHex, attestationToAggregate(attestation));
+      aggregateByRoot.set(attDataRootHex, attestationToAggregate(attestation));
       return InsertOutcome.NewData;
     }
   }

packages/beacon-node/src/chain/opPools/types.ts

@@ -11,6 +11,8 @@ export enum InsertOutcome {
   AlreadyKnown = "AlreadyKnown",
   /** Not existing in the pool but it's too old to add. No changes were made. */
   Old = "Old",
+  /** The pool has reached its limit. No changes were made. */
+  ReachLimit = "ReachLimit",
   /** Attestation comes to the pool at > 2/3 of slot. No changes were made */
   Late = "Late",
   /** The data is know, and the new participants have been added to the aggregated signature */

packages/beacon-node/src/chain/options.ts

@@ -6,6 +6,7 @@ import {LightClientServerOpts} from "./lightClient/index.js";
 
 export type IChainOptions = BlockProcessOpts &
   PoolOpts &
+  SeenCacheOpts &
   ForkChoiceOpts &
   ArchiverOpts &
   LightClientServerOpts & {
@@ -55,6 +56,13 @@ export type PoolOpts = {
   preaggregateSlotDistance?: number;
 };
 
+export type SeenCacheOpts = {
+  /**
+   * Slot distance from current slot to cache AttestationData
+   */
+  attDataCacheSlotDistance?: number;
+};
+
 export const defaultChainOptions: IChainOptions = {
   blsVerifyAllMainThread: false,
   blsVerifyAllMultiThread: false,

packages/beacon-node/src/chain/regen/queued.ts

@@ -10,6 +10,8 @@ import {StateRegenerator, RegenModules} from "./regen.js";
 import {RegenError, RegenErrorCode} from "./errors.js";
 
 const REGEN_QUEUE_MAX_LEN = 256;
+// TODO: Should this constant be lower than above? 256 feels high
+const REGEN_CAN_ACCEPT_WORK_THRESHOLD = 16;
 
 type QueuedStateRegeneratorModules = RegenModules & {
   signal: AbortSignal;
@@ -46,6 +48,10 @@ export class QueuedStateRegenerator implements IStateRegenerator {
     this.metrics = modules.metrics;
   }
 
+  canAcceptWork(): boolean {
+    return this.jobQueue.jobLen < REGEN_CAN_ACCEPT_WORK_THRESHOLD;
+  }
+
   /**
    * Get the state to run with `block`.
    * - State after `block.parentRoot` dialed forward to block.slot