update cli.

Author: alexliang

devops/scripts/build-docs.sh

@@ -1,5 +1,6 @@
 #!/usr/bin/env bash
 
+WORKSPACE=./
 wget https://repo.continuum.io/miniconda/Miniconda3-latest-Linux-x86_64.sh -O miniconda.sh
 bash miniconda.sh -b -p $WORKSPACE/miniconda
 hash -r

python/app/fedcv/object_detection/config/bootstrap.bat

@@ -5,8 +5,14 @@
 :: ### please customize your script in this region ####
 pip install opencv-python-headless pandas matplotlib seaborn addict
 set DATA_PATH=%userprofile%\fedcv_data
-if exist %DATA_PATH% (echo Exist %DATA_PATH%) else mkdir %DATA_PATH%
-.\..\data\coco128\download_coco128.bat
+if exist %DATA_PATH% (
+ echo Exist %DATA_PATH%
+) ^
+else (
+ mkdir %DATA_PATH%
+ set cur_dir=%cd%
+ %cur_dir%\..\data\coco128\download_coco128.bat
+)
 
 
 :: ### don't modify this part ###

python/app/fedcv/object_detection/config/bootstrap.sh

@@ -1,5 +1,6 @@
+#!/bin/bash
 ### don't modify this part ###
-set -x
+# set -x
 ##############################
 
 

python/examples/cross_silo/mqtt_s3_fedavg_mnist_lr_example/custom_data_and_model/config/fedml_config.yaml

@@ -40,13 +40,15 @@ device_args:
  gpu_mapping_key: mapping_default
 
 comm_args:
+ #backend: "MQTT_IPFS"
  backend: "MQTT_S3"
  mqtt_config_path: config/mqtt_config.yaml
  s3_config_path: config/s3_config.yaml
  # If you want to use your customized MQTT or s3 server as training backends, you should uncomment and set the following lines.
  #customized_training_mqtt_config: {'BROKER_HOST': 'your mqtt server address or domain name', 'MQTT_PWD': 'your mqtt password', 'BROKER_PORT': 1883, 'MQTT_KEEPALIVE': 180, 'MQTT_USER': 'your mqtt user'}
  #customized_training_s3_config: {'CN_S3_SAK': 'your s3 aws_secret_access_key', 'CN_REGION_NAME': 'your s3 region name', 'CN_S3_AKI': 'your s3 aws_access_key_id', 'BUCKET_NAME': 'your s3 bucket name'}
-
+ #customized_training_ipfs_config: {'token': 'your ipfs token at web3.storage', 'upload_uri': 'https://api.web3.storage/upload', 'download_uri': 'ipfs.w3s.link', }
+ #customized_training_ipfs_config: {'token': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJkaWQ6ZXRocjoweERiNzY5OTdGRUFBRjg3QjY0RDc3MWZBYjg2RUExMTg2ODRBOEI0MTUiLCJpc3MiOiJ3ZWIzLXN0b3JhZ2UiLCJpYXQiOjE2NjM0OTkwNzQ3MzQsIm5hbWUiOiJmZWRtbC13ZWIzIn0.VV8WHJaqe8cbShek8warIGFbS55KOUfpgtLMRevPIIY', 'upload_uri': 'https://api.web3.storage/upload', 'download_uri': 'ipfs.w3s.link', }
 
 tracking_args:
  # the default log path is at ~/fedml-client/fedml/logs/ and ~/fedml-server/fedml/logs/

python/examples/cross_silo/mqtt_s3_fedavg_mnist_lr_example/custom_data_and_model/torch_client.py

@@ -2,7 +2,9 @@
 
 import fedml
 from fedml import FedMLRunner
+import fedml
 from fedml.data.MNIST.data_loader import download_mnist, load_partition_data_mnist
+from fedml.core.alg_frame.context import Context
 
 
 def load_data(args):
@@ -57,6 +59,8 @@ def forward(self, x):
 
 
 if __name__ == "__main__":
+ Context().add("ipfs_secret_key", "12345678123456781234567812345678")
+
  # init FedML framework
  args = fedml.init()
 

python/examples/cross_silo/mqtt_s3_fedavg_mnist_lr_example/custom_data_and_model/torch_server.py

@@ -3,6 +3,7 @@
 import fedml
 from fedml import FedMLRunner
 from fedml.data.MNIST.data_loader import download_mnist, load_partition_data_mnist
+from fedml.core.alg_frame.context import Context
 
 
 def load_data(args):
@@ -57,6 +58,8 @@ def forward(self, x):
 
 
 if __name__ == "__main__":
+ Context().add("ipfs_secret_key", "12345678123456781234567812345678")
+
  # init FedML framework
  args = fedml.init()
 

python/fedml/cli/edge_deployment/client_constants.py

@@ -218,33 +218,33 @@ def exit_process(process):
  pass
 
  @staticmethod
- def exec_console_with_script(script_path, should_capture_stdout_err=False):
- if should_capture_stdout_err:
-  if platform.system() == 'Windows':
- script_process = subprocess.Popen(script_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-  else:
-  script_process = subprocess.Popen(['sh', '-c', script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ def exec_console_with_script(script_path, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ if platform.system() == 'Windows':
+ script_process = subprocess.Popen(script_path, stdout=stdout_flag, stderr=stderr_flag)
  else:
- if platform.system() == 'Windows':
- script_process = subprocess.Popen(script_path, stdout=sys.stdout, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen(['sh', '-c', script_path], stdout=sys.stdout, stderr=subprocess.PIPE)
+ script_process = subprocess.Popen(['sh', '-c', script_path], stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod
- def exec_console_with_shell(shell, script_path, should_capture_stdout_err=False):
- if should_capture_stdout_err:
- script_process = subprocess.Popen([shell, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen([shell, script_path], stdout=sys.stdout, stderr=subprocess.PIPE)
+ def exec_console_with_shell(shell, script_path, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ script_process = subprocess.Popen([shell, script_path], stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod
- def exec_console_with_shell_script_list(shell_script_list, should_capture_stdout_err=False):
- if should_capture_stdout_err:
- script_process = subprocess.Popen(shell_script_list, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen(shell_script_list, stdout=sys.stdout, stderr=sys.stderr)
+ def exec_console_with_shell_script_list(shell_script_list, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ script_process = subprocess.Popen(shell_script_list, stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod

python/fedml/cli/edge_deployment/client_daemon.py

@@ -44,5 +44,5 @@
  ]
  )
  ret_code, exec_out, exec_err = ClientConstants.get_console_sys_out_pipe_err_results(login_pid)
- time.sleep(15)
+ time.sleep(3)
 

python/fedml/cli/edge_deployment/client_runner.py

@@ -253,14 +253,18 @@ def build_dynamic_args(self, run_config, package_conf_object, base_dir):
  if os.path.exists(bootstrap_script_path):
  bootstrap_stat = os.stat(bootstrap_script_path)
  if platform.system() == 'Windows':
- os.chmod(bootstrap_script_path, bootstrap_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+ os.chmod(bootstrap_script_path,
+ bootstrap_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
  bootstrap_scripts = "{}".format(bootstrap_script_path)
  else:
- os.chmod(bootstrap_script_path, bootstrap_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
- bootstrap_scripts = "cd {}; ./{}".format(bootstrap_script_dir, os.path.basename(bootstrap_script_file))
+ os.chmod(bootstrap_script_path,
+ bootstrap_stat.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+ bootstrap_scripts = "cd {}; ./{}".format(bootstrap_script_dir,
+ os.path.basename(bootstrap_script_file))
  bootstrap_scripts = str(bootstrap_scripts).replace('\\', os.sep).replace('/', os.sep)
  logging.info("Bootstrap scripts are being executed...")
- process = ClientConstants.exec_console_with_script(bootstrap_scripts, should_capture_stdout_err=True)
+ process = ClientConstants.exec_console_with_script(bootstrap_scripts, should_capture_stdout=True,
+ should_capture_stderr=True)
  ret_code, out, err = ClientConstants.get_console_pipe_out_err_results(process)
  if out is not None:
  out_str = out.decode(encoding="utf-8")
@@ -340,12 +344,15 @@ def run(self):
  str(dynamic_args_config["rank"]),
  "--role",
  "client",
- ]
+ ],
+ should_capture_stdout=False,
+ should_capture_stderr=True
  )
  ClientConstants.save_learning_process(process.pid)
  self.release_client_mqtt_mgr()
- ret_code, out, err = ClientConstants.get_console_sys_out_pipe_err_results(process)
- if ret_code != 0 and err is not None and str(err.decode(encoding="utf-8")).find('__finish ') == -1:
+ ret_code, out, err = ClientConstants.get_console_pipe_out_err_results(process)
+ if ret_code != 0 and err is not None and str(err.decode(encoding="utf-8")).find('__finish ') == -1 and \
+ (out is not None and str(out.decode(encoding="utf-8")).find('__finish ') == -1):
  logging.error("Exception when executing client program: {}".format(err.decode(encoding="utf-8")))
  self.setup_client_mqtt_mgr()
  self.wait_client_mqtt_connected()
@@ -365,11 +372,6 @@ def stop_run(self):
 
  logging.info("Stop run successfully.")
 
- # Stop log processor for current run
- MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
-
- time.sleep(2)
-
  # Notify MLOps with the stopping message
  self.mlops_metrics.report_client_training_status(self.edge_id, ClientConstants.MSG_MLOPS_CLIENT_STATUS_STOPPING)
 
@@ -391,11 +393,6 @@ def stop_run_with_killed_status(self):
 
  logging.info("Stop run successfully.")
 
- # Stop log processor for current run
- MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
-
- time.sleep(2)
-
  # Notify MLOps with the stopping message
  self.mlops_metrics.report_client_training_status(self.edge_id, ClientConstants.MSG_MLOPS_CLIENT_STATUS_STOPPING)
 
@@ -435,11 +432,6 @@ def cleanup_run_when_starting_failed(self):
 
  logging.info("Cleanup run successfully when starting failed.")
 
- # Stop log processor for current run
- MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
-
- time.sleep(2)
-
  self.reset_devices_status(self.edge_id, ClientConstants.MSG_MLOPS_CLIENT_STATUS_FAILED)
 
  time.sleep(2)
@@ -465,11 +457,6 @@ def cleanup_run_when_finished(self):
 
  logging.info("Cleanup run successfully when finished.")
 
- # Stop log processor for current run
- MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(self.run_id, self.edge_id)
-
- time.sleep(2)
-
  self.reset_devices_status(self.edge_id, ClientConstants.MSG_MLOPS_CLIENT_STATUS_FINISHED)
 
  time.sleep(2)
@@ -616,6 +603,9 @@ def callback_stop_train(self, topic, payload):
  except Exception as e:
  pass
 
+ # Stop log processor for current run
+ MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(run_id, self.edge_id)
+
  def callback_exit_train_with_exception(self, topic, payload):
  logging.info("callback_exit_train_with_exception: topic = %s, payload = %s" % (topic, payload))
 
@@ -666,7 +656,12 @@ def callback_runner_id_status(self, topic, payload):
  run_id=run_id,
  )
  client_runner.device_status = status
- Process(target=client_runner.cleanup_client_with_status).start()
+ status_process = Process(target=client_runner.cleanup_client_with_status)
+ status_process.start()
+ status_process.join(15)
+
+ # Stop log processor for current run
+ MLOpsRuntimeLogDaemon.get_instance(self.args).stop_log_processor(run_id, edge_id)
 
  def report_client_status(self):
  self.send_agent_active_msg()
@@ -768,7 +763,7 @@ def GetUUID():
  def bind_account_and_device_id(self, url, account_id, device_id, os_name, role="client"):
  ip = requests.get('https://checkip.amazonaws.com').text.strip()
  fedml_ver, exec_path, os_ver, cpu_info, python_ver, torch_ver, mpi_installed, \
-  cpu_usage, available_mem, total_mem, gpu_info, gpu_available_mem, gpu_total_mem = get_sys_runner_info()
+ cpu_usage, available_mem, total_mem, gpu_info, gpu_available_mem, gpu_total_mem = get_sys_runner_info()
  json_params = {
  "accountid": account_id,
  "deviceid": device_id,
@@ -794,7 +789,7 @@ def bind_account_and_device_id(self, url, account_id, device_id, os_name, role="
  if gpu_available_mem is not None:
  json_params["extra_infos"]["gpu_available_mem"] = gpu_available_mem
  if gpu_total_mem is not None:
- json_params["extra_infos"]["gpu_total_mem"]  = gpu_total_mem
+ json_params["extra_infos"]["gpu_total_mem"] = gpu_total_mem
  else:
  json_params["gpu"] = "None"
 
@@ -803,12 +798,14 @@ def bind_account_and_device_id(self, url, account_id, device_id, os_name, role="
  try:
  requests.session().verify = cert_path
  response = requests.post(
- url, json=json_params, verify=True, headers={"content-type": "application/json", "Connection": "close"}
+ url, json=json_params, verify=True,
+ headers={"content-type": "application/json", "Connection": "close"}
  )
  except requests.exceptions.SSLError as err:
  MLOpsConfigs.install_root_ca_file()
  response = requests.post(
- url, json=json_params, verify=True, headers={"content-type": "application/json", "Connection": "close"}
+ url, json=json_params, verify=True,
+ headers={"content-type": "application/json", "Connection": "close"}
  )
  else:
  response = requests.post(url, json=json_params, headers={"Connection": "close"})
@@ -826,9 +823,9 @@ def send_agent_active_msg(self):
  active_topic = "/flclient_agent/active"
  status = MLOpsStatus.get_instance().get_client_agent_status(self.edge_id)
  if (
- status is not None
- and status != ClientConstants.MSG_MLOPS_CLIENT_STATUS_OFFLINE
- and status != ClientConstants.MSG_MLOPS_CLIENT_STATUS_IDLE
+  status is not None
+  and status != ClientConstants.MSG_MLOPS_CLIENT_STATUS_OFFLINE
+  and status != ClientConstants.MSG_MLOPS_CLIENT_STATUS_IDLE
  ):
  return
  status = ClientConstants.MSG_MLOPS_CLIENT_STATUS_IDLE
@@ -925,12 +922,11 @@ def setup_agent_mqtt_connection(self, service_config):
  MLOpsStatus.get_instance().set_client_agent_status(self.edge_id, ClientConstants.MSG_MLOPS_CLIENT_STATUS_IDLE)
  self.release_client_mqtt_mgr()
 
+ MLOpsRuntimeLogDaemon.get_instance(self.args).stop_all_log_processor()
+
  def start_agent_mqtt_loop(self):
  # Start MQTT message loop
  try:
  self.mqtt_mgr.loop_forever()
  except Exception as e:
  pass
-
-
-

python/fedml/cli/server_deployment/server_constants.py

@@ -192,33 +192,33 @@ def exit_process(process):
  pass
 
  @staticmethod
- def exec_console_with_script(script_path, should_capture_stdout_err=False):
- if should_capture_stdout_err:
-  if platform.system() == 'Windows':
- script_process = subprocess.Popen(script_path, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-  else:
-  script_process = subprocess.Popen(['sh', '-c', script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ def exec_console_with_script(script_path, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ if platform.system() == 'Windows':
+ script_process = subprocess.Popen(script_path, stdout=stdout_flag, stderr=stderr_flag)
  else:
- if platform.system() == 'Windows':
- script_process = subprocess.Popen(script_path, stdout=sys.stdout, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen(['sh', '-c', script_path], stdout=sys.stdout, stderr=subprocess.PIPE)
+ script_process = subprocess.Popen(['sh', '-c', script_path], stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod
- def exec_console_with_shell(shell, script_path, should_capture_stdout_err=False):
- if should_capture_stdout_err:
- script_process = subprocess.Popen([shell, script_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen([shell, script_path], stdout=sys.stdout, stderr=subprocess.PIPE)
+ def exec_console_with_shell(shell, script_path, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ script_process = subprocess.Popen([shell, script_path], stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod
- def exec_console_with_shell_script_list(shell_script_list, should_capture_stdout_err=False):
- if should_capture_stdout_err:
- script_process = subprocess.Popen(shell_script_list, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- else:
- script_process = subprocess.Popen(shell_script_list, stdout=sys.stdout, stderr=sys.stderr)
+ def exec_console_with_shell_script_list(shell_script_list, should_capture_stdout=False, should_capture_stderr=False):
+ stdout_flag = subprocess.PIPE if should_capture_stdout else sys.stdout
+ stderr_flag = subprocess.PIPE if should_capture_stderr else sys.stderr
+
+ script_process = subprocess.Popen(shell_script_list, stdout=stdout_flag, stderr=stderr_flag)
+
  return script_process
 
  @staticmethod