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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
//! The keyspaces a database serves: the default one, those named when it was opened, and those
//! created since.
use std::sync::Arc;
use tessarikv_types::KeyspaceId;
use tokio::sync::oneshot;
use super::Db;
use super::catalog::{Command, Create, DropKeyspace, check_new};
use super::installer::Flusher;
use super::view::Names;
use crate::memtable::Memtable;
use crate::options::DEFAULT_NAME;
use crate::{DbOptions, Error, KeyspaceOptions, Result};
impl Db {
/// Refuses a keyspace this database does not hold; reads the current view, never waits.
/// The default keyspace is always held and is never dropped.
pub(crate) fn check_keyspace(&self, keyspace: KeyspaceId) -> Result<()> {
if keyspace == KeyspaceId::DEFAULT || self.view.load().holds(keyspace.get()) {
Ok(())
} else {
Err(Error::UnknownKeyspace {
keyspace: keyspace.get(),
})
}
}
/// The id of the keyspace named `name`; `default` is the default keyspace.
#[must_use]
pub fn keyspace(&self, name: &str) -> Option<KeyspaceId> {
self.view.load().keyspace_named(name).map(KeyspaceId::new)
}
/// Every keyspace this database holds, the default one first, then by id, with its name.
#[must_use]
pub fn keyspaces(&self) -> Vec<(KeyspaceId, String)> {
let view = self.view.load();
std::iter::once((KeyspaceId::DEFAULT, DEFAULT_NAME.to_owned()))
.chain(
view.names()
.iter()
.map(|(id, name)| (KeyspaceId::new(*id), name.clone())),
)
.collect()
}
/// Creates the keyspace `name`, an independent set of keys with its own `options`, and
/// returns its id, which no other keyspace of this database ever takes. Writes and reads
/// may name it as soon as this returns, and a batch may span it and other keyspaces.
///
/// On storage the keyspace is recorded before this returns; its options are not, so every
/// later open must name it with [`crate::DbOptionsBuilder::keyspace`] or be refused.
///
/// # Errors
///
/// [`Error::InvalidArgument`] for an empty or overlong name, the default keyspace's name, a
/// name already held, or one keyspace past [`tessarikv_constants::MAX_KEYSPACES`]; on
/// storage, [`Error::FlushFailed`] or [`Error::CompactionFailed`] when the background work
/// stopped, and the storage's error when recording the keyspace failed.
pub async fn create_keyspace(
&self,
name: &str,
options: KeyspaceOptions,
) -> Result<KeyspaceId> {
check_new(name, std::iter::empty(), 0)?;
let Some(installer) = &self.installer else {
return self.create_in_memory(name, &options);
};
let (done, created) = oneshot::channel();
let stopped = || installer.signals.failure().unwrap_or(Error::FlushFailed);
installer
.requests
.send(Command::Create(Create {
name: name.to_owned(),
options,
done,
}))
.await
.map_err(|_| stopped())?;
created.await.map_err(|_| stopped())?
}
/// Drops `keyspace` with every key it holds: reads and writes naming it are refused from now
/// on, while a write that found it before still applies its other entries. Its table files
/// are removed once no reader holds them. Its name may be taken again, by a keyspace with a
/// new id. On storage the drop is recorded before this returns, and later opens must no longer
/// name it.
///
/// # Errors
///
/// [`Error::InvalidArgument`] for the default keyspace, which is never dropped,
/// [`Error::UnknownKeyspace`] for a keyspace this database does not hold; on storage,
/// [`Error::FlushFailed`] or [`Error::CompactionFailed`] when the background work stopped,
/// and the storage's error when recording the drop failed.
pub async fn drop_keyspace(&self, keyspace: KeyspaceId) -> Result<()> {
if keyspace == KeyspaceId::DEFAULT {
return Err(Error::InvalidArgument {
what: "dropping a keyspace",
reason: "the default keyspace is never dropped",
});
}
let Some(installer) = &self.installer else {
return self.drop_in_memory(keyspace.get());
};
let (done, dropped) = oneshot::channel();
let stopped = || installer.signals.failure().unwrap_or(Error::FlushFailed);
installer
.requests
.send(Command::Drop(DropKeyspace {
keyspace: keyspace.get(),
done,
}))
.await
.map_err(|_| stopped())?;
dropped.await.map_err(|_| stopped())?
}
/// Drops a keyspace of a database without a manifest, publishing the view without it by
/// compare-and-swap.
fn drop_in_memory(&self, keyspace: u32) -> Result<()> {
loop {
let current = self.view.load_full();
if !current.holds(keyspace) {
return Err(Error::UnknownKeyspace { keyspace });
}
let (next, _) = current.without_keyspace(keyspace);
let previous = self.view.compare_and_swap(¤t, Arc::new(next));
if Arc::ptr_eq(&previous, ¤t) {
return Ok(());
}
}
}
/// Creates a keyspace in a database without a manifest: the view naming it is published by
/// compare-and-swap, so creations racing each other each take their own id.
fn create_in_memory(&self, name: &str, options: &KeyspaceOptions) -> Result<KeyspaceId> {
loop {
let current = self.view.load_full();
let names = current.names();
check_new(
name,
names.iter().map(|(_, held)| held.as_str()),
names.len(),
)?;
let id = current.next_keyspace();
let table = Memtable::sized(options.memtable_bytes());
let next = Arc::new(current.with_keyspace(id, name, table)?);
let previous = self.view.compare_and_swap(¤t, next);
if Arc::ptr_eq(&previous, ¤t) {
return Ok(KeyspaceId::new(id));
}
}
}
}
/// One empty table per keyspace, the default one first, each sized by its options, and the
/// names of the keyspaces beside the default one; `held` gives their ids and names.
pub(super) fn tables_for<'a>(
options: &DbOptions,
held: impl Iterator<Item = (u32, &'a str)>,
) -> (Vec<(u32, Memtable)>, Names) {
let mut tables = vec![(
0,
Memtable::sized(options.default_keyspace().memtable_bytes()),
)];
let mut names = Vec::new();
for (id, name) in held {
let bytes = options
.keyspaces()
.iter()
.find(|(named, _)| named == name)
.map_or(options.default_keyspace(), |(_, keyspace)| keyspace)
.memtable_bytes();
tables.push((id, Memtable::sized(bytes)));
names.push((id, name.to_owned()));
}
(tables, Arc::from(names))
}
/// The options of each keyspace beside the default one, by id; `held` gives their ids and
/// names.
pub(super) fn options_by_id<'a>(
options: &DbOptions,
held: impl Iterator<Item = (u32, &'a str)>,
) -> Vec<(u32, KeyspaceOptions)> {
held.map(|(id, name)| {
let named = options.keyspaces().iter().find(|(named, _)| named == name);
let keyspace = named.map_or(options.default_keyspace(), |(_, keyspace)| keyspace);
(id, keyspace.clone())
})
.collect()
}
/// The options of keyspace `id` among `keyspaces`; `default` for the default keyspace.
pub(super) fn options_of<'a>(
keyspaces: &'a [(u32, KeyspaceOptions)],
default: &'a KeyspaceOptions,
id: u32,
) -> &'a KeyspaceOptions {
keyspaces
.iter()
.find(|(held, _)| *held == id)
.map_or(default, |(_, options)| options)
}
impl Flusher {
/// The options of `keyspace`; the default keyspace's for 0.
pub(super) fn keyspace_options(&self, keyspace: u32) -> &KeyspaceOptions {
options_of(&self.keyspaces, &self.keyspace, keyspace)
}
/// Whether `keyspace` is held: the default one, or one opened or created and not dropped.
pub(super) fn holds(&self, keyspace: u32) -> bool {
keyspace == 0 || self.keyspaces.iter().any(|(id, _)| *id == keyspace)
}
/// Every keyspace held, the default one first, with its options.
pub(super) fn all_keyspaces(&self) -> impl Iterator<Item = (u32, &KeyspaceOptions)> {
std::iter::once((0, &self.keyspace))
.chain(self.keyspaces.iter().map(|(id, options)| (*id, options)))
}
}