update

Author: luoluoyuyu

conf/config.yaml

@@ -49,6 +49,19 @@ wasm:
   # When cache exceeds this size, least recently used items will be evicted
   max_cache_size: 104857600
 
+# Streaming Runtime Configuration
+streaming:
+  # Global memory pool size for the streaming runtime (network buffering, backpressure).
+  # When not set, auto-detected as 70% of physical memory.
+  # Fallback: 256 MiB if detection fails.
+  # max_memory_bytes: 268435456
+  
+  # Memory budget per stateful operator (aggregation, join, window).
+  # Each operator gets its own independent memory controller with this limit.
+  # When exceeded, the operator spills state to disk automatically.
+  # Default: 67108864 (64 MiB)
+  per_operator_state_memory_bytes: 67108864
+
 # State Storage Configuration
 # Used to store runtime state data for tasks
 state_storage:

src/config/global_config.rs

@@ -21,9 +21,9 @@ use crate::config::wasm_config::WasmConfig;
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct StreamingConfig {
-    /// Maximum heap memory (in bytes) available to the streaming runtime's memory pool.
-    /// Defaults to 256 MiB when absent.
-    pub max_memory_bytes: Option<usize>,
+    pub max_memory_bytes: Option<u64>,
+    /// Total bytes for the global operator-state [`MemoryPool`](crate::runtime::memory::MemoryPool) (all stores share this quota).
+    pub per_operator_state_memory_bytes: Option<u64>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]

src/config/mod.rs

@@ -17,6 +17,7 @@ pub mod paths;
 pub mod python_config;
 pub mod service_config;
 pub mod storage;
+pub mod system;
 pub mod wasm_config;
 
 pub use global_config::GlobalConfig;

src/config/system.rs

@@ -0,0 +1,233 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::io;
+
+pub struct SystemMemoryInfo {
+    pub total_physical: u64,
+    pub available_physical: u64,
+    pub total_virtual: u64,
+    pub available_virtual: u64,
+}
+
+pub fn system_memory_info() -> io::Result<SystemMemoryInfo> {
+    sys::system_memory_info()
+}
+
+#[cfg(target_os = "linux")]
+mod sys {
+    use super::SystemMemoryInfo;
+    use std::io;
+
+    pub fn system_memory_info() -> io::Result<SystemMemoryInfo> {
+        let content = std::fs::read_to_string("/proc/meminfo")?;
+
+        let mut total_physical: Option<u64> = None;
+        let mut available_physical: Option<u64> = None;
+        let mut swap_total: u64 = 0;
+        let mut swap_free: u64 = 0;
+
+        for line in content.lines() {
+            if let Some(v) = parse_meminfo_kb(line, "MemTotal:") {
+                total_physical = Some(v);
+            } else if let Some(v) = parse_meminfo_kb(line, "MemAvailable:") {
+                available_physical = Some(v);
+            } else if let Some(v) = parse_meminfo_kb(line, "SwapTotal:") {
+                swap_total = v;
+            } else if let Some(v) = parse_meminfo_kb(line, "SwapFree:") {
+                swap_free = v;
+            }
+        }
+
+        let total_phys = total_physical.ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::NotFound,
+                "MemTotal not found in /proc/meminfo",
+            )
+        })?;
+        let avail_phys = available_physical.unwrap_or(0);
+
+        Ok(SystemMemoryInfo {
+            total_physical: total_phys,
+            available_physical: avail_phys,
+            total_virtual: total_phys + swap_total,
+            available_virtual: avail_phys + swap_free,
+        })
+    }
+
+    fn parse_meminfo_kb(line: &str, prefix: &str) -> Option<u64> {
+        let rest = line.strip_prefix(prefix)?;
+        let kb: u64 = rest.trim().trim_end_matches("kB").trim().parse().ok()?;
+        Some(kb * 1024)
+    }
+}
+
+#[cfg(target_os = "macos")]
+mod sys {
+    use super::SystemMemoryInfo;
+    use std::io;
+
+    pub fn system_memory_info() -> io::Result<SystemMemoryInfo> {
+        let total_physical = sysctl_u64("hw.memsize")?;
+
+        let page_size = sysctl_u64("hw.pagesize").unwrap_or(4096);
+        let vm_stats = read_vm_stat()?;
+
+        let free_pages = vm_stats.free + vm_stats.inactive + vm_stats.purgeable;
+        let available_physical = free_pages * page_size;
+
+        let swap = read_swap_usage();
+        let swap_total = swap.0;
+        let swap_free = swap_total.saturating_sub(swap.1);
+
+        Ok(SystemMemoryInfo {
+            total_physical,
+            available_physical,
+            total_virtual: total_physical + swap_total,
+            available_virtual: available_physical + swap_free,
+        })
+    }
+
+    fn sysctl_u64(name: &str) -> io::Result<u64> {
+        let output = std::process::Command::new("sysctl")
+            .arg("-n")
+            .arg(name)
+            .output()?;
+        if !output.status.success() {
+            return Err(io::Error::new(
+                io::ErrorKind::Other,
+                format!("sysctl {name} failed"),
+            ));
+        }
+        String::from_utf8_lossy(&output.stdout)
+            .trim()
+            .parse()
+            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))
+    }
+
+    struct VmPages {
+        free: u64,
+        inactive: u64,
+        purgeable: u64,
+    }
+
+    fn read_vm_stat() -> io::Result<VmPages> {
+        let output = std::process::Command::new("vm_stat").output()?;
+        let text = String::from_utf8_lossy(&output.stdout);
+
+        let mut free = 0u64;
+        let mut inactive = 0u64;
+        let mut purgeable = 0u64;
+
+        for line in text.lines() {
+            if let Some(v) = parse_vm_stat_line(line, "Pages free") {
+                free = v;
+            } else if let Some(v) = parse_vm_stat_line(line, "Pages inactive") {
+                inactive = v;
+            } else if let Some(v) = parse_vm_stat_line(line, "Pages purgeable") {
+                purgeable = v;
+            }
+        }
+
+        Ok(VmPages {
+            free,
+            inactive,
+            purgeable,
+        })
+    }
+
+    fn parse_vm_stat_line(line: &str, key: &str) -> Option<u64> {
+        if !line.contains(key) {
+            return None;
+        }
+        let val_str = line.rsplit(':').next()?.trim().trim_end_matches('.');
+        val_str.parse().ok()
+    }
+
+    fn read_swap_usage() -> (u64, u64) {
+        let output = match std::process::Command::new("sysctl")
+            .arg("-n")
+            .arg("vm.swapusage")
+            .output()
+        {
+            Ok(o) => o,
+            Err(_) => return (0, 0),
+        };
+        let text = String::from_utf8_lossy(&output.stdout);
+        let mut total = 0u64;
+        let mut used = 0u64;
+        for part in text.split_whitespace() {
+            if let Some(mb_str) = part.strip_suffix("M") {
+                if let Ok(mb) = mb_str.parse::<f64>() {
+                    if total == 0 {
+                        total = (mb * 1024.0 * 1024.0) as u64;
+                    } else if used == 0 {
+                        used = (mb * 1024.0 * 1024.0) as u64;
+                    }
+                }
+            }
+        }
+        (total, used)
+    }
+}
+
+#[cfg(target_os = "windows")]
+mod sys {
+    use super::SystemMemoryInfo;
+    use std::io;
+
+    #[repr(C)]
+    struct MemoryStatusEx {
+        dw_length: u32,
+        dw_memory_load: u32,
+        ull_total_phys: u64,
+        ull_avail_phys: u64,
+        ull_total_page_file: u64,
+        ull_avail_page_file: u64,
+        ull_total_virtual: u64,
+        ull_avail_virtual: u64,
+        ull_avail_extended_virtual: u64,
+    }
+
+    extern "system" {
+        fn GlobalMemoryStatusEx(lpBuffer: *mut MemoryStatusEx) -> i32;
+    }
+
+    pub fn system_memory_info() -> io::Result<SystemMemoryInfo> {
+        unsafe {
+            let mut status = std::mem::zeroed::<MemoryStatusEx>();
+            status.dw_length = std::mem::size_of::<MemoryStatusEx>() as u32;
+            if GlobalMemoryStatusEx(&mut status) == 0 {
+                return Err(io::Error::last_os_error());
+            }
+            Ok(SystemMemoryInfo {
+                total_physical: status.ull_total_phys,
+                available_physical: status.ull_avail_phys,
+                total_virtual: status.ull_total_virtual,
+                available_virtual: status.ull_avail_virtual,
+            })
+        }
+    }
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
+mod sys {
+    use super::SystemMemoryInfo;
+    use std::io;
+
+    pub fn system_memory_info() -> io::Result<SystemMemoryInfo> {
+        Err(io::Error::new(
+            io::ErrorKind::Unsupported,
+            "memory detection not supported on this platform",
+        ))
+    }
+}

src/coordinator/execution/executor.rs

@@ -30,7 +30,7 @@ use crate::coordinator::plan::{
 use crate::coordinator::statement::{ConfigSource, FunctionSource};
 use crate::runtime::streaming::job::JobManager;
 use crate::runtime::streaming::protocol::control::StopMode;
-use crate::runtime::taskexecutor::TaskManager;
+use crate::runtime::wasm::taskexecutor::TaskManager;
 use crate::sql::schema::show_create_catalog_table;
 use crate::sql::schema::table::Table as CatalogTable;
 use crate::storage::stream_catalog::CatalogManager;

src/coordinator/runtime_context.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 use anyhow::Result;
 
 use crate::runtime::streaming::job::JobManager;
-use crate::runtime::taskexecutor::TaskManager;
+use crate::runtime::wasm::taskexecutor::TaskManager;
 use crate::sql::schema::StreamSchemaProvider;
 use crate::storage::stream_catalog::CatalogManager;
 

src/runtime/memory/block.rs

@@ -0,0 +1,75 @@
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+//
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use super::pool::MemoryPool;
+use super::ticket::MemoryTicket;
+
+#[derive(Debug)]
+pub struct MemoryBlock {
+    capacity: u64,
+    available_bytes: AtomicU64,
+    pool: Arc<MemoryPool>,
+}
+
+impl MemoryBlock {
+    pub(crate) fn new(capacity: u64, pool: Arc<MemoryPool>) -> Arc<Self> {
+        Arc::new(Self {
+            capacity,
+            available_bytes: AtomicU64::new(capacity),
+            pool,
+        })
+    }
+
+    pub fn try_allocate(self: &Arc<Self>, bytes: u64) -> Option<MemoryTicket> {
+        if bytes == 0 {
+            return Some(MemoryTicket::new(0, self.clone()));
+        }
+
+        let mut current_available = self.available_bytes.load(Ordering::Acquire);
+        loop {
+            if current_available < bytes {
+                return None;
+            }
+
+            match self.available_bytes.compare_exchange_weak(
+                current_available,
+                current_available - bytes,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            ) {
+                Ok(_) => return Some(MemoryTicket::new(bytes, self.clone())),
+                Err(actual) => current_available = actual,
+            }
+        }
+    }
+
+    #[inline]
+    pub fn available_bytes(&self) -> u64 {
+        self.available_bytes.load(Ordering::Relaxed)
+    }
+
+    pub(crate) fn release_ticket(&self, bytes: u64) {
+        if bytes > 0 {
+            self.available_bytes.fetch_add(bytes, Ordering::Release);
+        }
+    }
+}
+
+impl Drop for MemoryBlock {
+    fn drop(&mut self) {
+        self.pool.release_block(self.capacity);
+    }
+}