From 0ce39fbee4a82971a6a9c49f826d737bce4b2ad8 Mon Sep 17 00:00:00 2001 From: Paras Jain Date: Sun, 16 Jan 2022 01:31:21 +0000 Subject: [PATCH 1/3] Fix GCP --- .../benchmark_num_connections.sh | 17 +-- scripts/requirements-gateway.txt | 4 + skylark/compute/aws/aws_server.py | 9 +- skylark/compute/azure/azure_server.py | 1 + skylark/compute/gcp/gcp_cloud_provider.py | 23 ++- skylark/compute/gcp/gcp_server.py | 1 + skylark/test/test_replicator_client.py | 136 ------------------ 7 files changed, 42 insertions(+), 149 deletions(-) delete mode 100644 skylark/test/test_replicator_client.py diff --git a/scripts/experiments/num_connections/benchmark_num_connections.sh b/scripts/experiments/num_connections/benchmark_num_connections.sh index c0fb9844e..ebfadd923 100644 --- a/scripts/experiments/num_connections/benchmark_num_connections.sh +++ b/scripts/experiments/num_connections/benchmark_num_connections.sh @@ -20,7 +20,7 @@ function benchmark_config { NUM_GATEWAYS=$2 # N_CHUNKS_PER_CONNECTION * NUM_CONNECTIONS N_CHUNKS=$((N_CHUNKS_PER_CONNECTION * NUM_CONNECTIONS)) - echo "skylark replicate-random $SRC_REGION $DST_REGION $INTER_REGION --chunk-size-mb $CHUNK_SIZE_MB --n-chunks $N_CHUNKS --num-gateways $NUM_GATEWAYS --num-outgoing-connections $NUM_CONNECTIONS --no-reuse-gateways --no-serve-web-dashboard" + echo "skylark replicate-random $SRC_REGION $DST_REGION $INTER_REGION --chunk-size-mb $CHUNK_SIZE_MB --n-chunks $N_CHUNKS --num-gateways $NUM_GATEWAYS --num-outgoing-connections $NUM_CONNECTIONS --no-reuse-gateways --no-serve-web-dashboard --gcp-project skylark-333700 --azure-subscription ab110d95-7b83-4cec-b9dc-400255f3166e" } # log function with message argument @@ -30,7 +30,8 @@ function log() { echo -e "${BGreen}$1${NC}" } -EXP_ID="$SRC_REGION-$DST_REGION-$(./scripts/utils/get_random_word_hash.sh)" +EXP_TAG=${EXP_TAG:-"$SRC_REGION+$DST_REGION+$INTER_REGION"} +EXP_ID="$EXP_TAG-$(./scripts/utils/get_random_word_hash.sh)" LOG_DIR=data/experiments/benchmark_num_connections/logs/$EXP_ID log "Creating log directory $LOG_DIR" log "Experiment ID: $EXP_ID" @@ -38,8 +39,8 @@ rm -rf $LOG_DIR mkdir -p $LOG_DIR touch $LOG_DIR/launch.log -log "Stopping existing instances" -skylark deprovision &>> $LOG_DIR/launch.log +# log "Stopping existing instances" +# skylark deprovision &>> $LOG_DIR/launch.log log "Building docker image" source scripts/pack_docker.sh &>> $LOG_DIR/launch.log @@ -58,7 +59,7 @@ fi # make list of commands to run with gnu parallel (one for each inter-region) and save to $PARALLEL_CMD_LIST (one command per line) PARALLEL_CMD_LIST="" # powers of 2 -for NUM_CONNECTIONS in 1 2 4 8 16 32 64 128 144; do +for NUM_CONNECTIONS in 128 96 64 32 16 8 4 2 1; do PARALLEL_CMD_LIST="$PARALLEL_CMD_LIST\n$(benchmark_config $NUM_CONNECTIONS $NUM_GATEWAYS) &> $LOG_DIR/$NUM_CONNECTIONS.log" done log "Running commands with gnu parallel:" @@ -66,11 +67,11 @@ echo -e "$PARALLEL_CMD_LIST\n" echo -e "$PARALLEL_CMD_LIST\n" >> $LOG_DIR/launch.log log "Parallel:" -parallel -j 8 --results $LOG_DIR/raw_logs --joblog $LOG_DIR/parallel_joblog.txt --eta < <(echo -e "$PARALLEL_CMD_LIST") +parallel -j 2 --results $LOG_DIR/raw_logs --joblog $LOG_DIR/parallel_joblog.txt --eta < <(echo -e "$PARALLEL_CMD_LIST") -log "Stopping instances" -skylark deprovision &>> $LOG_DIR/launch.log +# log "Stopping instances" +# skylark deprovision &>> $LOG_DIR/launch.log log "Done, results in $LOG_DIR" log "Experiment ID: $EXP_ID" \ No newline at end of file diff --git a/scripts/requirements-gateway.txt b/scripts/requirements-gateway.txt index c9ee145ec..97fbbc4ed 100644 --- a/scripts/requirements-gateway.txt +++ b/scripts/requirements-gateway.txt @@ -1,3 +1,7 @@ +azure-mgmt-resource +azure-mgmt-compute +azure-mgmt-network +azure-identity awscrt boto3 flask diff --git a/skylark/compute/aws/aws_server.py b/skylark/compute/aws/aws_server.py index c18f701b9..a1bee957f 100644 --- a/skylark/compute/aws/aws_server.py +++ b/skylark/compute/aws/aws_server.py @@ -117,5 +117,12 @@ def terminate_instance_impl(self): def get_ssh_client_impl(self): client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) - client.connect(self.public_ip(), username="ubuntu", key_filename=str(self.local_keyfile), look_for_keys=False, allow_agent=False) + client.connect( + self.public_ip(), + username="ubuntu", + key_filename=str(self.local_keyfile), + look_for_keys=False, + allow_agent=False, + banner_timeout=200, + ) return client diff --git a/skylark/compute/azure/azure_server.py b/skylark/compute/azure/azure_server.py index 4462c3293..41e5e83ac 100644 --- a/skylark/compute/azure/azure_server.py +++ b/skylark/compute/azure/azure_server.py @@ -156,5 +156,6 @@ def get_ssh_client_impl(self, uname=os.environ.get("USER"), ssh_key_password="sk key_filename=str(self.ssh_private_key), passphrase=ssh_key_password, look_for_keys=False, + banner_timeout=200, ) return ssh_client diff --git a/skylark/compute/gcp/gcp_cloud_provider.py b/skylark/compute/gcp/gcp_cloud_provider.py index 33234ee8c..89207b295 100644 --- a/skylark/compute/gcp/gcp_cloud_provider.py +++ b/skylark/compute/gcp/gcp_cloud_provider.py @@ -5,6 +5,8 @@ from typing import List import googleapiclient +from loguru import logger +from oslo_concurrency import lockutils import paramiko from skylark import key_root @@ -150,6 +152,15 @@ def configure_default_network(self): def configure_default_firewall(self, ip="0.0.0.0/0"): """Configure default firewall to allow access from all ports from all IPs (if not exists).""" compute = GCPServer.get_gcp_client() + + @lockutils.synchronized(f"gcp_configure_default_firewall", external=True, lock_path="/tmp/skylark_locks") + def create_firewall(body, update_firewall=False): + if update_firewall: + op = compute.firewalls().update(project=self.gcp_project, firewall="default", body=fw_body).execute() + else: + op = compute.firewalls().insert(project=self.gcp_project, body=fw_body).execute() + self.wait_for_operation_to_complete("global", op["name"]) + try: current_firewall = compute.firewalls().get(project=self.gcp_project, firewall="default").execute() except googleapiclient.errors.HttpError as e: @@ -157,6 +168,7 @@ def configure_default_firewall(self, ip="0.0.0.0/0"): current_firewall = None else: raise e + fw_body = { "name": "default", "allowed": [{"IPProtocol": "tcp", "ports": ["1-65535"]}, {"IPProtocol": "udp", "ports": ["1-65535"]}, {"IPProtocol": "icmp"}], @@ -164,10 +176,13 @@ def configure_default_firewall(self, ip="0.0.0.0/0"): "sourceRanges": [ip], } if current_firewall is None: - op = compute.firewalls().insert(project=self.gcp_project, body=fw_body).execute() - else: - op = compute.firewalls().update(project=self.gcp_project, firewall="default", body=fw_body).execute() - self.wait_for_operation_to_complete("global", op["name"]) + logger.warning(f"[GCP] Creating new firewall") + create_firewall(fw_body, update_firewall=False) + logger.debug(f"[GCP] Created new firewall") + elif current_firewall["allowed"] != fw_body["allowed"]: + logger.warning(f"[GCP] Updating firewall, current rules do not match") + create_firewall(fw_body, update_firewall=True) + logger.debug(f"[GCP] Updated firewall") def get_operation_state(self, zone, operation_name): compute = GCPServer.get_gcp_client() diff --git a/skylark/compute/gcp/gcp_server.py b/skylark/compute/gcp/gcp_server.py index ba77b413f..23410716d 100644 --- a/skylark/compute/gcp/gcp_server.py +++ b/skylark/compute/gcp/gcp_server.py @@ -101,5 +101,6 @@ def get_ssh_client_impl(self, uname=os.environ.get("USER"), ssh_key_password="sk key_filename=str(self.ssh_private_key), passphrase=ssh_key_password, look_for_keys=False, + banner_timeout=200, ) return ssh_client diff --git a/skylark/test/test_replicator_client.py b/skylark/test/test_replicator_client.py deleted file mode 100644 index 2aba2f392..000000000 --- a/skylark/test/test_replicator_client.py +++ /dev/null @@ -1,136 +0,0 @@ -import argparse -import atexit - -from loguru import logger -from skylark import GB, MB, print_header -from skylark.chunk import ChunkState - -from skylark.replicate.replication_plan import ReplicationJob, ReplicationTopology -from skylark.replicate.replicator_client import ReplicatorClient -from skylark.utils.utils import Timer -from skylark import skylark_root - - -def parse_args(): - parser = argparse.ArgumentParser(description="Run a replication job") - - # gateway path parameters - parser.add_argument("--src-region", default="aws:us-east-1", help="AWS region of source bucket") - parser.add_argument("--inter-region", default=None, help="AWS region of intermediate bucket") - parser.add_argument("--dest-region", default="aws:us-west-1", help="AWS region of destination bucket") - parser.add_argument("--num-gateways", default=1, type=int, help="Number of gateways to use") - parser.add_argument("--num-outgoing-connections", default=16, type=int, help="Number of outgoing connections from a gateway") - - # object information - parser.add_argument("--key-prefix", default="/test/direct_replication", help="S3 key prefix for all objects") - parser.add_argument("--chunk-size-mb", default=128, type=int, help="Chunk size in MB") - parser.add_argument("--n-chunks", default=16, type=int, help="Number of chunks in bucket") - parser.add_argument("--skip-upload", action="store_true", help="Skip uploading objects to S3") - - # gateway provisioning - parser.add_argument("--gcp-project", default="skylark-333700", help="GCP project ID") - parser.add_argument("--azure-subscription", default="", help="Azure subscription") - parser.add_argument("--gateway-docker-image", default="ghcr.io/parasj/skylark:main", help="Docker image for gateway instances") - parser.add_argument("--aws-instance-class", default="m5.4xlarge", help="AWS instance class") - parser.add_argument("--azure-instance-class", default="Standard_D2_v5", help="Azure instance class") - parser.add_argument("--gcp-instance-class", default="n2-standard-16", help="GCP instance class") - parser.add_argument("--copy-ssh-key", default=None, help="SSH public key to add to gateways") - parser.add_argument("--log-dir", default=None, help="Directory to write instance SSH logs to") - parser.add_argument("--gcp-use-premium-network", action="store_true", help="Use GCP premium network") - args = parser.parse_args() - - # add support for None arguments - if args.aws_instance_class == "None": - args.aws_instance_class = None - if args.azure_instance_class == "None": - args.azure_instance_class = None - if args.gcp_instance_class == "None": - args.gcp_instance_class = None - - return args - - -def main(args): - src_bucket, dst_bucket = f"skylark-{args.src_region.split(':')[1]}", f"skylark-{args.dest_region.split(':')[1]}" - # s3_interface_src = S3Interface(args.src_region.split(":")[1], src_bucket) - # s3_interface_dst = S3Interface(args.dest_region.split(":")[1], dst_bucket) - # s3_interface_src.create_bucket() - # s3_interface_dst.create_bucket() - - if not args.skip_upload: - # todo implement object store support - pass - # matching_src_keys = list(s3_interface_src.list_objects(prefix=args.key_prefix)) - # matching_dst_keys = list(s3_interface_dst.list_objects(prefix=args.key_prefix)) - # if matching_src_keys: - # logger.warning(f"Deleting objects from source bucket: {matching_src_keys}") - # s3_interface_src.delete_objects(matching_src_keys) - # if matching_dst_keys: - # logger.warning(f"Deleting objects from destination bucket: {matching_dst_keys}") - # s3_interface_dst.delete_objects(matching_dst_keys) - - # # create test objects w/ random data - # logger.info("Creating test objects") - # obj_keys = [] - # futures = [] - # with tempfile.NamedTemporaryFile() as f: - # f.write(os.urandom(int(MB * args.chunk_size_mb))) - # f.seek(0) - # for i in trange(args.n_chunks): - # k = f"{args.key_prefix}/{i}" - # futures.append(s3_interface_src.upload_object(f.name, k)) - # obj_keys.append(k) - # concurrent.futures.wait(futures) - else: - obj_keys = [f"{args.key_prefix}/{i}" for i in range(args.n_chunks)] - - # define the replication job and topology - if args.inter_region: - topo = ReplicationTopology(paths=[[args.src_region, args.inter_region, args.dest_region] for _ in range(args.num_gateways)]) - else: - topo = ReplicationTopology(paths=[[args.src_region, args.dest_region] for _ in range(args.num_gateways)]) - logger.info("Creating replication client") - rc = ReplicatorClient( - topo, - gcp_project=args.gcp_project, - azure_subscription=args.azure_subscription, - gateway_docker_image=args.gateway_docker_image, - aws_instance_class=args.aws_instance_class, - azure_instance_class=args.azure_instance_class, - gcp_instance_class=args.gcp_instance_class, - gcp_use_premium_network=args.gcp_use_premium_network, - ) - - # provision the gateway instances - logger.info("Provisioning gateway instances") - rc.provision_gateways( - reuse_instances=True, - log_dir=args.log_dir, - authorize_ssh_pub_key=args.copy_ssh_key, - num_outgoing_connections=args.num_outgoing_connections, - ) - for path in rc.bound_paths: - logger.info(f"Provisioned path {' -> '.join(path[i].region_tag for i in range(len(path)))}") - for gw in path: - logger.info(f"\t[{gw.region_tag}] {gw.gateway_log_viewer_url}") - - # run replication, monitor progress - job = ReplicationJob( - source_region=args.src_region, - source_bucket=src_bucket, - dest_region=args.dest_region, - dest_bucket=dst_bucket, - objs=obj_keys, - random_chunk_size_mb=args.chunk_size_mb, - ) - - total_bytes = args.n_chunks * args.chunk_size_mb * MB - crs = rc.run_replication_plan(job) - logger.info(f"{total_bytes / GB:.2f}GByte replication job launched") - stats = rc.monitor_transfer(crs, serve_web_dashboard=True) - logger.info(f"Replication completed in {stats['total_runtime_s']:.2f}s ({stats['throughput_gbits']:.2f}Gbit/s)") - - -if __name__ == "__main__": - print_header() - main(parse_args()) From 3234487e41912b4bac8b747e171a275a80866b60 Mon Sep 17 00:00:00 2001 From: Paras Jain Date: Sun, 16 Jan 2022 03:53:06 +0000 Subject: [PATCH 2/3] restore replicator client test --- skylark/test/test_replicator_client.py | 136 +++++++++++++++++++++++++ 1 file changed, 136 insertions(+) create mode 100644 skylark/test/test_replicator_client.py diff --git a/skylark/test/test_replicator_client.py b/skylark/test/test_replicator_client.py new file mode 100644 index 000000000..f547366f2 --- /dev/null +++ b/skylark/test/test_replicator_client.py @@ -0,0 +1,136 @@ +import argparse +import atexit + +from loguru import logger +from skylark import GB, MB, print_header +from skylark.chunk import ChunkState + +from skylark.replicate.replication_plan import ReplicationJob, ReplicationTopology +from skylark.replicate.replicator_client import ReplicatorClient +from skylark.utils.utils import Timer +from skylark import skylark_root + + +def parse_args(): + parser = argparse.ArgumentParser(description="Run a replication job") + + # gateway path parameters + parser.add_argument("--src-region", default="aws:us-east-1", help="AWS region of source bucket") + parser.add_argument("--inter-region", default=None, help="AWS region of intermediate bucket") + parser.add_argument("--dest-region", default="aws:us-west-1", help="AWS region of destination bucket") + parser.add_argument("--num-gateways", default=1, type=int, help="Number of gateways to use") + parser.add_argument("--num-outgoing-connections", default=16, type=int, help="Number of outgoing connections from a gateway") + + # object information + parser.add_argument("--key-prefix", default="/test/direct_replication", help="S3 key prefix for all objects") + parser.add_argument("--chunk-size-mb", default=128, type=int, help="Chunk size in MB") + parser.add_argument("--n-chunks", default=16, type=int, help="Number of chunks in bucket") + parser.add_argument("--skip-upload", action="store_true", help="Skip uploading objects to S3") + + # gateway provisioning + parser.add_argument("--gcp-project", default="skylark-333700", help="GCP project ID") + parser.add_argument("--azure-subscription", default="", help="Azure subscription") + parser.add_argument("--gateway-docker-image", default="ghcr.io/parasj/skylark:main", help="Docker image for gateway instances") + parser.add_argument("--aws-instance-class", default="m5.4xlarge", help="AWS instance class") + parser.add_argument("--azure-instance-class", default="Standard_D2_v5", help="Azure instance class") + parser.add_argument("--gcp-instance-class", default="n2-standard-16", help="GCP instance class") + parser.add_argument("--copy-ssh-key", default=None, help="SSH public key to add to gateways") + parser.add_argument("--log-dir", default=None, help="Directory to write instance SSH logs to") + parser.add_argument("--gcp-use-premium-network", action="store_true", help="Use GCP premium network") + args = parser.parse_args() + + # add support for None arguments + if args.aws_instance_class == "None": + args.aws_instance_class = None + if args.azure_instance_class == "None": + args.azure_instance_class = None + if args.gcp_instance_class == "None": + args.gcp_instance_class = None + + return args + + +def main(args): + src_bucket, dst_bucket = f"skylark-{args.src_region.split(':')[1]}", f"skylark-{args.dest_region.split(':')[1]}" + # s3_interface_src = S3Interface(args.src_region.split(":")[1], src_bucket) + # s3_interface_dst = S3Interface(args.dest_region.split(":")[1], dst_bucket) + # s3_interface_src.create_bucket() + # s3_interface_dst.create_bucket() + + if not args.skip_upload: + # todo implement object store support + pass + # matching_src_keys = list(s3_interface_src.list_objects(prefix=args.key_prefix)) + # matching_dst_keys = list(s3_interface_dst.list_objects(prefix=args.key_prefix)) + # if matching_src_keys: + # logger.warning(f"Deleting objects from source bucket: {matching_src_keys}") + # s3_interface_src.delete_objects(matching_src_keys) + # if matching_dst_keys: + # logger.warning(f"Deleting objects from destination bucket: {matching_dst_keys}") + # s3_interface_dst.delete_objects(matching_dst_keys) + + # # create test objects w/ random data + # logger.info("Creating test objects") + # obj_keys = [] + # futures = [] + # with tempfile.NamedTemporaryFile() as f: + # f.write(os.urandom(int(MB * args.chunk_size_mb))) + # f.seek(0) + # for i in trange(args.n_chunks): + # k = f"{args.key_prefix}/{i}" + # futures.append(s3_interface_src.upload_object(f.name, k)) + # obj_keys.append(k) + # concurrent.futures.wait(futures) + else: + obj_keys = [f"{args.key_prefix}/{i}" for i in range(args.n_chunks)] + + # define the replication job and topology + if args.inter_region: + topo = ReplicationTopology(paths=[[args.src_region, args.inter_region, args.dest_region] for _ in range(args.num_gateways)]) + else: + topo = ReplicationTopology(paths=[[args.src_region, args.dest_region] for _ in range(args.num_gateways)]) + logger.info("Creating replication client") + rc = ReplicatorClient( + topo, + gcp_project=args.gcp_project, + azure_subscription=args.azure_subscription, + gateway_docker_image=args.gateway_docker_image, + aws_instance_class=args.aws_instance_class, + azure_instance_class=args.azure_instance_class, + gcp_instance_class=args.gcp_instance_class, + gcp_use_premium_network=args.gcp_use_premium_network, + ) + + # provision the gateway instances + logger.info("Provisioning gateway instances") + rc.provision_gateways( + reuse_instances=True, + log_dir=args.log_dir, + authorize_ssh_pub_key=args.copy_ssh_key, + num_outgoing_connections=args.num_outgoing_connections, + ) + for path in rc.bound_paths: + logger.info(f"Provisioned path {' -> '.join(path[i].region_tag for i in range(len(path)))}") + for gw in path: + logger.info(f"\t[{gw.region_tag}] {gw.gateway_log_viewer_url}") + + # run replication, monitor progress + job = ReplicationJob( + source_region=args.src_region, + source_bucket=src_bucket, + dest_region=args.dest_region, + dest_bucket=dst_bucket, + objs=obj_keys, + random_chunk_size_mb=args.chunk_size_mb, + ) + + total_bytes = args.n_chunks * args.chunk_size_mb * MB + crs = rc.run_replication_plan(job) + logger.info(f"{total_bytes / GB:.2f}GByte replication job launched") + stats = rc.monitor_transfer(crs, serve_web_dashboard=True) + logger.info(f"Replication completed in {stats['total_runtime_s']:.2f}s ({stats['throughput_gbits']:.2f}Gbit/s)") + + +if __name__ == "__main__": + print_header() + main(parse_args()) \ No newline at end of file From 6b3be234eff9471e440f3bcaf02a334961985ae1 Mon Sep 17 00:00:00 2001 From: Paras Jain Date: Sun, 16 Jan 2022 03:54:30 +0000 Subject: [PATCH 3/3] update --- .../benchmark_num_connections.sh | 17 ++++++++--------- skylark/test/test_replicator_client.py | 2 +- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/scripts/experiments/num_connections/benchmark_num_connections.sh b/scripts/experiments/num_connections/benchmark_num_connections.sh index ebfadd923..c0fb9844e 100644 --- a/scripts/experiments/num_connections/benchmark_num_connections.sh +++ b/scripts/experiments/num_connections/benchmark_num_connections.sh @@ -20,7 +20,7 @@ function benchmark_config { NUM_GATEWAYS=$2 # N_CHUNKS_PER_CONNECTION * NUM_CONNECTIONS N_CHUNKS=$((N_CHUNKS_PER_CONNECTION * NUM_CONNECTIONS)) - echo "skylark replicate-random $SRC_REGION $DST_REGION $INTER_REGION --chunk-size-mb $CHUNK_SIZE_MB --n-chunks $N_CHUNKS --num-gateways $NUM_GATEWAYS --num-outgoing-connections $NUM_CONNECTIONS --no-reuse-gateways --no-serve-web-dashboard --gcp-project skylark-333700 --azure-subscription ab110d95-7b83-4cec-b9dc-400255f3166e" + echo "skylark replicate-random $SRC_REGION $DST_REGION $INTER_REGION --chunk-size-mb $CHUNK_SIZE_MB --n-chunks $N_CHUNKS --num-gateways $NUM_GATEWAYS --num-outgoing-connections $NUM_CONNECTIONS --no-reuse-gateways --no-serve-web-dashboard" } # log function with message argument @@ -30,8 +30,7 @@ function log() { echo -e "${BGreen}$1${NC}" } -EXP_TAG=${EXP_TAG:-"$SRC_REGION+$DST_REGION+$INTER_REGION"} -EXP_ID="$EXP_TAG-$(./scripts/utils/get_random_word_hash.sh)" +EXP_ID="$SRC_REGION-$DST_REGION-$(./scripts/utils/get_random_word_hash.sh)" LOG_DIR=data/experiments/benchmark_num_connections/logs/$EXP_ID log "Creating log directory $LOG_DIR" log "Experiment ID: $EXP_ID" @@ -39,8 +38,8 @@ rm -rf $LOG_DIR mkdir -p $LOG_DIR touch $LOG_DIR/launch.log -# log "Stopping existing instances" -# skylark deprovision &>> $LOG_DIR/launch.log +log "Stopping existing instances" +skylark deprovision &>> $LOG_DIR/launch.log log "Building docker image" source scripts/pack_docker.sh &>> $LOG_DIR/launch.log @@ -59,7 +58,7 @@ fi # make list of commands to run with gnu parallel (one for each inter-region) and save to $PARALLEL_CMD_LIST (one command per line) PARALLEL_CMD_LIST="" # powers of 2 -for NUM_CONNECTIONS in 128 96 64 32 16 8 4 2 1; do +for NUM_CONNECTIONS in 1 2 4 8 16 32 64 128 144; do PARALLEL_CMD_LIST="$PARALLEL_CMD_LIST\n$(benchmark_config $NUM_CONNECTIONS $NUM_GATEWAYS) &> $LOG_DIR/$NUM_CONNECTIONS.log" done log "Running commands with gnu parallel:" @@ -67,11 +66,11 @@ echo -e "$PARALLEL_CMD_LIST\n" echo -e "$PARALLEL_CMD_LIST\n" >> $LOG_DIR/launch.log log "Parallel:" -parallel -j 2 --results $LOG_DIR/raw_logs --joblog $LOG_DIR/parallel_joblog.txt --eta < <(echo -e "$PARALLEL_CMD_LIST") +parallel -j 8 --results $LOG_DIR/raw_logs --joblog $LOG_DIR/parallel_joblog.txt --eta < <(echo -e "$PARALLEL_CMD_LIST") -# log "Stopping instances" -# skylark deprovision &>> $LOG_DIR/launch.log +log "Stopping instances" +skylark deprovision &>> $LOG_DIR/launch.log log "Done, results in $LOG_DIR" log "Experiment ID: $EXP_ID" \ No newline at end of file diff --git a/skylark/test/test_replicator_client.py b/skylark/test/test_replicator_client.py index f547366f2..2aba2f392 100644 --- a/skylark/test/test_replicator_client.py +++ b/skylark/test/test_replicator_client.py @@ -133,4 +133,4 @@ def main(args): if __name__ == "__main__": print_header() - main(parse_args()) \ No newline at end of file + main(parse_args())