diff --git a/app/services/fragment_merger.ts b/app/services/fragment_merger.ts index 3aef3bb..16a9d1b 100644 --- a/app/services/fragment_merger.ts +++ b/app/services/fragment_merger.ts @@ -15,7 +15,7 @@ export interface MergeResult { } /** File keys per round. */ -const BATCH = 2000 +const BATCH = 20_000 /** * Files cut by a range's edge (REBUILD.md, fragments), put together once @@ -25,12 +25,18 @@ const BATCH = 2000 * incomplete, and dropped (counted on its ranges). Others wait. */ class FragmentMerger { + /** Where the next round goes on: waiting files must not take every round's places. */ + private cursor = '' + async run(): Promise { const result: MergeResult = { files: 0, completed: 0, incomplete: 0, waiting: 0 } const { rows: keys } = await db.rawQuery( - `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id WHERE r.status = 'done' LIMIT ?`, - [BATCH] + `SELECT DISTINCT f.file_key FROM fragments f JOIN ranges r ON r.id = f.range_id + WHERE r.status = 'done' AND f.file_key > ? ORDER BY f.file_key LIMIT ?`, + [this.cursor, BATCH] ) + // At the end of the keys: the next round starts over. + this.cursor = keys.length < BATCH ? '' : keys.at(-1).file_key if (keys.length === 0) return result const rules = new RuleSet(await RuleSets.current()) const rows = await db @@ -47,6 +53,7 @@ class FragmentMerger { const completed: { file: AggregatedFile; rangeId: number; ids: number[] }[] = [] const incomplete: { rangeId: number; ids: number[]; segments: number }[] = [] + const neighbours = await this.neighbourStatus([...byKey.values()]) for (const list of byKey.values()) { result.files++ const file = mergeFragments(list.map((r) => r.fragment as Fragment)) @@ -56,7 +63,7 @@ class FragmentMerger { completed.push({ file, rangeId, ids }) continue } - if (list.every((r) => r.status === 'done') && (await this.neighboursDone(list))) { + if (list.every((r) => r.status === 'done') && neighbours(list)) { incomplete.push({ rangeId, ids, segments: list.reduce((n, r) => n + (r.fragment as Fragment).segments.length, 0) }) } else { result.waiting++ @@ -96,21 +103,40 @@ class FragmentMerger { } /** - * Whether the ranges right before and after these fragments' ranges (where - * a missing part could be) are done. One that isn't planned yet counts as - * not done (newer articles come later), except before the group's oldest article. + * For each file's fragments: whether the ranges right before and after + * their ranges (where a missing part could be) are done, read in one + * query for the whole round (per file it was three, ~2,000 files a + * minute). A neighbour not planned yet counts as not done (newer articles + * come later), except before the group's oldest article. */ - private async neighboursDone(list: { group_id: number; start_id: number | string; end_id: number | string }[]) { - const groupId = list[0].group_id - const first = Math.min(...list.map((r) => Number(r.start_id))) - const last = Math.max(...list.map((r) => Number(r.end_id))) - const [before, after, group] = await Promise.all([ - db.from('ranges').where('group_id', groupId).where('end_id', first - 1).select('status').first(), - db.from('ranges').where('group_id', groupId).where('start_id', last + 1).select('status').first(), - db.from('groups').where('id', groupId).select('low').first(), - ]) - const beforeDone = before ? before.status === 'done' : group?.low === null || first <= Number(group?.low ?? 0) - return beforeDone && after?.status === 'done' + private async neighbourStatus(lists: { group_id: number; start_id: number | string; end_id: number | string }[][]) { + const span = (list: (typeof lists)[number]) => ({ + group: Number(list[0].group_id), + first: Math.min(...list.map((r) => Number(r.start_id))), + last: Math.max(...list.map((r) => Number(r.end_id))), + }) + const spans = lists.map(span) + const groups = [...new Set(spans.map((s) => s.group))] + const { rows } = await db.rawQuery( + `SELECT r.group_id, r.start_id, r.end_id, r.status FROM ranges r + JOIN unnest(?::int[], ?::bigint[], ?::bigint[]) AS s(group_id, before_end, after_start) + ON r.group_id = s.group_id AND (r.end_id = s.before_end OR r.start_id = s.after_start)`, + [spans.map((s) => s.group), spans.map((s) => s.first - 1), spans.map((s) => s.last + 1)] + ) + const byEnd = new Map() + const byStart = new Map() + for (const r of rows) { + byEnd.set(`${r.group_id}:${r.end_id}`, r.status) + byStart.set(`${r.group_id}:${r.start_id}`, r.status) + } + const lows = new Map((await db.from('groups').whereIn('id', groups).select('id', 'low')).map((g) => [Number(g.id), g.low === null ? null : Number(g.low)])) + return (list: (typeof lists)[number]) => { + const { group, first, last } = span(list) + const before = byEnd.get(`${group}:${first - 1}`) + const low = lows.get(group) + const beforeDone = before !== undefined ? before === 'done' : low === null || low === undefined || first <= low + return beforeDone && byStart.get(`${group}:${last + 1}`) === 'done' + } } } diff --git a/app/services/planner.ts b/app/services/planner.ts index 9ac9478..8af9888 100644 --- a/app/services/planner.ts +++ b/app/services/planner.ts @@ -1,4 +1,5 @@ import db from '@adonisjs/lucid/services/db' +import logger from '@adonisjs/core/services/logger' import coordinatorConfig from '#config/coordinator' import NntpPool from '#services/nntp_pool' @@ -56,7 +57,14 @@ class Planner { async planAll(): Promise<{ groups: number; ranges: number }> { const groups = await db.from('groups').where('active', true).orderBy('priority').select('id', 'name') let ranges = 0 - for (const group of groups) ranges += await this.planGroup({ id: Number(group.id), name: group.name }) + for (const group of groups) { + // One group failing (e.g. gone from the server) doesn't stop the others. + try { + ranges += await this.planGroup({ id: Number(group.id), name: group.name }) + } catch (error: any) { + logger.warn(`Planning ${group.name}: ${error.message}`) + } + } return { groups: groups.length, ranges } } } diff --git a/commands/coordinator_run.ts b/commands/coordinator_run.ts index a05c6c6..d46a9f1 100644 --- a/commands/coordinator_run.ts +++ b/commands/coordinator_run.ts @@ -9,6 +9,7 @@ import NntpPool from '#services/nntp_pool' const TICK_MS = 2000 const PLAN_EVERY_MS = 60 * 60 * 1000 const MERGE_EVERY_MS = 60 * 1000 +const MERGE_BUDGET_MS = 30 * 1000 /** * The coordinator's background work (REBUILD.md, part 1): the prefetcher, @@ -44,8 +45,16 @@ export default class CoordinatorRun extends BaseCommand { } if (Date.now() - mergedAt >= MERGE_EVERY_MS) { mergedAt = Date.now() - const merged = await FragmentMerger.run() - if (merged.completed || merged.incomplete) this.logger.info(`Fragments: ${merged.completed} files completed, ${merged.incomplete} incomplete, ${merged.waiting} waiting.`) + // Rounds until nothing more can be decided, or MERGE_BUDGET_MS. + const totals = { completed: 0, incomplete: 0, waiting: 0 } + while (!stopping && Date.now() - mergedAt < MERGE_BUDGET_MS) { + const merged = await FragmentMerger.run() + totals.completed += merged.completed + totals.incomplete += merged.incomplete + totals.waiting = merged.waiting + if (merged.files === 0) break + } + if (totals.completed || totals.incomplete) this.logger.info(`Fragments: ${totals.completed} files completed, ${totals.incomplete} incomplete, ${totals.waiting} waiting.`) } await prefetcher.tick() } catch (error: any) { diff --git a/commands/edge_bench.ts b/commands/edge_bench.ts index 01f9356..e2db663 100644 --- a/commands/edge_bench.ts +++ b/commands/edge_bench.ts @@ -101,7 +101,8 @@ export default class EdgeBench extends BaseCommand { let rangeId = 0 let nodeId = 0 if (this.store) { - const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000) }).returning('id') + // Inactive: the coordinator neither plans nor leases it. + const [g] = await db.table('groups').insert({ name: `bench:${group}:${Date.now()}`, priority: 1_000_000 + (Date.now() % 1_000_000), active: false }).returning('id') const [n] = await db.table('nodes').insert({ name: `bench-${Date.now()}` }).returning('id') nodeId = Number(n.id) const [r] = await db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: end, next_id: start, status: 'leased', node_id: nodeId }).returning('id')