Fetch terrain tiles concurrently with persistent connections
Serial fetches (no connection reuse) measured ~650ms/tile, almost all TCP+TLS handshake — a real bbox at z0..15 is tens of thousands of tiles, so this was hours for a modest region. Now: 16 worker threads share a queue (I/O-bound, so the GVL doesn't get in the way), each keeping its own persistent connection across every tile it fetches. Measured live against the real Terrarium source: ~40ms/tile, ~16x faster. write_terrarium_mbtiles's SQLite writes and progress updates stay on one thread (draining a results Queue the workers feed), so nothing needed its own locking beyond the two Queues themselves.
This commit is contained in:
1 parent
bbe9f38bda
commit
d041c21270
2 files changed
+69
-29
No files matched your search
@@ -28,6 +28,13 @@ module Maptool
|
||||
# source's own online maxzoom.
|
||||
TERRARIUM_URL_TEMPLATE = "https://s3.amazonaws.com/elevation-tiles-prod/terrarium/%<z>d/%<x>d/%<y>d.png"
|
||||
TERRAIN_MAX_ZOOM = 15
|
||||
# A real bbox at z0..15 is tens of thousands of tiles — serial fetches
|
||||
# (the original ponytail'd version of this) measured ~0.65s/tile, almost
|
||||
# all TCP+TLS handshake, no connection reuse: hours for a modest region.
|
||||
# This many workers, each keeping its own persistent connection (see
|
||||
# fetch_terrarium_tile), is a big enough win to matter without looking
|
||||
# like abuse against a public bucket.
|
||||
CONCURRENT_TILE_FETCHES = 16
|
||||
|
||||
OFFLINE_STATUS = { state: "idle" }
|
||||
TERRAIN_STATUS = { state: "idle" }
|
||||
@@ -141,10 +148,11 @@ module Maptool
|
||||
# public Terrarium-encoded PMTiles archive is known to exist to
|
||||
# `extract` from the way offline.pmtiles does above.
|
||||
#
|
||||
# ponytail: fetches tiles one at a time, serially — fine for a
|
||||
# park-sized region, slow for a big one. Add a small bounded-concurrency
|
||||
# fetch pool if regenerating a large area turns out to matter in
|
||||
# practice.
|
||||
# fetch_tile.call happens on CONCURRENT_TILE_FETCHES worker threads at
|
||||
# once (I/O-bound — Ruby's GVL is released while a thread blocks on
|
||||
# network I/O, so this genuinely overlaps); every SQLite write and
|
||||
# progress update stays on this thread, draining a results Queue the
|
||||
# workers feed, so nothing here needs its own locking.
|
||||
def self.write_terrarium_mbtiles(path, south, west, north, east, fetch_tile, progress)
|
||||
db = SQLite3::Database.new(path)
|
||||
db.execute_batch(<<~SQL)
|
||||
@@ -157,29 +165,48 @@ module Maptool
|
||||
end
|
||||
|
||||
# tile_range per zoom computed once upfront (cheap — no fetching, just
|
||||
# tile math) and reused below, so progress can report a real total
|
||||
# from the very first call instead of only once the run finishes.
|
||||
by_zoom = (0..TERRAIN_MAX_ZOOM).map { |z| [z, tile_range(z, south, west, north, east)] }
|
||||
total = by_zoom.sum { |_z, tiles| tiles.size }
|
||||
# tile math), so progress can report a real total from the very first
|
||||
# call instead of only once the run finishes.
|
||||
work = (0..TERRAIN_MAX_ZOOM).flat_map { |z| tile_range(z, south, west, north, east).map { |x, y| [z, x, y] } }
|
||||
total = work.size
|
||||
done = 0
|
||||
progress.call(done, total)
|
||||
|
||||
by_zoom.each do |z, tiles|
|
||||
tiles.each do |x, y|
|
||||
bytes = fetch_tile.call(z, x, y)
|
||||
queue = Queue.new
|
||||
work.each { |item| queue << item }
|
||||
CONCURRENT_TILE_FETCHES.times { queue << nil } # one stop signal per worker
|
||||
|
||||
results = Queue.new
|
||||
workers = CONCURRENT_TILE_FETCHES.times.map do
|
||||
Thread.new do
|
||||
while (item = queue.pop)
|
||||
z, x, y = item
|
||||
results << [z, x, y, fetch_tile.call(z, x, y)]
|
||||
end
|
||||
ensure
|
||||
# Only set when fetch_tile is fetch_terrarium_tile (a test fake
|
||||
# has no reason to touch this) — closes this worker's persistent
|
||||
# connection rather than leaving the socket for GC to eventually
|
||||
# notice.
|
||||
Thread.current[:maptool_terrarium_http]&.finish
|
||||
end
|
||||
end
|
||||
|
||||
total.times do
|
||||
z, x, y, bytes = results.pop
|
||||
done += 1
|
||||
progress.call(done, total)
|
||||
next unless bytes
|
||||
|
||||
# MBTiles uses TMS row numbering (0 at the south) — the opposite
|
||||
# of the XYZ scheme both the Terrarium source and this app's own
|
||||
# online raster-dem layer use (0 at the north) — flipped here so
|
||||
# go-pmtiles ends up serving the same tile at the same z/x/y this
|
||||
# app already requests online.
|
||||
# MBTiles uses TMS row numbering (0 at the south) — the opposite of
|
||||
# the XYZ scheme both the Terrarium source and this app's own online
|
||||
# raster-dem layer use (0 at the north) — flipped here so go-pmtiles
|
||||
# ends up serving the same tile at the same z/x/y this app already
|
||||
# requests online.
|
||||
tms_y = (2**z) - 1 - y
|
||||
db.execute("INSERT OR REPLACE INTO tiles (zoom_level, tile_column, tile_row, tile_data) VALUES (?, ?, ?, ?)", [z, x, tms_y, bytes.b])
|
||||
end
|
||||
end
|
||||
workers.each(&:join)
|
||||
ensure
|
||||
db&.close
|
||||
end
|
||||
@@ -204,8 +231,14 @@ module Maptool
|
||||
end
|
||||
private_class_method :lat_to_tile_y
|
||||
|
||||
# One persistent connection per calling Thread (each of
|
||||
# write_terrarium_mbtiles' worker threads calls this many times) instead
|
||||
# of Net::HTTP.get_response's usual "new TCP+TLS handshake per call" —
|
||||
# that handshake was most of the ~0.65s/tile this used to cost.
|
||||
def self.fetch_terrarium_tile(z, x, y)
|
||||
response = Net::HTTP.get_response(URI(format(TERRARIUM_URL_TEMPLATE, z: z, x: x, y: y)))
|
||||
uri = URI(format(TERRARIUM_URL_TEMPLATE, z: z, x: x, y: y))
|
||||
http = (Thread.current[:maptool_terrarium_http] ||= Net::HTTP.start(uri.host, uri.port, use_ssl: uri.scheme == "https"))
|
||||
response = http.get(uri.request_uri)
|
||||
response.is_a?(Net::HTTPSuccess) ? response.body : nil
|
||||
rescue StandardError
|
||||
nil
|
||||
|
||||
@@ -119,27 +119,34 @@ class PmtilesBuilderTest < Minitest::Test
|
||||
assert_equal status[:tiles_total], status[:tiles_done]
|
||||
end
|
||||
|
||||
# fetch_tile blocks after the very first call so the "running" progress
|
||||
# can be observed mid-flight, deterministically, rather than only ever
|
||||
# seeing the final done/total once the (very fast, faked) build finishes.
|
||||
# Every fetch_tile call blocks on a gate until explicitly told not to —
|
||||
# with CONCURRENT_TILE_FETCHES workers pulling from a shared queue, that
|
||||
# holds the whole pool at "in flight, none finished yet" deterministically
|
||||
# (a big enough bbox that z0..15 yields far more tiles than worker slots,
|
||||
# so most of the job is still queued up behind the blocked pool).
|
||||
def test_build_terrain_reports_progress_while_running
|
||||
ENV["PMTILES_BIN"] = @pmtiles_ok
|
||||
concurrency = Maptool::PmtilesBuilder::CONCURRENT_TILE_FETCHES
|
||||
gate = Queue.new
|
||||
first_call = true
|
||||
unblocked = false
|
||||
fetch_tile = lambda do |_z, _x, _y|
|
||||
gate.pop if first_call
|
||||
first_call = false
|
||||
gate.pop unless unblocked
|
||||
"bytes"
|
||||
end
|
||||
thread = Maptool::PmtilesBuilder.build_terrain([48.140, 17.100, 48.141, 17.101], data_dir: @data_dir, fetch_tile: fetch_tile)
|
||||
thread = Maptool::PmtilesBuilder.build_terrain([47.5, 16.5, 48.5, 17.5], data_dir: @data_dir, fetch_tile: fetch_tile)
|
||||
|
||||
20.times { break if Maptool::PmtilesBuilder.terrain_status[:tiles_total]; sleep 0.01 }
|
||||
# Let exactly one batch through (fills every worker's first call) then
|
||||
# check before releasing the rest — done should be capped at the pool
|
||||
# size, nowhere near the real total.
|
||||
concurrency.times { gate.push(nil) }
|
||||
sleep 0.05
|
||||
status = Maptool::PmtilesBuilder.terrain_status
|
||||
assert_equal "running", status[:state]
|
||||
assert_operator status[:tiles_total], :>, 0
|
||||
assert_equal 0, status[:tiles_done]
|
||||
assert_operator status[:tiles_total], :>, concurrency
|
||||
assert_operator status[:tiles_done], :<, status[:tiles_total]
|
||||
ensure
|
||||
gate.push(nil)
|
||||
unblocked = true
|
||||
concurrency.times { gate.push(nil) } # release whatever's still blocked in an in-progress gate.pop
|
||||
thread&.join
|
||||
end
|
||||
|
||||
|
||||
Reference in new issue
Block a user