From b64bc034c691f678c03fb9468bef7de0b0103cf2 Mon Sep 17 00:00:00 2001 From: Pierre-Henri Date: Tue, 14 May 2024 21:10:06 +0200 Subject: [PATCH] improve the debounce finalize --- src/lib/folder-watcher/FolderWatcher.ts | 76 +++++++++++-------- .../hash-compute/HashComputerIndexCache.ts | 2 +- src/lib/hash-compute/HashIndexManager.ts | 27 ++++--- 3 files changed, 61 insertions(+), 44 deletions(-) diff --git a/src/lib/folder-watcher/FolderWatcher.ts b/src/lib/folder-watcher/FolderWatcher.ts index 8cc3c71..1221d87 100644 --- a/src/lib/folder-watcher/FolderWatcher.ts +++ b/src/lib/folder-watcher/FolderWatcher.ts @@ -9,23 +9,28 @@ export class FolderWatcher { initialized = false; queue = new PQueue({concurrency: 100, autoStart: true}); fileProcessor: FileProcessorInterface; - queueSize = -1; + queueSize = 0; current = 0; processing = new Set(); private readonly finalizeDebounced: () => Promise; - constructor(fileProcessor:FileProcessorInterface,private WATCH_FOLDER_LIST: string, private config:{ + constructor(fileProcessor: FileProcessorInterface, private WATCH_FOLDER_LIST: string, private config: { interval?: number, stabilityThreshold?: number, pollInterval?: number, outputFolderUpdateIntervalMs?: number }) { this.fileProcessor = fileProcessor; - if(this.fileProcessor.finalize) { - const debounced = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true}); + if (this.fileProcessor.finalize) { + const debouncedImmediate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: true}); + const debouncedLate = debounce(() => this.fileProcessor.finalize(), this.config.outputFolderUpdateIntervalMs || 10000, {immediate: false}); this.finalizeDebounced = async () => { if (this.initialized) { - return debounced(); + const ret1 = debouncedImmediate(); + const ret2 = debouncedLate(); + this.current = 0; + this.queueSize = 0; + await Promise.all([ret1, ret2]); } } } @@ -73,15 +78,15 @@ export class FolderWatcher { return counts.reduce((acc, current) => acc + current, 0); } - private async processFile(filePath: string):Promise { - if(await this.fileProcessor.canProcessFile(filePath)) { - if(!this.processing.has(filePath)) { - this.processing.add(filePath);//avoid double processing in the queue + private async processFile(filePath: string): Promise { + if (!this.processing.has(filePath)) { + this.processing.add(filePath);//avoid double processing in the queue + if (await this.fileProcessor.canProcessFile(filePath)) { const current = ++this.current; await this.queue.add(() => this.fileProcessor.processFile(current, this.queueSize, filePath)); - this.processing.delete(filePath); } } + this.processing.delete(filePath); } /** @@ -89,12 +94,29 @@ export class FolderWatcher { * @param filePath * @private */ - private async processFileExtended(filePath: string):Promise { + private async processFileExtended(filePath: string): Promise { let dirname = path.dirname(filePath); - await this.processDirectory(dirname, []); + await this.processDirWithCount([dirname]); } - private chokidarWatch(folderList: string[]){ + private async processDirWithCount(folderList: string[]): Promise { + let countPromises = []; + for (let i = 0; i < folderList.length; i++) { + folderList[i] = path.normalize(folderList[i]); + countPromises.push(this.countFile(folderList[i])); + } + const count = (await Promise.all(countPromises)).reduce((acc, current) => acc + current, 0); + console.log(`Found ${count} files to process`); + this.queueSize += count; + let promises: Promise[] = []; + for (let i = 0; i < folderList.length; i++) { + folderList[i] = path.normalize(folderList[i]); + await this.processDirectory(folderList[i], promises); + } + await Promise.all(promises); + } + + private chokidarWatch(folderList: string[]) { const chokidarconfig = { ignoreInitial: true,//ignore the initial scan we have our own function persistent: true, @@ -120,7 +142,7 @@ export class FolderWatcher { .on('add', async (filePath) => { try { await this.processFileExtended(filePath); - if(this.finalizeDebounced) { + if (this.finalizeDebounced) { await this.finalizeDebounced(); } } catch (error) { @@ -130,7 +152,7 @@ export class FolderWatcher { .on('change', async (filePath) => { try { await this.processFileExtended(filePath); - if(this.finalizeDebounced) { + if (this.finalizeDebounced) { await this.finalizeDebounced(); } } catch (e) { @@ -139,11 +161,12 @@ export class FolderWatcher { }) .on('unlink', async (filePath) => { try { - if(this.fileProcessor.deleteFile) { + if (this.fileProcessor.deleteFile) { await this.fileProcessor.deleteFile(filePath); } + this.processing.delete(filePath); await this.processFileExtended(filePath); - if(this.finalizeDebounced) { + if (this.finalizeDebounced) { await this.finalizeDebounced(); } } catch (e) { @@ -163,29 +186,16 @@ export class FolderWatcher { } let folderList = folders.split(','); - if(this.fileProcessor.initialize){ + if (this.fileProcessor.initialize) { await this.fileProcessor.initialize(); } this.chokidarWatch(folderList); //start the initial scan - let countPromises = []; - for (let i = 0; i < folderList.length; i++) { - folderList[i] = path.normalize(folderList[i]); - countPromises.push(this.countFile(folderList[i])); - } - const count = (await Promise.all(countPromises)).reduce((acc, current) => acc + current, 0); - console.log(`Found ${count} files to process`); - this.queueSize = count; - let promises: Promise[] = []; - for (let i = 0; i < folderList.length; i++) { - folderList[i] = path.normalize(folderList[i]); - await this.processDirectory(folderList[i], promises); - } - await Promise.all(promises); + await this.processDirWithCount(folderList); this.initialized = true; - if(this.finalizeDebounced) { + if (this.finalizeDebounced) { await this.finalizeDebounced(); } diff --git a/src/lib/hash-compute/HashComputerIndexCache.ts b/src/lib/hash-compute/HashComputerIndexCache.ts index 0f3fb42..00547a7 100644 --- a/src/lib/hash-compute/HashComputerIndexCache.ts +++ b/src/lib/hash-compute/HashComputerIndexCache.ts @@ -16,7 +16,7 @@ export class HashComputerIndexCache implements ComputeInterface { public async computeMissingHash(filePath: string, metadata: MultiHashData): Promise { await this.hashIndexManager.init(); let stats = await stat(filePath); - if (this.hashIndexManager.getCache().has(path.basename(filePath))) { + if (this.hashIndexManager.hasFileInCache(path.basename(filePath),stats)) { const indexLine = this.hashIndexManager.getCidForFile(filePath, stats.size, stats.mtime.toISOString()); if(indexLine) { for (const hash of this.targetHash) { diff --git a/src/lib/hash-compute/HashIndexManager.ts b/src/lib/hash-compute/HashIndexManager.ts index 6084b2f..b98eb33 100644 --- a/src/lib/hash-compute/HashIndexManager.ts +++ b/src/lib/hash-compute/HashIndexManager.ts @@ -7,6 +7,7 @@ import path from "path"; import {clearInterval} from "node:timers"; import {CID_ALGORITHM_NAMES} from "./MultiHashData"; import {stat} from "fs/promises"; +import {Stats} from "node:fs"; interface IndexLine extends Partial> { path: string; @@ -17,7 +18,7 @@ interface IndexLine extends Partial> { export const INDEX_HEADERS = ['path', 'size', 'mtime']; export class HashIndexManager { - private cache: Map = new Map(); + private cache: Map = new Map();//the key is the file `${file_name}-${filesize}-${mtime_ISOstr}` private intervalId: any; private intervalTime: number = 30000; private lastIndexFileSize: { [key in CID_ALGORITHM_NAMES]?: number } = {}; //size of the index file last time it was read @@ -31,8 +32,12 @@ export class HashIndexManager { private targetHash: CID_ALGORITHM_NAMES[]) { } - public getCache(): Map { - return new Map(this.cache); + hasFileInCache(filePath: string,stats:Stats): boolean { + return this.cache.has(`${path.basename(filePath)}-${stats.size}-${stats.mtime.toISOString()}`); + } + + getCacheSize(): number { + return this.cache.size; } /** @@ -117,9 +122,10 @@ export class HashIndexManager { // Read existing file content and parse it const records: IndexLine[] = await this.readCsv(hash); for (const record of records) { - let indexLine = this.cache.get(record.path); + const cacheKey = `${record.path}-${record.size}-${record.mtime}`; + let indexLine = this.cache.get(cacheKey); if (!indexLine) { - this.cache.set(record.path, record); + this.cache.set(cacheKey, record); } else if(record[hash]){ //update the cache with the latest data indexLine[hash] = record[hash]; @@ -183,12 +189,12 @@ export class HashIndexManager { let didWrite=false; for (const hash of this.targetHash) { let existingRows: IndexLine[] = await this.loadIndex(hash); - let existingRowsAsMap: Map = new Map(existingRows.map(row => [row.path, row])); + let existingRowsAsMap: Map = new Map(existingRows.map(row => [row.path + '-' + row.size + '-' + row.mtime, row])); if (this.cache.size !== 0) { // Filter out cacheRows that are already in the file const newRows = cacheRows.filter(row => { //to be added a row must not exist in the file and must exist in the cache (with a hash) - const newRow = !existingRowsAsMap.has(row.path); + const newRow = !existingRowsAsMap.has(row.path + '-' + row.size + '-' + row.mtime); return newRow && !!row[hash]; }); @@ -235,7 +241,7 @@ export class HashIndexManager { public getCidForFile(filePath: string, fileSize: number, mtime: string): IndexLine { const fileName = path.basename(filePath); - let fileNameIndex = this.cache.get(fileName); + let fileNameIndex = this.cache.get(`${fileName}-${fileSize}-${mtime}`); if (fileNameIndex) { for (const hash of this.targetHash) { if (!fileNameIndex[hash]) { @@ -275,7 +281,8 @@ export class HashIndexManager { } const size = fileSize + ""; const baseName = path.basename(filePath); - let indexLine = this.cache.get(baseName); + const cacheKey = `${baseName}-${size}-${mtime}`; + let indexLine = this.cache.get(cacheKey); if (!indexLine) { //filter only the hashes we need let filteredHash = {}; @@ -283,7 +290,7 @@ export class HashIndexManager { filteredHash[hash] = hashs[hash]; } const data = {path: baseName, size: size, mtime: mtime, ...filteredHash}; - this.cache.set(baseName, data); + this.cache.set(cacheKey, data); } else { //update the cache with the latest data for (const hash of this.targetHash) {