Sync internal commits

.bleep

@@ -1 +1 @@
-deb3c5409e938ec9c7d0da9b7a2d331eabbb2cd5
+7d3baa7e49e9b5c7d76775971c9f57f604209f38

pingora-core/Cargo.toml

@@ -35,13 +35,13 @@ http = { workspace = true }
 log = { workspace = true }
 h2 = { workspace = true }
 lru = { workspace = true }
-nix = "0.24"
+nix = "~0.24.3"
 structopt = "0.3"
 once_cell = { workspace = true }
 serde = { version = "1.0", features = ["derive"] }
 serde_yaml = "0.8"
 libc = "0.2.70"
-chrono = { version = "0.4", features = ["alloc"], default-features = false }
+chrono = { version = "~0.4.31", features = ["alloc"], default-features = false }
 thread_local = "1.0"
 prometheus = "0.13"
 daemonize = "0.5.0"

pingora-core/src/protocols/http/v1/client.rs

@@ -293,7 +293,7 @@ impl HttpSession {
                         InvalidHTTPHeader,
                         format!("buf: {:?}", String::from_utf8_lossy(&buf)),
                         e,
-                    )
+                    );
                 }
             }
         }
@@ -428,10 +428,34 @@ impl HttpSession {
         is_buf_keepalive(self.get_header(header::CONNECTION).map(|v| v.as_bytes()))
     }
 
-    // `Keep-Alive: timeout=5, max=1000` => 5, 1000
+    /// `Keep-Alive: timeout=5, max=1000` => 5, 1000
+    /// This is defined in the below spec, this not part of any RFC, so
+    /// it's behavior is different on different platforms.
+    /// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Keep-Alive
     fn get_keepalive_values(&self) -> (Option<u64>, Option<usize>) {
-        // TODO: implement this parsing
-        (None, None)
+        let Some(keep_alive_header) = self.get_header("Keep-Alive") else {
+            return (None, None);
+        };
+
+        let Ok(header_value) = str::from_utf8(keep_alive_header.as_bytes()) else {
+            return (None, None);
+        };
+
+        let mut timeout = None;
+        let mut max = None;
+
+        for param in header_value.split(',') {
+            let mut parts = param.splitn(2, '=').map(|s| s.trim());
+            match (parts.next(), parts.next()) {
+                (Some("timeout"), Some(timeout_value)) => {
+                    timeout = timeout_value.trim().parse().ok()
+                }
+                (Some("max"), Some(max_value)) => max = max_value.trim().parse().ok(),
+                _ => {}
+            }
+        }
+
+        (timeout, max)
     }
 
     /// Close the connection abruptly. This allows to signal the server that the connection is closed
@@ -532,7 +556,7 @@ impl HttpSession {
     }
 
     fn init_req_body_writer(&mut self, header: &RequestHeader) {
-        if self.is_upgrade_req() {
+        if is_upgrade_req(header) {
             self.body_writer.init_http10();
         } else {
             self.init_body_writer_comm(&header.headers)
@@ -892,6 +916,26 @@ mod tests_stream {
         }
     }
 
+    #[tokio::test]
+    async fn init_body_for_upgraded_req() {
+        use crate::protocols::http::v1::body::BodyMode;
+
+        let wire =
+            b"GET / HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: WS\r\nContent-Length: 0\r\n\r\n";
+        let mock_io = Builder::new().write(wire).build();
+        let mut http_stream = HttpSession::new(Box::new(mock_io));
+        let mut new_request = RequestHeader::build("GET", b"/", None).unwrap();
+        new_request.insert_header("Connection", "Upgrade").unwrap();
+        new_request.insert_header("Upgrade", "WS").unwrap();
+        // CL is ignored when Upgrade presents
+        new_request.insert_header("Content-Length", "0").unwrap();
+        let _ = http_stream
+            .write_request_header(Box::new(new_request))
+            .await
+            .unwrap();
+        assert_eq!(http_stream.body_writer.body_mode, BodyMode::HTTP1_0(0));
+    }
+
     #[tokio::test]
     async fn read_switching_protocol() {
         init_log();
@@ -1061,6 +1105,65 @@ mod tests_stream {
                 .keepalive_timeout,
             KeepaliveStatus::Off
         );
+
+        async fn build_resp_with_keepalive_values(keep_alive: &str) -> HttpSession {
+            let input = format!("HTTP/1.1 200 OK\r\nKeep-Alive: {keep_alive}\r\n\r\n");
+            let mock_io = Builder::new().read(input.as_bytes()).build();
+            let mut http_stream = HttpSession::new(Box::new(mock_io));
+            let res = http_stream.read_response().await;
+            assert_eq!(input.len(), res.unwrap());
+            http_stream.respect_keepalive();
+            http_stream
+        }
+
+        assert_eq!(
+            build_resp_with_keepalive_values("timeout=5, max=1000")
+                .await
+                .get_keepalive_values(),
+            (Some(5), Some(1000))
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values("max=1000, timeout=5")
+                .await
+                .get_keepalive_values(),
+            (Some(5), Some(1000))
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values(" timeout = 5, max = 1000 ")
+                .await
+                .get_keepalive_values(),
+            (Some(5), Some(1000))
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values("timeout=5")
+                .await
+                .get_keepalive_values(),
+            (Some(5), None)
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values("max=1000")
+                .await
+                .get_keepalive_values(),
+            (None, Some(1000))
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values("a=b")
+                .await
+                .get_keepalive_values(),
+            (None, None)
+        );
+
+        assert_eq!(
+            build_resp_with_keepalive_values("")
+                .await
+                .get_keepalive_values(),
+            (None, None)
+        );
     }
 
     /* Note: body tests are covered in server.rs */

pingora-core/src/server/mod.rs

@@ -175,7 +175,8 @@ impl Server {
     ///
     /// Command line options can either be passed by parsing the command line arguments via
     /// `Opt::from_args()`, or be generated by other means.
-    pub fn new(opt: Option<Opt>) -> Result<Server> {
+    pub fn new(opt: impl Into<Option<Opt>>) -> Result<Server> {
+        let opt = opt.into();
         let (tx, rx) = watch::channel(false);
 
         let conf = if let Some(opt) = opt.as_ref() {

pingora-http/src/lib.rs

@@ -169,7 +169,7 @@ impl RequestHeader {
     /// Insert the header name and value to `self`.
     ///
     /// Different from [Self::append_header()], this method will replace all other existing headers
-    /// under the same name (case insensitive).
+    /// under the same name (case-insensitive).
     pub fn insert_header(
         &mut self,
         name: impl IntoCaseHeaderName,
@@ -476,21 +476,21 @@ fn clone_resp_parts(me: &RespParts) -> RespParts {
 
 // This function returns an upper bound on the size of the header map used inside the http crate.
 // As of version 0.2, there is a limit of 1 << 15 (32,768) items inside the map. There is an
-// assertion against this size inside the crate so we want to avoid panicking by not exceeding this
+// assertion against this size inside the crate, so we want to avoid panicking by not exceeding this
 // upper bound.
 fn http_header_map_upper_bound(size_hint: Option<usize>) -> usize {
     // Even though the crate has 1 << 15 as the max size, calls to `with_capacity` invoke a
     // function that returns the size + size / 3.
     //
     // See https://github.com/hyperium/http/blob/34a9d6bdab027948d6dea3b36d994f9cbaf96f75/src/header/map.rs#L3220
     //
-    // Therefore we set our max size to be even lower so we guarantee ourselves we won't hit that
+    // Therefore we set our max size to be even lower, so we guarantee ourselves we won't hit that
     // upper bound in the crate. Any way you cut it, 4,096 headers is insane.
     const PINGORA_MAX_HEADER_COUNT: usize = 4096;
     const INIT_HEADER_SIZE: usize = 8;
 
-    // We select the size hint or the max size here such that we pick a value substantially lower
-    // 1 << 15 with room to grow the header map.
+    // We select the size hint or the max size here, ensuring that we pick a value substantially lower
+    // than 1 << 15 with room to grow the header map.
     std::cmp::min(
         size_hint.unwrap_or(INIT_HEADER_SIZE),
         PINGORA_MAX_HEADER_COUNT,
@@ -509,7 +509,7 @@ fn append_header_value<T>(
         .as_slice()
         .try_into()
         .or_err(InvalidHTTPHeader, "invalid header name")?;
-    // storage the original case in the map
+    // store the original case in the map
     if let Some(name_map) = name_map {
         name_map.append(header_name.clone(), case_header_name);
     }
@@ -530,7 +530,7 @@ fn insert_header_value<T>(
         .try_into()
         .or_err(InvalidHTTPHeader, "invalid header name")?;
     if let Some(name_map) = name_map {
-        // storage the original case in the map
+        // store the original case in the map
         name_map.insert(header_name.clone(), case_header_name);
     }
     value_map.insert(header_name, value);
@@ -562,7 +562,7 @@ fn header_to_h1_wire(key_map: Option<&CaseMap>, value_map: &HMap, buf: &mut impl
         let iter = key_map.iter().zip(value_map.iter());
         for ((header, case_header), (header2, val)) in iter {
             if header != header2 {
-                // in case the header iter order changes in further version of HMap
+                // in case the header iteration order changes in future versions of HMap
                 panic!("header iter mismatch {}, {}", header, header2)
             }
             buf.put_slice(case_header.as_slice());

pingora-memory-cache/src/lib.rs

@@ -81,7 +81,7 @@ pub struct MemoryCache<K: Hash, T: Clone> {
     pub(crate) hasher: RandomState,
 }
 
-impl<K: Hash, T: Clone + Send + Sync> MemoryCache<K, T> {
+impl<K: Hash, T: Clone + Send + Sync + 'static> MemoryCache<K, T> {
     /// Create a new [MemoryCache] with the given size.
     pub fn new(size: usize) -> Self {
         MemoryCache {

pingora-memory-cache/src/read_through.rs

@@ -123,7 +123,7 @@ where
 impl<K, T, CB, S> RTCache<K, T, CB, S>
 where
     K: Hash + Send,
-    T: Clone + Send + Sync,
+    T: Clone + Send + Sync + 'static,
 {
     /// Create a new [RTCache] of given size. `lock_age` defines how long a lock is valid for.
     /// `lock_timeout` is used to stop a lookup from holding on to the key for too long.
@@ -142,7 +142,7 @@ where
 impl<K, T, CB, S> RTCache<K, T, CB, S>
 where
     K: Hash + Send,
-    T: Clone + Send + Sync,
+    T: Clone + Send + Sync + 'static,
     CB: Lookup<K, T, S>,
 {
     /// Query the cache for a given value. If it exists and no TTL is configured initially, it will
@@ -288,7 +288,7 @@ where
 impl<K, T, CB, S> RTCache<K, T, CB, S>
 where
     K: Hash + Send,
-    T: Clone + Send + Sync,
+    T: Clone + Send + Sync + 'static,
     CB: MultiLookup<K, T, S>,
 {
     /// Same behavior as [RTCache::get] but for an arbitrary amount of keys.

tinyufo/Cargo.toml

@@ -20,13 +20,15 @@ ahash = { workspace = true }
 flurry = "<0.5.0" # Try not to require Rust 1.71
 parking_lot = "0"
 crossbeam-queue = "0"
+crossbeam-skiplist = "0"
 
 [dev-dependencies]
 rand = "0"
 lru = "0"
 zipf = "7"
 moka = { version = "0", features = ["sync"] }
 dhat = "0"
+quick_cache = "0.4"
 
 [[bench]]
 name = "bench_perf"

tinyufo/README.md

@@ -38,12 +38,12 @@ Because of TinyUFO's lock-free design, it greatly outperforms the others.
 
 ### Memory overhead
 
-The table below show the memory allocation (in bytes) of the compared cache library under certain workloads to store zero-sized assets.
+TinyUFO provides a compact mode to trade raw read speed for more memory efficiency. Whether the saving worthy the trade off depends on the actual size and the work load. For small in-memory assets, the saved memory means more things can be cached.
 
-| cache size | TinyUFO | LRU | moka |
-| -------- | ------- | ------- | ------ |
-| 100 | 39,409 | 9,408 | 354,376
-| 1000 | 236,053 | 128,512 | 535,888
-| 10000 | 2,290,635 | 1,075,648 | 2,489,088
+The table below show the memory allocation (in bytes) of the compared cache library under certain workloads to store zero-sized assets.
 
-Whether these overheads matter depends on the actual sizes and volume of the assets. The more advanced algorithms are likely to be less memory efficient than the simple LRU.
+| cache size | TinyUFO | TinyUFO compact | LRU | moka |
+| -------- | ------- | ------- | ------- | ------ |
+| 100 | 39,409 | 19,000 | 9,408 | 354,376
+| 1000 | 236,053 | 86,352 | 128,512 | 535,888
+| 10000 | 2,290,635 | 766,024|  1,075,648 | 2,489,088