Disable GC by default

Cargo.lock

@@ -1090,7 +1090,8 @@ dependencies = [
 [[package]]
 name = "librocksdb-sys"
 version = "6.17.3"
-source = "git+https://github.com/rust-rocksdb/rust-rocksdb.git?rev=7dd6258b07861b9332f827b416e50e5aee69aea1#7dd6258b07861b9332f827b416e50e5aee69aea1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5da125e1c0f22c7cae785982115523a0738728498547f415c9054cb17c7e89f9"
 dependencies = [
  "bindgen",
  "cc",
@@ -1814,7 +1815,8 @@ dependencies = [
 [[package]]
 name = "rocksdb"
 version = "0.16.0"
-source = "git+https://github.com/rust-rocksdb/rust-rocksdb.git?rev=7dd6258b07861b9332f827b416e50e5aee69aea1#7dd6258b07861b9332f827b416e50e5aee69aea1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c749134fda8bfc90d0de643d59bfc841dcb3ac8a1062e12b6754bd60235c48b3"
 dependencies = [
  "libc",
  "librocksdb-sys",

pageserver/Cargo.toml

@@ -32,7 +32,7 @@ tokio-postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="a
 postgres-types = { git = "https://github.com/zenithdb/rust-postgres.git", rev="a0d067b66447951d1276a53fb09886539c3fa094" }
 postgres-protocol = { git = "https://github.com/zenithdb/rust-postgres.git", rev="a0d067b66447951d1276a53fb09886539c3fa094" }
 postgres = { git = "https://github.com/zenithdb/rust-postgres.git", rev="a0d067b66447951d1276a53fb09886539c3fa094" }
-rocksdb = { git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev="7dd6258b07861b9332f827b416e50e5aee69aea1" }
+rocksdb = "0.16.0"
 anyhow = "1.0"
 crc32c = "0.6.0"
 walkdir = "2"

pageserver/src/bin/pageserver.rs

@@ -19,7 +19,7 @@ use slog::Drain;
 
 use pageserver::{page_service, tui, zenith_repo_dir, PageServerConf};
 
-const DEFAULT_GC_HORIZON: u64 = 64 * 1024 * 1024;
+const DEFAULT_GC_HORIZON: u64 = 0; //64 * 1024 * 1024;
 
 fn main() -> Result<()> {
     let arg_matches = App::new("Zenith page server")
@@ -47,10 +47,15 @@ fn main() -> Result<()> {
         )
         .arg(
             Arg::with_name("gc_horizon")
-                .short("g")
                 .long("gc_horizon")
                 .takes_value(true)
-                .help("Garbage colletor horizon"),
+                .help("Distance from current LSN to perform all wal records cleanup"),
+        )
+        .arg(
+            Arg::with_name("gc_period")
+                .long("gc_period")
+                .takes_value(true)
+                .help("Interval between garbage collector iterations"),
         )
         .get_matches();
 

pageserver/src/page_cache.rs

@@ -148,15 +148,15 @@ pub fn get_or_restore_pagecache(
                     walredo::wal_redo_main(&conf_copy, timelineid);
                 })
                 .unwrap();
-
-            let conf_copy = conf.clone();
-            let _gc_thread = thread::Builder::new()
-                .name("Garbage collection thread".into())
-                .spawn(move || {
-                    gc_thread_main(&conf_copy, timelineid);
-                })
-                .unwrap();
-
+            if conf.gc_horizon != 0 {
+                let conf_copy = conf.clone();
+                let _gc_thread = thread::Builder::new()
+                    .name("Garbage collection thread".into())
+                    .spawn(move || {
+                        gc_thread_main(&conf_copy, timelineid);
+                    })
+                    .unwrap();
+            }
             Ok(result)
         }
     }
@@ -381,11 +381,13 @@ impl WALRecord {
 // Public interface functions
 
 impl PageCache {
-
     fn do_gc(&self, conf: &PageServerConf) -> anyhow::Result<Bytes> {
         let mut minbuf = BytesMut::new();
         let mut maxbuf = BytesMut::new();
-        let cf = self.db.cf_handle(rocksdb::DEFAULT_COLUMN_FAMILY_NAME).unwrap();
+        let cf = self
+            .db
+            .cf_handle(rocksdb::DEFAULT_COLUMN_FAMILY_NAME)
+            .unwrap();
         loop {
             thread::sleep(conf.gc_period);
             let last_lsn = self.get_last_valid_lsn();
@@ -406,9 +408,10 @@ impl PageCache {
                 loop {
                     maxbuf.clear();
                     maxkey.pack(&mut maxbuf);
-                    let mut iter = self
-                        .db
-                        .iterator(rocksdb::IteratorMode::From(&maxbuf[..], rocksdb::Direction::Reverse));
+                    let mut iter = self.db.iterator(rocksdb::IteratorMode::From(
+                        &maxbuf[..],
+                        rocksdb::Direction::Reverse,
+                    ));
                     if let Some((k, v)) = iter.next() {
                         minbuf.clear();
                         minbuf.extend_from_slice(&v);
@@ -436,9 +439,10 @@ impl PageCache {
 
                         if last_lsn > horizon {
                             // locate most recent record before horizon
-                            let mut iter = self
-                                .db
-                                .iterator(rocksdb::IteratorMode::From(&maxbuf[..], rocksdb::Direction::Reverse));
+                            let mut iter = self.db.iterator(rocksdb::IteratorMode::From(
+                                &maxbuf[..],
+                                rocksdb::Direction::Reverse,
+                            ));
                             if let Some((k, v)) = iter.next() {
                                 minbuf.clear();
                                 minbuf.extend_from_slice(&v);
@@ -544,9 +548,10 @@ impl PageCache {
 
         buf.clear();
         maxkey.pack(&mut buf);
-        let mut iter = self
-            .db
-            .iterator_opt(rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse), readopts);
+        let mut iter = self.db.iterator_opt(
+            rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse),
+            readopts,
+        );
         let entry_opt = iter.next();
 
         if entry_opt.is_none() {
@@ -612,9 +617,10 @@ impl PageCache {
 
         buf.clear();
         entry.key.pack(&mut buf);
-        let iter = self
-            .db
-            .iterator_opt(rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse), readopts);
+        let iter = self.db.iterator_opt(
+            rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse),
+            readopts,
+        );
 
         let mut base_img: Option<Bytes> = None;
         let mut records: Vec<WALRecord> = Vec::new();
@@ -825,9 +831,10 @@ impl PageCache {
         loop {
             buf.clear();
             key.pack(&mut buf);
-            let mut iter = self
-                .db
-                .iterator(rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse));
+            let mut iter = self.db.iterator(rocksdb::IteratorMode::From(
+                &buf[..],
+                rocksdb::Direction::Reverse,
+            ));
             if let Some((k, v)) = iter.next() {
                 buf.clear();
                 buf.extend_from_slice(&k);
@@ -866,9 +873,10 @@ impl PageCache {
         };
         let mut buf = BytesMut::new();
         key.pack(&mut buf);
-        let mut iter = self
-            .db
-            .iterator(rocksdb::IteratorMode::From(&buf[..], rocksdb::Direction::Reverse));
+        let mut iter = self.db.iterator(rocksdb::IteratorMode::From(
+            &buf[..],
+            rocksdb::Direction::Reverse,
+        ));
         if let Some((k, _v)) = iter.next() {
             buf.clear();
             buf.extend_from_slice(&k);

pageserver/src/walredo.rs

@@ -81,11 +81,11 @@ pub fn wal_redo_main(conf: &PageServerConf, timelineid: ZTimelineId) {
             let result = handle_apply_request(&pcache, &process, &runtime, request);
             if result.is_err() {
                 // On error, kill the process.
-                error!("Kill wal redo process on error");
                 break;
             }
         }
 
+        info!("killing WAL redo postgres process");
         let _ = runtime.block_on(process.stdin.get_mut().shutdown());
         let mut child = process.child;
         drop(process.stdin);