diff --git a/docs/reference/sparql-anything.md b/docs/reference/sparql-anything.md index 6887d23f..cf439fec 100644 --- a/docs/reference/sparql-anything.md +++ b/docs/reference/sparql-anything.md @@ -41,13 +41,14 @@ await converter.convert( ### Options -| Option | Type | Description | -| ------------ | ------------------ | ---------------------------------------------------------------------------------------------- | -| `jarPath` | `string` | Path to the SPARQL Anything CLI jar, as the task runner sees it | -| `workDir` | `string` | The task runner's working directory; see [Where files are written](#where-files-are-written) | -| `heap` | `string` | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory) | -| `cliArgs` | `string[]` | Further arguments for the SPARQL Anything CLI; see [Memory](#memory) | -| `taskRunner` | `TaskRunner` | Runs the SPARQL Anything process for each chunk | +| Option | Type | Description | +| ------------- | ------------------ | ----------------------------------------------------------------------------------------------------------------------------- | +| `jarPath` | `string` | Path to the SPARQL Anything CLI jar, as the task runner sees it | +| `workDir` | `string` | The task runner's working directory; see [Where files are written](#where-files-are-written) | +| `heap` | `string` | Maximum JVM heap per chunk process, as `-Xmx` takes it (default `'2g'`); see [Memory](#memory) | +| `cliArgs` | `string[]` | Further arguments for the SPARQL Anything CLI; see [Memory](#memory) | +| `concurrency` | `number` | How many chunks to convert at once (default `1`); see [Converting several chunks at once](#converting-several-chunks-at-once) | +| `taskRunner` | `TaskRunner` | Runs the SPARQL Anything process for each chunk | ### Jobs @@ -85,6 +86,19 @@ heap: '4g', // -Xmx4g Size it with the chunk size. A chunk that outgrows the heap fails loudly – the JVM's `OutOfMemoryError` arrives in the output of a non-zero exit, which aborts the conversion – where an uncapped JVM instead grows until the OOM killer takes the whole container. +### Converting several chunks at once + +`concurrency` is how many chunks are converted at the same time. Each one is a JVM of its own, so it multiplies against `heap`: a run needs `concurrency × heap`, on the machine the **task runner** uses – which is not this process's machine when the runner is Docker or remote. + +That is why the default is `1` rather than something derived from the CPU count or a memory limit: the converter cannot see the machine its processes run on, so the number is the caller's to choose. `map.sh` sizes its pool from `nproc` capped by the cgroup limit, budgeting ~3 GB a worker; a caller who knows their deployment can do the same arithmetic and pass the result. + +Chunks of every job are converted through one pool, in the order the jobs and their chunks were given – a long job and a short one pack together rather than draining in phases. The output is concatenated in that same order, however the processes happened to finish. + +The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. A process that cannot be stopped – one that has just exited, say – does not change what is reported: the conversion failure is the one worth reading. + +> [!WARNING] +> A `DockerTaskRunner` configured with a `containerName` cannot be used with `concurrency` above one. It force-removes any container of that name before starting a task, so each chunk would destroy the container of the chunk before it. Leave `containerName` unset for a converter that runs chunks in parallel. + `cliArgs` is the escape hatch for CLI flags the converter does not model itself, appended to the arguments it sets. It cannot repeat those: `-q`, `-f`, `-o` and `-l` are rejected, in their long and `--flag=value` forms too, because the converter reads back the `--output` it named, in the `--format` it asked for – overriding either leaves it reporting an empty conversion, or concatenating fragments that are not N-Triples. SPARQL Anything documents repetition only for `-v` and `-c`, so a repeated flag has no defined winner to rely on. ## How a conversion runs diff --git a/packages/sparql-anything/src/sparql-anything-converter.ts b/packages/sparql-anything/src/sparql-anything-converter.ts index ba2e3b45..5c1c8a29 100644 --- a/packages/sparql-anything/src/sparql-anything-converter.ts +++ b/packages/sparql-anything/src/sparql-anything-converter.ts @@ -84,6 +84,14 @@ export interface SparqlAnythingConverterOptions { * are the converter's own, and are rejected here. */ cliArgs?: string[]; + /** + * How many chunks to convert at once. Each one is a JVM of its own, so this + * multiplies against {@link heap}: the memory a run needs is `concurrency × + * heap`, and the machine that has to hold it is the task runner's, not this + * process's. Left at one, chunks are converted one after another. + * @default 1 + */ + concurrency?: number; /** Runs the SPARQL Anything process for each chunk. */ taskRunner: TaskRunner; } @@ -98,6 +106,7 @@ export class SparqlAnythingConverter { private readonly workDir: string; private readonly heap: string; private readonly cliArgs: string[]; + private readonly concurrency: number; private readonly taskRunner: TaskRunner; constructor(options: SparqlAnythingConverterOptions) { @@ -120,6 +129,13 @@ export class SparqlAnythingConverter { ); } this.cliArgs = options.cliArgs ?? []; + const concurrency = options.concurrency ?? 1; + if (!Number.isInteger(concurrency) || concurrency < 1) { + throw new Error( + `‘${concurrency}’ is not a number of chunks to convert at once; give a whole number of one or more`, + ); + } + this.concurrency = concurrency; this.taskRunner = options.taskRunner; } @@ -146,38 +162,114 @@ export class SparqlAnythingConverter { const runDir = await mkdtemp(join(this.workDir, 'sparql-anything-')); const runDirName = basename(runDir); try { - const outputs: string[] = []; - let index = 0; - for (const { job, query } of planned) { - // Interpolated per chunk, at the point of use: holding a copy of the - // query for every chunk up front would scale with the input. - for (const chunk of job.chunks ?? [undefined]) { - const queryPath = join(runDirName, `query-${index}.rq`); - await writeFile( - join(this.workDir, queryPath), - chunk === undefined - ? query - : // A replacer function, so `$&` and friends in a chunk path are - // the characters they look like rather than replacement patterns. - query.replaceAll(SOURCE_PLACEHOLDER, () => chunk), - ); - const processOutput = join(runDirName, `output-${index}.nt`); - const task = await this.taskRunner.run( - this.command(queryPath, processOutput, job), - ); - // wait() rejects on a non-zero exit, aborting convert() before the - // crashed process's missing output can be silently concatenated. - await this.taskRunner.wait(task); - const processOutputPath = join(this.workDir, processOutput); - await assertNonEmpty(processOutputPath, job, chunk); - outputs.push(processOutputPath); - index++; + const count = await this.runAll(planned, runDirName); + // By index, not by completion: the order the jobs and their chunks were + // given is the order of the triples, however the processes finished. + await concatenate( + Array.from({ length: count }, (_, index) => + join(this.workDir, runDirName, `output-${index}.nt`), + ), + outputPath, + ); + } finally { + await rm(runDir, { recursive: true, force: true }); + } + } + + /** + * Runs every chunk, at most {@link concurrency} at a time, and returns how + * many processes that was. + * + * The first failure aborts the run: no further chunk is started, and the + * processes still going are stopped rather than left writing into a + * directory this is about to delete. + */ + private async runAll( + planned: PlannedJob[], + runDirName: string, + ): Promise { + const pending = processesOf(planned); + const state: RunState = { inFlight: new Set() }; + + // Pulled one at a time rather than with `for...of`: leaving a for-of early + // closes the iterator, so the first worker to give up would end the queue + // for the others, whatever the reason it stopped. + const convertChunks = async (): Promise => { + while (state.failure === undefined) { + const next = pending.next(); + if (next.done === true) { + return; } + try { + await this.convertChunk(next.value, runDirName, state); + } catch (error) { + state.failure ??= error; + await this.stopInFlight(state); + return; + } + } + }; + + await Promise.all( + Array.from({ length: this.concurrency }, () => convertChunks()), + ); + if (state.failure !== undefined) { + throw state.failure; + } + return countOf(planned); + } + + /** Converts one chunk, writing `output-.nt` in the run directory. */ + private async convertChunk( + { index, job, chunk, query }: PlannedProcess, + runDirName: string, + state: RunState, + ): Promise { + const queryPath = join(runDirName, `query-${index}.rq`); + await writeFile( + join(this.workDir, queryPath), + chunk === undefined + ? query + : // A replacer function, so `$&` and friends in a chunk path are the + // characters they look like rather than replacement patterns. + query.replaceAll(SOURCE_PLACEHOLDER, () => chunk), + ); + const output = join(runDirName, `output-${index}.nt`); + const task = await this.taskRunner.run( + this.command(queryPath, output, job), + ); + state.inFlight.add(task); + try { + if (state.failure !== undefined) { + // Started in the window between another chunk failing and this worker + // seeing it, so it is not in the set that failure stopped. + await this.stopQuietly(task); + return; } - await concatenate(outputs, outputPath); + // wait() rejects on a non-zero exit, aborting convert() before the + // crashed chunk's missing output can be silently concatenated. + await this.taskRunner.wait(task); } finally { - await rm(runDir, { recursive: true, force: true }); + state.inFlight.delete(task); } + await assertNonEmpty(join(this.workDir, output), job, chunk); + } + + /** Stops every process still running, so none outlives the run. */ + private async stopInFlight(state: RunState): Promise { + await Promise.all( + [...state.inFlight].map((task) => this.stopQuietly(task)), + ); + } + + /** + * Stops a task, ignoring a failure to stop it. Stopping is best effort by + * nature – a process that has just exited cannot be stopped, and reporting + * that would replace the failure that is actually worth reporting, and leave + * the other workers unawaited while the run directory is deleted. + */ + private async stopQuietly(task: Task): Promise { + await this.taskRunner.stop(task).catch(() => undefined); } /** The SPARQL Anything invocation for one job. */ @@ -203,6 +295,46 @@ export class SparqlAnythingConverter { } } +/** One SPARQL Anything process: a job's query, to run over one of its chunks. */ +interface PlannedProcess { + /** Position in the run, which orders the outputs and names their files. */ + index: number; + job: ConversionJob; + chunk?: string; + /** The job's query as written, with `{SOURCE}` still in it. */ + query: string; +} + +/** + * The processes the planned jobs call for, in order: one per chunk, and one + * for a job that has none. Lazy, so the workers pulling from it hold one + * process each rather than the whole run. + */ +function* processesOf(planned: PlannedJob[]): Generator { + let index = 0; + for (const { job, query } of planned) { + for (const chunk of job.chunks ?? [undefined]) { + yield { index: index++, job, chunk, query }; + } + } +} + +/** How many processes the planned jobs call for. */ +function countOf(planned: PlannedJob[]): number { + return planned.reduce( + (total, { job }) => total + (job.chunks?.length ?? 1), + 0, + ); +} + +/** What the workers of one run share. */ +interface RunState { + /** Tasks that have been started and not yet finished. */ + inFlight: Set; + /** The first failure, which aborts the run. */ + failure?: unknown; +} + /** A job whose query has been read, and checked against its chunks. */ interface PlannedJob { job: ConversionJob; diff --git a/packages/sparql-anything/test/sparql-anything-converter.test.ts b/packages/sparql-anything/test/sparql-anything-converter.test.ts index bda1cb11..6eb2feb8 100644 --- a/packages/sparql-anything/test/sparql-anything-converter.test.ts +++ b/packages/sparql-anything/test/sparql-anything-converter.test.ts @@ -31,6 +31,14 @@ interface FakeTaskRunnerOptions { * cannot be inspected – a symlink to itself, which fails `stat` with ELOOP. */ unreadableOutputContaining?: string; + /** Milliseconds `wait()` takes, per output path, to order completions. */ + waitFor?: Record; + /** Milliseconds `wait()` takes for any output not named in `waitFor`. */ + waitForAll?: number; + /** Milliseconds `run()` takes, per output path, before the task exists. */ + runFor?: Record; + /** When set, `stop()` rejects, as a runner does for a container already gone. */ + stopFails?: boolean; } /** @@ -45,6 +53,11 @@ interface FakeTaskRunnerOptions { class FakeTaskRunner implements TaskRunner<{ command: string }> { readonly commands: string[] = []; readonly queries: string[] = []; + /** Commands that were stopped before they finished. */ + readonly stopped: string[] = []; + /** The most processes that were ever in flight at the same time. */ + peakInFlight = 0; + private inFlight = 0; constructor( private readonly workDir: string, @@ -60,6 +73,8 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { const outputFile = tokenAfter(command, '--output'); if (outputFile) { await this.writeOutput(outputFile); + const delay = delayFor(this.options.runFor, outputFile) ?? 0; + await new Promise((resolve) => setTimeout(resolve, delay)); } return { command }; } @@ -91,18 +106,49 @@ class FakeTaskRunner implements TaskRunner<{ command: string }> { } async wait(task: { command: string }): Promise { - const { failOutputContaining } = this.options; - if (matches(task.command, failOutputContaining)) { - throw new Error('Process failed with code 1'); + this.inFlight++; + this.peakInFlight = Math.max(this.peakInFlight, this.inFlight); + try { + const delay = + delayFor(this.options.waitFor, tokenAfter(task.command, '--output')) ?? + this.options.waitForAll ?? + 0; + await new Promise((resolve) => setTimeout(resolve, delay)); + if (matches(task.command, this.options.failOutputContaining)) { + throw new Error('Process failed with code 1'); + } + return ''; + } finally { + this.inFlight--; } - return ''; } - async stop(): Promise { + async stop(task: { command: string }): Promise { + this.stopped.push(task.command); + if (this.options.stopFails === true) { + throw new Error('No such container'); + } return null; } } +/** + * The delay configured for an output path. Keyed by file name, while the + * command names it relative to the run directory. + */ +function delayFor( + delays: Record | undefined, + outputPath?: string, +): number | undefined { + if (delays === undefined || outputPath === undefined) { + return undefined; + } + const named = Object.entries(delays).find(([name]) => + outputPath.endsWith(name), + ); + return named?.[1]; +} + /** Whether `needle` is configured and `haystack` contains it. */ function matches(haystack: string, needle?: string): boolean { return needle !== undefined && haystack.includes(needle); @@ -446,6 +492,134 @@ describe('SparqlAnythingConverter', () => { ).rejects.toThrow('would go unread'); }); + it('converts several chunks at once, up to the configured concurrency', async () => { + // A delay every chunk shares, so the overlap does not depend on timing. + const taskRunner = new FakeTaskRunner(workDir, { waitForAll: 20 }); + const chunks = await writeChunks(6); + + await new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')); + + expect(taskRunner.commands).toHaveLength(6); + expect(taskRunner.peakInFlight).toBe(3); + }); + + it('converts one chunk at a time by default', async () => { + const taskRunner = new FakeTaskRunner(workDir); + const chunks = await writeChunks(3); + + await converterFor(taskRunner).convert( + [{ queryFile, chunks }], + join(workDir, 'output.nt'), + ); + + expect(taskRunner.peakInFlight).toBe(1); + }); + + it('concatenates in the order given, not the order they finished', async () => { + const chunks = await writeChunks(3); + // The first chunk finishes last. + const taskRunner = new FakeTaskRunner(workDir, { + waitFor: { 'output-0.nt': 30, 'output-1.nt': 10 }, + }); + const outputPath = join(workDir, 'output.nt'); + + await new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], outputPath); + + expect(await readFile(outputPath, 'utf-8')).toMatch( + /^sparql-anything-\S+\/output-0\.nt\n\nsparql-anything-\S+\/output-1\.nt\n\nsparql-anything-\S+\/output-2\.nt\n$/, + ); + }); + + it('stops the chunks still running when one fails', async () => { + const chunks = await writeChunks(4); + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + // The others outlive the failure, so they have to be stopped rather than + // left writing into the directory convert() is about to delete. + waitFor: { 'output-1.nt': 200, 'output-2.nt': 200 }, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + + expect(taskRunner.stopped).toHaveLength(2); + // The fourth chunk was never started. + expect(taskRunner.commands).toHaveLength(3); + }); + + it('reports the conversion failure even when stopping the others fails', async () => { + const chunks = await writeChunks(3); + // A task runner cannot stop a container that has already gone; saying so + // must not replace the failure that is worth reporting. + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + waitFor: { 'output-1.nt': 100, 'output-2.nt': 100 }, + stopFails: true, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 3, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + }); + + it('stops a chunk that started while the run was failing', async () => { + const chunks = await writeChunks(2); + // The second chunk's process appears only after the first has failed, so + // it is not among the ones that failure stopped. + const taskRunner = new FakeTaskRunner(workDir, { + failOutputContaining: 'output-0.nt', + runFor: { 'output-1.nt': 50 }, + }); + + await expect( + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 2, + taskRunner, + }).convert([{ queryFile, chunks }], join(workDir, 'output.nt')), + ).rejects.toThrow('Process failed'); + + expect(taskRunner.stopped).toEqual([ + expect.stringContaining('output-1.nt'), + ]); + }); + + it('rejects a concurrency that is not a whole number of processes', () => { + const taskRunner = new FakeTaskRunner(workDir); + + expect( + () => + new SparqlAnythingConverter({ + jarPath: '/bin/sparql-anything.jar', + workDir, + concurrency: 0, + taskRunner, + }), + ).toThrow('is not a number of chunks to convert at once'); + }); + it('refuses an empty job list rather than writing an empty output', async () => { const taskRunner = new FakeTaskRunner(workDir); const outputPath = join(workDir, 'output.nt');