wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-18 12:40:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'releases/1.2.0' of github.com:ray-project/ray into releases/1.2.0

Browse Source
This commit is contained in:
Alex
2021-02-05 02:45:12 +00:00
parent 40beec569c 4c71f76b25
commit 75886c8e78
6 changed files with 172 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/tests/test_global_state.py
+26
View File
@@ -7,6 +7,7 @@ import time
import ray
import ray.ray_constants
import ray.services
import ray.test_utils
from ray._raylet import GlobalStateAccessor
@@ -332,6 +333,31 @@ def test_backlog_report(shutdown_only):
global_state_accessor.disconnect()
def test_heartbeat_ip(shutdown_only):
cluster = ray.init(
num_cpus=1, _system_config={
"report_worker_backlog": True,
})
global_state_accessor = GlobalStateAccessor(
cluster["redis_address"], ray.ray_constants.REDIS_DEFAULT_PASSWORD)
global_state_accessor.connect()
self_ip = ray.services.get_node_ip_address()
def self_ip_is_set():
message = global_state_accessor.get_all_resource_usage()
if message is None:
return False
resource_usage = ray.gcs_utils.ResourceUsageBatchData.FromString(
message)
resources_data = resource_usage.batch[0]
return resources_data.node_manager_address == self_ip
ray.test_utils.wait_for_condition(self_ip_is_set, timeout=2)
global_state_accessor.disconnect()
if __name__ == "__main__":
import pytest
import sys
python/ray/util/sgd/BUILD
+14
View File
@@ -241,6 +241,20 @@ py_test(
args = ["--smoke-test"]
)
# --------------------------------------------------------------------
# SGD related tests from the ../../../../release directory.
# Please keep these sorted alphabetically.
# --------------------------------------------------------------------
py_test(
name = "pytorch_pbt_failure",
size = "medium",
srcs = ["torch/examples/pytorch_pbt_failure.py"],
tags = ["exlusive", "pytorch", "release"],
deps = [":sgd_lib"],
args = ["--smoke-test"]
)
# This is a dummy test dependency that causes the above tests to be
# re-run if any of these files changes.
py_library(
python/ray/util/sgd/torch/examples/pytorch_pbt_failure.py
+128
View File
@@ -0,0 +1,128 @@
import argparse
import numpy as np
import os
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Subset
from torchvision.datasets import CIFAR10
import torchvision.transforms as transforms
import ray
from ray import tune
from ray.tune import CLIReporter
from ray.tune.schedulers import PopulationBasedTraining
from ray.tune.utils.mock import FailureInjectorCallback
from ray.util.sgd.torch import TorchTrainer, TrainingOperator
from ray.util.sgd.torch.resnet import ResNet18
from ray.util.sgd.utils import BATCH_SIZE
parser = argparse.ArgumentParser()
parser.add_argument(
"--smoke-test",
action="store_true",
default=False,
help="Finish quickly for training.")
args = parser.parse_args()
def initialization_hook():
# Need this for avoiding a connection restart issue on AWS.
os.environ["NCCL_SOCKET_IFNAME"] = "^docker0,lo"
os.environ["NCCL_LL_THRESHOLD"] = "0"
# set the below if needed
# print("NCCL DEBUG SET")
# os.environ["NCCL_DEBUG"] = "INFO"
def cifar_creator(config):
transform_train = transforms.Compose([
transforms.RandomCrop(32, padding=4),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010)),
]) # meanstd transformation
transform_test = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010)),
])
train_dataset = CIFAR10(
root="~/data", train=True, download=True, transform=transform_train)
validation_dataset = CIFAR10(
root="~/data", train=False, download=False, transform=transform_test)
if config.get("test_mode"):
train_dataset = Subset(train_dataset, list(range(64)))
validation_dataset = Subset(validation_dataset, list(range(64)))
train_loader = DataLoader(
train_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
validation_loader = DataLoader(
validation_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
return train_loader, validation_loader
def optimizer_creator(model, config):
"""Returns optimizer"""
return torch.optim.SGD(
model.parameters(),
lr=config.get("lr", 0.1),
momentum=config.get("momentum", 0.9))
ray.init(address="auto" if not args.smoke_test else None, log_to_driver=True)
num_training_workers = 1 if args.smoke_test else 3
CustomTrainingOperator = TrainingOperator.from_creators(
model_creator=ResNet18,
optimizer_creator=optimizer_creator,
data_creator=cifar_creator,
loss_creator=nn.CrossEntropyLoss)
TorchTrainable = TorchTrainer.as_trainable(
training_operator_cls=CustomTrainingOperator,
initialization_hook=initialization_hook,
num_workers=num_training_workers,
config={
"test_mode": args.smoke_test,
BATCH_SIZE: 128 * num_training_workers,
},
use_gpu=not args.smoke_test)
pbt_scheduler = PopulationBasedTraining(
time_attr="training_iteration",
metric="val_loss",
mode="min",
perturbation_interval=1,
hyperparam_mutations={
# distribution for resampling
"lr": lambda: np.random.uniform(0.001, 1),
# allow perturbations within this set of categorical values
"momentum": [0.8, 0.9, 0.99],
})
reporter = CLIReporter()
reporter.add_metric_column("val_loss", "loss")
reporter.add_metric_column("val_accuracy", "acc")
analysis = tune.run(
TorchTrainable,
num_samples=4,
config={
"lr": tune.choice([0.001, 0.01, 0.1]),
"momentum": 0.8,
"head_location": None,
"worker_locations": None
},
max_failures=-1, # used for fault tolerance
checkpoint_freq=2, # used for fault tolerance
progress_reporter=reporter,
scheduler=pbt_scheduler,
callbacks=[FailureInjectorCallback()],
queue_trials=True,
stop={"training_iteration": 1} if args.smoke_test else None)
print(analysis.get_best_config(metric="val_loss", mode="min"))
release/long_running_distributed_tests/workloads/pytorch_pbt_failure.py
-138
View File
@@ -1,138 +0,0 @@
import argparse
import numpy as np
import os
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Subset
from torchvision.datasets import CIFAR10
import torchvision.transforms as transforms
import ray
from ray import tune
from ray.tune import CLIReporter
from ray.tune.schedulers import PopulationBasedTraining
from ray.tune.utils.util import merge_dicts
from ray.tune.utils.mock import FailureInjectorCallback
from ray.util.sgd.torch import TorchTrainer, TrainingOperator
from ray.util.sgd.torch.resnet import ResNet18
from ray.util.sgd.utils import BATCH_SIZE
parser = argparse.ArgumentParser()
parser.add_argument(
"--smoke-test",
action="store_true",
default=False,
help="Finish quickly for training.")
args = parser.parse_args()
def initialization_hook():
# Need this for avoiding a connection restart issue on AWS.
os.environ["NCCL_SOCKET_IFNAME"] = "^docker0,lo"
os.environ["NCCL_LL_THRESHOLD"] = "0"
# set the below if needed
# print("NCCL DEBUG SET")
# os.environ["NCCL_DEBUG"] = "INFO"
def cifar_creator(config):
transform_train = transforms.Compose([
transforms.RandomCrop(32, padding=4),
transforms.RandomHorizontalFlip(),
transforms.ToTensor(),
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010)),
]) # meanstd transformation
transform_test = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.4914, 0.4822, 0.4465),
(0.2023, 0.1994, 0.2010)),
])
train_dataset = CIFAR10(
root="~/data", train=True, download=True, transform=transform_train)
validation_dataset = CIFAR10(
root="~/data", train=False, download=False, transform=transform_test)
if config.get("test_mode"):
train_dataset = Subset(train_dataset, list(range(64)))
validation_dataset = Subset(validation_dataset, list(range(64)))
train_loader = DataLoader(
train_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
validation_loader = DataLoader(
validation_dataset, batch_size=config[BATCH_SIZE], num_workers=2)
return train_loader, validation_loader
def optimizer_creator(model, config):
"""Returns optimizer"""
return torch.optim.SGD(
model.parameters(),
lr=config.get("lr", 0.1),
momentum=config.get("momentum", 0.9))
ray.init(address="auto" if not args.smoke_test else None, log_to_driver=True)
num_training_workers = 1 if args.smoke_test else 3
CustomTrainingOperator = TrainingOperator.from_creators(
model_creator=ResNet18,
optimizer_creator=optimizer_creator,
data_creator=cifar_creator,
loss_creator=nn.CrossEntropyLoss)
TorchTrainable = TorchTrainer.as_trainable(
training_operator_cls=CustomTrainingOperator,
initialization_hook=initialization_hook,
num_workers=num_training_workers,
config={
"test_mode": args.smoke_test,
BATCH_SIZE: 128 * num_training_workers,
},
use_gpu=not args.smoke_test)
class NoFaultToleranceTrainable(TorchTrainable):
def _train(self):
train_stats = self.trainer.train(max_retries=0, profile=True)
validation_stats = self.trainer.validate(profile=True)
stats = merge_dicts(train_stats, validation_stats)
return stats
pbt_scheduler = PopulationBasedTraining(
time_attr="training_iteration",
metric="val_loss",
mode="min",
perturbation_interval=1,
hyperparam_mutations={
# distribution for resampling
"lr": lambda: np.random.uniform(0.001, 1),
# allow perturbations within this set of categorical values
"momentum": [0.8, 0.9, 0.99],
})
reporter = CLIReporter()
reporter.add_metric_column("val_loss", "loss")
reporter.add_metric_column("val_accuracy", "acc")
analysis = tune.run(
NoFaultToleranceTrainable,
num_samples=4,
config={
"lr": tune.choice([0.001, 0.01, 0.1]),
"momentum": 0.8,
"head_location": None,
"worker_locations": None
},
max_failures=-1, # used for fault tolerance
checkpoint_freq=2, # used for fault tolerance
progress_reporter=reporter,
scheduler=pbt_scheduler,
callbacks=[FailureInjectorCallback()],
queue_trials=True,
stop={"training_iteration": 1} if args.smoke_test else None)
print(analysis.get_best_config(metric="val_loss", mode="min"))
release/long_running_distributed_tests/workloads/pytorch_pbt_failure.py
Symlink
+1
View File
@@ -0,0 +1 @@
../../../python/ray/util/sgd/torch/examples/pytorch_pbt_failure.py
src/ray/protobuf/gcs.proto
+2
View File
@@ -325,6 +325,8 @@ message ResourcesData {
ResourceLoad resource_load_by_shape = 7;
// Whether this node manager is requesting global GC.
bool should_global_gc = 8;
// IP address of the node.
string node_manager_address = 9;
}
message ResourceUsageBatchData {
src/ray/raylet/node_manager.cc
+1
View File
@@ -452,6 +452,7 @@ void NodeManager::Heartbeat() {
void NodeManager::ReportResourceUsage() {
auto resources_data = std::make_shared<rpc::ResourcesData>();
resources_data->set_node_id(self_node_id_.Binary());
resources_data->set_node_manager_address(initial_config_.node_manager_address);
// Update local chche from gcs remote cache, this is needed when gcs restart.
// We should always keep the cache view consistent.
cluster_resource_scheduler_->UpdateLastResourceUsage(