machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor indexer scheduling to avoid accumulating identical tasks

Browse Source
This commit is contained in:
Mononaut 2023-09-21 17:50:53 +01:00
parent ab8b557e73
commit 00887bc24b
No known key found for this signature in database
GPG Key ID: A3F058E41374C04E
3 changed files with 71 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
backend/src/api/blocks.ts
View File

@ -776,9 +776,7 @@ class Blocks {
this.updateTimerProgress(timer, `saved prices for ${this.currentBlockHeight}`); this.updateTimerProgress(timer, `saved prices for ${this.currentBlockHeight}`);
} else { } else {
logger.debug(`Cannot save block price for ${blockExtended.height} because the price updater hasnt completed yet. Trying again in 10 seconds.`, logger.tags.mining); logger.debug(`Cannot save block price for ${blockExtended.height} because the price updater hasnt completed yet. Trying again in 10 seconds.`, logger.tags.mining);
setTimeout(() => { indexer.scheduleSingleTask('blocksPrices', 10000);
indexer.runSingleTask('blocksPrices');
}, 10000);
} }
// Save blocks summary for visualization if it's enabled // Save blocks summary for visualization if it's enabled

2
backend/src/index.ts
View File

@ -206,7 +206,7 @@ class Server {
} }
const newMempool = await bitcoinApi.$getRawMempool(); const newMempool = await bitcoinApi.$getRawMempool();
const numHandledBlocks = await blocks.$updateBlocks(); const numHandledBlocks = await blocks.$updateBlocks();
const pollRate = config.MEMPOOL.POLL_RATE_MS * (indexer.indexerRunning ? 10 : 1); const pollRate = config.MEMPOOL.POLL_RATE_MS * (indexer.indexerIsRunning() ? 10 : 1);
if (numHandledBlocks === 0) { if (numHandledBlocks === 0) {
await memPool.$updateMempool(newMempool, pollRate); await memPool.$updateMempool(newMempool, pollRate);
} }

100
backend/src/indexer.ts
View File

@ -15,11 +15,18 @@ export interface CoreIndex {
best_block_height: number; best_block_height: number;
} }
type TaskName = 'blocksPrices' | 'coinStatsIndex';
class Indexer { class Indexer {
runIndexer = true; private runIndexer = true;
indexerRunning = false; private indexerRunning = false;
tasksRunning: string[] = []; private tasksRunning: { [key in TaskName]?: boolean; } = {};
coreIndexes: CoreIndex[] = []; private tasksScheduled: { [key in TaskName]?: NodeJS.Timeout; } = {};
private coreIndexes: CoreIndex[] = [];
public indexerIsRunning(): boolean {
return this.indexerRunning;
}
/** /**
* Check which core index is available for indexing * Check which core index is available for indexing
@ -69,38 +76,69 @@ class Indexer {
} }
} }
public async runSingleTask(task: 'blocksPrices' | 'coinStatsIndex'): Promise<void> { /**
if (!Common.indexingEnabled()) { * schedules a single task to run in `timeout` ms
* only one task of each type may be scheduled
*
* @param {TaskName} task - the type of task
* @param {number} timeout - delay in ms
* @param {boolean} replace - `true` replaces any already scheduled task (works like a debounce), `false` ignores subsequent requests (works like a throttle)
*/
public scheduleSingleTask(task: TaskName, timeout: number = 10000, replace = false): void {
if (this.tasksScheduled[task]) {
if (!replace) { //throttle
return;
} else { // debounce
clearTimeout(this.tasksScheduled[task]);
}
}
this.tasksScheduled[task] = setTimeout(async () => {
try {
await this.runSingleTask(task);
} catch (e) {
logger.err(`Unexpected error in scheduled task ${task}: ` + (e instanceof Error ? e.message : e));
} finally {
clearTimeout(this.tasksScheduled[task]);
}
}, timeout);
}
/**
* Runs a single task immediately
*
* (use `scheduleSingleTask` instead to queue a task to run after some timeout)
*/
public async runSingleTask(task: TaskName): Promise<void> {
if (!Common.indexingEnabled() || this.tasksRunning[task]) {
return; return;
} }
this.tasksRunning[task] = true;
if (task === 'blocksPrices' && !this.tasksRunning.includes(task) && !['testnet', 'signet'].includes(config.MEMPOOL.NETWORK)) { switch (task) {
this.tasksRunning.push(task); case 'blocksPrices': {
let lastestPriceId; if (!['testnet', 'signet'].includes(config.MEMPOOL.NETWORK)) {
try { let lastestPriceId;
lastestPriceId = await PricesRepository.$getLatestPriceId(); try {
} catch (e) { lastestPriceId = await PricesRepository.$getLatestPriceId();
logger.debug('failed to fetch latest price id from db: ' + (e instanceof Error ? e.message : e)); } catch (e) {
} logger.debug('failed to fetch latest price id from db: ' + (e instanceof Error ? e.message : e));
if (priceUpdater.historyInserted === false || lastestPriceId === null) { } if (priceUpdater.historyInserted === false || lastestPriceId === null) {
logger.debug(`Blocks prices indexer is waiting for the price updater to complete`, logger.tags.mining); logger.debug(`Blocks prices indexer is waiting for the price updater to complete`, logger.tags.mining);
setTimeout(() => { this.scheduleSingleTask(task, 10000);
this.tasksRunning = this.tasksRunning.filter(runningTask => runningTask !== task); } else {
this.runSingleTask('blocksPrices'); logger.debug(`Blocks prices indexer will run now`, logger.tags.mining);
}, 10000); await mining.$indexBlockPrices();
} else { }
logger.debug(`Blocks prices indexer will run now`, logger.tags.mining); }
await mining.$indexBlockPrices(); } break;
this.tasksRunning = this.tasksRunning.filter(runningTask => runningTask !== task);
} case 'coinStatsIndex': {
logger.debug(`Indexing coinStatsIndex now`);
await mining.$indexCoinStatsIndex();
} break;
} }
if (task === 'coinStatsIndex' && !this.tasksRunning.includes(task)) { this.tasksRunning[task] = false;
this.tasksRunning.push(task);
logger.debug(`Indexing coinStatsIndex now`);
await mining.$indexCoinStatsIndex();
this.tasksRunning = this.tasksRunning.filter(runningTask => runningTask !== task);
}
} }
public async $run(): Promise<void> { public async $run(): Promise<void> {