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

feat: update stable indexer method

Browse source
This commit is contained in:
Quorafind 2025-04-07 10:02:32 +08:00
parent 895a46e29f
commit 08f02a270e
11 changed files with 492 additions and 693 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
package.json
View file

@ -35,6 +35,7 @@
"@codemirror/state": "^6.0.0",
"@codemirror/stream-parser": "https://github.com/lishid/stream-parser",
"@codemirror/view": "^6.0.0",
"@datastructures-js/queue": "^4.2.3",
"@types/node": "^16.11.6",
"@typescript-eslint/eslint-plugin": "^5.2.0",
"@typescript-eslint/parser": "^5.2.0",

8
pnpm-lock.yaml
View file

@ -42,6 +42,9 @@ importers:
'@codemirror/view':
specifier: ^6.0.0
version: 6.36.3
'@datastructures-js/queue':
specifier: ^4.2.3
version: 4.2.3
'@types/node':
specifier: ^16.11.6
version: 16.18.126
@ -149,6 +152,9 @@ packages:
'@codemirror/view@6.36.3':
resolution: {integrity: sha512-N2bilM47QWC8Hnx0rMdDxO2x2ImJ1FvZWXubwKgjeoOrWwEiFrtpA7SFHcuZ+o2Ze2VzbkgbzWVj4+V18LVkeg==}
'@datastructures-js/queue@4.2.3':
resolution: {integrity: sha512-GWVMorC/xi2V2ta+Z/CPgPGHL2ZJozcj48g7y2nIX5GIGZGRrbShSHgvMViJwHJurUzJYOdIdRZnWDRrROFwJA==}
'@electron/get@2.0.3':
resolution: {integrity: sha512-Qkzpg2s9GnVV2I2BjRksUi43U5e6+zaQMcjoJy0C+C5oxaKl+fmckGDQFtRpZpZV0NQekuZZ+tGz7EA9TVnQtQ==}
engines: {node: '>=12'}
@ -1570,6 +1576,8 @@ snapshots:
style-mod: 4.1.2
w3c-keyname: 2.2.8
'@datastructures-js/queue@4.2.3': {}
'@electron/get@2.0.3':
dependencies:
debug: 4.4.0

5
src/utils/types/worker.d.ts vendored Normal file
View file

@ -0,0 +1,5 @@
/** @hidden */
declare module "TaskIndexWorker" {
const WorkerFactory: new () => Worker;
export default WorkerFactory;
}

171
src/utils/workers/TaskIndexWorker.ts
View file

@ -4,63 +4,15 @@
import { FileStats } from "obsidian";
import { Task } from "../types/TaskIndex";
/** Command to parse task from a file */
export interface ParseTasksCommand {
type: "parseTasks";
filePath: string;
content: string;
stats: FileStats;
}
/** Command to batch index multiple files */
export interface BatchIndexCommand {
type: "batchIndex";
files: {
path: string;
content: string;
stats: FileStats;
}[];
}
/** Available commands that can be sent to the worker */
export type IndexerCommand = ParseTasksCommand | BatchIndexCommand;
/** Result of task parsing */
export interface TaskParseResult {
type: "parseResult";
filePath: string;
tasks: Task[];
stats: {
totalTasks: number;
completedTasks: number;
processingTimeMs: number;
};
}
/** Result of batch indexing */
export interface BatchIndexResult {
type: "batchResult";
results: {
filePath: string;
taskCount: number;
}[];
stats: {
totalFiles: number;
totalTasks: number;
processingTimeMs: number;
};
}
/** Result of a worker operation or error */
export type IndexerResult = TaskParseResult | BatchIndexResult | ErrorResult;
/** Error response */
export interface ErrorResult {
type: "error";
error: string;
filePath?: string;
}
import {
BatchIndexCommand,
BatchIndexResult,
ErrorResult,
IndexerCommand,
IndexerResult,
ParseTasksCommand,
TaskParseResult
} from "./TaskIndexWorkerMessage";
/**
* Regular expressions for parsing task components
@ -239,12 +191,23 @@ function getIndentLevel(line: string): number {
function processFile(
filePath: string,
content: string,
stats: FileStats
stats: FileStats,
metadata?: { listItems?: any[] }
): TaskParseResult {
const startTime = performance.now();
try {
const tasks = parseTasksFromContent(filePath, content);
// 如果提供了 listItems 元数据，优先利用它来构建任务
let tasks: Task[] = [];
if (metadata?.listItems && metadata.listItems.length > 0) {
// 使用 Obsidian 的元数据缓存来构建任务
tasks = parseTasksFromListItems(filePath, content, metadata.listItems);
} else {
// 回退到正则表达式解析
tasks = parseTasksFromContent(filePath, content);
}
const completedTasks = tasks.filter((t) => t.completed).length;
return {
@ -263,6 +226,93 @@ function processFile(
}
}
/**
* Parse tasks from Obsidian's ListItemCache
*/
function parseTasksFromListItems(filePath: string, content: string, listItems: any[]): Task[] {
const tasks: Task[] = [];
const lines = content.split(/\r?\n/);
// 遍历所有列表项，找出任务项
for (const item of listItems) {
// 只处理任务项（有task属性的列表项）
if (item.task !== undefined) {
const line = item.position?.start?.line;
if (line === undefined) continue;
const lineContent = lines[line];
if (!lineContent) continue;
// 基本任务信息
const task: Task = {
id: `${filePath}-L${line}`,
content: extractTaskContent(lineContent),
filePath,
line,
completed: item.task !== ' ', // 空格表示未完成
originalMarkdown: lineContent,
tags: [],
children: [],
};
// 提取元数据
extractDates(task, task.content);
extractTags(task, task.content);
extractContext(task, task.content);
extractPriority(task, task.content);
tasks.push(task);
}
}
// 构建父子关系
buildTaskHierarchyFromListItems(tasks, listItems);
return tasks;
}
/**
* 从任务文本中提取实际内容（移除checkbox部分）
*/
function extractTaskContent(line: string): string {
const taskMatch = line.match(TASK_REGEX);
if (taskMatch) {
return taskMatch[3].trim();
}
return line.trim();
}
/**
* 从 ListItemCache 构建任务层级关系
*/
function buildTaskHierarchyFromListItems(tasks: Task[], listItems: any[]): void {
// 创建行号到任务的映射
const lineToTask = new Map<number, Task>();
tasks.forEach(task => {
lineToTask.set(task.line, task);
});
// 建立父子关系
for (const item of listItems) {
if (item.task !== undefined) {
const line = item.position?.start?.line;
if (line === undefined) continue;
const task = lineToTask.get(line);
if (!task) continue;
// 查找父任务
if (item.parent > 0) { // 正数表示父项的行号
const parentTask = lineToTask.get(item.parent);
if (parentTask) {
task.parent = parentTask.id;
parentTask.children.push(task.id);
}
}
}
}
}
/**
* Process multiple files in batch
*/
@ -316,7 +366,8 @@ self.onmessage = async (event) => {
const result = processFile(
message.filePath,
message.content,
message.stats
message.stats,
message.metadata
);
self.postMessage(result);
} else if (message.type === "batchIndex") {

116
src/utils/workers/TaskIndexWorkerMessage.ts Normal file
View file

@ -0,0 +1,116 @@
/**
* Message types for task indexing worker communication
*/
import { CachedMetadata, FileStats, ListItemCache } from "obsidian";
import { Task } from "../types/TaskIndex";
/**
* Command to parse tasks from a file
*/
export interface ParseTasksCommand {
type: "parseTasks";
/** The file path being processed */
filePath: string;
/** The file contents to parse */
content: string;
/** File stats information */
stats: FileStats;
/** Additional metadata from Obsidian cache */
metadata?: {
/** List items from Obsidian's metadata cache */
listItems?: ListItemCache[];
/** Full file metadata cache */
fileCache?: CachedMetadata;
};
}
/**
* Command to batch index multiple files
*/
export interface BatchIndexCommand {
type: "batchIndex";
/** Files to process in batch */
files: {
/** The file path */
path: string;
/** The file content */
content: string;
/** File stats */
stats: FileStats;
/** Optional metadata */
metadata?: {
listItems?: ListItemCache[];
fileCache?: CachedMetadata;
};
}[];
}
/**
* Available commands that can be sent to the worker
*/
export type IndexerCommand = ParseTasksCommand | BatchIndexCommand;
/**
* Result of task parsing
*/
export interface TaskParseResult {
type: "parseResult";
/** Path of the file that was processed */
filePath: string;
/** Tasks extracted from the file */
tasks: Task[];
/** Statistics about the parsing operation */
stats: {
/** Total number of tasks found */
totalTasks: number;
/** Number of completed tasks */
completedTasks: number;
/** Time taken to process in milliseconds */
processingTimeMs: number;
};
}
/**
* Result of batch indexing
*/
export interface BatchIndexResult {
type: "batchResult";
/** Results for each file processed */
results: {
/** File path */
filePath: string;
/** Number of tasks found */
taskCount: number;
}[];
/** Aggregated statistics */
stats: {
/** Total number of files processed */
totalFiles: number;
/** Total number of tasks found across all files */
totalTasks: number;
/** Total processing time in milliseconds */
processingTimeMs: number;
};
}
/**
* Error response
*/
export interface ErrorResult {
type: "error";
/** Error message */
error: string;
/** File path that caused the error (if available) */
filePath?: string;
}
/**
* All possible results from the worker
*/
export type IndexerResult = TaskParseResult | BatchIndexResult | ErrorResult;

543
src/utils/workers/TaskWorkerManager.ts
View file

@ -2,7 +2,7 @@
* Manager for task indexing web workers
*/
import { TFile, Vault } from "obsidian";
import { Component, ListItemCache, MetadataCache, TFile, Vault } from "obsidian";
import { Task } from "../types/TaskIndex";
import {
BatchIndexCommand,
@ -12,10 +12,14 @@ import {
IndexerResult,
ParseTasksCommand,
TaskParseResult,
} from "./TaskIndexWorker";
} from "./TaskIndexWorkerMessage";
// Import worker
import TaskWorker from "./TaskIndexWorker";
// Import worker and utilities
import TaskWorker from "TaskIndexWorker";
import { Deferred, deferred } from "./deferred";
// Using similar queue structure as importer.ts
import { Queue } from "@datastructures-js/queue";
/**
* Options for worker pool
@ -39,57 +43,64 @@ export const DEFAULT_WORKER_OPTIONS: WorkerPoolOptions = {
};
/**
* Task callback type
* A worker in the pool of executing workers
*/
export type TaskCallback<T> = (error: Error | null, result?: T) => void;
/**
* Task request interface
*/
interface TaskRequest {
/** Command to process */
command: IndexerCommand;
/** Callback to call with result */
callback: TaskCallback<IndexerResult>;
interface PoolWorker {
/** The id of this worker */
id: number;
/** The raw underlying worker */
worker: Worker;
/** UNIX time indicating the next time this worker is available for execution */
availableAt: number;
/** The active task this worker is processing, if any */
active?: [TFile, Deferred<any>, number];
}
/**
* Worker data for pool
* Task metadata from Obsidian cache
*/
interface WorkerData {
/** Worker instance */
worker: Worker;
/** Whether the worker is busy */
busy: boolean;
/** Last time this worker finished a task */
lastTaskFinished: number;
/** Number of tasks processed */
tasksProcessed: number;
/** Total processing time */
totalProcessingTime: number;
interface TaskMetadata {
/** List item cache information */
listItems?: ListItemCache[];
/** Raw file content */
content: string;
/** File stats */
stats: {
ctime: number;
mtime: number;
size: number;
};
}
/**
* Worker pool for task processing
*/
export class TaskWorkerManager {
export class TaskWorkerManager extends Component {
/** Worker pool */
private workers: WorkerData[] = [];
/** Queue of pending tasks */
private taskQueue: TaskRequest[] = [];
private workers: Map<number, PoolWorker> = new Map();
/** Task queue */
private queue: Queue<[TFile, Deferred<any>]> = new Queue();
/** Map of outstanding tasks by file path */
private outstanding: Map<string, Promise<any>> = new Map();
/** Whether the pool is currently active */
private active: boolean = true;
/** Options for the worker pool */
/** Worker pool options */
private options: WorkerPoolOptions;
/** Vault instance */
private vault: Vault;
/** Metadata cache for accessing file metadata */
private metadataCache: MetadataCache;
/** Next worker ID to assign */
private nextWorkerId: number = 0;
/**
* Create a new worker pool
*/
constructor(vault: Vault, options: Partial<WorkerPoolOptions> = {}) {
constructor(vault: Vault, metadataCache: MetadataCache, options: Partial<WorkerPoolOptions> = {}) {
super();
this.options = { ...DEFAULT_WORKER_OPTIONS, ...options };
this.vault = vault;
this.metadataCache = metadataCache;
// Initialize workers up to max
this.initializeWorkers();
@ -106,336 +117,282 @@ export class TaskWorkerManager {
for (let i = 0; i < workerCount; i++) {
try {
const worker = new Worker(
URL.createObjectURL(
new Blob([`importScripts('${TaskWorker}')`], {
type: "application/javascript",
})
)
);
// Setup worker message handler
worker.onmessage = (event) => {
this.handleWorkerMessage(worker, event.data);
};
worker.onerror = (event) => {
console.error("Worker error:", event);
};
this.workers.push({
worker,
busy: false,
lastTaskFinished: Date.now(),
tasksProcessed: 0,
totalProcessingTime: 0,
});
this.log(`Initialized worker #${i + 1}`);
const worker = this.newWorker();
this.workers.set(worker.id, worker);
this.log(`Initialized worker #${worker.id}`);
} catch (error) {
console.error("Failed to initialize worker:", error);
}
}
this.log(
`Initialized ${this.workers.length} workers (requested ${workerCount})`
`Initialized ${this.workers.size} workers (requested ${workerCount})`
);
// Check if we have any workers
if (this.workers.length === 0) {
if (this.workers.size === 0) {
console.warn(
"No workers could be initialized, falling back to main thread processing"
);
}
}
/**
* Create a new worker
*/
private newWorker(): PoolWorker {
const workerId = this.nextWorkerId++;
let worker = new Worker(
URL.createObjectURL(
new Blob([`importScripts('${TaskWorker}')`], {
type: "application/javascript",
})
)
);
const poolWorker: PoolWorker = {
id: workerId,
worker,
availableAt: Date.now(),
};
worker.onmessage = (evt) => this.finish(poolWorker, evt.data);
worker.onerror = (event) => {
console.error("Worker error:", event);
// If there's an active task, reject it
if (poolWorker.active) {
poolWorker.active[1].reject("Worker error");
poolWorker.active = undefined;
}
};
return poolWorker;
}
/**
* Process a single file for tasks
*/
public processFile(file: TFile, callback: TaskCallback<Task[]>): void {
this.vault
.cachedRead(file)
.then((content) => {
// Create file stats (adapting to the worker's needs)
const stats = {
ctime: file.stat.ctime,
mtime: file.stat.mtime,
size: file.stat.size,
};
public processFile(file: TFile): Promise<Task[]> {
// De-bounce repeated requests for the same file
let existing = this.outstanding.get(file.path);
if (existing) return existing;
const command: ParseTasksCommand = {
type: "parseTasks",
filePath: file.path,
content,
stats,
};
let promise = deferred<Task[]>();
this.queueTask(command, (error, result) => {
if (error) {
callback(error);
return;
}
if (!result || result.type === "error") {
callback(
new Error(
(result as ErrorResult)?.error ||
"Unknown error"
)
);
return;
}
if (result.type === "parseResult") {
const parseResult = result as TaskParseResult;
callback(null, parseResult.tasks);
} else {
callback(
new Error(`Unexpected result type: ${result.type}`)
);
}
});
})
.catch((error) => {
callback(error);
});
this.outstanding.set(file.path, promise);
this.queue.enqueue([file, promise]);
this.schedule();
return promise;
}
/**
* Process multiple files in a batch
*/
public processBatch(
files: TFile[],
callback: TaskCallback<Map<string, Task[]>>
): void {
// Read all files first
Promise.all(
files.map((file) =>
this.vault.cachedRead(file).then((content) => ({
file,
content,
}))
)
)
.then((fileContents) => {
const command: BatchIndexCommand = {
type: "batchIndex",
files: fileContents.map(({ file, content }) => ({
path: file.path,
content,
stats: {
ctime: file.stat.ctime,
mtime: file.stat.mtime,
size: file.stat.size,
},
})),
};
this.queueTask(command, (error, result) => {
if (error) {
callback(error);
return;
}
if (!result || result.type === "error") {
callback(
new Error(
(result as ErrorResult)?.error ||
"Unknown error"
)
);
return;
}
if (result.type === "batchResult") {
const batchResult = result as BatchIndexResult;
// This is just a summary, not the actual tasks
callback(
new Error("Batch result does not include tasks")
);
} else if (result.type === "parseResult") {
// Single file result, convert to map
const parseResult = result as TaskParseResult;
const resultMap = new Map<string, Task[]>();
resultMap.set(parseResult.filePath, parseResult.tasks);
callback(null, resultMap);
} else {
callback(
new Error(`Unexpected result type: ${result.type}`)
);
}
});
})
.catch((error) => {
callback(error);
public processBatch(files: TFile[]): Promise<Map<string, Task[]>> {
const promises: Promise<Task[]>[] = [];
// Queue each file for processing
for (const file of files) {
promises.push(this.processFile(file));
}
// Combine all results into a map
return Promise.all(promises).then(results => {
const resultMap = new Map<string, Task[]>();
files.forEach((file, index) => {
resultMap.set(file.path, results[index]);
});
}
/**
* Queue a task for processing by a worker
*/
private queueTask(
command: IndexerCommand,
callback: TaskCallback<IndexerResult>
): void {
this.taskQueue.push({
command,
callback,
return resultMap;
});
this.processQueue();
}
/**
* Process the task queue
* Get task metadata from the file and Obsidian cache
*/
private processQueue(): void {
if (!this.active || this.taskQueue.length === 0) {
return;
}
private async getTaskMetadata(file: TFile): Promise<TaskMetadata> {
// Get file content
const content = await this.vault.cachedRead(file);
// Get file metadata from Obsidian cache
const fileCache = this.metadataCache.getFileCache(file);
return {
listItems: fileCache?.listItems,
content,
stats: {
ctime: file.stat.ctime,
mtime: file.stat.mtime,
size: file.stat.size,
}
};
}
// Try to find an available worker
const availableWorker = this.getAvailableWorker();
if (!availableWorker) {
// No available workers, will retry when a worker becomes available
return;
}
/**
* Execute next task from the queue
*/
private schedule(): void {
if (this.queue.size() === 0 || !this.active) return;
// Get the next task
const task = this.taskQueue.shift();
if (!task) {
return;
}
const worker = this.availableWorker();
if (!worker) return;
// Mark the worker as busy
availableWorker.busy = true;
const [file, promise] = this.queue.dequeue()!;
worker.active = [file, promise, Date.now()];
try {
// Send the task to the worker
availableWorker.worker.postMessage(task.command);
// Store the callback with the worker so we can call it when the worker responds
(availableWorker as any).currentTask = task;
(availableWorker as any).taskStartTime = Date.now();
this.log(`Sent task to worker: ${task.command.type}`);
this.getTaskMetadata(file).then(metadata => {
const command: ParseTasksCommand = {
type: "parseTasks",
filePath: file.path,
content: metadata.content,
stats: metadata.stats,
metadata: {
listItems: metadata.listItems || [],
fileCache: this.metadataCache.getFileCache(file) || undefined
}
};
worker.worker.postMessage(command);
}).catch(error => {
console.error(`Error reading file ${file.path}:`, error);
promise.reject(error);
worker.active = undefined;
// Try to process next task
this.schedule();
});
} catch (error) {
// Failed to send task to worker
console.error("Failed to send task to worker:", error);
// Mark the worker as available again
availableWorker.busy = false;
// Call the callback with the error
task.callback(error as Error);
// Try to process the next task
this.processQueue();
console.error(`Error processing file ${file.path}:`, error);
promise.reject(error);
worker.active = undefined;
// Try to process next task
this.schedule();
}
}
/**
* Handle a message from a worker
* Handle worker completion and process result
*/
private handleWorkerMessage(worker: Worker, data: IndexerResult): void {
// Find the worker data
const workerData = this.workers.find((w) => w.worker === worker);
if (!workerData) {
console.error("Received message from unknown worker:", data);
private finish(worker: PoolWorker, data: IndexerResult): void {
if (!worker.active) {
console.log(
"Received a stale worker message. Ignoring.",
data
);
return;
}
// Get the task that was being processed
const task = (workerData as any).currentTask as TaskRequest | undefined;
const taskStartTime = (workerData as any).taskStartTime as
| number
| undefined;
const [file, promise, start] = worker.active;
// Update worker stats
workerData.busy = false;
workerData.lastTaskFinished = Date.now();
workerData.tasksProcessed++;
if (taskStartTime) {
const processingTime = Date.now() - taskStartTime;
workerData.totalProcessingTime += processingTime;
// Apply throttling based on CPU utilization
const delay = Math.round(
(processingTime * (1 - this.options.cpuUtilization)) /
this.options.cpuUtilization
);
if (delay > 0) {
// Delay before this worker processes another task
setTimeout(() => {
// Process next task if there are any
this.processQueue();
}, delay);
this.log(
`Worker throttled for ${delay}ms (processed in ${processingTime}ms)`
);
return;
}
}
// Clear the current task
(workerData as any).currentTask = undefined;
(workerData as any).taskStartTime = undefined;
// Call the callback with the result
if (task) {
if (data.type === "error") {
task.callback(new Error((data as ErrorResult).error));
} else {
task.callback(null, data);
}
// Resolve or reject the promise based on result
if (data.type === "error") {
promise.reject(new Error((data as ErrorResult).error));
} else if (data.type === "parseResult") {
const parseResult = data as TaskParseResult;
promise.resolve(parseResult.tasks);
} else if (data.type === "batchResult") {
// For batch results, we handle differently as we don't have tasks directly
promise.reject(new Error("Batch results should be handled by processBatch"));
} else {
console.error(
"Received message from worker with no associated task:",
data
);
promise.reject(new Error(`Unexpected result type: ${(data as any).type}`));
}
// Process next task if there are any
this.processQueue();
// Remove from outstanding tasks
this.outstanding.delete(file.path);
// Check if we should remove this worker (if we're over capacity)
if (this.workers.size > this.options.maxWorkers) {
this.workers.delete(worker.id);
this.terminate(worker);
} else {
// Calculate delay based on CPU utilization target
const now = Date.now();
const processingTime = now - start;
const throttle = Math.max(0.1, this.options.cpuUtilization) - 1.0;
const delay = processingTime * throttle;
worker.active = undefined;
if (delay <= 0) {
worker.availableAt = now;
this.schedule();
} else {
worker.availableAt = now + delay;
setTimeout(() => this.schedule(), delay);
}
}
}
/**
* Get an available worker
*/
private getAvailableWorker(): WorkerData | undefined {
// First, look for a non-busy worker
for (const worker of this.workers) {
if (!worker.busy) {
private availableWorker(): PoolWorker | undefined {
const now = Date.now();
// Find a worker that's not busy and is available
for (const worker of this.workers.values()) {
if (!worker.active && worker.availableAt <= now) {
return worker;
}
}
// Create a new worker if we haven't reached capacity
if (this.workers.size < this.options.maxWorkers) {
const worker = this.newWorker();
this.workers.set(worker.id, worker);
return worker;
}
return undefined;
}
/**
* Terminate a worker
*/
private terminate(worker: PoolWorker): void {
worker.worker.terminate();
if (worker.active) {
worker.active[1].reject("Terminated");
worker.active = undefined;
}
this.log(`Terminated worker #${worker.id}`);
}
/**
* Reset throttling for all workers
*/
public unthrottle(): void {
const now = Date.now();
for (const worker of this.workers.values()) {
worker.availableAt = now;
}
this.schedule();
}
/**
* Shutdown the worker pool
*/
public shutdown(): void {
public onunload(): void {
this.active = false;
// Terminate all workers
for (const worker of this.workers) {
worker.worker.terminate();
for (const worker of this.workers.values()) {
this.terminate(worker);
this.workers.delete(worker.id);
}
// Clear the workers array
this.workers = [];
// Clear the task queue and call all callbacks with an error
for (const task of this.taskQueue) {
task.callback(new Error("Worker pool shut down"));
// Clear all remaining queued tasks and reject their promises
while (!this.queue.isEmpty()) {
const [_, promise] = this.queue.dequeue()!;
promise.reject("Terminated");
}
this.taskQueue = [];
this.log("Worker pool shut down");
}
@ -443,7 +400,7 @@ export class TaskWorkerManager {
* Get the number of pending tasks
*/
public getPendingTaskCount(): number {
return this.taskQueue.length;
return this.queue.size();
}
/**

234
src/utils/workers/importer.ts
View file

@ -1,234 +0,0 @@
/** Controls and creates Dataview file importers, allowing for asynchronous loading and parsing of files. */
import ImportWorker from "index/web-worker/importer.worker";
import { Component, FileManager, MetadataCache, TFile, Vault } from "obsidian";
import { CanvasImport, MarkdownImport } from "index/web-worker/message";
import { Deferred, deferred } from "utils/deferred";
import { Queue } from "@datastructures-js/queue";
/** Settings for throttling import. */
export interface ImportThrottle {
/** The number of workers to use for imports. */
workers: number;
/** A number between 0.1 and 1 which indicates total cpu utilization target; 0.1 means spend 10% of time */
utilization: number;
}
/** Default throttle configuration. */
export const DEFAULT_THROTTLE: ImportThrottle = {
workers: 2,
utilization: 0.75,
};
/** Multi-threaded file parser which debounces rapid file requests automatically. */
export class FileImporter extends Component {
/* Background workers which do the actual file parsing. */
workers: Map<number, PoolWorker>;
/** The next worker ID to hand out. */
nextWorkerId: number;
/** If true, the importer is now inactive and will not process further files. */
shutdown: boolean;
/** List of files which have been queued for a reload. */
queue: Queue<[TFile, Deferred<any>]>;
/** Outstanding loads indexed by path. */
outstanding: Map<string, Promise<any>>;
/** Throttle settings. */
throttle: () => ImportThrottle;
public constructor(
public vault: Vault,
public fileManager: FileManager,
public metadataCache: MetadataCache,
throttle?: () => ImportThrottle
) {
super();
this.workers = new Map();
this.shutdown = false;
this.nextWorkerId = 0;
this.throttle = throttle ?? (() => DEFAULT_THROTTLE);
this.queue = new Queue();
this.outstanding = new Map();
}
/**
* Queue the given file for importing. Multiple import requests for the same file in a short time period will be de-bounced
* and all be resolved by a single actual file reload.
*/
public import<T>(file: TFile): Promise<T> {
// De-bounce repeated requests for the same file.
let existing = this.outstanding.get(file.path);
if (existing) return existing;
let promise = deferred<T>();
this.outstanding.set(file.path, promise);
this.queue.enqueue([file, promise]);
this.schedule();
return promise;
}
/** Reset any active throttles on the importer (such as if the utilization changes). */
public unthrottle() {
for (let worker of this.workers.values()) {
worker.availableAt = Date.now();
}
}
/** Poll from the queue and execute if there is an available worker. */
private async schedule() {
if (this.queue.size() == 0 || this.shutdown) return;
const worker = this.availableWorker();
if (!worker) return;
const [file, promise] = this.queue.dequeue()!;
worker.active = [file, promise, Date.now()];
try {
switch (file.extension) {
case "markdown":
case "md": {
const contents = await this.vault.cachedRead(file);
worker!.worker.postMessage({
type: "markdown",
path: file.path,
contents: contents,
stat: file.stat,
metadata: this.metadataCache.getFileCache(file),
} as MarkdownImport);
break;
}
case "canvas": {
const contents = await this.vault.cachedRead(file);
worker!.worker.postMessage({
type: "canvas",
path: file.path,
contents: contents,
stat: file.stat,
index: this.fileManager.linkUpdaters.canvas.canvas.index
.index[file.path],
} as CanvasImport);
break;
}
}
} catch (ex) {
console.log("Datacore: Background file reloading failed. " + ex);
// Message failed, release this worker.
worker.active = undefined;
}
}
/** Finish the parsing of a file, potentially queueing a new file. */
private finish(worker: PoolWorker, data: any) {
if (!worker.active) {
console.log(
"Datacore: Received a stale worker message. Ignoring.",
data
);
return;
}
const [file, promise, start] = worker.active!;
// Resolve promises to let users know this file has finished.
if ("$error" in data) promise.reject(data["$error"]);
else promise.resolve(data);
// Remove file from outstanding.
this.outstanding.delete(file.path);
// Remove this worker if we are over capacity.
// Otherwise, notify the queue this file is available for new work.
if (this.workers.size > this.throttle().workers) {
this.workers.delete(worker.id);
terminate(worker);
} else {
const now = Date.now();
const throttle = Math.max(0.1, this.throttle().utilization) - 1.0;
const delay = (now - start) * throttle;
worker.active = undefined;
if (delay <= 1e-10) {
worker.availableAt = now;
this.schedule();
} else {
worker.availableAt = now + delay;
// Note: I'm pretty sure this will garauntee that this executes AFTER delay milliseconds,
// so this should be fine; if it's not, we'll have to swap to an external timeout loop
// which infinitely reschedules itself to the next available execution time.
setTimeout(this.schedule.bind(this), delay);
}
}
}
/** Obtain an available worker, returning undefined if one does not exist. */
private availableWorker(): PoolWorker | undefined {
const now = Date.now();
for (let worker of this.workers.values()) {
if (!worker.active && worker.availableAt <= now) {
return worker;
}
}
// Make a new worker if we can.
if (this.workers.size < this.throttle().workers) {
let worker = this.newWorker();
this.workers.set(worker.id, worker);
return worker;
}
return undefined;
}
/** Create a new worker bound to this importer. */
private newWorker(): PoolWorker {
let worker: PoolWorker = {
id: this.nextWorkerId++,
availableAt: Date.now(),
worker: new ImportWorker(),
};
worker.worker.onmessage = (evt) => this.finish(worker, evt.data);
return worker;
}
/** Reject all outstanding promises and close all workers on close. */
public onunload(): void {
for (let worker of this.workers.values()) {
terminate(worker);
}
while (!this.queue.isEmpty()) {
const [_file, promise] = this.queue.pop()!;
promise.reject("Terminated");
}
this.shutdown = true;
}
}
/** A worker in the pool of executing workers. */
interface PoolWorker {
/** The id of this worker. */
id: number;
/** The raw underlying worker. */
worker: Worker;
/** UNIX time indicating the next time this worker is available for execution according to target utilization. */
availableAt: number;
/** The active promise this worker is working on, if any. */
active?: [TFile, Deferred<any>, number];
}
/** Terminate a pool worker. */
function terminate(worker: PoolWorker) {
worker.worker.terminate();
if (worker.active) worker.active[1].reject("Terminated");
worker.active = undefined;
}

47
src/utils/workers/importer.worker.ts
View file

@ -1,47 +0,0 @@
import { canvasImport } from "index/import/canvas";
import { markdownImport } from "index/import/markdown";
import {
CanvasImportResult,
ImportCommand,
MarkdownImportResult,
} from "index/web-worker/message";
/** Web worker entry point for importing. */
onmessage = async (event) => {
try {
const message = event.data as ImportCommand;
if (message.type === "markdown") {
const markdown = markdownImport(
message.path,
message.contents,
message.metadata,
message.stat
);
postMessage({
type: "markdown",
result: markdown,
} as MarkdownImportResult);
} else if (message.type === "canvas") {
const canvas = canvasImport(
message.path,
message.contents,
message.index,
message.stat
);
postMessage({
type: "canvas",
result: canvas,
} as CanvasImportResult);
} else {
postMessage({ $error: "Unsupported import method." });
}
} catch (error) {
console.error(
`Datacore Indexer failed to index ${event.data.path}: ${error}`
);
postMessage({ $error: error.message });
}
};

58
src/utils/workers/message.ts
View file

@ -1,58 +0,0 @@
import { CanvasMetadataIndex, JsonCanvas } from "index/types/json/canvas";
import { JsonMarkdownPage } from "index/types/json/markdown";
import { CachedMetadata, FileStats } from "obsidian";
/** A command to import a markdown file. */
export interface MarkdownImport {
type: "markdown";
/** The path we are importing. */
path: string;
/** The file contents to import. */
contents: string;
/** The stats for the file. */
stat: FileStats;
/** Metadata for the file. */
metadata: CachedMetadata;
}
/** A command to import a canvas file. */
export interface CanvasImport {
type: "canvas";
/** The path we are importing. */
path: string;
/** The raw JSON contents we are importing. */
contents: string;
/** The stats for the file. */
stat: FileStats;
/** the canvas's metadata cache */
index: CanvasMetadataIndex["any"];
}
/** Available import commands to be sent to an import web worker. */
export type ImportCommand = MarkdownImport | CanvasImport;
/** The result of importing a file of some variety. */
export interface MarkdownImportResult {
/** The type of import. */
type: "markdown";
/** The result of importing. */
result: JsonMarkdownPage;
}
export interface CanvasImportResult {
type: "canvas";
result: JsonCanvas;
}
export interface ImportFailure {
/** Failed to import. */
type: "error";
/** The error that the worker indicated on failure. */
$error: string;
}
export type ImportResult =
| MarkdownImportResult
| CanvasImportResult
| ImportFailure;

0
src/utils/workers/normalizers.ts
View file

2
tsconfig.json
View file

@ -13,5 +13,5 @@
"allowSyntheticDefaultImports": true,
"types": ["obsidian-typings"]
},
"include": ["**/*.ts"]
"include": ["**/*.ts", "src/utils/types/**/*.d.ts"]
}