Bound processing pipeline memory use#3191
Conversation
poodlewars
added
patch
poodlewars
force-pushed
the
aseaton/12201416606/column-stats-memory-no-scripts
branch
from
c4fe938 to
cf370d6
Compare
ArcticDB Code Review SummaryThe implementation was reworked since the previous review (force-push + squash). The earlier design hooked residency tracking into I re-reviewed the new design in full. No blocking correctness, memory-safety, or concurrency issues found. Specifically verified: PR Title and Description
Code Quality (non-blocking)
|
poodlewars
changed the title
|
Do some more convinving testing of the new memory use Always use the admission handler (polymorphic or semaphore size = n processing units) Is it valid to use IO thread count for the number of processing units to load? Should we also consider the CPU thread count? We could take the max of the current setting and the CPU pool size, or twice the CPU pool size to give slack like the current approach (the idea being to keep the hopper full, try not to have consumers sat waiting on an empty queue). This would help if processing units have a large size (very col sliced) such that the IO pool creates only one processing unit at a time. The CPU pool works on units, so would be mostly idle with this workload. Later |
poodlewars
changed the title
…pipeline Column stats creation previously read and processed every row slice of a symbol concurrently, with no bound on how many decompressed segments could be resident at once. This adds an admission handler that windows how many processing units are live at a time, tracks segment residency via SegmentInMemoryImpl, and applies it throughout the pipeline rather than just column stats. A config kill switch (VersionStore.NumProcessingUnitsLive=0) disables the bound entirely.
poodlewars
force-pushed
the
aseaton/12201416606/column-stats-memory-no-scripts
branch
from
ddbb212 to
16ac857
Compare
1 participant
In the processing pipeline we currently kick off reads eagerly, meaning that if the in-memory processing does not complete quickly, large numbers of
SegmentInMemoryobjects accumulate in the CPU executor's queue. Creating column stats manually considers all the data in a symbol, so we can end up holding the entire symbol's data in memory.Implementation
This PR adds a new "admission handler" to control memory use in the processing pipeline. This affects all QueryBuilder operations, not just column stats. We now create a special "admission handler" object that provides segment read futures. Importantly, it creates these as promises and only kicks off work for them later, unlike the current eager read pipeline. This design means that everything downstream of the segment load futures in the processing pipeline can be used as-is.
The admission handler only allows K processing units to be live in memory at a time. As each unit finishes it calls
on_unit_complete(when theMemSegmentProcessingTaskcompletes). This kicks off the load of the next unit's segments. Importantly this is not a blocking operation, else it could deadlock.To give an example, with processing units with an overlap:
we first call
admit_initialto load processing unit0, i.e. segments0, 1. When this unit is complete, the component manager frees segment0but not segment1. At this pointon_unit_completefires. This callsadmit(processing_unit=1)which callsfire(segment=1)(no-op) andfire(segment=2)which kicks off a segment load.We also make sure that we effectively
folly::windowover the IO work that we kick off, rather than letting it all pile up on the IO queues when K is large, which gives worse queuing behaviour and regressed performance compared tofolly::window.For testing, I add a class
SegmentResidencyTrackerthat measures how many segments were live in memory over an operation. This is compiled in to release builds but is a no-op - it only does anything in testing. Its cost is just reading a boolean that's always false.Impact
Measurements
Each table cell is peak
ru_maxrss(MiB) / wall time (s) for one operation. Every cell is a separate process running a single operation, soru_maxrss(peak RSS over the process lifetime) is a clean high-water mark for that one operation.masterhas no admission bound, so it has a single column.fixesvaries the residency budgetK(VersionStore.NumProcessingUnitsLive, the max processing units in flight):K=0is the kill switch (bound disabled),K=8a tight bound,K=defaultthe computed default; the read windowW(VersionStore.SegmentReadWindow) is left at its default of2*io.Each build is run under three process allocators: glibc default, glibc with
MALLOC_ARENA_MAX=4, and tcmalloc (LD_PRELOAD).IO:CPUis the IO/CPU threadpool sizes (VersionStore.NumIOThreads/NumCPUThreads).Data
One symbol,
f64_100m_100c_zeros, on PURE (S3): 100,000,000 rows x 100float64data columns (f_0..f_99), all zeros, plus anint64categorycolumn cycling through values 1-5,indexed by a nanosecond
DatetimeIndexat 1-second frequency. Written with default slicing (100k rows/segment, all columns in one column slice), giving 1000 row-slice segments.Each segment is about 80MiB decoded, the symbol is about 80GiB decoded.
Operations
readwithQueryBuilder()[q["f_0"] == -1.0].-1.0never occurs (data is zeros), so the result is empty and does not contribute to RSSreadwithQueryBuilder().resample("30D").agg(sum)summing all 100 float columns into ~30-day buckets (a few dozen output rows).readwithQueryBuilder().groupby("category").agg(sum)summing all 100 float columns across the 5categorygroups (5 output rows).Findings
(column_stats IO=64 master: 33745 glibc -> 10575 arena4 -> 7472 tcmalloc).
The best config is the bound plus tcmalloc: filter 64:96 K=8 tcmalloc = 713 MiB vs 4782 master glibc (6.7x); column_stats K=8 tcmalloc = 1641 vs 33745 (20x).
filter
glibc default — peak MiB / wall s
MALLOC_ARENA_MAX=4 — peak MiB / wall s
tcmalloc — peak MiB / wall s
column_stats
glibc default — peak MiB / wall s
MALLOC_ARENA_MAX=4 — peak MiB / wall s
tcmalloc — peak MiB / wall s
resample
glibc default — peak MiB / wall s
MALLOC_ARENA_MAX=4 — peak MiB / wall s
tcmalloc — peak MiB / wall s
groupby
glibc default — peak MiB / wall s
MALLOC_ARENA_MAX=4 — peak MiB / wall s
tcmalloc — peak MiB / wall s
Alternatives
#3138 - discarded as it leaves the CPU pool idle, and is difficult to read (in that the mechanism that bounds memory use is not explicit).
Originally I wanted to pass a semaphore in to
batch_read_uncompressed, and acquire it for segment reads, and release it after a segment is processed, like compacting staged data does, always allowing at leastn_col_slicesto be loaded. This did not work because we kick off all the segment reads in parallel so yourn_col_slicesbudget can be split across segments in different processing units while leaving no processing unit able to load completely and continue. This PR is instead explicit about allowing loads per processing unit.