Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

refactor(hydroflow_plus)!: defer network instantiation until after finalizing IR #1377

Merged
merged 1 commit into from
Aug 14, 2024
Merged

refactor(hydroflow_plus)!: defer network instantiation until after finalizing IR #1377

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 11 additions & 15 deletions docs/docs/hydroflow_plus/quickstart/clusters.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,28 +89,26 @@ use hydroflow_plus_cli_integration::{DeployProcessSpec, DeployClusterSpec};

#[tokio::main]
async fn main() {
let deployment = RefCell::new(Deployment::new());
let localhost = deployment.borrow_mut().Localhost();
let mut deployment = Deployment::new();
let localhost = deployment.Localhost();
let profile = "dev";

let builder = hydroflow_plus::FlowBuilder::new();
let localhost_1 = localhost.clone();
let localhost_2 = localhost.clone();
flow::broadcast::broadcast(
&builder,
&DeployProcessSpec::new(|| {
let mut deployment = deployment.borrow_mut();
deployment.add_service(
HydroflowCrate::new(".", localhost.clone())
.bin("broadcast")
.profile(profile)
.display_name("leader"),
)
&DeployProcessSpec::new(move |deployment| {
HydroflowCrate::new(".", localhost_1.clone())
.bin("broadcast")
.profile(profile)
.display_name("leader"),
}),
&DeployClusterSpec::new(|| {
let mut deployment = deployment.borrow_mut();
&DeployClusterSpec::new(move |deployment| {
(0..2)
.map(|idx| {
deployment.add_service(
HydroflowCrate::new(".", localhost.clone())
HydroflowCrate::new(".", localhost_2.clone())
.bin("broadcast")
.profile(profile)
.display_name(format!("worker/{}", idx)),
Expand All @@ -120,8 +118,6 @@ async fn main() {
}),
);

let mut deployment = deployment.into_inner();

deployment.deploy().await.unwrap();

deployment.start().await.unwrap();
Expand Down
134 changes: 89 additions & 45 deletions hydro_deploy/hydroflow_plus_cli_integration/src/deploy.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ use hydro_deploy::hydroflow_crate::ports::{
};
use hydro_deploy::hydroflow_crate::HydroflowCrateService;
use hydro_deploy::{Deployment, Host};
use hydroflow_plus::lang::graph::HydroflowGraph;
use hydroflow_plus::location::{
Cluster, ClusterSpec, Deploy, HfSendManyToMany, HfSendManyToOne, HfSendOneToMany,
HfSendOneToOne, Location, ProcessSpec,
Expand All @@ -23,6 +24,7 @@ pub struct HydroDeploy {}

impl<'a> Deploy<'a> for HydroDeploy {
type ClusterId = u32;
type InstantiateEnv = Deployment;
type Process = DeployNode;
type Cluster = DeployCluster;
type Meta = HashMap<usize, Vec<u32>>;
Expand Down Expand Up @@ -68,12 +70,13 @@ pub trait DeployCrateWrapper {
pub struct DeployNode {
id: usize,
next_port: Rc<RefCell<usize>>,
underlying: Arc<RwLock<HydroflowCrateService>>,
node_fn: Rc<CrateBuilder<'static>>,
underlying: Rc<RefCell<Option<Arc<RwLock<HydroflowCrateService>>>>>,
}

impl DeployCrateWrapper for DeployNode {
fn underlying(&self) -> Arc<RwLock<HydroflowCrateService>> {
self.underlying.clone()
self.underlying.borrow().as_ref().unwrap().clone()
}
}

Expand All @@ -99,7 +102,7 @@ impl DeployPort<DeployCluster> {
on: &Arc<impl Host + 'static>,
) -> Vec<CustomClientPort> {
let mut out = vec![];
for member in &self.node.members {
for member in self.node.members() {
out.push(member.create_sender(&self.port, deployment, on).await);
}

Expand All @@ -110,6 +113,7 @@ impl DeployPort<DeployCluster> {
impl Location for DeployNode {
type Port = DeployPort<Self>;
type Meta = HashMap<usize, Vec<u32>>;
type InstantiateEnv = Deployment;

fn id(&self) -> usize {
self.id
Expand All @@ -126,13 +130,23 @@ impl Location for DeployNode {
}

fn update_meta(&mut self, meta: &Self::Meta) {
let mut n = self.underlying.try_write().unwrap();
let underlying_node = self.underlying.borrow();
let mut n = underlying_node.as_ref().unwrap().try_write().unwrap();
n.update_meta(HydroflowPlusMeta {
clusters: meta.clone(),
cluster_id: None,
subgraph_id: self.id,
});
}

fn instantiate(
&self,
env: &mut Self::InstantiateEnv,
_meta: &mut Self::Meta,
_graph: HydroflowGraph,
) {
*self.underlying.borrow_mut() = Some((self.node_fn.as_ref())(env));
}
}

#[derive(Clone)]
Expand All @@ -146,16 +160,26 @@ impl DeployCrateWrapper for DeployClusterNode {
}
}

type CreateClusterFn = dyn Fn(&mut Deployment) -> Vec<Arc<RwLock<HydroflowCrateService>>> + 'static;

#[derive(Clone)]
pub struct DeployCluster {
id: usize,
next_port: Rc<RefCell<usize>>,
pub members: Vec<DeployClusterNode>,
cluster_fn: Rc<RefCell<Box<CreateClusterFn>>>,
members: Rc<RefCell<Vec<DeployClusterNode>>>,
}

impl DeployCluster {
pub fn members(&self) -> Vec<DeployClusterNode> {
self.members.borrow().clone()
}
}

impl Location for DeployCluster {
type Port = DeployPort<Self>;
type Meta = HashMap<usize, Vec<u32>>;
type InstantiateEnv = Deployment;

fn id(&self) -> usize {
self.id
Expand All @@ -171,15 +195,29 @@ impl Location for DeployCluster {
}
}

fn instantiate(
&self,
_env: &mut Self::InstantiateEnv,
meta: &mut Self::Meta,
_graph: HydroflowGraph,
) {
let cluster_nodes = (self.cluster_fn.borrow_mut())(_env);
meta.insert(self.id, (0..(cluster_nodes.len() as u32)).collect());
*self.members.borrow_mut() = cluster_nodes
.into_iter()
.map(|n| DeployClusterNode { underlying: n })
.collect();
}

fn update_meta(&mut self, meta: &Self::Meta) {
self.members.iter().enumerate().for_each(|(cluster_id, n)| {
let mut n = n.underlying.try_write().unwrap();
n.update_meta(&HydroflowPlusMeta {
for (cluster_id, node) in self.members.borrow().iter().enumerate() {
let mut n = node.underlying.try_write().unwrap();
n.update_meta(HydroflowPlusMeta {
clusters: meta.clone(),
cluster_id: Some(cluster_id as u32),
subgraph_id: self.id,
});
});
}
}
}

Expand All @@ -202,17 +240,19 @@ impl HfSendOneToOne<DeployNode> for DeployNode {
source_port: &DeployPort<DeployNode>,
recipient_port: &DeployPort<DeployNode>,
) {
let source_port = self
.underlying
let self_underlying_borrow = self.underlying.borrow();
let self_underlying = self_underlying_borrow.as_ref().unwrap();
let source_port = self_underlying
.try_read()
.unwrap()
.get_port(source_port.port.clone(), &self.underlying);
.get_port(source_port.port.clone(), self_underlying);

let recipient_port = other
.underlying
let other_underlying_borrow = other.underlying.borrow();
let other_underlying = other_underlying_borrow.as_ref().unwrap();
let recipient_port = other_underlying
.try_read()
.unwrap()
.get_port(recipient_port.port.clone(), &other.underlying);
.get_port(recipient_port.port.clone(), other_underlying);

source_port.send_to(&recipient_port);
}
Expand All @@ -233,14 +273,15 @@ impl HfSendManyToOne<DeployNode, u32> for DeployCluster {
source_port: &DeployPort<DeployCluster>,
recipient_port: &DeployPort<DeployNode>,
) {
let recipient_port = other
.underlying
let other_underlying_borrow = other.underlying.borrow();
let other_underlying = other_underlying_borrow.as_ref().unwrap();
let recipient_port = other_underlying
.try_read()
.unwrap()
.get_port(recipient_port.port.clone(), &other.underlying)
.get_port(recipient_port.port.clone(), other_underlying)
.merge();

for (i, node) in self.members.iter().enumerate() {
for (i, node) in self.members.borrow().iter().enumerate() {
let source_port = node
.underlying
.try_read()
Expand Down Expand Up @@ -271,15 +312,17 @@ impl HfSendOneToMany<DeployCluster, u32> for DeployNode {
source_port: &DeployPort<DeployNode>,
recipient_port: &DeployPort<DeployCluster>,
) {
let source_port = self
.underlying
let self_underlying_borrow = self.underlying.borrow();
let self_underlying = self_underlying_borrow.as_ref().unwrap();
let source_port = self_underlying
.try_read()
.unwrap()
.get_port(source_port.port.clone(), &self.underlying);
.get_port(source_port.port.clone(), self_underlying);

let recipient_port = DemuxSink {
demux: other
.members
.borrow()
.iter()
.enumerate()
.map(|(id, c)| {
Expand Down Expand Up @@ -315,7 +358,7 @@ impl HfSendManyToMany<DeployCluster, u32> for DeployCluster {
source_port: &DeployPort<DeployCluster>,
recipient_port: &DeployPort<DeployCluster>,
) {
for (i, sender) in self.members.iter().enumerate() {
for (i, sender) in self.members.borrow().iter().enumerate() {
let source_port = sender
.underlying
.try_read()
Expand All @@ -325,6 +368,7 @@ impl HfSendManyToMany<DeployCluster, u32> for DeployCluster {
let recipient_port = DemuxSink {
demux: other
.members
.borrow()
.iter()
.enumerate()
.map(|(id, c)| {
Expand Down Expand Up @@ -360,48 +404,48 @@ impl HfSendManyToMany<DeployCluster, u32> for DeployCluster {
}
}

type CrateBuilder<'a> = dyn FnMut() -> Arc<RwLock<HydroflowCrateService>> + 'a;
type CrateBuilder<'a> = dyn Fn(&mut Deployment) -> Arc<RwLock<HydroflowCrateService>> + 'a;

pub struct DeployProcessSpec<'a>(RefCell<Box<CrateBuilder<'a>>>);
pub struct DeployProcessSpec(Rc<CrateBuilder<'static>>);

impl<'a> DeployProcessSpec<'a> {
pub fn new<F: FnMut() -> Arc<RwLock<HydroflowCrateService>> + 'a>(f: F) -> Self {
Self(RefCell::new(Box::new(f)))
impl DeployProcessSpec {
pub fn new<F: Fn(&mut Deployment) -> Arc<RwLock<HydroflowCrateService>> + 'static>(
f: F,
) -> Self {
Self(Rc::new(f))
}
}

impl<'a: 'b, 'b> ProcessSpec<'a, HydroDeploy> for DeployProcessSpec<'b> {
fn build(&self, id: usize, _meta: &mut HashMap<usize, Vec<u32>>) -> DeployNode {
impl<'a> ProcessSpec<'a, HydroDeploy> for DeployProcessSpec {
fn build(&self, id: usize) -> DeployNode {
DeployNode {
id,
next_port: Rc::new(RefCell::new(0)),
underlying: (self.0.borrow_mut())(),
node_fn: self.0.clone(),
underlying: Rc::new(RefCell::new(None)),
}
}
}

type ClusterSpecFn<'a> = dyn FnMut() -> Vec<Arc<RwLock<HydroflowCrateService>>> + 'a;
type ClusterSpecFn = dyn Fn(&mut Deployment) -> Vec<Arc<RwLock<HydroflowCrateService>>> + 'static;

pub struct DeployClusterSpec<'a>(RefCell<Box<ClusterSpecFn<'a>>>);
pub struct DeployClusterSpec(Rc<RefCell<Box<ClusterSpecFn>>>);

impl<'a> DeployClusterSpec<'a> {
pub fn new<F: FnMut() -> Vec<Arc<RwLock<HydroflowCrateService>>> + 'a>(f: F) -> Self {
Self(RefCell::new(Box::new(f)))
impl DeployClusterSpec {
pub fn new<F: Fn(&mut Deployment) -> Vec<Arc<RwLock<HydroflowCrateService>>> + 'static>(
f: F,
) -> Self {
Self(Rc::new(RefCell::new(Box::new(f))))
}
}

impl<'a: 'b, 'b> ClusterSpec<'a, HydroDeploy> for DeployClusterSpec<'b> {
fn build(&self, id: usize, meta: &mut HashMap<usize, Vec<u32>>) -> DeployCluster {
let cluster_nodes = (self.0.borrow_mut())();
meta.insert(id, (0..(cluster_nodes.len() as u32)).collect());

impl<'a> ClusterSpec<'a, HydroDeploy> for DeployClusterSpec {
fn build(&self, id: usize) -> DeployCluster {
DeployCluster {
id,
next_port: Rc::new(RefCell::new(0)),
members: cluster_nodes
.into_iter()
.map(|u| DeployClusterNode { underlying: u })
.collect(),
cluster_fn: self.0.clone(),
members: Rc::new(RefCell::new(vec![])),
}
}
}
Loading
Loading