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.
5.9 KiB
Vendored
Migrations Tooling
The migrations/ directory is split into four path-referenced gems:
core/—Migrations::*: CLI framework, UI, SQLite schemas, DB infrastructure, IntermediateDB models, and the conversion framework (Migrations::Conversion::*).tooling/—Migrations::Tooling::*: the schema DSL,disco schemacommands, benchmarks.converters/—Migrations::Converters::*: public converter implementations + source adapters.importer/—Migrations::Importer::*: the row importer and the uploads importer.
All four are wired into the root Gemfile via path: in the optional :migrations group.
Command line interface
The single binary is migrations/bin/disco (commands register dynamically via
Migrations::CLI::Registry). Run it without arguments — or with --help — for the
authoritative, always-current list of commands:
migrations/bin/disco --help
Rails is booted lazily: only commands that declare requires_rails! (import, upload, schema)
load the Discourse app.
Converters
Public converters live in converters/lib/migrations/converters/. To run a private
(closed-source) converter, put its code in a subdirectory of private/converters/
(or point MIGRATIONS_PRIVATE_CONVERTERS_PATH at it).
Source DB adapters and fork safety
Worker processes inherit the source DB connection's socket from the main process. Whether
that's dangerous depends on the client library: a destructor that only closes the file
descriptor is harmless (the parent still holds it, so the kernel sends nothing over the
wire), but a destructor that writes a protocol goodbye kills the parent's session as soon
as a worker exits — libpq sends a Terminate message, MySQL clients send COM_QUIT.
Adapter::Postgres handles this by registering a ForkManager.after_fork_child hook that
calls discard! in each worker: the inherited socket is redirected to /dev/null, and any
later use of the adapter in the worker raises DiscardedError. New adapters should follow
the same pattern. The discard mechanism itself is library-specific — mysql2 has
automatic_close = false, trilogy has a native discard!. To check whether a library
needs one at all: connect, fork an empty child that exits normally, wait for it, and query
again from the parent (see the fork-safety specs in postgres_spec.rb).
Partitioning large steps
Most steps run in a single worker. A handful are large enough that it's worth
splitting them across CPU cores, so the framework can run one worker per chunk of
the source. A step opts in from its source block:
source do
reads_table "topic_users", where: "user_id > 0"
partition_by :topic_id
end
reads_table is the part that reads a whole table: it defines items
(SELECT * FROM topic_users WHERE …) and max_progress (the row count), filtered
by where. It works on its own, without partitioning — a plain table-copy step
declares just reads_table and writes neither method. partition_by adds the
split: it takes the key (normally a single indexed column, so each chunk is an
index range scan; pass an array for a composite key) and reuses the table and
filter from reads_table, so it only needs the column. When both are present the
generated queries add the chunk to their WHERE automatically.
Override items when you need specific columns, a join, or a particular order —
and then add partition_slice to its WHERE yourself:
def items
@source_db.query("SELECT id, name FROM topic_users WHERE #{partition_slice} AND name IS NOT NULL")
end
The framework does the rest. Before forking, it asks the adapter for the chunk
boundaries — evenly sized chunks over a numeric key, or a sorted-key scan for a
text/UUID/composite key. It then forks one worker per chunk; each worker reads
its [lower, upper) slice (that's what partition_slice expands to), writes its
own SQLite shard, and the shards are merged back into the run database.
Two things to get right:
- In a custom query, add
partition_sliceto theWHERE. Miss it and each worker reads the whole source instead of its slice — duplicated work and wrong counts. - Only partition order-independent steps. Workers run concurrently and their
output is merged, so there is no global order across the step. A running total
or a sequence number across all rows can't be partitioned. Deduplication can,
but do it in the source query (
DISTINCT ON, a window function, a view) and partition on the dedup key, rather than keeping state inprocess.
Schema DSL
The schema DSL lives in migrations/tooling/lib/migrations/tooling/schema/dsl/. Config sources
are in migrations/tooling/config/schema/. Generated artifacts (SQL, models, enums) are written
into migrations/core/.
Key files:
table_builder.rb- DSL for defining table configsschema_resolver.rb- Resolves DSL config + DB introspection into final schemaconventions_builder.rb- Global column conventions (renames, type overrides)generator.rb- Generates SQL, models, and enums from resolved schemavalidator.rb- Validates DSL configresolved_schema_validator.rb- Validates resolved schema before generation
Development
Installing gems
bundle config set --local with migrations
bundle install
Updating gems
bundle update --group migrations
Running tests
Each gem has an isolated, no-Rails suite, run from the gem directory:
cd migrations/core && bundle exec rspec
cd migrations/tooling && bundle exec rspec
cd migrations/converters && bundle exec rspec
cd migrations/importer && bundle exec rspec
Specs that need a booted Rails environment are tagged :rails. They are excluded by default and
run from the host app's bundle:
cd migrations/<gem> && BUNDLE_GEMFILE=../../Gemfile MIGRATIONS_RAILS=1 bundle exec rspec --tag rails
Linting
bin/lint path/to/file
bin/lint --fix path/to/file
Uses both rubocop and syntax_tree. Always lint changed files.
