wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-09-11 12:43:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Clean up top level Ray dir (#5404)

Browse Source
This commit is contained in:
Eric Liang
2019-08-08 23:35:55 -07:00
committed by GitHub
parent d9b45cceec
commit 1a8fa5d2fa
60 changed files with 17 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/examples/cython/cython_examples/__init__.py
+31
View File
@@ -0,0 +1,31 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from .cython_simple import simple_func, fib, fib_int, \
fib_cpdef, fib_cdef, simple_class
from .masked_log import masked_log
from .cython_blas import \
compute_self_corr_for_voxel_sel, \
compute_kernel_matrix, \
compute_single_self_corr_syrk, \
compute_single_self_corr_gemm, \
compute_corr_vectors, \
compute_single_matrix_multiplication
__all__ = [
"simple_func",
"fib",
"fib_int",
"fib_cpdef",
"fib_cdef",
"simple_class",
"masked_log",
"compute_self_corr_for_voxel_sel",
"compute_kernel_matrix",
"compute_single_self_corr_syrk",
"compute_single_self_corr_gemm",
"compute_corr_vectors",
"compute_single_matrix_multiplication"
]
doc/examples/cython/cython_examples/cython_blas.pyx
+571
View File
@@ -0,0 +1,571 @@
#!python
# cython: embedsignature=True, binding=True
# Copyright 2016 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Authors: Yida Wang
# (Intel Labs), 2016
cimport scipy.linalg.cython_blas as blas
def compute_self_corr_for_voxel_sel(py_trans_a, py_trans_b, py_m, py_n, py_k,
py_alpha, py_a, py_lda, int py_start_voxel,
py_b, py_ldb, py_beta, py_c, py_ldc,
int py_start_epoch):
""" use blas API sgemm wrapped by scipy to compute correlation
This method is limited to process self-correlation.
The blas APIs process matrices in column-major,
but our matrices are in row-major,
so we play the transpose trick here, i.e. A*B=(B^T*A^T)^T.
The resulting matrix in shape [num_assigned_voxels, num_voxels]
is stored in an alternate way to make sure that
the correlation vectors of the same voxel stored continuously
Parameters
----------
py_trans_a: str
do transpose or not for the first matrix A
py_trans_b: str
do transpose or not for the first matrix B
py_m: int
the row of the resulting matrix C
in our case, is num_voxels
py_n: int
the column of the resulting matrix C
in our case, is num_assigned_voxels
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
py_alpha: float
the weight applied to the first matrix A
py_a: 2D array in shape [epoch_length, num_voxels]
It is the activity data of an epoch, part 1 of the data to be
correlated with. Note that py_a can point to the same location of py_b.
py_lda: int
the stride of the first matrix A
py_start_voxel: int
the starting voxel of assigned voxels
used to locate the second matrix B
py_b: 2D array in shape [epoch_length, num_voxels]
It is the activity data of an epoch, part 2 of the data to be
correlated with. Note that py_a can point to the same location of py_b.
py_ldb: int
the stride of the second matrix B
py_beta: float
the weight applied to the resulting matrix C
py_c: 3D array in shape [num_selected_voxels, num_epochs, num_voxels]
place to store the resulting correlation values
py_ldc: int
the stride of the resulting matrix
in our case, num_voxels*num_epochs
py_start_epoch: int
the epoch over which the correlation is computed
Returns
-------
py_c: 3D array in shape [num_selected_voxels, num_epochs, num_voxels]
write the resulting correlation values in an alternate way
for the processing epoch
"""
cdef bytes by_trans_a=py_trans_a.encode()
cdef bytes by_trans_b=py_trans_b.encode()
cdef char* trans_a = by_trans_a
cdef char* trans_b = by_trans_b
cdef int M, N, K, lda, ldb, ldc
M = py_m
N = py_n
K = py_k
lda = py_lda
ldb = py_ldb
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, ::1] A
A = py_a
cdef float[:, ::1] B
B = py_b
cdef float[:, :, ::1] C
C = py_c
blas.sgemm(trans_a, trans_b, &M, &N, &K, &alpha, &A[0, 0], &lda,
&B[0, py_start_voxel], &ldb, &beta,
&C[0, py_start_epoch, 0], &ldc)
def compute_kernel_matrix(py_uplo, py_trans, py_n, py_k, py_alpha, py_a,
int py_start_voxel, py_lda,
py_beta, py_c, py_ldc):
""" use blas API syrk wrapped by scipy to compute kernel matrix of SVM
The blas APIs process matrices in column-major, but our matrices are
in row-major, so we play the transpose trick here, i.e. A*B=(B^T*A^T)^T
In SVM with linear kernel, the distance of two samples
is essentially the dot product of them.
Therefore, the kernel matrix can be obtained by matrix multiplication.
Since the kernel matrix is symmetric, ssyrk is used,
the other half of the matrix is assigned later.
In our case, the dimension of samples is much larger than
the number samples, so we proportionally shrink the values of
the kernel matrix for getting more robust alpha values in SVM iteration.
Parameters
----------
py_uplo: str
getting the upper or lower triangle of the matrix
py_trans: str
do transpose or not for the input matrix A
py_n: int
the row and column of the resulting matrix C
in our case, is num_epochs
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
in our case, is num_voxels
py_alpha: float
the weight applied to the input matrix A
py_a: 3D array in shape [num_assigned_voxels, num_epochs, num_voxels]
in our case the normalized correlation values of a voxel
py_start_voxel: int
the processed voxel
used to locate the input matrix A
py_lda: int
the stride of the input matrix A
py_beta: float
the weight applied to the resulting matrix C
py_c: 2D array in shape [num_epochs, num_epochs]
place to store the resulting kernel matrix
py_ldc: int
the stride of the resulting matrix
Returns
-------
py_c: 2D array in shape [num_epochs, num_epochs]
write the resulting kernel_matrix
for the processing voxel
"""
cdef bytes by_uplo=py_uplo.encode()
cdef bytes by_trans=py_trans.encode()
cdef char* uplo = by_uplo
cdef char* trans = by_trans
cdef int N, K, lda, ldc
N = py_n
K = py_k
lda = py_lda
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, :, ::1] A
A = py_a
cdef float[:, ::1] C
C = py_c
blas.ssyrk(uplo, trans, &N, &K, &alpha, &A[py_start_voxel, 0, 0], &lda,
&beta, &C[0, 0], &ldc)
# complete the other half of the kernel matrix
if py_uplo == 'L':
for j in range(py_c.shape[0]):
for k in range(j):
py_c[j, k] = py_c[k, j]
else:
for j in range(py_c.shape[0]):
for k in range(j):
py_c[k, j] = py_c[j, k]
def compute_single_self_corr_syrk(py_uplo, py_trans, py_n, py_k,
py_alpha, py_a, py_lda,
py_beta, py_c, py_ldc,
int py_start_sample):
""" use blas API syrk wrapped by scipy to compute correlation matrix
This is to compute the correlation between selected voxels for
final training and classification. Since the resulting correlation
matrix is symmetric, syrk is used. However, it looks like that in most
cases, syrk performs much worse than gemm (the next function).
Here we assume that the resulting matrix is stored in a compact way,
i.e. py_ldc == py_n.
Parameters
----------
py_uplo: str
getting the upper or lower triangle of the matrix
py_trans: str
do transpose or not for the input matrix A
py_n: int
the row and column of the resulting matrix C
in our case, is num_selected_voxels
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
in our case, is num_TRs
py_alpha: float
the weight applied to the input matrix A
py_a: 2D array in shape [num_TRs, num_selected_voxels]
in our case the normalized activity values
py_lda: int
the stride of the input matrix A
py_beta: float
the weight applied to the resulting matrix C
py_c: 3D array
in shape [num_samples, num_selected_voxels, num_selected_voxels]
place to store the resulting kernel matrix
py_ldc: int
the stride of the resulting matrix
py_start_sample: int
the processed sample
used to locate the resulting matrix C
Returns
-------
py_c: 3D array
in shape [num_samples, num_selected_voxels, num_selected_voxels]
write the resulting correlation matrices
for the processed sample
"""
cdef bytes by_uplo=py_uplo.encode()
cdef bytes by_trans=py_trans.encode()
cdef char* uplo = by_uplo
cdef char* trans = by_trans
cdef int N, K, lda, ldc
N = py_n
K = py_k
lda = py_lda
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, ::1] A
A = py_a
cdef float[:, :, ::1] C
C = py_c
blas.ssyrk(uplo, trans, &N, &K, &alpha, &A[0, 0], &lda,
&beta, &C[py_start_sample, 0, 0], &ldc)
# complete the other half of the kernel matrix
if py_uplo == 'L':
for j in range(py_c.shape[1]):
for k in range(j):
py_c[py_start_sample, j, k] = py_c[py_start_sample, k, j]
else:
for j in range(py_c.shape[1]):
for k in range(j):
py_c[py_start_sample, k, j] = py_c[py_start_sample, j, k]
def compute_single_self_corr_gemm(py_trans_a, py_trans_b, py_m, py_n,
py_k, py_alpha, py_a, py_lda,
py_ldb, py_beta, py_c, py_ldc,
int py_start_sample):
""" use blas API gemm wrapped by scipy to compute correlation matrix
This is to compute the correlation between selected voxels for
final training and classification. Although the resulting correlation
matrix is symmetric, in most cases, gemm performs better than syrk.
Here we assume that the resulting matrix is stored in a compact way,
i.e. py_ldc == py_n.
Parameters
----------
py_trans_a: str
do transpose or not for the first matrix A
py_trans_b: str
do transpose or not for the first matrix B
py_m: int
the row of the resulting matrix C
in our case, is num_selected_voxels
py_n: int
the column of the resulting matrix C
in our case, is num_selected_voxels
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
in our case, is num_TRs
py_alpha: float
the weight applied to the input matrix A
py_a: 2D array in shape [num_TRs, num_selected_voxels]
in our case the normalized activity values
both multipliers are specified here as the same one
py_lda: int
the stride of the input matrix A
py_ldb: int
the stride of the input matrix B
in our case, the same as py_lda
py_beta: float
the weight applied to the resulting matrix C
py_c: 3D array
in shape [num_samples, num_selected_voxels, num_selected_voxels]
place to store the resulting kernel matrix
py_ldc: int
the stride of the resulting matrix
py_start_sample: int
the processed sample
used to locate the resulting matrix C
Returns
-------
py_c: 3D array
in shape [num_samples, num_selected_voxels, num_selected_voxels]
write the resulting correlation matrices
for the processed sample
"""
cdef bytes by_trans_a=py_trans_a.encode()
cdef bytes by_trans_b=py_trans_b.encode()
cdef char* trans_a = by_trans_a
cdef char* trans_b = by_trans_b
cdef int M, N, K, lda, ldb, ldc
M = py_m
N = py_n
K = py_k
lda = py_lda
ldb = py_ldb
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, ::1] A
A = py_a
cdef float[:, :, ::1] C
C = py_c
blas.sgemm(trans_a, trans_b, &M, &N, &K, &alpha, &A[0, 0], &lda,
&A[0, 0], &ldb, &beta, &C[py_start_sample, 0, 0], &ldc)
def compute_corr_vectors(py_trans_a, py_trans_b, py_m, py_n,
py_k, py_alpha, py_a, py_lda,
py_b, py_ldb, py_beta, py_c, py_ldc,
int py_start_voxel,
int py_start_sample):
""" use blas API gemm wrapped by scipy to construct a correlation vector
The correlation vector is essentially correlation matrices computed
from two activity matrices. It will be placed in the corresponding place
of the resulting correlation data set.
The blas APIs process matrices in column-major,
but our matrices are in row-major, so we play the transpose trick here,
i.e. A*B=(B^T*A^T)^T
py_trans_a: str
do transpose or not for the first matrix A
py_trans_b: str
do transpose or not for the first matrix B
py_m: int
the row of the resulting matrix C
py_n: int
the column of the resulting matrix C
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
py_alpha: float
the weight applied to the input matrix A
py_a: 2D array
py_lda: int
the stride of the input matrix A
py_b: 2D array
py_ldb: int
the stride of the input matrix B
py_beta: float
the weight applied to the resulting matrix C
py_c: 2D array
in shape [py_m, py_n] of column-major
in fact it is
in shape [py_n, py_m] of row-major
py_ldc: int
the stride of the resulting matrix
py_start_voxel: int
the starting voxel of assigned voxels
used to locate the second matrix B
py_start_sample: int
the processed sample
used to locate the resulting matrix C
Returns
-------
py_c: 2D array
in shape [py_m, py_n] of column-major
write the resulting matrix to the place indicated by py_start_sample
"""
cdef bytes by_trans_a=py_trans_a.encode()
cdef bytes by_trans_b=py_trans_b.encode()
cdef char* trans_a = by_trans_a
cdef char* trans_b = by_trans_b
cdef int M, N, K, lda, ldb, ldc
M = py_m
N = py_n
K = py_k
lda = py_lda
ldb = py_ldb
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, ::1] A
A = py_a
cdef float[:, ::1] B
B = py_b
cdef float[:, :, ::1] C
C = py_c
blas.sgemm(trans_a, trans_b, &M, &N, &K, &alpha, &A[0, 0], &lda,
&B[0, py_start_voxel], &ldb, &beta,
&C[py_start_sample, 0, 0], &ldc)
def compute_single_matrix_multiplication(py_trans_a, py_trans_b, py_m, py_n,
py_k, py_alpha, py_a, py_lda,
py_b, py_ldb, py_beta, py_c, py_ldc):
""" use blas API gemm wrapped by scipy to do matrix multiplication
This is to compute the matrix multiplication.
The blas APIs process matrices in column-major,
but our matrices are in row-major, so we play the transpose trick here,
i.e. A*B=(B^T*A^T)^T
Parameters
----------
py_trans_a: str
do transpose or not for the first matrix A
py_trans_b: str
do transpose or not for the first matrix B
py_m: int
the row of the resulting matrix C
py_n: int
the column of the resulting matrix C
py_k: int
the collapsed dimension of the multiplying matrices
i.e. the column of the first matrix after transpose if necessary
the row of the second matrix after transpose if necessary
py_alpha: float
the weight applied to the input matrix A
py_a: 2D array
py_lda: int
the stride of the input matrix A
py_b: 2D array
py_ldb: int
the stride of the input matrix B
py_beta: float
the weight applied to the resulting matrix C
py_c: 2D array
in shape [py_m, py_n] of column-major
in fact it is
in shape [py_n, py_m] of row-major
py_ldc: int
the stride of the resulting matrix
Returns
-------
py_c: 2D array
in shape [py_m, py_n] of column-major
write the resulting matrix
"""
cdef bytes by_trans_a=py_trans_a.encode()
cdef bytes by_trans_b=py_trans_b.encode()
cdef char* trans_a = by_trans_a
cdef char* trans_b = by_trans_b
cdef int M, N, K, lda, ldb, ldc
M = py_m
N = py_n
K = py_k
lda = py_lda
ldb = py_ldb
ldc = py_ldc
cdef float alpha, beta
alpha = py_alpha
beta = py_beta
cdef float[:, ::1] A
A = py_a
cdef float[:, ::1] B
B = py_b
cdef float[:, ::1] C
C = py_c
blas.sgemm(trans_a, trans_b, &M, &N, &K, &alpha, &A[0, 0], &lda,
&B[0, 0], &ldb, &beta, &C[0, 0], &ldc)
doc/examples/cython/cython_examples/cython_simple.pyx
+48
View File
@@ -0,0 +1,48 @@
#!python
# cython: embedsignature=True, binding=True
def simple_func(x, y, z):
return x + y + z
# Cython code directly callable from Python
def fib(n):
if n < 2:
return n
return fib(n-2) + fib(n-1)
# Typed Cython code
def fib_int(int n):
if n < 2:
return n
return fib_int(n-2) + fib_int(n-1)
# Cython-Python code
cpdef fib_cpdef(int n):
if n < 2:
return n
return fib_cpdef(n-2) + fib_cpdef(n-1)
# C code
def fib_cdef(int n):
return fib_in_c(n)
cdef int fib_in_c(int n):
if n < 2:
return n
return fib_in_c(n-2) + fib_in_c(n-1)
# Simple class
class simple_class(object):
def __init__(self):
self.value = 0
def increment(self):
self.value += 1
return self.value
doc/examples/cython/cython_examples/masked_log.pyx
+48
View File
@@ -0,0 +1,48 @@
#!python
# cython: embedsignature=True, binding=True
# Copyright 2016 Intel Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from libc.math cimport log
import numpy as np
cimport numpy as np
def masked_log(x):
"""Compute natural logarithm while accepting nonpositive input
For nonpositive elements, return -inf.
Modified slightly from the original BrainIAK code to support
Python 2.
Parameters
----------
x: ndarray[T]
Returns
-------
ndarray[Union[T, np.float64]]
"""
y = np.empty(x.shape, dtype=np.float64)
lim = x.shape[0]
for i in range(lim):
if x[i] <= 0:
y[i] = float("-inf")
else:
y[i] = log(x[i])
return y
doc/examples/cython/cython_main.py
+120
View File
@@ -0,0 +1,120 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import ray
import click
import inspect
import numpy as np
import cython_examples as cyth
def run_func(func, *args, **kwargs):
"""Helper function for running examples"""
ray.init()
func = ray.remote(func)
# NOTE: kwargs not allowed for now
result = ray.get(func.remote(*args))
# Inspect the stack to get calling example
caller = inspect.stack()[1][3]
print("%s: %s" % (caller, str(result)))
return result
@click.group(context_settings={"help_option_names": ["-h", "--help"]})
def cli():
"""Working with Cython actors and functions in Ray"""
@cli.command()
def example1():
"""Cython def function"""
run_func(cyth.simple_func, 1, 2, 3)
@cli.command()
def example2():
"""Cython def function, recursive"""
run_func(cyth.fib, 10)
@cli.command()
def example3():
"""Cython def function, built-in typed parameter"""
# NOTE: Cython will attempt to cast argument to correct type
# NOTE: Floats will be cast to int, but string, for example will error
run_func(cyth.fib_int, 10)
@cli.command()
def example4():
"""Cython cpdef function"""
run_func(cyth.fib_cpdef, 10)
@cli.command()
def example5():
"""Cython wrapped cdef function"""
# NOTE: cdef functions are not exposed to Python
run_func(cyth.fib_cdef, 10)
@cli.command()
def example6():
"""Cython simple class"""
ray.init()
cls = ray.remote(cyth.simple_class)
a1 = cls.remote()
a2 = cls.remote()
result1 = ray.get(a1.increment.remote())
result2 = ray.get(a2.increment.remote())
print(result1, result2)
@cli.command()
def example7():
"""Cython with function from BrainIAK (masked log)"""
run_func(cyth.masked_log, np.array([-1.0, 0.0, 1.0, 2.0]))
@cli.command()
def example8():
"""Cython with blas. NOTE: requires scipy"""
# See cython_blas.pyx for argument documentation
mat = np.array([[[2.0, 2.0], [2.0, 2.0]], [[2.0, 2.0], [2.0, 2.0]]],
dtype=np.float32)
result = np.zeros((2, 2), np.float32, order="C")
run_func(cyth.compute_kernel_matrix,
"L",
"T",
2,
2,
1.0,
mat,
0,
2,
1.0,
result,
2
)
if __name__ == "__main__":
cli()
doc/examples/cython/setup.py
+35
View File
@@ -0,0 +1,35 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import os
from setuptools import setup
from Cython.Build import cythonize
import numpy
pkg_dir = "cython_examples"
modules = ["cython_simple.pyx", "masked_log.pyx"]
install_requires = ["cython", "numpy"]
include_dirs = [numpy.get_include()]
# TODO: Need scipy to run BrainIAK example, but don't want to add additional
# dependencies
try:
import scipy # noqa
modules.append("cython_blas.pyx")
install_requires.append("scipy")
except ImportError as e: # noqa
pass
modules = [os.path.join(pkg_dir, module) for module in modules]
setup(
name=pkg_dir,
version="0.0.1",
description="Cython examples for Ray",
packages=[pkg_dir],
ext_modules=cythonize(modules),
install_requires=install_requires,
include_dirs=include_dirs
)
doc/examples/hyperopt/hyperopt_adaptive.py
+154
View File
@@ -0,0 +1,154 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
from collections import defaultdict
import numpy as np
import ray
from tensorflow.examples.tutorials.mnist import input_data
import objective
parser = argparse.ArgumentParser(description="Run the hyperparameter "
"optimization example.")
parser.add_argument("--num-starting-segments", default=5, type=int,
help="The number of training segments to start in "
"parallel.")
parser.add_argument("--num-segments", default=10, type=int,
help="The number of additional training segments to "
"perform.")
parser.add_argument("--steps-per-segment", default=20, type=int,
help="The number of steps of training to do per training "
"segment.")
parser.add_argument("--redis-address", default=None, type=str,
help="The Redis address of the cluster.")
if __name__ == "__main__":
args = parser.parse_args()
ray.init(redis_address=args.redis_address)
# The number of training passes over the dataset to use for network.
steps = args.steps_per_segment
# Load the mnist data and turn the data into remote objects.
print("Downloading the MNIST dataset. This may take a minute.")
mnist = input_data.read_data_sets("MNIST_data", one_hot=True)
train_images = ray.put(mnist.train.images)
train_labels = ray.put(mnist.train.labels)
validation_images = ray.put(mnist.validation.images)
validation_labels = ray.put(mnist.validation.labels)
# Keep track of the accuracies that we've seen at different numbers of
# iterations.
accuracies_by_num_steps = defaultdict(lambda: [])
# Define a method to determine if an experiment looks promising or not.
def is_promising(experiment_info):
accuracies = experiment_info["accuracies"]
total_num_steps = experiment_info["total_num_steps"]
comparable_accuracies = accuracies_by_num_steps[total_num_steps]
if len(comparable_accuracies) == 0:
if len(accuracies) == 1:
# This means that we haven't seen anything finish yet, so keep
# running this experiment.
return True
else:
# The experiment is promising if the second half of the
# accuracies are better than the first half of the accuracies.
return (np.mean(accuracies[:len(accuracies) // 2]) <
np.mean(accuracies[len(accuracies) // 2:]))
# Otherwise, continue running the experiment if it is in the top half
# of experiments we've seen so far at this point in time.
return np.mean(accuracy > np.array(comparable_accuracies)) > 0.5
# Keep track of all of the experiment segments that we're running. This
# dictionary uses the object ID of the experiment as the key.
experiment_info = {}
# Keep track of the curently running experiment IDs.
remaining_ids = []
# Keep track of the best hyperparameters and the best accuracy.
best_hyperparameters = None
best_accuracy = 0
# A function for generating random hyperparameters.
def generate_hyperparameters():
return {"learning_rate": 10 ** np.random.uniform(-5, 5),
"batch_size": np.random.randint(1, 100),
"dropout": np.random.uniform(0, 1),
"stddev": 10 ** np.random.uniform(-5, 5)}
# Launch some initial experiments.
for _ in range(args.num_starting_segments):
hyperparameters = generate_hyperparameters()
experiment_id = objective.train_cnn_and_compute_accuracy.remote(
hyperparameters, steps, train_images, train_labels,
validation_images, validation_labels)
experiment_info[experiment_id] = {"hyperparameters": hyperparameters,
"total_num_steps": steps,
"accuracies": []}
remaining_ids.append(experiment_id)
for _ in range(args.num_segments):
# Wait for a segment of an experiment to finish.
ready_ids, remaining_ids = ray.wait(remaining_ids, num_returns=1)
experiment_id = ready_ids[0]
# Get the accuracy and the weights.
accuracy, weights = ray.get(experiment_id)
# Update the experiment info.
previous_info = experiment_info[experiment_id]
previous_info["accuracies"].append(accuracy)
# Update the best accuracy and best hyperparameters.
if accuracy > best_accuracy:
best_hyperparameters = previous_info["hyperparameters"]
best_accuracy = accuracy
if is_promising(previous_info):
# If the experiment still looks promising, then continue running
# it.
print("Continuing to run the experiment with hyperparameters {}."
.format(previous_info["hyperparameters"]))
new_hyperparameters = previous_info["hyperparameters"]
new_info = {"hyperparameters": new_hyperparameters,
"total_num_steps": (previous_info["total_num_steps"] +
steps),
"accuracies": previous_info["accuracies"][:]}
starting_weights = weights
else:
# If the experiment does not look promising, start a new
# experiment.
print("Ending the experiment with hyperparameters {}."
.format(previous_info["hyperparameters"]))
new_hyperparameters = generate_hyperparameters()
new_info = {"hyperparameters": new_hyperparameters,
"total_num_steps": steps,
"accuracies": []}
starting_weights = None
# Start running the next segment.
new_experiment_id = objective.train_cnn_and_compute_accuracy.remote(
new_hyperparameters, steps, train_images, train_labels,
validation_images, validation_labels, weights=starting_weights)
experiment_info[new_experiment_id] = new_info
remaining_ids.append(new_experiment_id)
# Update the set of all accuracies that we've seen.
accuracies_by_num_steps[previous_info["total_num_steps"]].append(
accuracy)
# Record the best performing set of hyperparameters.
print("""Best accuracy was {:.3} with
learning_rate: {:.2}
batch_size: {}
dropout: {:.2}
stddev: {:.2}
""".format(100 * best_accuracy,
best_hyperparameters["learning_rate"],
best_hyperparameters["batch_size"],
best_hyperparameters["dropout"],
best_hyperparameters["stddev"]))
doc/examples/hyperopt/hyperopt_simple.py
+100
View File
@@ -0,0 +1,100 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import numpy as np
import ray
import argparse
from tensorflow.examples.tutorials.mnist import input_data
import objective
parser = argparse.ArgumentParser(description="Run the hyperparameter "
"optimization example.")
parser.add_argument("--trials", default=2, type=int,
help="The number of random trials to do.")
parser.add_argument("--steps", default=10, type=int,
help="The number of steps of training to do per network.")
parser.add_argument("--redis-address", default=None, type=str,
help="The Redis address of the cluster.")
if __name__ == "__main__":
args = parser.parse_args()
ray.init(redis_address=args.redis_address)
# The number of sets of random hyperparameters to try.
trials = args.trials
# The number of training passes over the dataset to use for network.
steps = args.steps
# Load the mnist data and turn the data into remote objects.
print("Downloading the MNIST dataset. This may take a minute.")
mnist = input_data.read_data_sets("MNIST_data", one_hot=True)
train_images = ray.put(mnist.train.images)
train_labels = ray.put(mnist.train.labels)
validation_images = ray.put(mnist.validation.images)
validation_labels = ray.put(mnist.validation.labels)
# Keep track of the best hyperparameters and the best accuracy.
best_hyperparameters = None
best_accuracy = 0
# This list holds the object IDs for all of the experiments that we have
# launched and that have not yet been processed.
remaining_ids = []
# This is a dictionary mapping the object ID of an experiment to the
# hyerparameters used for that experiment.
hyperparameters_mapping = {}
# A function for generating random hyperparameters.
def generate_hyperparameters():
return {"learning_rate": 10 ** np.random.uniform(-5, 5),
"batch_size": np.random.randint(1, 100),
"dropout": np.random.uniform(0, 1),
"stddev": 10 ** np.random.uniform(-5, 5)}
# Randomly generate some hyperparameters, and launch a task for each set.
for i in range(trials):
hyperparameters = generate_hyperparameters()
accuracy_id = objective.train_cnn_and_compute_accuracy.remote(
hyperparameters, steps, train_images, train_labels,
validation_images, validation_labels)
remaining_ids.append(accuracy_id)
# Keep track of which hyperparameters correspond to this experiment.
hyperparameters_mapping[accuracy_id] = hyperparameters
# Fetch and print the results of the tasks in the order that they complete.
for i in range(trials):
# Use ray.wait to get the object ID of the first task that completes.
ready_ids, remaining_ids = ray.wait(remaining_ids)
# Process the output of this task.
result_id = ready_ids[0]
hyperparameters = hyperparameters_mapping[result_id]
accuracy, _ = ray.get(result_id)
print("""We achieve accuracy {:.3}% with
learning_rate: {:.2}
batch_size: {}
dropout: {:.2}
stddev: {:.2}
""".format(100 * accuracy,
hyperparameters["learning_rate"],
hyperparameters["batch_size"],
hyperparameters["dropout"],
hyperparameters["stddev"]))
if accuracy > best_accuracy:
best_hyperparameters = hyperparameters
best_accuracy = accuracy
# Record the best performing set of hyperparameters.
print("""Best accuracy over {} trials was {:.3} with
learning_rate: {:.2}
batch_size: {}
dropout: {:.2}
stddev: {:.2}
""".format(trials, 100 * best_accuracy,
best_hyperparameters["learning_rate"],
best_hyperparameters["batch_size"],
best_hyperparameters["dropout"],
best_hyperparameters["stddev"]))
doc/examples/hyperopt/objective.py
+127
View File
@@ -0,0 +1,127 @@
# Most of the tensorflow code is adapted from Tensorflow's tutorial on using
# CNNs to train MNIST
# https://www.tensorflow.org/versions/r0.9/tutorials/mnist/pros/index.html#build-a-multilayer-convolutional-network. # noqa: E501
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import tensorflow as tf
import ray
import ray.experimental.tf_utils
def get_batch(data, batch_index, batch_size):
# This method currently drops data when num_data is not divisible by
# batch_size.
num_data = data.shape[0]
num_batches = num_data // batch_size
batch_index %= num_batches
return data[(batch_index * batch_size):((batch_index + 1) * batch_size)]
def weight(shape, stddev):
initial = tf.truncated_normal(shape, stddev=stddev)
return tf.Variable(initial)
def bias(shape):
initial = tf.constant(0.1, shape=shape)
return tf.Variable(initial)
def conv2d(x, W):
return tf.nn.conv2d(x, W, strides=[1, 1, 1, 1], padding="SAME")
def max_pool_2x2(x):
return tf.nn.max_pool(
x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding="SAME")
def cnn_setup(x, y, keep_prob, lr, stddev):
first_hidden = 32
second_hidden = 64
fc_hidden = 1024
W_conv1 = weight([5, 5, 1, first_hidden], stddev)
B_conv1 = bias([first_hidden])
x_image = tf.reshape(x, [-1, 28, 28, 1])
h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + B_conv1)
h_pool1 = max_pool_2x2(h_conv1)
W_conv2 = weight([5, 5, first_hidden, second_hidden], stddev)
b_conv2 = bias([second_hidden])
h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2) + b_conv2)
h_pool2 = max_pool_2x2(h_conv2)
W_fc1 = weight([7 * 7 * second_hidden, fc_hidden], stddev)
b_fc1 = bias([fc_hidden])
h_pool2_flat = tf.reshape(h_pool2, [-1, 7 * 7 * second_hidden])
h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, W_fc1) + b_fc1)
h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)
W_fc2 = weight([fc_hidden, 10], stddev)
b_fc2 = bias([10])
y_conv = tf.nn.softmax(tf.matmul(h_fc1_drop, W_fc2) + b_fc2)
cross_entropy = tf.reduce_mean(
-tf.reduce_sum(y * tf.log(y_conv), reduction_indices=[1]))
correct_pred = tf.equal(tf.argmax(y_conv, 1), tf.argmax(y, 1))
return (tf.train.AdamOptimizer(lr).minimize(cross_entropy),
tf.reduce_mean(tf.cast(correct_pred, tf.float32)), cross_entropy)
# Define a remote function that takes a set of hyperparameters as well as the
# data, consructs and trains a network, and returns the validation accuracy.
@ray.remote
def train_cnn_and_compute_accuracy(params,
steps,
train_images,
train_labels,
validation_images,
validation_labels,
weights=None):
# Extract the hyperparameters from the params dictionary.
learning_rate = params["learning_rate"]
batch_size = params["batch_size"]
keep = 1 - params["dropout"]
stddev = params["stddev"]
# Create the network and related variables.
with tf.Graph().as_default():
# Create the input placeholders for the network.
x = tf.placeholder(tf.float32, shape=[None, 784])
y = tf.placeholder(tf.float32, shape=[None, 10])
keep_prob = tf.placeholder(tf.float32)
# Create the network.
train_step, accuracy, loss = cnn_setup(x, y, keep_prob, learning_rate,
stddev)
# Do the training and evaluation.
with tf.Session() as sess:
# Use the TensorFlowVariables utility. This is only necessary if we
# want to set and get the weights.
variables = ray.experimental.tf_utils.TensorFlowVariables(
loss, sess)
# Initialize the network weights.
sess.run(tf.global_variables_initializer())
# If some network weights were passed in, set those.
if weights is not None:
variables.set_weights(weights)
# Do some steps of training.
for i in range(1, steps + 1):
# Fetch the next batch of data.
image_batch = get_batch(train_images, i, batch_size)
label_batch = get_batch(train_labels, i, batch_size)
# Do one step of training.
sess.run(
train_step,
feed_dict={
x: image_batch,
y: label_batch,
keep_prob: keep
})
# Training is done, so compute the validation accuracy and the
# current weights and return.
totalacc = accuracy.eval(feed_dict={
x: validation_images,
y: validation_labels,
keep_prob: 1.0
})
new_weights = variables.get_weights()
return float(totalacc), new_weights
doc/examples/lbfgs/driver.py
+150
View File
@@ -0,0 +1,150 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import numpy as np
import os
import scipy.optimize
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import ray
import ray.experimental.tf_utils
class LinearModel(object):
"""Simple class for a one layer neural network.
Note that this code does not initialize the network weights. Instead
weights are set via self.variables.set_weights.
Example:
net = LinearModel([10, 10])
weights = [np.random.normal(size=[10, 10]),
np.random.normal(size=[10])]
variable_names = [v.name for v in net.variables]
net.variables.set_weights(dict(zip(variable_names, weights)))
Attributes:
x (tf.placeholder): Input vector.
w (tf.Variable): Weight matrix.
b (tf.Variable): Bias vector.
y_ (tf.placeholder): Input result vector.
cross_entropy (tf.Operation): Final layer of network.
cross_entropy_grads (tf.Operation): Gradient computation.
sess (tf.Session): Session used for training.
variables (TensorFlowVariables): Extracted variables and methods to
manipulate them.
"""
def __init__(self, shape):
"""Creates a LinearModel object."""
x = tf.placeholder(tf.float32, [None, shape[0]])
w = tf.Variable(tf.zeros(shape))
b = tf.Variable(tf.zeros(shape[1]))
self.x = x
self.w = w
self.b = b
y = tf.nn.softmax(tf.matmul(x, w) + b)
y_ = tf.placeholder(tf.float32, [None, shape[1]])
self.y_ = y_
cross_entropy = tf.reduce_mean(
-tf.reduce_sum(y_ * tf.log(y), reduction_indices=[1]))
self.cross_entropy = cross_entropy
self.cross_entropy_grads = tf.gradients(cross_entropy, [w, b])
self.sess = tf.Session()
# In order to get and set the weights, we pass in the loss function to
# Ray's TensorFlowVariables to automatically create methods to modify
# the weights.
self.variables = ray.experimental.tf_utils.TensorFlowVariables(
cross_entropy, self.sess)
def loss(self, xs, ys):
"""Computes the loss of the network."""
return float(
self.sess.run(
self.cross_entropy, feed_dict={
self.x: xs,
self.y_: ys
}))
def grad(self, xs, ys):
"""Computes the gradients of the network."""
return self.sess.run(
self.cross_entropy_grads, feed_dict={
self.x: xs,
self.y_: ys
})
@ray.remote
class NetActor(object):
def __init__(self, xs, ys):
os.environ["CUDA_VISIBLE_DEVICES"] = ""
with tf.device("/cpu:0"):
self.net = LinearModel([784, 10])
self.xs = xs
self.ys = ys
# Compute the loss on a batch of data.
def loss(self, theta):
net = self.net
net.variables.set_flat(theta)
return net.loss(self.xs, self.ys)
# Compute the gradient of the loss on a batch of data.
def grad(self, theta):
net = self.net
net.variables.set_flat(theta)
gradients = net.grad(self.xs, self.ys)
return np.concatenate([g.flatten() for g in gradients])
def get_flat_size(self):
return self.net.variables.get_flat_size()
# Compute the loss on the entire dataset.
def full_loss(theta):
theta_id = ray.put(theta)
loss_ids = [actor.loss.remote(theta_id) for actor in actors]
return sum(ray.get(loss_ids))
# Compute the gradient of the loss on the entire dataset.
def full_grad(theta):
theta_id = ray.put(theta)
grad_ids = [actor.grad.remote(theta_id) for actor in actors]
# The float64 conversion is necessary for use with fmin_l_bfgs_b.
return sum(ray.get(grad_ids)).astype("float64")
if __name__ == "__main__":
ray.init()
# From the perspective of scipy.optimize.fmin_l_bfgs_b, full_loss is simply
# a function which takes some parameters theta, and computes a loss.
# Similarly, full_grad is a function which takes some parameters theta, and
# computes the gradient of the loss. Internally, these functions use Ray to
# distribute the computation of the loss and the gradient over the data
# that is represented by the remote object IDs x_batches and y_batches and
# which is potentially distributed over a cluster. However, these details
# are hidden from scipy.optimize.fmin_l_bfgs_b, which simply uses it to run
# the L-BFGS algorithm.
# Load the mnist data and turn the data into remote objects.
print("Downloading the MNIST dataset. This may take a minute.")
mnist = input_data.read_data_sets("MNIST_data", one_hot=True)
num_batches = 10
batch_size = mnist.train.num_examples // num_batches
batches = [mnist.train.next_batch(batch_size) for _ in range(num_batches)]
print("Putting MNIST in the object store.")
actors = [NetActor.remote(xs, ys) for (xs, ys) in batches]
# Initialize the weights for the network to the vector of all zeros.
dim = ray.get(actors[0].get_flat_size.remote())
theta_init = 1e-2 * np.random.normal(size=dim)
# Use L-BFGS to minimize the loss function.
print("Running L-BFGS.")
result = scipy.optimize.fmin_l_bfgs_b(
full_loss, theta_init, maxiter=10, fprime=full_grad, disp=True)
doc/examples/newsreader/server.py
+85
View File
@@ -0,0 +1,85 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import atoma
from flask import Flask, jsonify, request
from flask_cors import CORS
import requests
import sqlite3
import ray
@ray.remote
class NewsServer(object):
def __init__(self):
self.conn = sqlite3.connect("newsreader.db")
c = self.conn.cursor()
c.execute("""CREATE TABLE IF NOT EXISTS news
(title text, link text,
description text, published timestamp,
feed url, liked bool)""")
self.conn.commit()
def retrieve_feed(self, url):
response = requests.get(url)
feed = atoma.parse_rss_bytes(response.content)
items = []
c = self.conn.cursor()
for item in feed.items:
items.append({"title": item.title,
"link": item.link,
"description": item.description,
"description_text": item.description,
"pubDate": str(item.pub_date)})
c.execute("""INSERT INTO news (title, link, description,
published, feed, liked) values
(?, ?, ?, ?, ?, ?)""", (
item.title, item.link, item.description,
item.pub_date, feed.link, False))
self.conn.commit()
return {"channel": {"title": feed.title,
"link": feed.link,
"url": feed.link},
"items": items}
def like_item(self, url, is_faved):
c = self.conn.cursor()
if is_faved:
c.execute("UPDATE news SET liked = 1 WHERE link = ?", (url,))
else:
c.execute("UPDATE news SET liked = 0 WHERE link = ?", (url,))
self.conn.commit()
# instantiate the app
app = Flask(__name__)
app.config.from_object(__name__)
# enable CORS
CORS(app)
@app.route("/api", methods=["POST"])
def dispatcher():
req = request.get_json()
method_name = req["method_name"]
method_args = req["method_args"]
if hasattr(dispatcher.server, method_name):
method = getattr(dispatcher.server, method_name)
# Doing a blocking ray.get right after submitting the task
# might be bad for performance if the task is expensive.
result = ray.get(method.remote(*method_args))
return jsonify(result)
else:
return jsonify(
{"error": "method_name '" + method_name + "' not found"})
if __name__ == "__main__":
ray.init(num_cpus=2)
dispatcher.server = NewsServer.remote()
app.run()
doc/examples/parameter_server/async_parameter_server.py
+80
View File
@@ -0,0 +1,80 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
import time
import ray
import model
parser = argparse.ArgumentParser(description="Run the asynchronous parameter "
"server example.")
parser.add_argument("--num-workers", default=4, type=int,
help="The number of workers to use.")
parser.add_argument("--redis-address", default=None, type=str,
help="The Redis address of the cluster.")
@ray.remote
class ParameterServer(object):
def __init__(self, keys, values):
# These values will be mutated, so we must create a copy that is not
# backed by the object store.
values = [value.copy() for value in values]
self.weights = dict(zip(keys, values))
def push(self, keys, values):
for key, value in zip(keys, values):
self.weights[key] += value
def pull(self, keys):
return [self.weights[key] for key in keys]
@ray.remote
def worker_task(ps, worker_index, batch_size=50):
# Download MNIST.
mnist = model.download_mnist_retry(seed=worker_index)
# Initialize the model.
net = model.SimpleCNN()
keys = net.get_weights()[0]
while True:
# Get the current weights from the parameter server.
weights = ray.get(ps.pull.remote(keys))
net.set_weights(keys, weights)
# Compute an update and push it to the parameter server.
xs, ys = mnist.train.next_batch(batch_size)
gradients = net.compute_update(xs, ys)
ps.push.remote(keys, gradients)
if __name__ == "__main__":
args = parser.parse_args()
ray.init(redis_address=args.redis_address)
# Create a parameter server with some random weights.
net = model.SimpleCNN()
all_keys, all_values = net.get_weights()
ps = ParameterServer.remote(all_keys, all_values)
# Start some training tasks.
worker_tasks = [worker_task.remote(ps, i) for i in range(args.num_workers)]
# Download MNIST.
mnist = model.download_mnist_retry()
i = 0
while True:
# Get and evaluate the current model.
current_weights = ray.get(ps.pull.remote(all_keys))
net.set_weights(all_keys, current_weights)
test_xs, test_ys = mnist.test.next_batch(1000)
accuracy = net.compute_accuracy(test_xs, test_ys)
print("Iteration {}: accuracy is {}".format(i, accuracy))
i += 1
time.sleep(1)
doc/examples/parameter_server/model.py
+203
View File
@@ -0,0 +1,203 @@
# Most of the tensorflow code is adapted from Tensorflow's tutorial on using
# CNNs to train MNIST
# https://www.tensorflow.org/get_started/mnist/pros#build-a-multilayer-convolutional-network. # noqa: E501
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import time
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import ray
import ray.experimental.tf_utils
def download_mnist_retry(seed=0, max_num_retries=20):
for _ in range(max_num_retries):
try:
return input_data.read_data_sets(
"MNIST_data", one_hot=True, seed=seed)
except tf.errors.AlreadyExistsError:
time.sleep(1)
raise Exception("Failed to download MNIST.")
class SimpleCNN(object):
def __init__(self, learning_rate=1e-4):
with tf.Graph().as_default():
# Create the model
self.x = tf.placeholder(tf.float32, [None, 784])
# Define loss and optimizer
self.y_ = tf.placeholder(tf.float32, [None, 10])
# Build the graph for the deep net
self.y_conv, self.keep_prob = deepnn(self.x)
with tf.name_scope("loss"):
cross_entropy = tf.nn.softmax_cross_entropy_with_logits(
labels=self.y_, logits=self.y_conv)
self.cross_entropy = tf.reduce_mean(cross_entropy)
with tf.name_scope("adam_optimizer"):
self.optimizer = tf.train.AdamOptimizer(learning_rate)
self.train_step = self.optimizer.minimize(self.cross_entropy)
with tf.name_scope("accuracy"):
correct_prediction = tf.equal(
tf.argmax(self.y_conv, 1), tf.argmax(self.y_, 1))
correct_prediction = tf.cast(correct_prediction, tf.float32)
self.accuracy = tf.reduce_mean(correct_prediction)
self.sess = tf.Session(
config=tf.ConfigProto(
intra_op_parallelism_threads=1,
inter_op_parallelism_threads=1))
self.sess.run(tf.global_variables_initializer())
# Helper values.
self.variables = ray.experimental.tf_utils.TensorFlowVariables(
self.cross_entropy, self.sess)
self.grads = self.optimizer.compute_gradients(self.cross_entropy)
self.grads_placeholder = [(tf.placeholder(
"float", shape=grad[1].get_shape()), grad[1])
for grad in self.grads]
self.apply_grads_placeholder = self.optimizer.apply_gradients(
self.grads_placeholder)
def compute_update(self, x, y):
# TODO(rkn): Computing the weights before and after the training step
# and taking the diff is awful.
weights = self.get_weights()[1]
self.sess.run(
self.train_step,
feed_dict={
self.x: x,
self.y_: y,
self.keep_prob: 0.5
})
new_weights = self.get_weights()[1]
return [x - y for x, y in zip(new_weights, weights)]
def compute_gradients(self, x, y):
return self.sess.run(
[grad[0] for grad in self.grads],
feed_dict={
self.x: x,
self.y_: y,
self.keep_prob: 0.5
})
def apply_gradients(self, gradients):
feed_dict = {}
for i in range(len(self.grads_placeholder)):
feed_dict[self.grads_placeholder[i][0]] = gradients[i]
self.sess.run(self.apply_grads_placeholder, feed_dict=feed_dict)
def compute_accuracy(self, x, y):
return self.sess.run(
self.accuracy,
feed_dict={
self.x: x,
self.y_: y,
self.keep_prob: 1.0
})
def set_weights(self, variable_names, weights):
self.variables.set_weights(dict(zip(variable_names, weights)))
def get_weights(self):
weights = self.variables.get_weights()
return list(weights.keys()), list(weights.values())
def deepnn(x):
"""deepnn builds the graph for a deep net for classifying digits.
Args:
x: an input tensor with the dimensions (N_examples, 784), where 784 is
the number of pixels in a standard MNIST image.
Returns:
A tuple (y, keep_prob). y is a tensor of shape (N_examples, 10), with
values equal to the logits of classifying the digit into one of 10
classes (the digits 0-9). keep_prob is a scalar placeholder for the
probability of dropout.
"""
# Reshape to use within a convolutional neural net.
# Last dimension is for "features" - there is only one here, since images
# are grayscale -- it would be 3 for an RGB image, 4 for RGBA, etc.
with tf.name_scope("reshape"):
x_image = tf.reshape(x, [-1, 28, 28, 1])
# First convolutional layer - maps one grayscale image to 32 feature maps.
with tf.name_scope("conv1"):
W_conv1 = weight_variable([5, 5, 1, 32])
b_conv1 = bias_variable([32])
h_conv1 = tf.nn.relu(conv2d(x_image, W_conv1) + b_conv1)
# Pooling layer - downsamples by 2X.
with tf.name_scope("pool1"):
h_pool1 = max_pool_2x2(h_conv1)
# Second convolutional layer -- maps 32 feature maps to 64.
with tf.name_scope("conv2"):
W_conv2 = weight_variable([5, 5, 32, 64])
b_conv2 = bias_variable([64])
h_conv2 = tf.nn.relu(conv2d(h_pool1, W_conv2) + b_conv2)
# Second pooling layer.
with tf.name_scope("pool2"):
h_pool2 = max_pool_2x2(h_conv2)
# Fully connected layer 1 -- after 2 round of downsampling, our 28x28 image
# is down to 7x7x64 feature maps -- maps this to 1024 features.
with tf.name_scope("fc1"):
W_fc1 = weight_variable([7 * 7 * 64, 1024])
b_fc1 = bias_variable([1024])
h_pool2_flat = tf.reshape(h_pool2, [-1, 7 * 7 * 64])
h_fc1 = tf.nn.relu(tf.matmul(h_pool2_flat, W_fc1) + b_fc1)
# Dropout - controls the complexity of the model, prevents co-adaptation of
# features.
with tf.name_scope("dropout"):
keep_prob = tf.placeholder(tf.float32)
h_fc1_drop = tf.nn.dropout(h_fc1, keep_prob)
# Map the 1024 features to 10 classes, one for each digit
with tf.name_scope("fc2"):
W_fc2 = weight_variable([1024, 10])
b_fc2 = bias_variable([10])
y_conv = tf.matmul(h_fc1_drop, W_fc2) + b_fc2
return y_conv, keep_prob
def conv2d(x, W):
"""conv2d returns a 2d convolution layer with full stride."""
return tf.nn.conv2d(x, W, strides=[1, 1, 1, 1], padding="SAME")
def max_pool_2x2(x):
"""max_pool_2x2 downsamples a feature map by 2X."""
return tf.nn.max_pool(
x, ksize=[1, 2, 2, 1], strides=[1, 2, 2, 1], padding="SAME")
def weight_variable(shape):
"""weight_variable generates a weight variable of a given shape."""
initial = tf.truncated_normal(shape, stddev=0.1)
return tf.Variable(initial)
def bias_variable(shape):
"""bias_variable generates a bias variable of a given shape."""
initial = tf.constant(0.1, shape=shape)
return tf.Variable(initial)
doc/examples/parameter_server/sync_parameter_server.py
+76
View File
@@ -0,0 +1,76 @@
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
import numpy as np
import ray
import model
parser = argparse.ArgumentParser(description="Run the synchronous parameter "
"server example.")
parser.add_argument("--num-workers", default=4, type=int,
help="The number of workers to use.")
parser.add_argument("--redis-address", default=None, type=str,
help="The Redis address of the cluster.")
@ray.remote
class ParameterServer(object):
def __init__(self, learning_rate):
self.net = model.SimpleCNN(learning_rate=learning_rate)
def apply_gradients(self, *gradients):
self.net.apply_gradients(np.mean(gradients, axis=0))
return self.net.variables.get_flat()
def get_weights(self):
return self.net.variables.get_flat()
@ray.remote
class Worker(object):
def __init__(self, worker_index, batch_size=50):
self.worker_index = worker_index
self.batch_size = batch_size
self.mnist = model.download_mnist_retry(seed=worker_index)
self.net = model.SimpleCNN()
def compute_gradients(self, weights):
self.net.variables.set_flat(weights)
xs, ys = self.mnist.train.next_batch(self.batch_size)
return self.net.compute_gradients(xs, ys)
if __name__ == "__main__":
args = parser.parse_args()
ray.init(redis_address=args.redis_address)
# Create a parameter server.
net = model.SimpleCNN()
ps = ParameterServer.remote(1e-4 * args.num_workers)
# Create workers.
workers = [Worker.remote(worker_index)
for worker_index in range(args.num_workers)]
# Download MNIST.
mnist = model.download_mnist_retry()
i = 0
current_weights = ps.get_weights.remote()
while True:
# Compute and apply gradients.
gradients = [worker.compute_gradients.remote(current_weights)
for worker in workers]
current_weights = ps.apply_gradients.remote(*gradients)
if i % 10 == 0:
# Evaluate the current model.
net.variables.set_flat(ray.get(current_weights))
test_xs, test_ys = mnist.test.next_batch(1000)
accuracy = net.compute_accuracy(test_xs, test_ys)
print("Iteration {}: accuracy is {}".format(i, accuracy))
i += 1
doc/examples/resnet/cifar_input.py
+116
View File
@@ -0,0 +1,116 @@
"""CIFAR dataset input module, with the majority taken from
https://github.com/tensorflow/models/tree/master/resnet.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import tensorflow as tf
def build_data(data_path, size, dataset):
"""Creates the queue and preprocessing operations for the dataset.
Args:
data_path: Filename for cifar10 data.
size: The number of images in the dataset.
dataset: The dataset we are using.
Returns:
queue: A Tensorflow queue for extracting the images and labels.
"""
image_size = 32
if dataset == "cifar10":
label_bytes = 1
label_offset = 0
elif dataset == "cifar100":
label_bytes = 1
label_offset = 1
depth = 3
image_bytes = image_size * image_size * depth
record_bytes = label_bytes + label_offset + image_bytes
def load_transform(value):
# Convert these examples to dense labels and processed images.
record = tf.reshape(tf.decode_raw(value, tf.uint8), [record_bytes])
label = tf.cast(tf.slice(record, [label_offset], [label_bytes]),
tf.int32)
# Convert from string to [depth * height * width] to
# [depth, height, width].
depth_major = tf.reshape(
tf.slice(record, [label_bytes], [image_bytes]),
[depth, image_size, image_size])
# Convert from [depth, height, width] to [height, width, depth].
image = tf.cast(tf.transpose(depth_major, [1, 2, 0]), tf.float32)
return (image, label)
# Read examples from files in the filename queue.
data_files = tf.gfile.Glob(data_path)
data = tf.contrib.data.FixedLengthRecordDataset(data_files,
record_bytes=record_bytes)
data = data.map(load_transform)
data = data.batch(size)
iterator = data.make_one_shot_iterator()
return iterator.get_next()
def build_input(data, batch_size, dataset, train):
"""Build CIFAR image and labels.
Args:
data_path: Filename for cifar10 data.
batch_size: Input batch size.
train: True if we are training and false if we are testing.
Returns:
images: Batches of images of size
[batch_size, image_size, image_size, 3].
labels: Batches of labels of size [batch_size, num_classes].
Raises:
ValueError: When the specified dataset is not supported.
"""
image_size = 32
depth = 3
num_classes = 10 if dataset == "cifar10" else 100
images, labels = data
num_samples = images.shape[0] - images.shape[0] % batch_size
dataset = tf.contrib.data.Dataset.from_tensor_slices(
(images[:num_samples], labels[:num_samples]))
def map_train(image, label):
image = tf.image.resize_image_with_crop_or_pad(image, image_size + 4,
image_size + 4)
image = tf.random_crop(image, [image_size, image_size, 3])
image = tf.image.random_flip_left_right(image)
image = tf.image.per_image_standardization(image)
return (image, label)
def map_test(image, label):
image = tf.image.resize_image_with_crop_or_pad(image, image_size,
image_size)
image = tf.image.per_image_standardization(image)
return (image, label)
dataset = dataset.map(map_train if train else map_test)
dataset = dataset.batch(batch_size)
dataset = dataset.repeat()
if train:
dataset = dataset.shuffle(buffer_size=16 * batch_size)
images, labels = dataset.make_one_shot_iterator().get_next()
images = tf.reshape(images, [batch_size, image_size, image_size, depth])
labels = tf.reshape(labels, [batch_size, 1])
indices = tf.reshape(tf.range(0, batch_size, 1), [batch_size, 1])
labels = tf.sparse_to_dense(
tf.concat([indices, labels], 1),
[batch_size, num_classes], 1.0, 0.0)
assert len(images.get_shape()) == 4
assert images.get_shape()[0] == batch_size
assert images.get_shape()[-1] == 3
assert len(labels.get_shape()) == 2
assert labels.get_shape()[0] == batch_size
assert labels.get_shape()[1] == num_classes
if not train:
tf.summary.image("images", images)
return images, labels
doc/examples/resnet/resnet_main.py
+257
View File
@@ -0,0 +1,257 @@
"""ResNet training script, with some code from
https://github.com/tensorflow/models/tree/master/resnet.
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
import os
import numpy as np
import ray
import tensorflow as tf
import cifar_input
import resnet_model
# Tensorflow must be at least version 1.2.0 for the example to work.
tf_major = int(tf.__version__.split(".")[0])
tf_minor = int(tf.__version__.split(".")[1])
if (tf_major < 1) or (tf_major == 1 and tf_minor < 2):
raise Exception("Your Tensorflow version is less than 1.2.0. Please "
"update Tensorflow to the latest version.")
parser = argparse.ArgumentParser(description="Run the ResNet example.")
parser.add_argument(
"--dataset",
default="cifar10",
type=str,
help="Dataset to use: cifar10 or cifar100.")
parser.add_argument(
"--train_data_path",
default="cifar-10-batches-bin/data_batch*",
type=str,
help="Data path for the training data.")
parser.add_argument(
"--eval_data_path",
default="cifar-10-batches-bin/test_batch.bin",
type=str,
help="Data path for the testing data.")
parser.add_argument(
"--eval_dir",
default="/tmp/resnet-model/eval",
type=str,
help="Data path for the tensorboard logs.")
parser.add_argument(
"--eval_batch_count",
default=50,
type=int,
help="Number of batches to evaluate over.")
parser.add_argument(
"--num_gpus",
default=0,
type=int,
help="Number of GPUs to use for training.")
parser.add_argument(
"--redis-address",
default=None,
type=str,
help="The Redis address of the cluster.")
FLAGS = parser.parse_args()
# Determines if the actors require a gpu or not.
use_gpu = 1 if int(FLAGS.num_gpus) > 0 else 0
@ray.remote
def get_data(path, size, dataset):
# Retrieves all preprocessed images and labels using a tensorflow queue.
# This only uses the cpu.
os.environ["CUDA_VISIBLE_DEVICES"] = ""
with tf.device("/cpu:0"):
dataset = cifar_input.build_data(path, size, dataset)
sess = tf.Session()
images, labels = sess.run(dataset)
sess.close()
return images, labels
@ray.remote(num_gpus=use_gpu)
class ResNetTrainActor(object):
def __init__(self, data, dataset, num_gpus):
if num_gpus > 0:
os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(
[str(i) for i in ray.get_gpu_ids()])
hps = resnet_model.HParams(
batch_size=128,
num_classes=100 if dataset == "cifar100" else 10,
min_lrn_rate=0.0001,
lrn_rate=0.1,
num_residual_units=5,
use_bottleneck=False,
weight_decay_rate=0.0002,
relu_leakiness=0.1,
optimizer="mom",
num_gpus=num_gpus)
# We seed each actor differently so that each actor operates on a
# different subset of data.
if num_gpus > 0:
tf.set_random_seed(ray.get_gpu_ids()[0] + 1)
else:
# Only a single actor in this case.
tf.set_random_seed(1)
with tf.device("/gpu:0" if num_gpus > 0 else "/cpu:0"):
# Build the model.
images, labels = cifar_input.build_input(data, hps.batch_size,
dataset, True)
self.model = resnet_model.ResNet(hps, images, labels, "train")
self.model.build_graph()
config = tf.ConfigProto(allow_soft_placement=True)
config.gpu_options.allow_growth = True
sess = tf.Session(config=config)
self.model.variables.set_session(sess)
init = tf.global_variables_initializer()
sess.run(init)
self.steps = 10
def compute_steps(self, weights):
# This method sets the weights in the network, trains the network
# self.steps times, and returns the new weights.
self.model.variables.set_weights(weights)
for i in range(self.steps):
self.model.variables.sess.run(self.model.train_op)
return self.model.variables.get_weights()
def get_weights(self):
# Note that the driver cannot directly access fields of the class,
# so helper methods must be created.
return self.model.variables.get_weights()
@ray.remote
class ResNetTestActor(object):
def __init__(self, data, dataset, eval_batch_count, eval_dir):
os.environ["CUDA_VISIBLE_DEVICES"] = ""
hps = resnet_model.HParams(
batch_size=100,
num_classes=100 if dataset == "cifar100" else 10,
min_lrn_rate=0.0001,
lrn_rate=0.1,
num_residual_units=5,
use_bottleneck=False,
weight_decay_rate=0.0002,
relu_leakiness=0.1,
optimizer="mom",
num_gpus=0)
with tf.device("/cpu:0"):
# Builds the testing network.
images, labels = cifar_input.build_input(data, hps.batch_size,
dataset, False)
self.model = resnet_model.ResNet(hps, images, labels, "eval")
self.model.build_graph()
config = tf.ConfigProto(allow_soft_placement=True)
config.gpu_options.allow_growth = True
sess = tf.Session(config=config)
self.model.variables.set_session(sess)
init = tf.global_variables_initializer()
sess.run(init)
# Initializing parameters for tensorboard.
self.best_precision = 0.0
self.eval_batch_count = eval_batch_count
self.summary_writer = tf.summary.FileWriter(eval_dir, sess.graph)
# The IP address where tensorboard logs will be on.
self.ip_addr = ray.services.get_node_ip_address()
def accuracy(self, weights, train_step):
# Sets the weights, computes the accuracy and other metrics
# over eval_batches, and outputs to tensorboard.
self.model.variables.set_weights(weights)
total_prediction, correct_prediction = 0, 0
model = self.model
sess = self.model.variables.sess
for _ in range(self.eval_batch_count):
summaries, loss, predictions, truth = sess.run(
[model.summaries, model.cost, model.predictions, model.labels])
truth = np.argmax(truth, axis=1)
predictions = np.argmax(predictions, axis=1)
correct_prediction += np.sum(truth == predictions)
total_prediction += predictions.shape[0]
precision = 1.0 * correct_prediction / total_prediction
self.best_precision = max(precision, self.best_precision)
precision_summ = tf.Summary()
precision_summ.value.add(tag="Precision", simple_value=precision)
self.summary_writer.add_summary(precision_summ, train_step)
best_precision_summ = tf.Summary()
best_precision_summ.value.add(
tag="Best Precision", simple_value=self.best_precision)
self.summary_writer.add_summary(best_precision_summ, train_step)
self.summary_writer.add_summary(summaries, train_step)
tf.logging.info("loss: %.3f, precision: %.3f, best precision: %.3f" %
(loss, precision, self.best_precision))
self.summary_writer.flush()
return precision
def get_ip_addr(self):
# As above, a helper method must be created to access the field from
# the driver.
return self.ip_addr
def train():
num_gpus = FLAGS.num_gpus
if FLAGS.redis_address is None:
ray.init(num_gpus=num_gpus)
else:
ray.init(redis_address=FLAGS.redis_address)
train_data = get_data.remote(FLAGS.train_data_path, 50000, FLAGS.dataset)
test_data = get_data.remote(FLAGS.eval_data_path, 10000, FLAGS.dataset)
# Creates an actor for each gpu, or one if only using the cpu. Each actor
# has access to the dataset.
if FLAGS.num_gpus > 0:
train_actors = [
ResNetTrainActor.remote(train_data, FLAGS.dataset, num_gpus)
for _ in range(num_gpus)
]
else:
train_actors = [ResNetTrainActor.remote(train_data, FLAGS.dataset, 0)]
test_actor = ResNetTestActor.remote(test_data, FLAGS.dataset,
FLAGS.eval_batch_count, FLAGS.eval_dir)
print("The log files for tensorboard are stored at ip {}.".format(
ray.get(test_actor.get_ip_addr.remote())))
step = 0
weight_id = train_actors[0].get_weights.remote()
acc_id = test_actor.accuracy.remote(weight_id, step)
# Correction for dividing the weights by the number of gpus.
if num_gpus == 0:
num_gpus = 1
print("Starting training loop. Use Ctrl-C to exit.")
try:
while True:
all_weights = ray.get([
actor.compute_steps.remote(weight_id) for actor in train_actors
])
mean_weights = {
k: (sum(weights[k] for weights in all_weights) / num_gpus)
for k in all_weights[0]
}
weight_id = ray.put(mean_weights)
step += 10
if step % 200 == 0:
# Retrieves the previously computed accuracy and launches a new
# testing task with the current weights every 200 steps.
acc = ray.get(acc_id)
acc_id = test_actor.accuracy.remote(weight_id, step)
print("Step {}: {:.6f}".format(step - 200, acc))
except KeyboardInterrupt:
pass
if __name__ == "__main__":
train()
doc/examples/resnet/resnet_model.py
+317
View File
@@ -0,0 +1,317 @@
"""ResNet model with most of the code taken from
https://github.com/tensorflow/models/tree/master/resnet.
Related papers:
https://arxiv.org/pdf/1603.05027v2.pdf
https://arxiv.org/pdf/1512.03385v1.pdf
https://arxiv.org/pdf/1605.07146v1.pdf
"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from collections import namedtuple
import numpy as np
import tensorflow as tf
from tensorflow.python.training import moving_averages
import ray
import ray.experimental.tf_utils
HParams = namedtuple(
"HParams", "batch_size, num_classes, min_lrn_rate, lrn_rate, "
"num_residual_units, use_bottleneck, weight_decay_rate, "
"relu_leakiness, optimizer, num_gpus")
class ResNet(object):
"""ResNet model."""
def __init__(self, hps, images, labels, mode):
"""ResNet constructor.
Args:
hps: Hyperparameters.
images: Batches of images of size [batch_size, image_size,
image_size, 3].
labels: Batches of labels of size [batch_size, num_classes].
mode: One of 'train' and 'eval'.
"""
self.hps = hps
self._images = images
self.labels = labels
self.mode = mode
self._extra_train_ops = []
def build_graph(self):
"""Build a whole graph for the model."""
self.global_step = tf.Variable(0, trainable=False)
self._build_model()
if self.mode == "train":
self._build_train_op()
else:
# Additional initialization for the test network.
self.variables = ray.experimental.tf_utils.TensorFlowVariables(
self.cost)
self.summaries = tf.summary.merge_all()
def _stride_arr(self, stride):
"""Map a stride scalar to the stride array for tf.nn.conv2d."""
return [1, stride, stride, 1]
def _build_model(self):
"""Build the core model within the graph."""
with tf.variable_scope("init"):
x = self._conv("init_conv", self._images, 3, 3, 16,
self._stride_arr(1))
strides = [1, 2, 2]
activate_before_residual = [True, False, False]
if self.hps.use_bottleneck:
res_func = self._bottleneck_residual
filters = [16, 64, 128, 256]
else:
res_func = self._residual
filters = [16, 16, 32, 64]
with tf.variable_scope("unit_1_0"):
x = res_func(x, filters[0], filters[1], self._stride_arr(
strides[0]), activate_before_residual[0])
for i in range(1, self.hps.num_residual_units):
with tf.variable_scope("unit_1_%d" % i):
x = res_func(x, filters[1], filters[1], self._stride_arr(1),
False)
with tf.variable_scope("unit_2_0"):
x = res_func(x, filters[1], filters[2], self._stride_arr(
strides[1]), activate_before_residual[1])
for i in range(1, self.hps.num_residual_units):
with tf.variable_scope("unit_2_%d" % i):
x = res_func(x, filters[2], filters[2], self._stride_arr(1),
False)
with tf.variable_scope("unit_3_0"):
x = res_func(x, filters[2], filters[3], self._stride_arr(
strides[2]), activate_before_residual[2])
for i in range(1, self.hps.num_residual_units):
with tf.variable_scope("unit_3_%d" % i):
x = res_func(x, filters[3], filters[3], self._stride_arr(1),
False)
with tf.variable_scope("unit_last"):
x = self._batch_norm("final_bn", x)
x = self._relu(x, self.hps.relu_leakiness)
x = self._global_avg_pool(x)
with tf.variable_scope("logit"):
logits = self._fully_connected(x, self.hps.num_classes)
self.predictions = tf.nn.softmax(logits)
with tf.variable_scope("costs"):
xent = tf.nn.softmax_cross_entropy_with_logits(
logits=logits, labels=self.labels)
self.cost = tf.reduce_mean(xent, name="xent")
self.cost += self._decay()
if self.mode == "eval":
tf.summary.scalar("cost", self.cost)
def _build_train_op(self):
"""Build training specific ops for the graph."""
num_gpus = self.hps.num_gpus if self.hps.num_gpus != 0 else 1
# The learning rate schedule is dependent on the number of gpus.
boundaries = [int(20000 * i / np.sqrt(num_gpus)) for i in range(2, 5)]
values = [0.1, 0.01, 0.001, 0.0001]
self.lrn_rate = tf.train.piecewise_constant(self.global_step,
boundaries, values)
tf.summary.scalar("learning rate", self.lrn_rate)
if self.hps.optimizer == "sgd":
optimizer = tf.train.GradientDescentOptimizer(self.lrn_rate)
elif self.hps.optimizer == "mom":
optimizer = tf.train.MomentumOptimizer(self.lrn_rate, 0.9)
apply_op = optimizer.minimize(self.cost, global_step=self.global_step)
train_ops = [apply_op] + self._extra_train_ops
self.train_op = tf.group(*train_ops)
self.variables = ray.experimental.tf_utils.TensorFlowVariables(
self.train_op)
def _batch_norm(self, name, x):
"""Batch normalization."""
with tf.variable_scope(name):
params_shape = [x.get_shape()[-1]]
beta = tf.get_variable(
"beta",
params_shape,
tf.float32,
initializer=tf.constant_initializer(0.0, tf.float32))
gamma = tf.get_variable(
"gamma",
params_shape,
tf.float32,
initializer=tf.constant_initializer(1.0, tf.float32))
if self.mode == "train":
mean, variance = tf.nn.moments(x, [0, 1, 2], name="moments")
moving_mean = tf.get_variable(
"moving_mean",
params_shape,
tf.float32,
initializer=tf.constant_initializer(0.0, tf.float32),
trainable=False)
moving_variance = tf.get_variable(
"moving_variance",
params_shape,
tf.float32,
initializer=tf.constant_initializer(1.0, tf.float32),
trainable=False)
self._extra_train_ops.append(
moving_averages.assign_moving_average(
moving_mean, mean, 0.9))
self._extra_train_ops.append(
moving_averages.assign_moving_average(
moving_variance, variance, 0.9))
else:
mean = tf.get_variable(
"moving_mean",
params_shape,
tf.float32,
initializer=tf.constant_initializer(0.0, tf.float32),
trainable=False)
variance = tf.get_variable(
"moving_variance",
params_shape,
tf.float32,
initializer=tf.constant_initializer(1.0, tf.float32),
trainable=False)
tf.summary.histogram(mean.op.name, mean)
tf.summary.histogram(variance.op.name, variance)
# elipson used to be 1e-5. Maybe 0.001 solves NaN problem in deeper
# net.
y = tf.nn.batch_normalization(x, mean, variance, beta, gamma,
0.001)
y.set_shape(x.get_shape())
return y
def _residual(self,
x,
in_filter,
out_filter,
stride,
activate_before_residual=False):
"""Residual unit with 2 sub layers."""
if activate_before_residual:
with tf.variable_scope("shared_activation"):
x = self._batch_norm("init_bn", x)
x = self._relu(x, self.hps.relu_leakiness)
orig_x = x
else:
with tf.variable_scope("residual_only_activation"):
orig_x = x
x = self._batch_norm("init_bn", x)
x = self._relu(x, self.hps.relu_leakiness)
with tf.variable_scope("sub1"):
x = self._conv("conv1", x, 3, in_filter, out_filter, stride)
with tf.variable_scope("sub2"):
x = self._batch_norm("bn2", x)
x = self._relu(x, self.hps.relu_leakiness)
x = self._conv("conv2", x, 3, out_filter, out_filter, [1, 1, 1, 1])
with tf.variable_scope("sub_add"):
if in_filter != out_filter:
orig_x = tf.nn.avg_pool(orig_x, stride, stride, "VALID")
orig_x = tf.pad(
orig_x,
[[0, 0], [0, 0], [0, 0], [(out_filter - in_filter) // 2,
(out_filter - in_filter) // 2]])
x += orig_x
return x
def _bottleneck_residual(self,
x,
in_filter,
out_filter,
stride,
activate_before_residual=False):
"""Bottleneck residual unit with 3 sub layers."""
if activate_before_residual:
with tf.variable_scope("common_bn_relu"):
x = self._batch_norm("init_bn", x)
x = self._relu(x, self.hps.relu_leakiness)
orig_x = x
else:
with tf.variable_scope("residual_bn_relu"):
orig_x = x
x = self._batch_norm("init_bn", x)
x = self._relu(x, self.hps.relu_leakiness)
with tf.variable_scope("sub1"):
x = self._conv("conv1", x, 1, in_filter, out_filter / 4, stride)
with tf.variable_scope("sub2"):
x = self._batch_norm("bn2", x)
x = self._relu(x, self.hps.relu_leakiness)
x = self._conv("conv2", x, 3, out_filter / 4, out_filter / 4,
[1, 1, 1, 1])
with tf.variable_scope("sub3"):
x = self._batch_norm("bn3", x)
x = self._relu(x, self.hps.relu_leakiness)
x = self._conv("conv3", x, 1, out_filter / 4, out_filter,
[1, 1, 1, 1])
with tf.variable_scope("sub_add"):
if in_filter != out_filter:
orig_x = self._conv("project", orig_x, 1, in_filter,
out_filter, stride)
x += orig_x
return x
def _decay(self):
"""L2 weight decay loss."""
costs = []
for var in tf.trainable_variables():
if var.op.name.find(r"DW") > 0:
costs.append(tf.nn.l2_loss(var))
return tf.multiply(self.hps.weight_decay_rate, tf.add_n(costs))
def _conv(self, name, x, filter_size, in_filters, out_filters, strides):
"""Convolution."""
with tf.variable_scope(name):
n = filter_size * filter_size * out_filters
kernel = tf.get_variable(
"DW", [filter_size, filter_size, in_filters, out_filters],
tf.float32,
initializer=tf.random_normal_initializer(
stddev=np.sqrt(2.0 / n)))
return tf.nn.conv2d(x, kernel, strides, padding="SAME")
def _relu(self, x, leakiness=0.0):
"""Relu, with optional leaky support."""
return tf.where(tf.less(x, 0.0), leakiness * x, x, name="leaky_relu")
def _fully_connected(self, x, out_dim):
"""FullyConnected layer for final output."""
x = tf.reshape(x, [self.hps.batch_size, -1])
w = tf.get_variable(
"DW", [x.get_shape()[1], out_dim],
initializer=tf.uniform_unit_scaling_initializer(factor=1.0))
b = tf.get_variable(
"biases", [out_dim], initializer=tf.constant_initializer())
return tf.nn.xw_plus_b(x, w, b)
def _global_avg_pool(self, x):
assert x.get_shape().ndims == 4
return tf.reduce_mean(x, [1, 2])
doc/examples/rl_pong/driver.py
+213
View File
@@ -0,0 +1,213 @@
# This code is copied and adapted from Andrej Karpathy's code for learning to
# play Pong https://gist.github.com/karpathy/a4166c7fe253700972fcbc77e4ea32c5.
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import argparse
import numpy as np
import os
import ray
import time
import gym
# Define some hyperparameters.
# The number of hidden layer neurons.
H = 200
learning_rate = 1e-4
# Discount factor for reward.
gamma = 0.99
# The decay factor for RMSProp leaky sum of grad^2.
decay_rate = 0.99
# The input dimensionality: 80x80 grid.
D = 80 * 80
def sigmoid(x):
# Sigmoid "squashing" function to interval [0, 1].
return 1.0 / (1.0 + np.exp(-x))
def preprocess(img):
"""Preprocess 210x160x3 uint8 frame into 6400 (80x80) 1D float vector."""
# Crop the image.
img = img[35:195]
# Downsample by factor of 2.
img = img[::2, ::2, 0]
# Erase background (background type 1).
img[img == 144] = 0
# Erase background (background type 2).
img[img == 109] = 0
# Set everything else (paddles, ball) to 1.
img[img != 0] = 1
return img.astype(np.float).ravel()
def discount_rewards(r):
"""take 1D float array of rewards and compute discounted reward"""
discounted_r = np.zeros_like(r)
running_add = 0
for t in reversed(range(0, r.size)):
# Reset the sum, since this was a game boundary (pong specific!).
if r[t] != 0:
running_add = 0
running_add = running_add * gamma + r[t]
discounted_r[t] = running_add
return discounted_r
def policy_forward(x, model):
h = np.dot(model["W1"], x)
h[h < 0] = 0 # ReLU nonlinearity.
logp = np.dot(model["W2"], h)
p = sigmoid(logp)
# Return probability of taking action 2, and hidden state.
return p, h
def policy_backward(eph, epx, epdlogp, model):
"""backward pass. (eph is array of intermediate hidden states)"""
dW2 = np.dot(eph.T, epdlogp).ravel()
dh = np.outer(epdlogp, model["W2"])
# Backprop relu.
dh[eph <= 0] = 0
dW1 = np.dot(dh.T, epx)
return {"W1": dW1, "W2": dW2}
@ray.remote
class PongEnv(object):
def __init__(self):
# Tell numpy to only use one core. If we don't do this, each actor may
# try to use all of the cores and the resulting contention may result
# in no speedup over the serial version. Note that if numpy is using
# OpenBLAS, then you need to set OPENBLAS_NUM_THREADS=1, and you
# probably need to do it from the command line (so it happens before
# numpy is imported).
os.environ["MKL_NUM_THREADS"] = "1"
self.env = gym.make("Pong-v0")
def compute_gradient(self, model):
# Reset the game.
observation = self.env.reset()
# Note that prev_x is used in computing the difference frame.
prev_x = None
xs, hs, dlogps, drs = [], [], [], []
reward_sum = 0
done = False
while not done:
cur_x = preprocess(observation)
x = cur_x - prev_x if prev_x is not None else np.zeros(D)
prev_x = cur_x
aprob, h = policy_forward(x, model)
# Sample an action.
action = 2 if np.random.uniform() < aprob else 3
# The observation.
xs.append(x)
# The hidden state.
hs.append(h)
y = 1 if action == 2 else 0 # A "fake label".
# The gradient that encourages the action that was taken to be
# taken (see http://cs231n.github.io/neural-networks-2/#losses if
# confused).
dlogps.append(y - aprob)
observation, reward, done, info = self.env.step(action)
reward_sum += reward
# Record reward (has to be done after we call step() to get reward
# for previous action).
drs.append(reward)
epx = np.vstack(xs)
eph = np.vstack(hs)
epdlogp = np.vstack(dlogps)
epr = np.vstack(drs)
# Reset the array memory.
xs, hs, dlogps, drs = [], [], [], []
# Compute the discounted reward backward through time.
discounted_epr = discount_rewards(epr)
# Standardize the rewards to be unit normal (helps control the gradient
# estimator variance).
discounted_epr -= np.mean(discounted_epr)
discounted_epr /= np.std(discounted_epr)
# Modulate the gradient with advantage (the policy gradient magic
# happens right here).
epdlogp *= discounted_epr
return policy_backward(eph, epx, epdlogp, model), reward_sum
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Train an RL agent on Pong.")
parser.add_argument(
"--batch-size",
default=10,
type=int,
help="The number of rollouts to do per batch.")
parser.add_argument(
"--redis-address",
default=None,
type=str,
help="The Redis address of the cluster.")
parser.add_argument(
"--iterations",
default=-1,
type=int,
help="The number of model updates to perform. By "
"default, training will not terminate.")
args = parser.parse_args()
batch_size = args.batch_size
ray.init(redis_address=args.redis_address)
# Run the reinforcement learning.
running_reward = None
batch_num = 1
model = {}
# "Xavier" initialization.
model["W1"] = np.random.randn(H, D) / np.sqrt(D)
model["W2"] = np.random.randn(H) / np.sqrt(H)
# Update buffers that add up gradients over a batch.
grad_buffer = {k: np.zeros_like(v) for k, v in model.items()}
# Update the rmsprop memory.
rmsprop_cache = {k: np.zeros_like(v) for k, v in model.items()}
actors = [PongEnv.remote() for _ in range(batch_size)]
iteration = 0
while iteration != args.iterations:
iteration += 1
model_id = ray.put(model)
actions = []
# Launch tasks to compute gradients from multiple rollouts in parallel.
start_time = time.time()
for i in range(batch_size):
action_id = actors[i].compute_gradient.remote(model_id)
actions.append(action_id)
for i in range(batch_size):
action_id, actions = ray.wait(actions)
grad, reward_sum = ray.get(action_id[0])
# Accumulate the gradient over batch.
for k in model:
grad_buffer[k] += grad[k]
running_reward = (reward_sum if running_reward is None else
running_reward * 0.99 + reward_sum * 0.01)
end_time = time.time()
print("Batch {} computed {} rollouts in {} seconds, "
"running mean is {}".format(batch_num, batch_size,
end_time - start_time,
running_reward))
for k, v in model.items():
g = grad_buffer[k]
rmsprop_cache[k] = (
decay_rate * rmsprop_cache[k] + (1 - decay_rate) * g**2)
model[k] += learning_rate * g / (np.sqrt(rmsprop_cache[k]) + 1e-5)
# Reset the batch gradient buffer.
grad_buffer[k] = np.zeros_like(v)
batch_num += 1
doc/examples/streaming/articles.txt
+8
View File
@@ -0,0 +1,8 @@
New York City
Berlin
London
Paris
United States
Germany
France
United Kingdom
doc/examples/streaming/streaming.py
+104
View File
@@ -0,0 +1,104 @@
import argparse
from collections import Counter, defaultdict
import heapq
import numpy as np
import os
import ray
import wikipedia
parser = argparse.ArgumentParser()
parser.add_argument("--num-mappers",
help="number of mapper actors used", default=3)
parser.add_argument("--num-reducers",
help="number of reducer actors used", default=4)
@ray.remote
class Mapper(object):
def __init__(self, title_stream):
self.title_stream = title_stream
self.num_articles_processed = 0
self.articles = []
self.word_counts = []
def get_new_article(self):
# Get the next wikipedia article.
article = wikipedia.page(self.title_stream.next()).content
# Count the words and store the result.
self.word_counts.append(Counter(article.split(" ")))
self.num_articles_processed += 1
def get_range(self, article_index, keys):
# Process more articles if this Mapper hasn't processed enough yet.
while self.num_articles_processed < article_index + 1:
self.get_new_article()
# Return the word counts from within a given character range.
return [(k, v) for k, v in self.word_counts[article_index].items()
if len(k) >= 1 and k[0] >= keys[0] and k[0] <= keys[1]]
@ray.remote
class Reducer(object):
def __init__(self, keys, *mappers):
self.mappers = mappers
self.keys = keys
def next_reduce_result(self, article_index):
word_count_sum = defaultdict(lambda: 0)
# Get the word counts for this Reducer's keys from all of the Mappers
# and aggregate the results.
count_ids = [mapper.get_range.remote(article_index, self.keys)
for mapper in self.mappers]
# TODO(rkn): We should process these out of order using ray.wait.
for count_id in count_ids:
for k, v in ray.get(count_id):
word_count_sum[k] += v
return word_count_sum
class Stream(object):
def __init__(self, elements):
self.elements = elements
def next(self):
i = np.random.randint(0, len(self.elements))
return self.elements[i]
if __name__ == "__main__":
args = parser.parse_args()
ray.init()
# Create one streaming source of articles per mapper.
directory = os.path.dirname(os.path.realpath(__file__))
streams = []
for _ in range(args.num_mappers):
with open(os.path.join(directory, "articles.txt")) as f:
streams.append(Stream([line.strip() for line in f.readlines()]))
# Partition the keys among the reducers.
chunks = np.array_split([chr(i) for i in range(ord("a"), ord("z") + 1)],
args.num_reducers)
keys = [[chunk[0], chunk[-1]] for chunk in chunks]
# Create a number of mappers.
mappers = [Mapper.remote(stream) for stream in streams]
# Create a number of reduces, each responsible for a different range of
# keys. This gives each Reducer actor a handle to each Mapper actor.
reducers = [Reducer.remote(key, *mappers) for key in keys]
article_index = 0
while True:
print("article index = {}".format(article_index))
wordcounts = {}
counts = ray.get([reducer.next_reduce_result.remote(article_index)
for reducer in reducers])
for count in counts:
wordcounts.update(count)
most_frequent_words = heapq.nlargest(10, wordcounts,
key=wordcounts.get)
for word in most_frequent_words:
print(" ", word, wordcounts[word])
article_index += 1