Indexer workers

jest.config.js

@@ -170,7 +170,7 @@ module.exports = {
   // ],
 
   // The regexp pattern or array of patterns that Jest uses to detect test files
-  testRegex: '.*\\.(spec|test)\\.ts$',
+  testRegex: '.*\\.spec\\.ts$',
 
   // This option allows the use of a custom results processor
   // testResultsProcessor: undefined,

packages/node/CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 - Make handler data types generic (#1194)
 
+### Added
+
+- Support for worker threads. This will move block fetching and processing into a worker, this can increase performance by up to 4 times. By default this feature is disabled, you can enable it with the `--workers=<number>` flag. The number of workers will be capped to the number of CPU cores. (#1103)
+
 ## [1.5.1] - 2022-07-15
 ### Fixed
 - Unable to create ds processor from template when project hosted on IPFS (#1190)

packages/node/src/app.module.ts

@@ -6,11 +6,9 @@ import { EventEmitterModule } from '@nestjs/event-emitter';
 import { ScheduleModule } from '@nestjs/schedule';
 import { ConfigureModule } from './configure/configure.module';
 import { DbModule } from './db/db.module';
-import { IndexerModule } from './indexer/indexer.module';
+import { FetchModule } from './indexer/fetch.module';
 import { MetaModule } from './meta/meta.module';
 
-export class NodeOption {}
-
 @Module({
   imports: [
     DbModule.forRoot({
@@ -23,7 +21,7 @@ export class NodeOption {}
     EventEmitterModule.forRoot(),
     ConfigureModule.register(),
     ScheduleModule.forRoot(),
-    IndexerModule,
+    FetchModule,
     MetaModule,
   ],
   controllers: [],

packages/node/src/configure/NodeConfig.ts

@@ -32,6 +32,7 @@ export interface IConfig {
   readonly mmrPath?: string;
   readonly ipfs?: string;
   readonly dictionaryTimeout: number;
+  readonly workers?: number;
 }
 
 export type MinConfig = Partial<Omit<IConfig, 'subquery'>> &
@@ -109,6 +110,7 @@ export class NodeConfig implements IConfig {
   get debug(): boolean {
     return this._config.debug;
   }
+
   get preferRange(): boolean {
     return this._config.preferRange;
   }
@@ -152,6 +154,10 @@ export class NodeConfig implements IConfig {
     return this._config.dbSchema ?? this.subqueryName;
   }
 
+  get workers(): number {
+    return this._config.workers;
+  }
+
   merge(config: Partial<IConfig>): this {
     assign(this._config, config);
     return this;

packages/node/src/indexer/PoiBlock.ts

@@ -1,6 +1,6 @@
 // Copyright 2020-2022 OnFinality Limited authors & contributors
 // SPDX-License-Identifier: Apache-2.0
-// eslint-disable-next-line header/header
+
 import { u8aConcat, numberToU8a, hexToU8a, isHex, isU8a } from '@polkadot/util';
 import { blake2AsU8a } from '@polkadot/util-crypto';
 import { ProofOfIndex } from './entities/Poi.entity';

packages/node/src/indexer/benchmark.service.ts

@@ -66,13 +66,13 @@ export class BenchmarkService {
   }
 
   @OnEvent(IndexerEvent.BlockProcessing)
-  handleProcessingBlock(blockPayload: ProcessBlockPayload) {
+  handleProcessingBlock(blockPayload: ProcessBlockPayload): void {
     this.currentProcessingHeight = blockPayload.height;
     this.currentProcessingTimestamp = blockPayload.timestamp;
   }
 
   @OnEvent(IndexerEvent.BlockTarget)
-  handleTargetBlock(blockPayload: TargetBlockPayload) {
+  handleTargetBlock(blockPayload: TargetBlockPayload): void {
     this.targetHeight = blockPayload.height;
   }
 }

packages/node/src/indexer/dynamic-ds.service.ts

@@ -53,7 +53,7 @@ export class DynamicDsService {
 
       return ds;
     } catch (e) {
-      logger.error(e.message);
+      logger.error(e, 'Failed to create dynamic ds');
       process.exit(1);
     }
   }
@@ -67,7 +67,7 @@ export class DynamicDsService {
           params.map((params) => this.getDatasource(params)),
         );
       } catch (e) {
-        logger.error(`Unable to get dynamic datasources:\n${e.message}`);
+        logger.error(e, `Unable to get dynamic datasources`);
         process.exit(1);
       }
     }

packages/node/src/indexer/fetch.module.ts

@@ -0,0 +1,109 @@
+// Copyright 2020-2022 OnFinality Limited authors & contributors
+// SPDX-License-Identifier: Apache-2.0
+
+import { Module } from '@nestjs/common';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { SchedulerRegistry } from '@nestjs/schedule';
+import { NodeConfig } from '../configure/NodeConfig';
+import { SubqueryProject } from '../configure/SubqueryProject';
+import { DbModule } from '../db/db.module';
+import { getYargsOption } from '../yargs';
+import { ApiService } from './api.service';
+import { BenchmarkService } from './benchmark.service';
+import { DictionaryService } from './dictionary.service';
+import { DsProcessorService } from './ds-processor.service';
+import { DynamicDsService } from './dynamic-ds.service';
+import { FetchService } from './fetch.service';
+import { IndexerManager } from './indexer.manager';
+import { MmrService } from './mmr.service';
+import { PoiService } from './poi.service';
+import { ProjectService } from './project.service';
+import { SandboxService } from './sandbox.service';
+import { StoreService } from './store.service';
+import {
+  BlockDispatcherService,
+  WorkerBlockDispatcherService,
+  IBlockDispatcher,
+} from './worker/block-dispatcher.service';
+
+const { argv } = getYargsOption();
+
+@Module({
+  imports: [DbModule.forFeature(['Subquery'])],
+  providers: [
+    StoreService,
+    {
+      provide: ApiService,
+      useFactory: async (
+        project: SubqueryProject,
+        eventEmitter: EventEmitter2,
+      ) => {
+        const apiService = new ApiService(project, eventEmitter);
+        await apiService.init();
+        return apiService;
+      },
+      inject: [SubqueryProject, EventEmitter2],
+    },
+    IndexerManager,
+    {
+      provide: 'IBlockDispatcher',
+      useClass: argv.workers
+        ? WorkerBlockDispatcherService
+        : BlockDispatcherService,
+    },
+    {
+      provide: FetchService,
+      useFactory: async (
+        apiService: ApiService,
+        nodeConfig: NodeConfig,
+        project: SubqueryProject,
+        blockDispatcher: IBlockDispatcher,
+        dictionaryService: DictionaryService,
+        dsProcessorService: DsProcessorService,
+        eventEmitter: EventEmitter2,
+        projectService: ProjectService,
+        dynamicDsService: DynamicDsService,
+        schedulerRegistry: SchedulerRegistry,
+      ) => {
+        await projectService.init();
+
+        const fetchService = new FetchService(
+          apiService,
+          nodeConfig,
+          project,
+          blockDispatcher,
+          dictionaryService,
+          dsProcessorService,
+          dynamicDsService,
+          eventEmitter,
+          schedulerRegistry,
+        );
+
+        await fetchService.init(projectService.startHeight);
+        return fetchService;
+      },
+      inject: [
+        ApiService,
+        NodeConfig,
+        SubqueryProject,
+        'IBlockDispatcher',
+        DictionaryService,
+        DsProcessorService,
+        EventEmitter2,
+        ProjectService,
+        DynamicDsService,
+        SchedulerRegistry,
+      ],
+    },
+    BenchmarkService,
+    DictionaryService,
+    SandboxService,
+    DsProcessorService,
+    DynamicDsService,
+    PoiService,
+    MmrService,
+    ProjectService,
+  ],
+  exports: [StoreService, MmrService],
+})
+export class FetchModule {}