chore: perf of flamepy.

Cargo.toml

@@ -69,3 +69,10 @@ gethostname = "1"
 arrow = "58"
 arrow-flight = "58"
 arrow-ipc = { version = "58", features = ["zstd"] }
+
+# profiling and HTTP
+axum = "0.8"
+pprof = { version = "0.14", features = ["flamegraph", "prost-codec"] }
+
+# parallelism
+rayon = "1.10"

common/Cargo.toml

@@ -33,6 +33,9 @@ rustix = { version = "1.1" , features = ["system"] }
 num_cpus = "1.17"
 bytesize = "1.3"
 
+axum = { workspace = true }
+pprof = { workspace = true }
+
 
 [dev-dependencies]
 tempfile = { workspace = true }

common/src/apis/from_rpc.rs

@@ -76,7 +76,7 @@ impl From<rpc::Event> for Event {
         Self {
             code: event.code,
             message: event.message,
-            creation_time: DateTime::from_timestamp(event.creation_time, 0).unwrap_or_default(),
+            creation_time: DateTime::from_timestamp_millis(event.creation_time).unwrap_or_default(),
         }
     }
 }
@@ -189,7 +189,7 @@ impl TryFrom<&rpc::Application> for Application {
             name: metadata.name.clone(),
             version: 0,
             state: ApplicationState::from(status.state()),
-            creation_time: DateTime::<Utc>::from_timestamp(status.creation_time, 0).ok_or(
+            creation_time: DateTime::<Utc>::from_timestamp_millis(status.creation_time).ok_or(
                 FlameError::InvalidState("invalid creation time".to_string()),
             )?,
             shim: Shim::from(spec.shim()),

common/src/apis/session.rs

@@ -11,8 +11,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-use std::collections::HashMap;
-
 use stdng::lock_ptr;
 
 use super::types::*;
@@ -123,39 +121,3 @@ impl Session {
         Ok(())
     }
 }
-
-impl Clone for Session {
-    fn clone(&self) -> Self {
-        let mut ssn = Session {
-            id: self.id.clone(),
-            application: self.application.clone(),
-            slots: self.slots,
-            version: self.version,
-            common_data: self.common_data.clone(),
-            tasks: HashMap::new(),
-            tasks_index: HashMap::new(),
-            creation_time: self.creation_time,
-            completion_time: self.completion_time,
-            events: self.events.clone(),
-            status: self.status.clone(),
-            min_instances: self.min_instances,
-            max_instances: self.max_instances,
-            batch_size: self.batch_size,
-        };
-
-        for (id, t) in &self.tasks {
-            match t.lock() {
-                Ok(t) => {
-                    if let Err(e) = ssn.update_task(&t) {
-                        tracing::error!("Failed to update task: <{id}> for session: <{ssn_id}>, ignore it during clone: {e}", ssn_id = self.id);
-                    }
-                }
-                Err(_) => {
-                    tracing::error!("Failed to lock task: <{id}> for session: <{ssn_id}>, ignore it during clone.", ssn_id = self.id);
-                }
-            }
-        }
-
-        ssn
-    }
-}

common/src/apis/to_rpc.rs

@@ -89,7 +89,7 @@ impl From<Event> for rpc::Event {
         Self {
             code: event.code,
             message: event.message,
-            creation_time: event.creation_time.timestamp(),
+            creation_time: event.creation_time.timestamp_millis(),
         }
     }
 }
@@ -147,8 +147,8 @@ impl From<&Task> for rpc::Task {
         });
         let status = Some(rpc::TaskStatus {
             state: task.state as i32,
-            creation_time: task.creation_time.timestamp(),
-            completion_time: task.completion_time.map(|s| s.timestamp()),
+            creation_time: task.creation_time.timestamp_millis(),
+            completion_time: task.completion_time.map(|s| s.timestamp_millis()),
             events: task.events.clone().into_iter().map(Event::into).collect(),
         });
         rpc::Task {
@@ -169,8 +169,8 @@ impl From<&Session> for rpc::Session {
     fn from(ssn: &Session) -> Self {
         let mut status = rpc::SessionStatus {
             state: ssn.status.state as i32,
-            creation_time: ssn.creation_time.timestamp(),
-            completion_time: ssn.completion_time.map(|s| s.timestamp()),
+            creation_time: ssn.creation_time.timestamp_millis(),
+            completion_time: ssn.completion_time.map(|s| s.timestamp_millis()),
             failed: 0,
             pending: 0,
             running: 0,
@@ -250,7 +250,7 @@ impl From<&Application> for rpc::Application {
 
         let status = Some(rpc::ApplicationStatus {
             state: app.state.into(),
-            creation_time: app.creation_time.timestamp(),
+            creation_time: app.creation_time.timestamp_millis(),
         });
         rpc::Application {
             metadata,

common/src/apis/types.rs

@@ -169,7 +169,7 @@ pub struct SessionStatus {
     pub state: SessionState,
 }
 
-#[derive(Debug, Default)]
+#[derive(Debug, Default, Clone)]
 pub struct Session {
     pub id: SessionID,
     pub application: String,
@@ -203,7 +203,7 @@ impl TaskState {
     }
 }
 
-#[derive(Clone, Debug, Default, PartialEq, Eq, Hash)]
+#[derive(Clone, Debug, Default, Eq, PartialEq, Hash)]
 pub struct TaskGID {
     pub ssn_id: SessionID,
     pub task_id: TaskID,

common/src/ctx.rs

@@ -30,7 +30,7 @@ const DEFAULT_SLOT: &str = "cpu=1,mem=2g";
 const DEFAULT_POLICY: &str = "proportion";
 const DEFAULT_STORAGE: &str = "sqlite://flame.db";
 const DEFAULT_MAX_EXECUTORS_PER_NODE: u32 = 128;
-const DEFAULT_SCHEDULE_INTERVAL: u64 = 500;
+const DEFAULT_SCHEDULE_INTERVAL: u64 = 100;
 const DEFAULT_SHIM: &str = "host";
 const DEFAULT_FLAME_CACHE_ENDPOINT: &str = "http://127.0.0.1:9090";
 const DEFAULT_FLAME_CACHE_NETWORK_INTERFACE: &str = "eth0";
@@ -62,6 +62,8 @@ struct FlameClusterYaml {
     pub tls: Option<FlameTlsYaml>,
     /// Resource limits configuration
     pub limits: Option<FlameLimitsYaml>,
+    /// pprof profiling configuration
+    pub pprof: Option<FlamePprofYaml>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -111,6 +113,11 @@ struct FlameEvictionYaml {
     pub max_objects: Option<usize>,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct FlamePprofYaml {
+    pub port: Option<u16>,
+}
+
 // ============================================================
 // Runtime structs (validated, with helper methods)
 // ============================================================
@@ -128,14 +135,11 @@ pub struct FlameCluster {
     pub slot: ResourceRequirement,
     pub policy: String,
     pub storage: String,
-    /// Schedule interval in milliseconds for the session scheduler loop
     pub schedule_interval: u64,
-    /// Executors configuration
     pub executors: FlameExecutors,
-    /// TLS configuration for Session Manager
     pub tls: Option<FlameTls>,
-    /// Resource limits configuration
     pub limits: FlameLimits,
+    pub pprof: Option<FlamePprof>,
 }
 
 #[derive(Debug, Clone, Default)]
@@ -149,6 +153,13 @@ pub struct FlameLimits {
     pub max_executors: u32,
 }
 
+#[derive(Debug, Clone)]
+pub struct FlamePprof {
+    pub port: u16,
+}
+
+const DEFAULT_PPROF_PORT: u16 = 6060;
+
 /// TLS configuration for Flame services.
 ///
 /// When this struct is present and valid (cert_file + key_file configured),
@@ -366,6 +377,8 @@ impl TryFrom<FlameClusterYaml> for FlameCluster {
 
         let limits = cluster.limits.map(FlameLimits::from).unwrap_or_default();
 
+        let pprof = cluster.pprof.map(FlamePprof::from);
+
         Ok(FlameCluster {
             name: cluster.name,
             endpoint: cluster.endpoint,
@@ -378,6 +391,7 @@ impl TryFrom<FlameClusterYaml> for FlameCluster {
             executors,
             tls,
             limits,
+            pprof,
         })
     }
 }
@@ -409,6 +423,14 @@ impl Default for FlameLimits {
     }
 }
 
+impl From<FlamePprofYaml> for FlamePprof {
+    fn from(yaml: FlamePprofYaml) -> Self {
+        FlamePprof {
+            port: yaml.port.unwrap_or(DEFAULT_PPROF_PORT),
+        }
+    }
+}
+
 impl Default for FlameCluster {
     fn default() -> Self {
         FlameCluster {
@@ -421,6 +443,7 @@ impl Default for FlameCluster {
             executors: FlameExecutors::default(),
             tls: None,
             limits: FlameLimits::default(),
+            pprof: None,
         }
     }
 }

common/src/lib.rs

@@ -13,6 +13,7 @@ limitations under the License.
 
 pub mod apis;
 pub mod ctx;
+pub mod pprof;
 pub mod storage;
 
 use std::string::FromUtf8Error;