metastar/meta-hash
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

improve the debounce finalize

Browse Source
This commit is contained in:
Pierre-Henri
2024-05-14 21:10:06 +02:00
parent b9cc9c9775
commit b64bc034c6
3 changed files with 61 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lib/folder-watcher/FolderWatcher.ts
+43 -33
View File
@@ -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<string>();
private readonly finalizeDebounced: () => Promise<void>;
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<void> {
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<void> {
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<void> {
private async processFileExtended(filePath: string): Promise<void> {
let dirname = path.dirname(filePath);
await this.processDirectory(dirname, []);
await this.processDirWithCount([dirname]);
}
private chokidarWatch(folderList: string[]){
private async processDirWithCount(folderList: string[]): Promise<void> {
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<any>[] = [];
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<any>[] = [];
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();
}
src/lib/hash-compute/HashComputerIndexCache.ts
+1 -1
View File
@@ -16,7 +16,7 @@ export class HashComputerIndexCache implements ComputeInterface {
public async computeMissingHash(filePath: string, metadata: MultiHashData): Promise<void> {
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) {
src/lib/hash-compute/HashIndexManager.ts
+17 -10
View File
@@ -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<Record<CID_ALGORITHM_NAMES, string>> {
path: string;
@@ -17,7 +18,7 @@ interface IndexLine extends Partial<Record<CID_ALGORITHM_NAMES, string>> {
export const INDEX_HEADERS = ['path', 'size', 'mtime'];
export class HashIndexManager {
private cache: Map<string, IndexLine> = new Map<string, IndexLine>();
private cache: Map<string, IndexLine> = new Map<string, IndexLine>();//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<string, IndexLine> {
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<string, IndexLine> = new Map(existingRows.map(row => [row.path, row]));
let existingRowsAsMap: Map<string, IndexLine> = 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) {