mirror of
https://github.com/discourse/discourse.git
synced 2026-08-06 13:08:40 +08:00
12a0656627
This is the big one I've been building up to across the step-concurrency series: the converter now uses every core instead of grinding through one step at a time on a single CPU. The win scales with the number of cores, so the figure depends on the machine. On my laptop, running all the currently implemented Discourse converter steps dropped from 102s to 18s, about 5.5x faster, with all cores busy the whole time. Two things were slow before: steps ran one after another, and the few genuinely large steps (topic_users above all) ran single-threaded even when nothing else was happening. This PR tackles both. Concurrent steps. A dependency-aware scheduler runs independent steps at the same time, each with its own source connection. It's event-driven: as soon as a step finishes, it frees its fork and the next ready step starts, so the cores don't sit idle between steps. How many run at once is bounded by the cores it can use, and `--max-parallel-steps` lowers it. A pull model for workers. Instead of the parent reading every row and streaming items down a pipe to the workers, each worker now opens its own source connection and reads its own slice directly. Only progress goes back over the pipe, and only once every thousand items, so it's never the bottleneck. The parent's single-threaded read was the real ceiling, and this removes it, along with a fair bit of machinery (the old worker pool, the item serialization, the handshake). Partitioning the heavy steps. A step opts in with `partition_by`, and the scheduler splits it across forks, each reading one chunk of the key range. I kept it general: a dense numeric key is divided into even chunks straight from a cheap MIN/MAX, while a sparse numeric key or a text/UUID or composite key falls back to a sorted-key scan so every chunk holds a similar number of rows whatever the key type. The dialect-specific SQL lives in the Postgres adapter, so other sources can fill in their own later. One caveat I documented on `partition_by`: only use it on steps whose processing is order-independent, since the forks run concurrently and their output is merged. Sharded writes. Each worker writes to its own SQLite shard, and a background consolidator folds finished shards back into the run database off the step's critical path, so a step no longer lingers at 100% while its merge runs. Reliability. A bad row or a worker that dies mid-step no longer hangs the run or takes down the other steps running at the same time: failures are caught and surfaced per step. Debugging. Forks make a debugger awkward, so there's a `--no-fork` flag: it runs each step inline in the main process, one at a time, so a breakpoint in a step's `process` stops where you can actually use it. The data path is the same (it still writes a shard the consolidator merges), only the fork is gone. Output order is no longer deterministic with concurrency, so to check correctness I added a small dev script under `migrations/tooling/scripts` that compares two IntermediateDBs order-insensitively and used it to confirm a parallel run produces the same data as a serial one. The gem suites are green.
69 lines
1.9 KiB
Ruby
Vendored
69 lines
1.9 KiB
Ruby
Vendored
# frozen_string_literal: true
|
||
|
||
# Compares two IntermediateDB files for order-insensitive equality. The
|
||
# concurrent converter writes rows in no fixed order, so we sort each table's
|
||
# rows and compare a digest instead of diffing the files.
|
||
#
|
||
# Usage (from the repo root, under the migrations bundle):
|
||
#
|
||
# cd migrations/core
|
||
# bundle exec ruby ../tooling/scripts/compare_intermediate_dbs.rb \
|
||
# /path/to/sequential.db /path/to/concurrent.db
|
||
|
||
require "extralite"
|
||
require "digest"
|
||
|
||
# `schema_migrations.applied_at` is a wall-clock time and always differs.
|
||
IGNORED_TABLES = %w[schema_migrations].freeze
|
||
|
||
def tables(db)
|
||
(db.tables - IGNORED_TABLES).sort
|
||
end
|
||
|
||
def quote(identifier)
|
||
%("#{identifier.gsub('"', '""')}")
|
||
end
|
||
|
||
def table_digest(db, table)
|
||
rows = []
|
||
count = 0
|
||
db.query_array("SELECT * FROM #{quote(table)}") do |row|
|
||
rows << row.map { |value| value.is_a?(String) ? value.b : value.inspect }.join("")
|
||
count += 1
|
||
end
|
||
rows.sort!
|
||
[count, Digest::SHA256.hexdigest(rows.join(""))]
|
||
end
|
||
|
||
seq_path, conc_path = ARGV
|
||
abort "usage: compare_intermediate_dbs.rb SEQUENTIAL_DB CONCURRENT_DB" unless seq_path && conc_path
|
||
|
||
seq = Extralite::Database.new(seq_path)
|
||
conc = Extralite::Database.new(conc_path)
|
||
|
||
seq_tables = tables(seq)
|
||
conc_tables = tables(conc)
|
||
|
||
ok = true
|
||
|
||
if seq_tables != conc_tables
|
||
ok = false
|
||
warn "Table sets differ:"
|
||
warn " only in #{seq_path}: #{(seq_tables - conc_tables).join(", ")}"
|
||
warn " only in #{conc_path}: #{(conc_tables - seq_tables).join(", ")}"
|
||
end
|
||
|
||
(seq_tables & conc_tables).each do |table|
|
||
seq_count, seq_digest = table_digest(seq, table)
|
||
conc_count, conc_digest = table_digest(conc, table)
|
||
|
||
if seq_digest == conc_digest
|
||
puts " ok #{table.ljust(32)} #{seq_count} rows"
|
||
else
|
||
ok = false
|
||
puts " DIFF #{table.ljust(32)} sequential=#{seq_count} rows, concurrent=#{conc_count} rows"
|
||
end
|
||
end
|
||
|
||
puts(ok ? "\nIdentical (order-insensitive)." : "\nDifferences found.")
|
||
exit(ok ? 0 : 1)
|
