[water_http] updating water_http version and add cached queries test to the default config (#10641)

* Add water-http framework

* benchmark_config.json update

* updating water_http version to release one

* updating water_http release version

* version update

* fix water_http errors in update queries and update water_http version

* adding cached-queries to default test

* [water-http] remove unnecessary buffers

* add maintainers list into benchmark_config.json

Author: HassanSharara

.gitignore

@@ -1,7 +1,7 @@
 installs
 node_modules/
 .travis.bak
-
+t.bat
 # Added for pedestal framework test
 .lein-deps-sum
 

frameworks/Rust/water-http/Cargo.toml

@@ -1,12 +1,12 @@
 [package]
 name = "water-http"
 version = "0.1.0"
-edition = "2018"
+edition = "2024"
 
 [dependencies]
 askama = "0.14.0"
 tokio = { version = "1.47.1", features = ["full"] }
-water_http = { version = "3.2.1-beta.14" ,optional = true , features = ["use_only_http1"]}
+water_http = { features = ["use_io_uring","use_only_http1"],optional = true , version = "3.4.2-beta.4" }
 smallvec = "1.15.1"
 nanorand = "0.8.0"
 tokio-postgres = "0.7.15"

frameworks/Rust/water-http/benchmark_config.json

@@ -1,15 +1,17 @@
 
 {
   "framework": "water-http",
+  "maintainers" : ["HassanSharara"],
   "tests": [
     {
-      "default": {
+        "default": {
         "json_url": "/json",
         "plaintext_url": "/plaintext",
         "fortune_url": "/fortunes",
         "db_url": "/db",
         "query_url": "/queries?q=",
         "update_url": "/updates?q=",
+        "cached_query_url": "/cached-queries?q=",
         "port": 8080,
         "approach": "Realistic",
         "classification": "Fullstack",
@@ -24,5 +26,5 @@
         "display_name": "water_http"
       }
     }
-  ]
+    ]
 }

frameworks/Rust/water-http/src/cached.rs

@@ -170,9 +170,9 @@ pub struct ThreadSharedStruct{
 impl ThreadSharedStruct {
 
     #[inline(always)]
-    pub fn get_value(id:i32)->&'static i32{
+    pub fn get_value(id:i32)->Option<&'static i32>{
         let map = unsafe {CACHED_VALUES.as_ref().unwrap().get(&id)} ;
-        map.unwrap()
+        map
     }
     pub fn get_cached_queries(&self,num:usize)->&[u8]{
         let buf = unsafe{&mut *(self.writing_buffer.get())};
@@ -181,8 +181,11 @@ impl ThreadSharedStruct {
         let mut writer = BytesMuteWriter(buf);
         let mut rn = self.rng.clone();
         for _ in 0..num {
-            let rd: i32 = (rn.generate::<u32>() & 0x3FFF) as i32 % 10_000 + 1;
-            let v = Self::get_value(rd);
+            let rd = (rn.generate::<u32>() % 10_000 ) as i32;
+            let v = match Self::get_value(rd) {
+                None => {continue}
+                Some(c) => {c}
+            };
             writer.extend_from_slice(br"{");
             _ = write!(writer, r#""id":{},"randomnumber":{}"#, rd, v);
             writer.extend_from_slice(br"},");

frameworks/Rust/water-http/src/db.rs

@@ -1,17 +1,19 @@
 #![cfg(any(feature = "db",feature = "all"))]
-use std::{borrow::Cow, io};
+use std::{borrow::Cow, io, ptr};
 use std::fmt::Arguments;
 use std::io::Write;
 use std::mem::MaybeUninit;
 use std::rc::Rc;
 use std::cell::UnsafeCell;
+use std::collections::HashMap;
 use bytes::Buf;
 use nanorand::{Rng, WyRand};
-use tokio_postgres::{connect, Client, Statement, NoTls};
+use tokio_postgres::{connect, Client, Statement, NoTls, Error};
 use tokio_postgres::types::private::BytesMut;
 use crate::models::{Fortune, FortuneTemplate, World};
 use sonic_rs::prelude::WriteExt;
 use yarte::TemplateBytesTrait;
+pub static  mut CACHED_VALUES:Option<HashMap<i32,i32>> = None;
 
 /// Database connection pool with thread-local RNG
 pub struct DbConnectionPool {
@@ -60,8 +62,6 @@ impl DbConnectionPool {
 /// Reusable buffer pool per connection
 struct BufferPool {
     body: BytesMut,
-    worlds: Vec<World>,
-    numbers: Vec<i32>,
     fortunes: Vec<Fortune>,
     fortune_output: Vec<u8>,
 }
@@ -70,8 +70,6 @@ impl BufferPool {
     fn new() -> Self {
         Self {
             body: BytesMut::with_capacity(4096),
-            worlds: Vec::with_capacity(501),
-            numbers: Vec::with_capacity(501),
 
             fortunes: Vec::with_capacity(501),
             fortune_output: Vec::with_capacity(4096),
@@ -130,7 +128,7 @@ impl PgConnection {
     }    /// Connect to the database
 
     #[inline(always)]
-  pub fn generate_update_values_stmt(batch_size: usize) -> String {
+    pub fn generate_update_values_stmt(batch_size: usize) -> String {
 
         let mut sql = String::from("UPDATE world SET randomNumber = w.r FROM (VALUES ");
 
@@ -156,38 +154,34 @@ impl PgConnection {
     /// Get a single random world - optimized with buffer reuse
     #[inline]
     pub async fn get_world(&self) -> &[u8] {
-        let rd = (self.rang.clone().generate::<u32>() % 10_000 + 1) as i32;
+        let rd = (self.rang.clone().generate::<u32>() % 10_000 ) as i32;
         let row = self.cl.query_one(&self.world, &[&rd]).await.unwrap();
-
         let buffers = self.buffers();
         buffers.body.clear();
-
         sonic_rs::to_writer(
             BytesMuteWriter(&mut buffers.body),
             &World {
                 id: row.get(0),
                 randomnumber: row.get(1),
             },
         ).unwrap();
-
         buffers.body.chunk()
     }
 
     /// Get multiple random worlds - optimized with buffer reuse
     pub async fn get_worlds(&self, num: usize) -> &[u8] {
         let buffers = self.buffers();
-        buffers.worlds.clear();
-        let mut rn = self.rang.clone();
+        let mut worlds = Vec::with_capacity(num);
         for _ in 0..num {
-            let id: i32 = (rn.generate::<u32>() & 0x3FFF) as i32 % 10_000 + 1;
+            let id = (self.rang.clone().generate::<u32>() % 10_000 ) as i32;
             let row = self.cl.query_one(&self.world, &[&id]).await.unwrap();
-            buffers.worlds.push(World {
+           worlds.push(World {
                 id: row.get(0),
                 randomnumber: row.get(1),
             });
         }
         buffers.body.clear();
-        sonic_rs::to_writer(BytesMuteWriter(&mut buffers.body), &buffers.worlds).unwrap();
+        sonic_rs::to_writer(BytesMuteWriter(&mut buffers.body), &worlds).unwrap();
         buffers.body.chunk()
     }
     /// Update worlds in batch - optimized with buffer reuse
@@ -215,23 +209,24 @@ impl PgConnection {
         futures.extend(ids.iter().map(|x| async move {self.cl.query_one(&self.world,&[&x]).await}));
         futures_util::future::join_all(futures).await;
         ids.sort_unstable();
-        buffers.worlds.clear();
+        let mut worlds = Vec::with_capacity(num);
+        let mut numbers = Vec::with_capacity(num);
         for index in 0..num {
             let s_id = (rng.generate::<u32>() % 10_000 + 1 ) as i32;
-            buffers.worlds.push(World{
+            worlds.push(World{
                 id:ids[index],
                 randomnumber:s_id
             });
-            buffers.numbers.push(s_id);
+           numbers.push(s_id);
         }
         buffers.body.clear();
         for index in 0..num {
             params.push(&ids[index]);
-            params.push(&buffers.numbers[index]);
+            params.push(&numbers[index]);
         }
 
         _=self.cl.execute(&self.updates[num - 1], &params).await.unwrap();
-        sonic_rs::to_writer(BytesMuteWriter(&mut buffers.body), &buffers.worlds).unwrap();
+        sonic_rs::to_writer(BytesMuteWriter(&mut buffers.body), &worlds).unwrap();
         buffers.body.chunk()
     }
 
@@ -264,11 +259,50 @@ impl PgConnection {
         // Return reference to buffer - zero-copy!
         Ok(&buffers.fortune_output)
     }
+
+
+    pub fn get_cached_queries(&self,num:usize)->&[u8]{
+        let buf = self.buffers();
+        let buf = &mut buf.body;
+        buf.clear();
+        buf.extend_from_slice(br#"["#);
+        let mut writer = BytesMuteWriter(buf);
+        for _ in 0..num {
+            let rd = (self.rang.clone().generate::<u32>() % 10_000 ) as i32;
+            let v = match self.get_world_id_for_cache(rd){
+                None => {continue}
+                Some(e)=>{e}
+            };
+            writer.extend_from_slice(br"{");
+            _ = write!(writer, r#""id":{},"randomnumber":{}"#, rd, v);
+            writer.extend_from_slice(br"},");
+        }
+        if buf.len() >1  {buf.truncate(buf.len() - 1);}
+        buf.extend_from_slice(b"]");
+        return &buf[..]
+    }
+
+    fn get_world_id_for_cache(&self, id: i32) -> Option<&i32> {
+        unsafe {
+            let ptr = ptr::addr_of!(CACHED_VALUES);
+
+            match &*ptr {
+                Some(map) => map.get(&id),
+                None => None,
+            }
+        }
+    }
 }
 
 /// Zero-copy writer for BytesMut
 pub struct BytesMuteWriter<'a>(pub &'a mut BytesMut);
+impl BytesMuteWriter<'_> {
 
+    #[inline(always)]
+    pub fn extend_from_slice(&mut self,data:&[u8]){
+        self.0.extend_from_slice(data);
+    }
+}
 impl Write for BytesMuteWriter<'_> {
     #[inline(always)]
     fn write(&mut self, src: &[u8]) -> Result<usize, io::Error> {

frameworks/Rust/water-http/src/server.rs

@@ -1,13 +1,18 @@
-
+use std::collections::HashMap;
 use std::pin::Pin;
+use std::ptr;
 use std::rc::Rc;
+use tokio::task::LocalSet;
 use water_http::{InitControllersRoot, RunServer, WaterController};
-use water_http::server::ServerConfigurations;
-use crate::db::{DbConnectionPool};
+use water_http::http::{HttpSender, ResponseData};
+use water_http::server::{HttpContext, ServerConfigurations};
+use crate::db::{CACHED_VALUES, DbConnectionPool};
 InitControllersRoot! {
     name:ROOT,
     holder_type:MainType,
     shared_type:SharedType,
+    headers_length:6,
+    queries_length:3
 }
 
 pub struct ThreadSharedStruct{
@@ -22,6 +27,43 @@ pub type SharedType = Rc<ThreadSharedStruct>;
 
 pub    fn run_server(){
 
+
+    _= std::thread::spawn(
+        ||{
+            let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+            rt.block_on(async move {
+                const URL:&'static str = "postgres://benchmarkdbuser:benchmarkdbpass@tfb-database/hello_world";
+                // const URL:&'static str = "postgres://postgres:root@localhost:5432/techmpower";
+
+                let  mut pool = DbConnectionPool{
+                    connections:Vec::with_capacity( 1
+                    ),
+                    next:0.into(),
+                    // rt:tokio::runtime::Builder::new_multi_thread().enable_all().worker_threads(cpu_nums).build().unwrap()
+                };
+
+                let local_set = LocalSet::new();
+
+                _= local_set.run_until(async move {
+                    tokio::task::spawn_local(async move {
+                        pool.fill_pool(URL, 1).await;
+                        let connection = pool.get_connection();
+                        let statement = connection.cl.prepare("SELECT id,randomnumber FROM World").await.unwrap();
+                        let res = connection.cl.query(&statement,&[]).await.unwrap();
+                        let mut map = HashMap::new();
+                        for row in res {
+                            map.insert(row.get(0),row.get(1));
+                        }
+                        unsafe {
+                            let ptr = ptr::addr_of_mut!(CACHED_VALUES);
+                            ptr.write(Some(map));
+                        }
+                    }).await
+                }).await;
+
+            });
+        }
+    ).join();
     let cpu_nums = num_cpus::get();
 
 
@@ -71,6 +113,9 @@ const JSON_RESPONSE:&'static [u8] = br#"{"message":"Hello, World!"}"#;
 #[cfg(any(feature = "json_plaintext",feature = "all"))]
 const P:&'static [u8] = br#"Hello, World!"#;
 
+
+
+
 #[cfg(feature = "all")]
 WaterController! {
     holder -> super::MainType,
@@ -151,6 +196,27 @@ WaterController! {
         }
 
 
+        GET -> "cached-queries" -> cached(context)async {
+              let q = context
+        .get_from_path_query("q")
+        .and_then(|v| v.parse::<usize>().ok()) // safely parse
+        .unwrap_or(1)                          // default to 1 if missing or invalid
+        .clamp(1, 500);
+
+    let   connection:Shared = context.thread_shared_struct.clone().unwrap().clone();
+    let connection = connection.pg_connection.get_connection();
+    let data = connection.get_cached_queries(q);
+    let mut sender= context.sender();
+    sender.set_header_ef("Content-Type","application/json");
+    sender.set_header_ef("Server","water");
+    let date = httpdate::fmt_http_date(std::time::SystemTime::now());
+    sender.set_header_ef("Date",date);
+    _= sender.send_data_as_final_response(
+        http::ResponseData::Slice(data)
+    ).await;
+        }
+
+
         GET -> fortunes -> ft (context){
 
             let   connection:Shared = context.thread_shared_struct.clone().unwrap().clone();

frameworks/Rust/water-http/water-http-cached.dockerfile

@@ -0,0 +1,12 @@
+FROM rust:1.93
+
+RUN apt-get update -yqq && apt-get install -yqq cmake g++
+
+WORKDIR /water
+COPY . .
+RUN cargo clean
+RUN RUSTFLAGS="-C target-cpu=native" cargo build --release --bin cache --features cache
+
+EXPOSE 8080
+
+CMD ./target/release/cache