diff --git a/Cargo.lock b/Cargo.lock index 0a07a71..96c9115 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2325,6 +2325,7 @@ dependencies = [ "prost 0.11.9", "prost-build", "qp2p", + "rand 0.8.5", "regex", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 53317aa..038da4f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,8 @@ wasmedge-sdk = { version = "0.10.1" } async-channel = "2.1.0" sysinfo = "0.29.10" ssh2 = "0.9.4" +rand = "0.8.5" +numeric = "0.1.4" # slog-envlogger = { version = "2.1.0", optional = true } diff --git a/src/schedule/master.rs b/src/schedule/master.rs index 94677cd..30e9574 100644 --- a/src/schedule/master.rs +++ b/src/schedule/master.rs @@ -1,11 +1,14 @@ use std::{collections::hash_map::DefaultHasher, hash::Hasher}; +use async_raft::NodeId; use async_trait::async_trait; use axum::{ http::StatusCode, response::{IntoResponse, Redirect, Response}, }; +use bincode::Options; use ws_derive::LogicalModule; +// use use super::{executor::Executor, http_handler::RequestHandler}; use crate::{ @@ -24,6 +27,41 @@ pub struct ScheMaster { // view: ScheMasterView, } +trait NodeWeighteFetcher { + // NOTE: get weight return node weight + // larger is better + fn get_node_weight(id:NodeId) -> f64; +} + +struct StrawNodeSelector { + weight_fetcher:NodeWeighteFetcher, + rng:rand::Rng +} + +impl StrawNodeSelector { + fn new(weight_fetcher: NodeWeighteFetcher) -> Self { Self { weight_fetcher,rng:rand::thread_rng() } } +} + +// NOTE: Straw2 algorithm +impl NodeSelector for StrawNodeSelector { + fn select_node(&self, all_node_cnt: usize, fn_name: &str) -> NodeID { + // NOTE: node id is [1,all_node_cnt] + let mut maxStraw:f64 = -1; + let mut nodeId:NodeID = 1; + let range = Uniform::new(0, 65536); + for i in 1..all_node_cnt+1 { + let weight = self.weight_fetcher.get_node_weight(i); + let mut straw:f64 = range.sample(&mut self.rng); + straw = (straw / 65536).ln() / weight; + if maxStraw == -1 || maxStraw < straw { + maxStraw = straw; + nodeId = i; + } + } + return nodeId; + } +} + trait NodeSelector: Send + Sync + 'static { fn select_node(&self, all_node_cnt: usize, fn_name: &str) -> NodeID; }