1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
//! The aggregate's map of places to slots, and the pool the maps go back to.
//!
//! Every instance of a grouping sink seeds a map on the key's ends before it has read a row, up to
//! a quarter of a million places, and clearing it was the largest single cost an instance paid
//! before its first chunk. On ClickBench 28 at six threads it was 120 of 1416 samples, nearly all of
//! it the kernel handing out fresh pages for the clear to write to. The map a finished instance
//! drops is the same size as the one the next instance asks for, so it is kept here instead, with
//! only the places it wrote set back, and the next instance takes it as it is.
//!
//! The places written are few next to the map. A group is written to the map once, when the row
//! that found it missed, so an instance that saw four thousand groups sets four thousand places
//! back rather than clearing a quarter of a million. A map that wrote more than a sixteenth of its
//! places is cleared whole, which is still no new pages.
use std::ops::Deref;
use std::sync::Mutex;
use crate::table::UNSEEN;
/// How many maps the pool keeps. One per thread that folds at once is all it ever needs, and a
/// map past this is freed the ordinary way.
const KEPT: usize = 64;
/// The widest map the pool takes back, four million places or 16 MB, so that one query with a very
/// wide key does not leave that much behind for the life of the process.
const WIDEST: usize = 1 << 22;
/// Maps dropped by instances that finished, every place [`UNSEEN`].
static POOL: Mutex<Vec<Vec<u32>>> = Mutex::new(Vec::new());
/// One slot per place, or [`UNSEEN`], which remembers which places it wrote.
///
/// It reads as a slice, and writes only through [`Places::set`], so that nothing reaches a place
/// without it being counted.
#[derive(Debug, Default)]
pub(crate) struct Places {
map: Vec<u32>,
/// The places written since the map was last all [`UNSEEN`], while there are few enough of
/// them to be worth setting back one at a time.
written: Vec<u32>,
/// Set when `written` stopped counting, so that the whole map is cleared instead.
crowded: bool,
}
impl Places {
/// A map of `len` places, every one [`UNSEEN`], taken from the pool when it has one.
pub(crate) fn seeded(len: usize) -> Self {
let mut map = POOL.lock().ok().and_then(|mut pool| pool.pop()).unwrap_or_default();
map.resize(len, UNSEEN);
Self { map, written: Vec::new(), crowded: false }
}
/// Writes `held` at `place`.
#[inline(always)]
pub(crate) fn set(&mut self, place: usize, held: u32) {
self.map[place] = held;
if !self.crowded {
match u32::try_from(place) {
Ok(place) if self.written.len() < self.map.len() / 16 => self.written.push(place),
_ => self.crowded = true,
}
}
}
/// Makes the map `len` places, every one [`UNSEEN`].
pub(crate) fn reset(&mut self, len: usize) {
self.map.clear();
self.map.resize(len, UNSEEN);
self.written.clear();
self.crowded = false;
}
/// Widens the map from `span` places to `len`, keeping every place but the last, the null
/// place, which moves to the new last place.
pub(crate) fn widen(&mut self, span: usize, len: usize) {
let null = std::mem::replace(&mut self.map[span - 1], UNSEEN);
self.map.resize(len, UNSEEN);
self.set(len - 1, null);
}
/// Sets every written place back to [`UNSEEN`].
fn clean(&mut self) {
if self.crowded {
self.map.fill(UNSEEN);
} else {
for &place in &self.written {
self.map[place as usize] = UNSEEN;
}
}
}
}
impl Deref for Places {
type Target = [u32];
fn deref(&self) -> &[u32] {
&self.map
}
}
impl Drop for Places {
fn drop(&mut self) {
if self.map.is_empty() || self.map.len() > WIDEST {
return;
}
self.clean();
debug_assert!(self.map.iter().all(|&held| held == UNSEEN));
if let Ok(mut pool) = POOL.lock()
&& pool.len() < KEPT
{
pool.push(std::mem::take(&mut self.map));
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_map_comes_back_clear() {
let mut places = Places::seeded(1_000);
places.set(3, 7);
places.set(999, 8);
places.widen(1_000, 1_200);
assert_eq!(places[1_199], 8, "the null place moved to the new end");
assert_eq!(places[999], UNSEEN);
drop(places);
// Other tests share the pool, so every map in it is checked rather than the one just given.
for map in POOL.lock().unwrap().iter() {
assert!(map.iter().all(|&held| held == UNSEEN));
}
let places = Places::seeded(500);
assert_eq!(places.len(), 500);
assert!(places.iter().all(|&held| held == UNSEEN));
}
#[test]
fn a_crowded_map_is_cleared_whole() {
let mut places = Places::seeded(64);
for place in 0..64 {
places.set(place, place as u32);
}
assert!(places.crowded);
places.clean();
assert!(places.iter().all(|&held| held == UNSEEN));
}
}