feat: persist network thread profile

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

@@ -77,6 +77,11 @@ export type LodestarNodePeer = NodePeer & {
 export type Api = {
  /** Trigger to write a heapdump to disk at `dirpath`. May take > 1min */
  writeHeapdump(dirpath?: string): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: {filepath: string}}}>>;
+ /** Trigger to write 10m network thread profile to disk */
+ writeNetworkThreadProfile(
+ duration?: number,
+ dirpath?: string
+ ): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: {filepath: string}}}>>;
  /** TODO: description */
  getLatestWeakSubjectivityCheckpointEpoch(): Promise<ApiClientResponse<{[HttpStatusCode.OK]: {data: Epoch}}>>;
  /** TODO: description */
@@ -125,6 +130,7 @@ export type Api = {
  */
 export const routesData: RoutesData<Api> = {
  writeHeapdump: {url: "/eth/v1/lodestar/writeheapdump", method: "POST"},
+ writeNetworkThreadProfile: {url: "/eth/v1/lodestar/write_network_thread_profile", method: "POST"},
  getLatestWeakSubjectivityCheckpointEpoch: {url: "/eth/v1/lodestar/ws_epoch", method: "GET"},
  getSyncChainsDebugState: {url: "/eth/v1/lodestar/sync-chains-debug-state", method: "GET"},
  getGossipQueueItems: {url: "/eth/v1/lodestar/gossip-queue-items/:gossipType", method: "GET"},
@@ -145,6 +151,7 @@ export const routesData: RoutesData<Api> = {
 
 export type ReqTypes = {
  writeHeapdump: {query: {dirpath?: string}};
+ writeNetworkThreadProfile: {query: {duration?: number; dirpath?: string}};
  getLatestWeakSubjectivityCheckpointEpoch: ReqEmpty;
  getSyncChainsDebugState: ReqEmpty;
  getGossipQueueItems: {params: {gossipType: string}};
@@ -170,6 +177,11 @@ export function getReqSerializers(): ReqSerializers<Api, ReqTypes> {
  parseReq: ({query}) => [query.dirpath],
  schema: {query: {dirpath: Schema.String}},
  },
+ writeNetworkThreadProfile: {
+ writeReq: (duration, dirpath) => ({query: {duration, dirpath}}),
+ parseReq: ({query}) => [query.duration, query.dirpath],
+ schema: {query: {dirpath: Schema.String}},
+ },
  getLatestWeakSubjectivityCheckpointEpoch: reqEmpty,
  getSyncChainsDebugState: reqEmpty,
  getGossipQueueItems: {
@@ -212,6 +224,7 @@ export function getReqSerializers(): ReqSerializers<Api, ReqTypes> {
 export function getReturnTypes(): ReturnTypes<Api> {
  return {
  writeHeapdump: sameType(),
+ writeNetworkThreadProfile: sameType(),
  getLatestWeakSubjectivityCheckpointEpoch: sameType(),
  getSyncChainsDebugState: jsonType("snake"),
  getGossipQueueItems: jsonType("snake"),

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

@@ -19,6 +19,7 @@ export function getLodestarApi({
  sync,
 }: Pick<ApiModules, "chain" | "config" | "db" | "network" | "sync">): ServerApi<routes.lodestar.Api> {
  let writingHeapdump = false;
+ let writingNetworkProfile = false;
 
  return {
  async writeHeapdump(dirpath = ".") {
@@ -47,6 +48,19 @@ export function getLodestarApi({
  }
  },
 
+ async writeNetworkThreadProfile(durationMs?: number, dirpath?: string) {
+ if (writingNetworkProfile) {
+ throw Error("Already writing network profile");
+ }
+
+ try {
+ const filepath = await network.writeNetworkThreadProfile(durationMs, dirpath);
+ return {data: {filepath}};
+ } finally {
+ writingNetworkProfile = false;
+ }
+ },
+
  async getLatestWeakSubjectivityCheckpointEpoch() {
  const state = chain.getHeadState();
  return {data: getLatestWeakSubjectivityCheckpointEpoch(config, state)};

packages/beacon-node/src/network/core/networkCore.ts

@@ -412,6 +412,10 @@ export class NetworkCore implements INetworkCore {
  return meshPeers;
  }
 
+ async writeNetworkThreadProfile(): Promise<string> {
+ throw new Error("Method not implemented, please configure network thread");
+ }
+
  /**
  * Handle subscriptions through fork transitions, @see FORK_EPOCH_LOOKAHEAD
  */

packages/beacon-node/src/network/core/networkCoreWorker.ts

@@ -1,15 +1,19 @@
 import worker from "node:worker_threads";
+import fs from "node:fs";
+import path from "node:path";
 import {createFromProtobuf} from "@libp2p/peer-id-factory";
 import {expose} from "@chainsafe/threads/worker";
 import {chainConfigFromJson, createBeaconConfig} from "@lodestar/config";
 import {getNodeLogger} from "@lodestar/logger/node";
 import type {WorkerModule} from "@chainsafe/threads/dist/types/worker.js";
+import {SLOTS_PER_EPOCH} from "@lodestar/params";
 import {collectNodeJSMetrics, RegistryMetricCreator} from "../../metrics/index.js";
 import {AsyncIterableBridgeCaller, AsyncIterableBridgeHandler} from "../../util/asyncIterableToEvents.js";
 import {Clock} from "../../util/clock.js";
 import {wireEventsOnWorkerThread} from "../../util/workerEvents.js";
 import {NetworkEventBus, NetworkEventData, networkEventDirection} from "../events.js";
 import {peerIdToString} from "../../util/peerId.js";
+import {profileNodeJS} from "../../util/profile.js";
 import {getNetworkCoreWorkerMetrics} from "./metrics.js";
 import {NetworkWorkerApi, NetworkWorkerData} from "./types.js";
 import {NetworkCore} from "./networkCore.js";
@@ -32,6 +36,7 @@ if (!parentPort) throw Error("parentPort must be defined");
 
 const config = createBeaconConfig(chainConfigFromJson(workerData.chainConfigJson), workerData.genesisValidatorsRoot);
 const peerId = await createFromProtobuf(workerData.peerIdProto);
+const DEFAULT_PROFILE_DURATION = SLOTS_PER_EPOCH * config.SECONDS_PER_SLOT * 1000;
 
 // TODO: Pass options from main thread for logging
 // TODO: Logging won't be visible in file loggers
@@ -147,6 +152,12 @@ const libp2pWorkerApi: NetworkWorkerApi = {
  dumpGossipPeerScoreStats: () => core.dumpGossipPeerScoreStats(),
  dumpDiscv5KadValues: () => core.dumpDiscv5KadValues(),
  dumpMeshPeers: () => core.dumpMeshPeers(),
+ writeProfile: async (durationMs = DEFAULT_PROFILE_DURATION, dirpath = ".") => {
+ const profile = await profileNodeJS(durationMs);
+ const filePath = path.join(dirpath, `network_thread_${new Date().toISOString()}.cpuprofile`);
+ fs.writeFileSync(filePath, profile);
+ return filePath;
+ },
 };
 
 expose(libp2pWorkerApi as WorkerModule<keyof NetworkWorkerApi>);

packages/beacon-node/src/network/core/networkCoreWorkerHandler.ts

@@ -210,6 +210,9 @@ export class WorkerNetworkCore implements INetworkCore {
  dumpMeshPeers(): Promise<Record<string, string[]>> {
  return this.getApi().dumpMeshPeers();
  }
+ writeNetworkThreadProfile(durationMs?: number, dirpath?: string): Promise<string> {
+ return this.getApi().writeProfile(durationMs, dirpath);
+ }
 
  private getApi(): NetworkWorkerApi {
  return this.modules.workerApi;

packages/beacon-node/src/network/core/types.ts

@@ -61,6 +61,7 @@ export interface INetworkCore extends INetworkCorePublic {
 
  close(): Promise<void>;
  scrapeMetrics(): Promise<string>;
+ writeNetworkThreadProfile(durationMs?: number, dirpath?: string): Promise<string>;
 }
 
 /**
@@ -99,6 +100,7 @@ export type NetworkWorkerApi = INetworkCorePublic & {
 
  close(): Promise<void>;
  scrapeMetrics(): Promise<string>;
+ writeProfile(durationMs?: number, dirpath?: string): Promise<string>;
 
  // TODO: ReqResp outgoing
  // TODO: ReqResp incoming

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

@@ -60,6 +60,7 @@ export interface INetwork extends INetworkCorePublic {
 
  // Debug
  dumpGossipQueue(gossipType: GossipType): Promise<PendingGossipsubMessage[]>;
+ writeNetworkThreadProfile(durationMs?: number, dirpath?: string): Promise<string>;
 }
 
 export type PeerDirection = Connection["stat"]["direction"];

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

@@ -544,6 +544,10 @@ export class Network implements INetwork {
  return this.networkProcessor.dumpGossipQueue(gossipType);
  }
 
+ async writeNetworkThreadProfile(durationMs?: number, dirpath?: string): Promise<string> {
+ return this.core.writeNetworkThreadProfile(durationMs, dirpath);
+ }
+
  private onLightClientFinalityUpdate = async (finalityUpdate: allForks.LightClientFinalityUpdate): Promise<void> => {
  // TODO: Review is OK to remove if (this.hasAttachedSyncCommitteeMember())
 

packages/beacon-node/src/util/profile.ts

@@ -0,0 +1,32 @@
+import {sleep} from "@lodestar/utils";
+
+/**
+ * Take 10m profile of the current thread without promise tracking.
+ */
+export async function profileNodeJS(durationMs: number): Promise<string> {
+ const inspector = await import("node:inspector");
+
+ // due to some typing issues, not able to use promisify here
+ return new Promise<string>((resolve, reject) => {
+ // Start the inspector and connect to it
+ const session = new inspector.Session();
+ session.connect();
+
+ session.post("Profiler.enable", () => {
+ session.post("Profiler.start", async () => {
+ await sleep(durationMs);
+ session.post("Profiler.stop", (err, {profile}) => {
+ if (!err) {
+ resolve(JSON.stringify(profile));
+ } else {
+ reject(err);
+ }
+
+ // Detach from the inspector and close the session
+ session.post("Profiler.disable");
+ session.disconnect();
+ });
+ });
+ });
+ });
+}

packages/beacon-node/test/e2e/network/onWorker/dataSerialization.test.ts

@@ -140,6 +140,7 @@ describe("data serialization through worker boundary", function () {
  publishGossip: ["test-topic", bytes, {allowPublishToZeroPeers: true, ignoreDuplicatePublishError: true}],
  close: [],
  scrapeMetrics: [],
+ writeProfile: [0, ""],
  };
 
  const lodestarPeer: routes.lodestar.LodestarNodePeer = {
@@ -201,6 +202,7 @@ describe("data serialization through worker boundary", function () {
  publishGossip: 1,
  close: null,
  scrapeMetrics: "test-metrics",
+ writeProfile: "",
  };
 
  type TestCase = {id: string; data: unknown; shouldFail?: boolean};