Recovery

config/segcache_test.toml

@@ -0,0 +1,94 @@
+daemonize = false
+
+[admin]
+# interfaces listening on
+host = "0.0.0.0"
+# port listening on
+port = "9999"
+
+[server]
+# interfaces listening on
+host = "0.0.0.0"
+# port listening on
+port = "12321"
+# epoll timeout in milliseconds
+timeout = 100
+# epoll max events returned
+nevent = 1024
+
+[worker]
+# epoll timeout in milliseconds
+timeout = 100
+# epoll max events returned
+nevent = 1024
+# number of worker threads
+threads = 1
+
+# storage configuration
+[seg]
+# hash power adjusts how many items can be held in the hashtable
+hash_power = 3
+# total bytes to use for item storage - 32 segments * 1024 segment size
+heap_size = 32768
+# size of each segment in bytes
+segment_size = 1024
+# number of segments for a non-evict compaction
+compact_target = 2
+# number of segments to merge in one merge eviction pass
+merge_target = 4
+# max number of segments to merge in one pass
+merge_max = 8
+# use merge based eviction
+eviction = "Merge"
+# optionally, set a file path to back the data datapool
+datapool_path = "/home/users/u6632448/file_for_data"
+#datapool_path = "/mnt/pmem1.0/cassy/data"
+# set a file path to back the metadata datapool
+metadata_path = "/home/users/u6632448/file_for_metadata"
+#metadata_path = "/mnt/pmem1.0/cassy/metadata"
+# state whether cache will be restored
+restore = false
+# state whether cache will be flushed upon shutdown
+graceful_shutdown = true
+
+[time]
+time_type = "Memcache"
+
+[buf]
+
+[debug]
+# choose from: error, warn, info, debug, trace
+log_level = "warn"
+# optionally, log to the file below instead of standard out
+# log_file = "segcache.log"
+# backup file name for use with log rotation
+log_backup = "segcache.log.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+log_max_size = 1073741824
+
+[klog]
+# optionally, log commands to the file below
+# file = "segcache.cmd"
+# backup file name for use with log rotation
+backup = "segcache.cmd.old"
+# trigger log rotation when the file grows beyond this size (in bytes). Set this
+# option to '0' to disable log rotation.
+max_size = 1073741824
+# specify the sampling ratio, 1 in N commands will be logged. Setting to '0'
+# will disable command logging.
+sample = 100
+
+[sockio]
+
+[tcp]
+
+[tls]
+# certificate chain used to validate client certificate
+# certificate_chain = "client.chain"
+# server certificate
+# certificate = "server.crt"
+# server private key
+# private_key = "server.key"
+# ca certificate file used as the root of trust
+# ca_file = "ca.crt"

src/rust/common/src/signal.rs

@@ -5,4 +5,5 @@
 #[derive(Clone)]
 pub enum Signal {
     Shutdown,
+    Stop,
 }

src/rust/config/src/seg.rs

@@ -8,6 +8,10 @@ use serde::{Deserialize, Serialize};
 
 const MB: usize = 1024 * 1024;
 
+// restore and graceful shutdown options
+const RESTORE: bool = false;
+const GRACEFUL_SHUTDOWN: bool = false;
+
 // defaults for hashtable
 const HASH_POWER: u8 = 16;
 const OVERFLOW_FACTOR: f64 = 1.0;
@@ -24,9 +28,12 @@ const COMPACT_TARGET: usize = 2;
 const MERGE_TARGET: usize = 4;
 const MERGE_MAX: usize = 8;
 
-// datapool
+// datapool (`Segments.data`)
 const DATAPOOL_PATH: Option<&str> = None;
 
+// hashtable
+const HASHTABLE_PATH: Option<&str> = None;
+
 #[derive(Copy, Clone, Debug, Serialize, Deserialize)]
 pub enum Eviction {
     None,
@@ -39,6 +46,14 @@ pub enum Eviction {
 }
 
 // helper functions for default values
+fn restore() -> bool {
+    RESTORE
+}
+
+fn graceful_shutdown() -> bool {
+    GRACEFUL_SHUTDOWN
+}
+
 fn hash_power() -> u8 {
     HASH_POWER
 }
@@ -75,9 +90,17 @@ fn datapool_path() -> Option<String> {
     DATAPOOL_PATH.map(|v| v.to_string())
 }
 
+fn metadata_path() -> Option<String> {
+    HASHTABLE_PATH.map(|v| v.to_string())
+}
+
 // definitions
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Seg {
+    #[serde(default = "restore")]
+    restore: bool,
+    #[serde(default = "graceful_shutdown")]
+    graceful_shutdown: bool,
     #[serde(default = "hash_power")]
     hash_power: u8,
     #[serde(default = "overflow_factor")]
@@ -96,11 +119,15 @@ pub struct Seg {
     compact_target: usize,
     #[serde(default = "datapool_path")]
     datapool_path: Option<String>,
+    #[serde(default = "metadata_path")]
+    metadata_path: Option<String>,
 }
 
 impl Default for Seg {
     fn default() -> Self {
         Self {
+            restore: restore(),
+            graceful_shutdown: graceful_shutdown(),
             hash_power: hash_power(),
             overflow_factor: overflow_factor(),
             heap_size: heap_size(),
@@ -110,12 +137,28 @@ impl Default for Seg {
             merge_max: merge_max(),
             compact_target: compact_target(),
             datapool_path: datapool_path(),
+            metadata_path: metadata_path(),
         }
     }
 }
 
 // implementation
 impl Seg {
+    // Determines if the `Seg` will be restored. The restoration will be
+    // successful if `datapool_path` and `metadata_path` are valid paths.
+    // Otherwise, the `Seg` will be created as
+    //new.
+    pub fn restore(&self) -> bool {
+        self.restore
+    }
+
+    // Determines if the `Seg` will be gracefully shutdown. The graceful
+    // shutdown will be successful if the cache is file backed and
+    // metadata_path` is a valid path to save the relevant `Seg` fields to.
+    // Otherwise, the relevant `Seg` fields will not be saved.
+    pub fn graceful_shutdown(&self) -> bool {
+        self.graceful_shutdown
+    }
     pub fn hash_power(&self) -> u8 {
         self.hash_power
     }
@@ -151,6 +194,10 @@ impl Seg {
     pub fn datapool_path(&self) -> Option<PathBuf> {
         self.datapool_path.as_ref().map(|v| Path::new(v).to_owned())
     }
+
+    pub fn metadata_path(&self) -> Option<PathBuf> {
+        self.metadata_path.as_ref().map(|v| Path::new(v).to_owned())
+    }
 }
 
 // trait definitions

src/rust/core/server/src/lib.rs

@@ -114,6 +114,11 @@ pub const DEFAULT_BUFFER_SIZE: usize = 16 * 1024; // 16KB
 // specific upper bounds.
 const ADMIN_MAX_BUFFER_SIZE: usize = 2 * 1024 * 1024; // 1MB
 
+// TODO(bmartin): this *should* be plenty safe, the queue should rarely ever be
+// full, and a single wakeup should drain at least one message and make room for
+// the response. A stat to prove that this is sufficient would be good.
+const QUEUE_RETRIES: usize = 3;
+
 const THREAD_PREFIX: &str = "pelikan";
 
 metrics::test_no_duplicates!();

src/rust/core/server/src/threads/admin.rs

@@ -7,6 +7,7 @@
 
 use super::EventLoop;
 use crate::poll::{Poll, LISTENER_TOKEN, WAKER_TOKEN};
+use crate::QUEUE_RETRIES;
 use crate::TCP_ACCEPT_EX;
 use common::signal::Signal;
 use common::ssl::{HandshakeError, MidHandshakeSslStream, Ssl, SslContext, SslStream};
@@ -324,9 +325,9 @@ impl Admin {
                         self.do_accept();
                     }
                     WAKER_TOKEN => {
-                        #[allow(clippy::never_loop)]
                         // check if we have received signals from any sibling
                         // thread
+                        #[allow(clippy::never_loop)]
                         while let Ok(signal) = self.signal_queue.recv_from(0) {
                             match signal {
                                 Signal::Shutdown => {
@@ -341,6 +342,11 @@ impl Admin {
                                     let _ = self.log_drain.flush();
                                     return;
                                 }
+                                Signal::Stop => {
+                                    warn!("received stop");
+                                    let _ = self.log_drain.flush();
+                                    return;
+                                }
                             }
                         }
                     }
@@ -434,6 +440,25 @@ impl EventLoop for Admin {
 
                         match request {
                             AdminRequest::FlushAll => {}
+                            AdminRequest::Stop => {
+
+                                for _ in 0..QUEUE_RETRIES {
+                                    // Send Stop to all other threads
+                                    if self.signal_queue.broadcast(Signal::Stop).is_ok() {
+                                        warn!("sending stop signal to all threads");
+                                        break;
+                                    }
+                                }
+                                for _ in 0..QUEUE_RETRIES {
+                                    if self.signal_queue.wake_all().is_ok() {
+                                        break;
+                                    }
+                                }
+
+                                let _ = session.write(b"OK\r\n");
+                                session.finalize_response();
+                                ADMIN_RESPONSE_COMPOSE.increment();
+                            }
                             AdminRequest::Stats => {
                                 Self::handle_stats_request(session);
                             }

src/rust/core/server/src/threads/listener.rs

@@ -224,6 +224,10 @@ impl Listener {
                                 Signal::Shutdown => {
                                     return;
                                 }
+                                Signal::Stop => {
+                                    warn!("received stop");
+                                    return;
+                                }
                             }
                         }
                     }

src/rust/core/server/src/threads/worker/multi.rs

@@ -12,6 +12,7 @@ use super::*;
 use crate::poll::Poll;
 use crate::threads::worker::StorageWorker;
 use crate::threads::worker::TokenWrapper;
+use crate::QUEUE_RETRIES;
 use common::signal::Signal;
 use common::time::Instant;
 use config::WorkerConfig;
@@ -26,11 +27,6 @@ use session::Session;
 use std::io::{BufRead, Write};
 use std::sync::Arc;
 
-// TODO(bmartin): this *should* be plenty safe, the queue should rarely ever be
-// full, and a single wakeup should drain at least one message and make room for
-// the request. A stat to prove that this is sufficient would be good.
-const QUEUE_RETRIES: usize = 3;
-
 const WAKER_TOKEN: usize = usize::MAX;
 
 /// A `MultiWorker` handles events on `Session`s and routes storage requests to
@@ -110,15 +106,19 @@ where
                         self.handle_new_sessions();
                         self.handle_storage_queue();
 
-                        #[allow(clippy::never_loop)]
                         // check if we received any signals from the admin thread
+                        #[allow(clippy::never_loop)]
                         while let Ok(signal) = self.signal_queue.recv_from(0) {
                             match signal {
                                 Signal::Shutdown => {
                                     // if we received a shutdown, we can return
                                     // and stop processing events
                                     return;
                                 }
+                                Signal::Stop => {
+                                    warn!("received stop");
+                                    return;
+                                }
                             }
                         }
                     }