pub fn with_mode(mode: env::DeploymentMode) -> Result<Arc<Self>> {
        let next_rdd_id = Arc::new(AtomicUsize::new(0));
        let next_shuffle_id = Arc::new(AtomicUsize::new(0));
        use Schedulers::*;
        match mode {
            env::DeploymentMode::Distributed => {
                let mut port: u16 = 10000;
                let mut address_map: Vec<(String, u16)> = Vec::new();
                if env::Configuration::get().is_master {
                    let uuid = Uuid::new_v4().to_string();
                    initialize_loggers(format!("/tmp/master-{}", uuid));
                    let binary_path =
                        std::env::current_exe().map_err(|_| Error::CurrentBinaryPath)?;
                    let binary_path_str = binary_path
                        .to_str()
                        .ok_or_else(|| Error::PathToString(binary_path.clone()))?
                        .into();
                    let binary_name = binary_path
                        .file_name()
                        .ok_or(Error::CurrentBinaryName)?
                        .to_os_string()
                        .into_string()
                        .map_err(Error::OsStringToString)?;
                    for address in &hosts::Hosts::get()?.slaves {
                        info!("deploying executor at address {:?}", address);
                        let address_cli = address
                            .split('@')
                            .nth(1)
                            .ok_or_else(|| Error::ParseSlaveAddress(address.into()))?
                            .to_string();
                        address_map.push((address_cli, port));
                        let local_dir_root = "/tmp";
                        let uuid = Uuid::new_v4();
                        let local_dir_uuid = uuid.to_string();
                        let local_dir =
                            format!("{}/spark-binary-{}", local_dir_root, local_dir_uuid);
                        let mkdir_output = Command::new("ssh")
                            .args(&[address, "mkdir", &local_dir.clone()])
                            .output()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "ssh mkdir".into(),
                            })?;
                        let remote_path = format!("{}:{}/{}", address, local_dir, binary_name);
                        let scp_output = Command::new("scp")
                            .args(&[&binary_path_str, &remote_path])
                            .output()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "scp executor".into(),
                            })?;
                        let path = format!("{}/{}", local_dir, binary_name);
                        info!("remote path {}", path);
                        // Deploy a remote slave
                        Command::new("ssh")
                            .args(&[
                                address,
                                &path,
                                &env::SLAVE_DEPLOY_CMD.to_string(),
                                &format!("--port={}", port),
                            ])
                            .spawn()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "ssh run".into(),
                            })?;
                        port += 5000;
                    }
                    Ok(Arc::new(Context {
                        next_rdd_id,
                        next_shuffle_id,
                        scheduler: Distributed(DistributedScheduler::new(
                            4,
                            20,
                            true,
                            Some(address_map.clone()),
                            10000,
                        )),
                        address_map,
                        distributed_master: true,
                    }))
                } else {
                    let uuid = Uuid::new_v4().to_string();
                    initialize_loggers(format!("/tmp/executor-{}", uuid));
                    info!("started client");
                    let executor = Executor::new(
                        env::Configuration::get()
                            .port
                            .ok_or(Error::GetOrCreateConfig("executor port not set"))?,
                    );
                    executor.worker();
                    info!("initiated executor worker exit");
                    executor.exit_signal();
                    info!("got executor end signal");
                    std::process::exit(0);
                }
            }
            env::DeploymentMode::Local => {
                let uuid = Uuid::new_v4().to_string();
                initialize_loggers(format!("/tmp/master-{}", uuid));
                let scheduler = Local(LocalScheduler::new(num_cpus::get(), 20, true));
                Ok(Arc::new(Context {
                    next_rdd_id,
                    next_shuffle_id,
                    scheduler,
                    address_map: Vec::new(),
                    distributed_master: false,
                }))
            }
        }
    }
// Sends the binary to all nodes present in hosts.conf and starts them
Sends the binary to all nodes present in hosts.conf and starts them
Sends the binary to all nodes present in hosts . conf and starts them
pub fn with_mode(mode: env::DeploymentMode) -> Result<Arc<Self>> {
        let next_rdd_id = Arc::new(AtomicUsize::new(0));
        let next_shuffle_id = Arc::new(AtomicUsize::new(0));
        use Schedulers::*;
        match mode {
            env::DeploymentMode::Distributed => {
                let mut port: u16 = 10000;
                let mut address_map: Vec<(String, u16)> = Vec::new();
                if env::Configuration::get().is_master {
                    let uuid = Uuid::new_v4().to_string();
                    initialize_loggers(format!("/tmp/master-{}", uuid));
                    let binary_path =
                        std::env::current_exe().map_err(|_| Error::CurrentBinaryPath)?;
                    let binary_path_str = binary_path
                        .to_str()
                        .ok_or_else(|| Error::PathToString(binary_path.clone()))?
                        .into();
                    let binary_name = binary_path
                        .file_name()
                        .ok_or(Error::CurrentBinaryName)?
                        .to_os_string()
                        .into_string()
                        .map_err(Error::OsStringToString)?;
                    for address in &hosts::Hosts::get()?.slaves {
                        info!("deploying executor at address {:?}", address);
                        let address_cli = address
                            .split('@')
                            .nth(1)
                            .ok_or_else(|| Error::ParseSlaveAddress(address.into()))?
                            .to_string();
                        address_map.push((address_cli, port));
                        let local_dir_root = "/tmp";
                        let uuid = Uuid::new_v4();
                        let local_dir_uuid = uuid.to_string();
                        let local_dir =
                            format!("{}/spark-binary-{}", local_dir_root, local_dir_uuid);
                        let mkdir_output = Command::new("ssh")
                            .args(&[address, "mkdir", &local_dir.clone()])
                            .output()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "ssh mkdir".into(),
                            })?;
                        let remote_path = format!("{}:{}/{}", address, local_dir, binary_name);
                        let scp_output = Command::new("scp")
                            .args(&[&binary_path_str, &remote_path])
                            .output()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "scp executor".into(),
                            })?;
                        let path = format!("{}/{}", local_dir, binary_name);
                        info!("remote path {}", path);
                        // Deploy a remote slave
                        Command::new("ssh")
                            .args(&[
                                address,
                                &path,
                                &env::SLAVE_DEPLOY_CMD.to_string(),
                                &format!("--port={}", port),
                            ])
                            .spawn()
                            .map_err(|e| Error::CommandOutput {
                                source: e,
                                command: "ssh run".into(),
                            })?;
                        port += 5000;
                    }
                    Ok(Arc::new(Context {
                        next_rdd_id,
                        next_shuffle_id,
                        scheduler: Distributed(DistributedScheduler::new(
                            4,
                            20,
                            true,
                            Some(address_map.clone()),
                            10000,
                        )),
                        address_map,
                        distributed_master: true,
                    }))
                } else {
                    let uuid = Uuid::new_v4().to_string();
                    initialize_loggers(format!("/tmp/executor-{}", uuid));
                    info!("started client");
                    let executor = Executor::new(
                        env::Configuration::get()
                            .port
                            .ok_or(Error::GetOrCreateConfig("executor port not set"))?,
                    );
                    executor.worker();
                    info!("initiated executor worker exit");
                    executor.exit_signal();
                    info!("got executor end signal");
                    std::process::exit(0);
                }
            }
            env::DeploymentMode::Local => {
                let uuid = Uuid::new_v4().to_string();
                initialize_loggers(format!("/tmp/master-{}", uuid));
                let scheduler = Local(LocalScheduler::new(num_cpus::get(), 20, true));
                Ok(Arc::new(Context {
                    next_rdd_id,
                    next_shuffle_id,
                    scheduler,
                    address_map: Vec::new(),
                    distributed_master: false,
                }))
            }
        }
    }
pub fn with_mode ( mode : env :: DeploymentMode ) -> Result < Arc < Self > > { let next_rdd_id = Arc :: new ( AtomicUsize :: new ( 0 ) ) ; let next_shuffle_id = Arc :: new ( AtomicUsize :: new ( 0 ) ) ; use Schedulers :: * ; match mode { env :: DeploymentMode :: Distributed => { let mut port : u16 = 10000 ; let mut address_map : Vec < ( String , u16 ) > = Vec :: new ( ) ; if env :: Configuration :: get ( ) . is_master { let uuid = Uuid :: new_v4 ( ) . to_string ( ) ; initialize_loggers ( format ! ( " " uuid ) ) ; let binary_path = std :: env :: current_exe ( ) . map_err ( | _ | Error :: CurrentBinaryPath ) ? ; let binary_path_str = binary_path . to_str ( ) . ok_or_else ( | | Error :: PathToString ( binary_path . clone ( ) ) ) ? . into ( ) ; let binary_name = binary_path . file_name ( ) . ok_or ( Error :: CurrentBinaryName ) ? . to_os_string ( ) . into_string ( ) . map_err ( Error :: OsStringToString ) ? ; for address in & hosts :: Hosts :: get ( ) ? . slaves { info ! ( " " address ) ; let address_cli = address . split ( '@' ) . nth ( 1 ) . ok_or_else ( | | Error :: ParseSlaveAddress ( address . into ( ) ) ) ? . to_string ( ) ; address_map . push ( ( address_cli , port ) ) ; let local_dir_root = " " ; let uuid = Uuid :: new_v4 ( ) ; let local_dir_uuid = uuid . to_string ( ) ; let local_dir = format ! ( " " local_dir_root local_dir_uuid ) ; let mkdir_output = Command :: new ( " " ) . args ( & [ address , " " , & local_dir . clone ( ) ] ) . output ( ) . map_err ( | e | Error :: CommandOutput { source : e , command : " " . into ( ) , } ) ? ; let remote_path = format ! ( " " address local_dir binary_name ) ; let scp_output = Command :: new ( " " ) . args ( & [ & binary_path_str , & remote_path ] ) . output ( ) . map_err ( | e | Error :: CommandOutput { source : e , command : " " . into ( ) , } ) ? ; let path = format ! ( " " local_dir binary_name ) ; info ! ( " " path ) ; // Deploy a remote slave Command :: new ( " " ) . args ( & [ address , & path , & env :: SLAVE_DEPLOY_CMD . to_string ( ) , & format ! ( " " port ) , ] ) . spawn ( ) . map_err ( | e | Error :: CommandOutput { source : e , command : " " . into ( ) , } ) ? ; port += 5000 ; } Ok ( Arc :: new ( Context { next_rdd_id , next_shuffle_id , scheduler : Distributed ( DistributedScheduler :: new ( 4 , 20 , true , Some ( address_map . clone ( ) ) , 10000 , ) ) , address_map , distributed_master : true , } ) ) } else { let uuid = Uuid :: new_v4 ( ) . to_string ( ) ; initialize_loggers ( format ! ( " " uuid ) ) ; info ! ( " " ) ; let executor = Executor :: new ( env :: Configuration :: get ( ) . port . ok_or ( Error :: GetOrCreateConfig ( " " ) ) ? , ) ; executor . worker ( ) ; info ! ( " " ) ; executor . exit_signal ( ) ; info ! ( " " ) ; std :: process :: exit ( 0 ) ; } } env :: DeploymentMode :: Local => { let uuid = Uuid :: new_v4 ( ) . to_string ( ) ; initialize_loggers ( format ! ( " " uuid ) ) ; let scheduler = Local ( LocalScheduler :: new ( num_cpus :: get ( ) , 20 , true ) ) ; Ok ( Arc :: new ( Context { next_rdd_id , next_shuffle_id , scheduler , address_map : Vec :: new ( ) , distributed_master : false , } ) ) } } }
Sends the binary to all nodes present in hosts.conf and starts them
Sends the binary to all nodes present in hosts . conf and starts them
mode
env::DeploymentMode