cerc-io/watcher-ts
10
1
Fork 0
mirror of https://github.com/cerc-io/watcher-ts synced 2025-07-28 11:02:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix use of prefetchBlocksInMem flag in block processing (#240)

Browse Source 
* Fix use of prefetchBlocksInMem flag in block processing

* Rename prefetchedBlocksMap to blockAndEventsMap
This commit is contained in:
nikugogoi 2022-11-18 11:01:09 +05:30 committed by GitHub
parent 92fd3cac03
commit cc8fcffaa1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 47 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
packages/util/src/common.ts
View File

@ -52,17 +52,25 @@ export const processBlockByNumberWithCache = async (
* @param job * @param job
* @param indexer * @param indexer
* @param jobQueueConfig * @param jobQueueConfig
* @param prefetchedBlocksMap * @param blockAndEventsMap
*/ */
export const fetchBlocksAtHeight = async ( export const fetchBlocksAtHeight = async (
job: any, job: any,
indexer: IndexerInterface, indexer: IndexerInterface,
jobQueueConfig: JobQueueConfig, jobQueueConfig: JobQueueConfig,
prefetchedBlocksMap: Map<string, PrefetchedBlock> blockAndEventsMap: Map<string, PrefetchedBlock>
): Promise<DeepPartial<BlockProgressInterface>[]> => { ): Promise<DeepPartial<BlockProgressInterface>[]> => {
const { blockNumber } = job.data; const { blockNumber } = job.data;
let blocks = []; let blocks = [];
// Check for blocks in cache if prefetchBlocksInMem flag set.
if (jobQueueConfig.prefetchBlocksInMem) {
// Get blocks prefetched in memory.
blocks = getPrefetchedBlocksAtHeight(blockAndEventsMap, blockNumber);
log('size:common#fetchBlocksAtHeight-prefetch-_blockAndEventsMap-size:', blockAndEventsMap.size);
}
if (!blocks.length) {
// Try fetching blocks from the db. // Try fetching blocks from the db.
const blockProgressEntities = await indexer.getBlocksAtHeight(blockNumber, false); const blockProgressEntities = await indexer.getBlocksAtHeight(blockNumber, false);
blocks = blockProgressEntities.map((block: any) => { blocks = blockProgressEntities.map((block: any) => {
@ -70,27 +78,29 @@ export const fetchBlocksAtHeight = async (
return block; return block;
}); });
}
// If blocks not found in the db: if (jobQueueConfig.prefetchBlocksInMem && !blocks.length) {
if (!blocks.length) { // If blocks not found in the db and cache, fetch next batch.
// Check for blocks in cache if prefetchBlocksInMem flag set.
if (jobQueueConfig.prefetchBlocksInMem) {
// Get blocks prefetched in memory.
blocks = getPrefetchedBlocksAtHeight(prefetchedBlocksMap, blockNumber);
// If not found in cache, fetch the next batch.
if (!blocks.length) {
log(`common#cache-miss-${blockNumber}`); log(`common#cache-miss-${blockNumber}`);
// Wait for blocks to be prefetched. // Wait for blocks to be prefetched.
console.time('time:common#fetchBlocks-_prefetchBlocks'); console.time('time:common#fetchBlocks-_prefetchBlocks');
await _prefetchBlocks(blockNumber, indexer, jobQueueConfig, prefetchedBlocksMap); await _prefetchBlocks(blockNumber, indexer, jobQueueConfig, blockAndEventsMap);
console.timeEnd('time:common#fetchBlocks-_prefetchBlocks'); console.timeEnd('time:common#fetchBlocks-_prefetchBlocks');
blocks = getPrefetchedBlocksAtHeight(prefetchedBlocksMap, blockNumber); blocks = getPrefetchedBlocksAtHeight(blockAndEventsMap, blockNumber);
} }
log('size:common#_fetchBlocks-_prefetchedBlocksMap-size:', prefetchedBlocksMap.size); // Try fetching blocks from eth-server until found.
while (!blocks.length) {
console.time('time:common#_fetchBlocks-eth-server');
blocks = await indexer.getBlocks({ blockNumber });
console.timeEnd('time:common#_fetchBlocks-eth-server');
if (!blocks.length) {
log(`No blocks fetched for block number ${blockNumber}, retrying after ${jobQueueConfig.blockDelayInMilliSecs} ms delay.`);
await wait(jobQueueConfig.blockDelayInMilliSecs);
} }
} }
@ -118,10 +128,10 @@ export const _prefetchBlocks = async (
blockNumber: number, blockNumber: number,
indexer: IndexerInterface, indexer: IndexerInterface,
jobQueueConfig: JobQueueConfig, jobQueueConfig: JobQueueConfig,
prefetchedBlocksMap: Map<string, PrefetchedBlock> blockAndEventsMap: Map<string, PrefetchedBlock>
): Promise<void> => { ): Promise<void> => {
// Clear cache of any remaining blocks. // Clear cache of any remaining blocks.
prefetchedBlocksMap.clear(); blockAndEventsMap.clear();
const blocksWithEvents = await _fetchBatchBlocks( const blocksWithEvents = await _fetchBatchBlocks(
indexer, indexer,
@ -131,7 +141,7 @@ export const _prefetchBlocks = async (
); );
blocksWithEvents.forEach(({ blockProgress, events }) => { blocksWithEvents.forEach(({ blockProgress, events }) => {
prefetchedBlocksMap.set(blockProgress.blockHash, { block: blockProgress, events }); blockAndEventsMap.set(blockProgress.blockHash, { block: blockProgress, events });
}); });
}; };
@ -360,8 +370,8 @@ export const createCheckpointJob = async (jobQueue: JobQueue, blockHash: string,
); );
}; };
const getPrefetchedBlocksAtHeight = (prefetchedBlocksMap: Map<string, PrefetchedBlock>, blockNumber: number):any[] => { const getPrefetchedBlocksAtHeight = (blockAndEventsMap: Map<string, PrefetchedBlock>, blockNumber: number):any[] => {
return Array.from(prefetchedBlocksMap.values()) return Array.from(blockAndEventsMap.values())
.filter(({ block }) => Number(block.blockNumber) === blockNumber) .filter(({ block }) => Number(block.blockNumber) === blockNumber)
.map(prefetchedBlock => prefetchedBlock.block); .map(prefetchedBlock => prefetchedBlock.block);
}; };

16
packages/util/src/job-runner.ts
View File

@ -39,7 +39,7 @@ export class JobRunner {
_endBlockProcessTimer?: () => void _endBlockProcessTimer?: () => void
_shutDown = false _shutDown = false
_signalCount = 0 _signalCount = 0
_prefetchedBlocksMap: Map<string, PrefetchedBlock> = new Map() _blockAndEventsMap: Map<string, PrefetchedBlock> = new Map()
constructor (jobQueueConfig: JobQueueConfig, indexer: IndexerInterface, jobQueue: JobQueue) { constructor (jobQueueConfig: JobQueueConfig, indexer: IndexerInterface, jobQueue: JobQueue) {
this._indexer = indexer; this._indexer = indexer;
@ -56,7 +56,7 @@ export class JobRunner {
job, job,
this._indexer, this._indexer,
this._jobQueueConfig, this._jobQueueConfig,
this._prefetchedBlocksMap this._blockAndEventsMap
); );
const indexBlockPromises = blocksToBeIndexed.map(blockToBeIndexed => this._indexBlock(job, blockToBeIndexed)); const indexBlockPromises = blocksToBeIndexed.map(blockToBeIndexed => this._indexBlock(job, blockToBeIndexed));
await Promise.all(indexBlockPromises); await Promise.all(indexBlockPromises);
@ -352,7 +352,7 @@ export class JobRunner {
} }
if (!blockProgress) { if (!blockProgress) {
const prefetchedBlock = this._prefetchedBlocksMap.get(blockHash); const prefetchedBlock = this._blockAndEventsMap.get(blockHash);
if (prefetchedBlock) { if (prefetchedBlock) {
({ block: blockProgress } = prefetchedBlock); ({ block: blockProgress } = prefetchedBlock);
@ -365,7 +365,7 @@ export class JobRunner {
[blockProgress] = await this._indexer.saveBlockAndFetchEvents({ cid, blockHash, blockNumber, parentHash, blockTimestamp }); [blockProgress] = await this._indexer.saveBlockAndFetchEvents({ cid, blockHash, blockNumber, parentHash, blockTimestamp });
console.timeEnd('time:job-runner#_indexBlock-saveBlockAndFetchEvents'); console.timeEnd('time:job-runner#_indexBlock-saveBlockAndFetchEvents');
this._prefetchedBlocksMap.set(blockHash, { block: blockProgress, events: [] }); this._blockAndEventsMap.set(blockHash, { block: blockProgress, events: [] });
} }
} }
@ -382,16 +382,16 @@ export class JobRunner {
async _processEvents (job: any): Promise<void> { async _processEvents (job: any): Promise<void> {
const { blockHash } = job.data; const { blockHash } = job.data;
if (!this._prefetchedBlocksMap.has(blockHash)) { if (!this._blockAndEventsMap.has(blockHash)) {
console.time('time:job-runner#_processEvents-get-block-progress'); console.time('time:job-runner#_processEvents-get-block-progress');
const block = await this._indexer.getBlockProgress(blockHash); const block = await this._indexer.getBlockProgress(blockHash);
console.timeEnd('time:job-runner#_processEvents-get-block-progress'); console.timeEnd('time:job-runner#_processEvents-get-block-progress');
assert(block); assert(block);
this._prefetchedBlocksMap.set(blockHash, { block, events: [] }); this._blockAndEventsMap.set(blockHash, { block, events: [] });
} }
const prefetchedBlock = this._prefetchedBlocksMap.get(blockHash); const prefetchedBlock = this._blockAndEventsMap.get(blockHash);
assert(prefetchedBlock); assert(prefetchedBlock);
const { block } = prefetchedBlock; const { block } = prefetchedBlock;
@ -404,7 +404,7 @@ export class JobRunner {
lastProcessedBlockNumber.set(block.blockNumber); lastProcessedBlockNumber.set(block.blockNumber);
lastBlockNumEvents.set(block.numEvents); lastBlockNumEvents.set(block.numEvents);
this._prefetchedBlocksMap.delete(block.blockHash); this._blockAndEventsMap.delete(block.blockHash);
if (this._endBlockProcessTimer) { if (this._endBlockProcessTimer) {
this._endBlockProcessTimer(); this._endBlockProcessTimer();