obsidian-plugin/taskgenius_taskgenius-plugin
1
0
Fork 0
mirror of https://github.com/taskgenius/taskgenius-plugin.git synced 2026-07-22 06:40:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor(dataflow): optimize worker parallelization and fix tgProject handling

Browse source 
- Remove project data pre-computation bottleneck from worker processing
- Implement parallel computation of project data alongside file parsing
- Move tgProject handling from Workers to Augmentor per architecture design
- Fix Augmentor to properly set tgProject attribute on tasks
- Update DataflowOrchestrator to include tgProject in projectMeta context
- Align with dataflow architecture principle: parsers extract raw data, augmentor handles enhancement
- Improve performance by eliminating sequential project data computation

This change eliminates the performance bottleneck where task parsing had to wait
for all project data to be computed first. Now workers focus solely on raw task
extraction while project enhancement happens in the Augmentor phase, fully
aligning with the documented dataflow architecture design.
This commit is contained in:
Quorafind 2025-08-20 00:02:27 +08:00
parent b84389ee95
commit 4e783829f9
5 changed files with 196 additions and 120 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

131
src/dataflow/Orchestrator.ts
View file

@ -173,7 +173,10 @@ export class DataflowOrchestrator {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
projectMeta: {
...projectData.enhancedMetadata,
tgProject: projectData.tgProject // Include tgProject in projectMeta
},
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
@ -241,12 +244,113 @@ export class DataflowOrchestrator {
}
/**
* Process multiple files in batch
* Process multiple files in batch using workers for parallel processing
*/
async processBatch(files: TFile[]): Promise<void> {
async processBatch(files: TFile[], useWorkers: boolean = true): Promise<void> {
const updates = new Map<string, Task[]>();
let skippedCount = 0;
// Decide whether to use workers based on batch size and configuration
const shouldUseWorkers = useWorkers && files.length > 5; // Use workers for batches > 5 files
if (shouldUseWorkers) {
// Use WorkerOrchestrator for parallel processing
console.log(`[DataflowOrchestrator] Using workers to process ${files.length} files in parallel`);
try {
// Configure worker manager with plugin settings
const taskWorkerManager = this.workerOrchestrator['taskWorkerManager'] as TaskWorkerManager;
if (taskWorkerManager) {
taskWorkerManager.updateSettings({
preferMetadataFormat: this.plugin.settings.preferMetadataFormat || 'tasks',
customDateFormats: this.plugin.settings.customDateFormats,
fileMetadataInheritance: this.plugin.settings.fileMetadataInheritance,
projectConfig: this.plugin.settings.projectConfig
});
}
// Parse all files in parallel using workers (raw parsing only, no project data)
console.log(`[DataflowOrchestrator] Parsing ${files.length} files with workers (raw extraction)...`);
const parsedResults = await this.workerOrchestrator.batchParse(files, "normal");
// Compute project data in parallel with storage operations
const projectDataPromises = new Map<string, Promise<any>>();
for (const file of files) {
projectDataPromises.set(file.path, this.projectResolver.get(file.path));
}
// Process each parsed result
for (const [filePath, rawTasks] of parsedResults) {
try {
const file = files.find(f => f.path === filePath);
if (!file) continue;
// Get file modification time for caching
const fileStat = await this.vault.adapter.stat(filePath);
const mtime = fileStat?.mtime;
const fileContent = await this.vault.cachedRead(file);
// Store parsed tasks with mtime (can happen in parallel)
const storePromise = this.storage.storeRaw(filePath, rawTasks, fileContent, mtime);
// Get project data for augmentation (already computing in parallel)
const projectData = await projectDataPromises.get(filePath);
// Wait for storage to complete
await storePromise;
// Augment tasks with project data
const fileMetadata = this.metadataCache.getFileCache(file);
const augmentContext: AugmentContext = {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData?.tgProject?.name,
projectMeta: projectData ? {
...projectData.enhancedMetadata || {},
tgProject: projectData.tgProject // Include tgProject in projectMeta
} : {},
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
updates.set(filePath, augmentedTasks);
} catch (error) {
console.error(`Error processing parsed result for ${filePath}:`, error);
}
}
console.log(`[DataflowOrchestrator] Worker processing complete, parsed ${parsedResults.size} files`);
} catch (error) {
console.error("[DataflowOrchestrator] Worker processing failed, falling back to sequential:", error);
// Fall back to sequential processing
await this.processBatchSequential(files, updates, skippedCount);
}
} else {
// Use sequential processing for small batches or when workers are disabled
await this.processBatchSequential(files, updates, skippedCount);
}
if (skippedCount > 0) {
console.log(`[DataflowOrchestrator] Skipped ${skippedCount} unchanged files`);
}
// Update repository in batch
if (updates.size > 0) {
await this.repository.updateBatch(updates);
}
}
/**
* Process files sequentially (fallback or for small batches)
*/
private async processBatchSequential(
files: TFile[],
updates: Map<string, Task[]>,
skippedCount: number
): Promise<number> {
let localSkippedCount = 0;
for (const file of files) {
try {
const filePath = file.path;
@ -275,7 +379,10 @@ export class DataflowOrchestrator {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
projectMeta: {
...projectData.enhancedMetadata,
tgProject: projectData.tgProject // Include tgProject in projectMeta
},
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
@ -296,7 +403,10 @@ export class DataflowOrchestrator {
filePath,
fileMeta: fileMetadata?.frontmatter || {},
projectName: projectData.tgProject?.name,
projectMeta: projectData.enhancedMetadata,
projectMeta: {
...projectData.enhancedMetadata,
tgProject: projectData.tgProject // Include tgProject in projectMeta
},
tasks: rawTasks
};
const augmentedTasks = await this.augmentor.merge(augmentContext);
@ -305,18 +415,11 @@ export class DataflowOrchestrator {
}
} catch (error) {
console.error(`Error processing file ${file.path} in batch:`, error);
console.error(`Error processing file ${file.path} sequentially:`, error);
}
}
if (skippedCount > 0) {
console.log(`[DataflowOrchestrator] Skipped ${skippedCount} unchanged files`);
}
// Update repository in batch
if (updates.size > 0) {
await this.repository.updateBatch(updates);
}
return localSkippedCount;
}
/**

21
src/dataflow/augment/Augmentor.ts
View file

@ -230,13 +230,20 @@ export class Augmentor {
}
// Set TgProject if project metadata is available
if (ctx.projectMeta && ctx.projectName) {
metadata.tgProject = {
type: ctx.projectMeta.type || 'metadata',
name: ctx.projectName,
source: ctx.projectMeta.source,
readonly: ctx.projectMeta.readonly || false
} as TgProject;
// The tgProject should come from ProjectResolver which includes all necessary fields
if (ctx.projectMeta) {
// Check if we have a tgProject object in the project metadata
if (ctx.projectMeta.tgProject) {
metadata.tgProject = ctx.projectMeta.tgProject;
} else if (ctx.projectName) {
// Fallback: construct from available data
metadata.tgProject = {
type: ctx.projectMeta.type || 'metadata',
name: ctx.projectName,
source: ctx.projectMeta.source || ctx.projectMeta.configSource || 'unknown',
readonly: ctx.projectMeta.readonly || false
} as TgProject;
}
}
}

96
src/dataflow/workers/TaskIndex.worker.ts
View file

@ -43,100 +43,14 @@ function parseTasksWithConfigurableParser(
const parser = new MarkdownTaskParser(config);
// Enhanced parsing: use pre-computed data if available
let enhancedFileMetadata = fileMetadata;
let projectConfigData: Record<string, any> | undefined;
let tgProject: TgProject | undefined;
// Only process enhanced project data if enhanced project is enabled
if (
settings.enhancedProjectData &&
settings.projectConfig?.enableEnhancedProject
) {
// Use pre-computed enhanced metadata if available (this already contains MetadataMapping transforms)
const precomputedMetadata =
settings.enhancedProjectData.fileMetadataMap[filePath];
if (precomputedMetadata) {
// Use the pre-computed metadata directly since it already includes the original metadata + mappings
enhancedFileMetadata = precomputedMetadata;
}
// Use pre-computed project config data
const dirPath = filePath.substring(0, filePath.lastIndexOf("/"));
projectConfigData =
settings.enhancedProjectData.projectConfigMap[dirPath];
// Use pre-computed tgProject
const projectInfo =
settings.enhancedProjectData.fileProjectMap[filePath];
if (projectInfo) {
// The projectInfo.source contains either the actual type or the specific source
// We need to determine the type and appropriate display source
let actualType: "metadata" | "path" | "config" | "default";
let displaySource: string;
// If source is one of the type values, use it directly
if (
["metadata", "path", "config", "default"].includes(
projectInfo.source
)
) {
actualType = projectInfo.source as
| "metadata"
| "path"
| "config"
| "default";
}
// Otherwise, infer type from source characteristics
else if (
projectInfo.source &&
projectInfo.source.includes("/")
) {
// Path patterns contain "/"
actualType = "path";
} else if (
projectInfo.source &&
projectInfo.source.includes(".")
) {
// Config files contain "."
actualType = "config";
} else {
// Metadata keys are simple strings without "/" or "."
actualType = "metadata";
}
// Set appropriate display source based on type
switch (actualType) {
case "path":
displaySource = "path-mapping";
break;
case "metadata":
displaySource = "frontmatter";
break;
case "config":
displaySource = "config-file";
break;
case "default":
displaySource = "default-naming";
break;
}
tgProject = {
type: actualType,
name: projectInfo.project,
source: displaySource,
readonly: projectInfo.readonly,
};
}
}
// Use the parseLegacy method with enhanced data
// Raw parsing only - no project enhancement per dataflow architecture
// Project data will be handled by Augmentor in main thread
const tasks = parser.parseLegacy(
content,
filePath,
enhancedFileMetadata,
projectConfigData,
tgProject
fileMetadata,
undefined, // No project config in worker
undefined // No tgProject in worker
);
// Apply heading filters if specified

21
src/dataflow/workers/TaskWorkerManager.ts
View file

@ -973,15 +973,28 @@ export class TaskWorkerManager extends Component {
}
/**
* Set enhanced project data for worker processing
* @deprecated Project data is now handled by Augmentor in main thread per dataflow architecture.
* Workers only perform raw task extraction without project enhancement.
*/
public setEnhancedProjectData(
enhancedProjectData: import("./task-index-message").EnhancedProjectData
): void {
// Update the settings with enhanced project data
// NO-OP: Project data is handled by Augmentor, not Workers
// This method is kept for backward compatibility but does nothing
}
/**
* Update worker settings dynamically
*/
public updateSettings(settings: Partial<{
preferMetadataFormat: "dataview" | "tasks";
customDateFormats?: string[];
fileMetadataInheritance?: any;
projectConfig?: any;
}>): void {
// Update the settings
if (this.options.settings) {
(this.options.settings as any).enhancedProjectData =
enhancedProjectData;
Object.assign(this.options.settings, settings);
}
}

47
src/dataflow/workers/WorkerOrchestrator.ts
View file

@ -275,10 +275,49 @@ export class WorkerOrchestrator {
*/
private async parseFileTasksMainThread(file: TFile): Promise<Task[]> {
this.metrics.fallbackToMainThread++;
// This would typically delegate to the original synchronous parsers
// For now, return empty array as a safe fallback
console.warn(`WorkerOrchestrator: Main thread parsing not implemented for ${file.path}`);
return [];
console.warn(`WorkerOrchestrator: Falling back to main thread parsing for ${file.path}`);
// Import and use ConfigurableTaskParser for fallback
const { ConfigurableTaskParser } = await import("../core/ConfigurableTaskParser");
const { parseCanvas } = await import("../parsers/CanvasEntry");
const { parseFileMeta } = await import("../parsers/FileMetaEntry");
const extension = file.extension.toLowerCase();
let tasks: Task[] = [];
if (extension === "md") {
// Get necessary data
const vault = (this.taskWorkerManager as any).vault;
const metadataCache = (this.taskWorkerManager as any).metadataCache;
const content = await vault.cachedRead(file);
const fileCache = metadataCache.getFileCache(file);
const fileMetadata = fileCache?.frontmatter || {};
// Create parser with default settings
const parser = new ConfigurableTaskParser({
parseMetadata: true,
parseTags: true,
parseComments: true,
parseHeadings: true
});
// Parse tasks - raw extraction only, no project enhancement
// Project data will be handled by Augmentor per dataflow architecture
const markdownTasks = parser.parseLegacy(
content,
file.path,
fileMetadata,
undefined, // No project config in fallback
undefined // No tgProject in fallback
);
tasks.push(...markdownTasks);
} else if (extension === "canvas") {
// For canvas files, we need plugin instance
console.warn(`WorkerOrchestrator: Canvas parsing requires plugin instance, returning empty`);
}
return tasks;
}
private async batchParseMainThread(files: TFile[]): Promise<Map<string, Task[]>> {