aboutsummaryrefslogtreecommitdiffstats
path: root/log_db/src/primary_memtable.rs
diff options
context:
space:
mode:
Diffstat (limited to 'log_db/src/primary_memtable.rs')
-rw-r--r--log_db/src/primary_memtable.rs79
1 files changed, 79 insertions, 0 deletions
diff --git a/log_db/src/primary_memtable.rs b/log_db/src/primary_memtable.rs
new file mode 100644
index 0000000..0c94ceb
--- /dev/null
+++ b/log_db/src/primary_memtable.rs
@@ -0,0 +1,79 @@
+use super::common::*;
+use priority_queue::PriorityQueue;
+use std::collections::BTreeMap;
+
+pub struct PrimaryMemtable {
+ capacity: usize,
+ /// Running counter of memtable operations, used as priority
+ /// in evict_queue.
+ n_operations: u64,
+ records: BTreeMap<IndexableValue, Record>,
+ /// A max heap priority queue of keys. Note: n_operations must
+ /// be negated upon append to evict oldest values first.
+ evict_queue: PriorityQueue<IndexableValue, i64>,
+ evict_policy: MemtableEvictPolicy,
+}
+
+impl PrimaryMemtable {
+ pub fn new(capacity: usize, evict_policy: MemtableEvictPolicy) -> PrimaryMemtable {
+ PrimaryMemtable {
+ capacity,
+ n_operations: 0,
+ records: BTreeMap::new(),
+ evict_queue: PriorityQueue::new(),
+ evict_policy,
+ }
+ }
+
+ pub fn set(&mut self, key: &IndexableValue, value: &Record) {
+ if self.capacity == 0 {
+ return;
+ }
+
+ debug!(
+ "Inserting/updating record in primary memtable with key {:?} = {:?}",
+ &key, &value,
+ );
+
+ if self.records.len() >= self.capacity {
+ let (evict_key, _prio) = self.evict_queue.pop().expect("Evict queue was empty");
+ self.records.remove(&evict_key);
+ }
+
+ self.records.insert(key.clone(), value.clone());
+
+ if self.evict_policy == MemtableEvictPolicy::LeastWritten
+ || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten
+ {
+ self.set_priority(&key);
+ }
+ }
+
+ pub fn get(&mut self, key: &IndexableValue) -> Option<&Record> {
+ if self.evict_policy == MemtableEvictPolicy::LeastRead
+ || self.evict_policy == MemtableEvictPolicy::LeastReadOrWritten
+ {
+ self.set_priority(&key);
+ }
+
+ self.records.get(key)
+ }
+
+ fn set_priority(&mut self, key: &IndexableValue) {
+ let priority = self.get_and_increment_current_priority();
+ match self.evict_queue.get(key) {
+ Some(_) => {
+ self.evict_queue.change_priority(key, priority);
+ }
+ None => {
+ self.evict_queue.push(key.clone(), priority);
+ }
+ }
+ }
+
+ fn get_and_increment_current_priority(&mut self) -> i64 {
+ let ret = -(self.n_operations as i64);
+ self.n_operations += 1;
+ ret
+ }
+}