Skip to content

Commit 6ab8d6f

Browse files
authored
feat: support multiple optimizers (#16)
1 parent c94a01e commit 6ab8d6f

7 files changed

Lines changed: 139 additions & 30 deletions

File tree

epl/ir/graph.py

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -93,8 +93,8 @@ def _reset(self):
9393
self._vars_related_op_names = None
9494
self._dataset_related_ops = None
9595
self._clone_dataset_related_ops = False
96-
# TODO(jiangle.jl): Support multi-optimizer.
9796
self._gradients = []
97+
self._variables = []
9898
self.primitive_init_op = None
9999
# TODO(jiangle.jl): Support train_and_evaluate.
100100
self._dataset_api_op = []
@@ -232,13 +232,20 @@ def num_constructors(self):
232232
"""Number of constructors defined in epl.Cluster."""
233233
return len(self.constructor_task)
234234

235+
def add_grads_and_vars(self, grads_and_vars):
236+
"""Add gradients for parallelism. For each variable, only add corresponding grad once."""
237+
for grad, var in grads_and_vars:
238+
if var not in self._variables:
239+
self._variables.append(var)
240+
self._gradients.append(grad)
241+
235242
@property
236243
def gradients(self):
237244
return self._gradients
238245

239-
@gradients.setter
240-
def gradients(self, gradients):
241-
self._gradients = gradients
246+
@property
247+
def variables(self):
248+
return self._variables
242249

243250
@property
244251
def dataset_api_op(self):

epl/parallel/hooks.py

Lines changed: 12 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -124,10 +124,6 @@ def gradients_impl_gradients_helper(fn):
124124
def gradient_helper(*args, **kwargs):
125125
"""Check is epl context empty before gradients function. Clear
126126
epl context after gradeints function. Get gradients results."""
127-
if Env.get().strategy_context:
128-
warnings.warn("EPL ignores the context of backward operations, "
129-
"as it collocates the backward operations with their "
130-
"corresponding forward operations automatically.")
131127

132128
config = Env.get().config
133129
if config.auto.auto_parallel:
@@ -176,7 +172,8 @@ def gradient_helper(*args, **kwargs):
176172
if Env.get().config.communication.clip_after_allreduce:
177173
if not ga_enabled():
178174
warnings.warn("Clip gradients by norm after allreduce is enabled.")
179-
Graph.get().gradients = gradients
175+
variables = args[1]
176+
Graph.get().add_grads_and_vars(list(zip(gradients, variables)))
180177
else:
181178
warnings.warn("Gradient accumulation does not support opt_clip_after_allreduce by now, ignore it.")
182179
return gradients
@@ -188,18 +185,15 @@ def optimizer_apply_gradients(fn):
188185
"""Hook apply function to get apply operations."""
189186
def apply_gradients(self, grads_and_vars, *args, **kwargs):
190187
"""Get apply operations by set model phase to APPLY."""
191-
if Env.get().strategy_context:
192-
warnings.warn("EPL ignores the context of apply operations, "
193-
"as it collocates the apply operations with their "
194-
"corresponding forward operations automatically.")
195-
188+
# Convert to list for py3
189+
if not isinstance(grads_and_vars, list):
190+
grads_and_vars = list(grads_and_vars)
196191
num_apply_group = Env.get().config.optimizer.num_apply_group
197-
192+
# apply optimizations only if all variables are not updated before.
193+
apply_opt = all(v not in Graph.get().variables for g, v in grads_and_vars)
198194
if not Env.get().config.communication.clip_after_allreduce:
199-
# TODO(jiangle.jl): Suppoort multi-optimizers.
200195
if not ga_enabled():
201-
Graph.get().gradients += [gv[0] for gv in grads_and_vars]
202-
196+
Graph.get().add_grads_and_vars(grads_and_vars)
203197

204198
global_step = None
205199
name = None
@@ -214,8 +208,9 @@ def apply_gradients(self, grads_and_vars, *args, **kwargs):
214208
ga_iters = ga_iter_num()
215209

216210
with ModelPhase(ModelPhase.APPLY):
217-
apply_fn = None
218-
if zero_enabled():
211+
if not apply_opt:
212+
apply_fn = lambda: fn(self, grads_and_vars, *args, **kwargs)
213+
elif zero_enabled():
219214
apply_fn = lambda: apply_zero(self, fn, grads_and_vars,
220215
global_step, ga_iters,
221216
num_apply_group, name)
@@ -230,7 +225,7 @@ def apply_gradients(self, grads_and_vars, *args, **kwargs):
230225
else:
231226
apply_fn = lambda: fn(self, grads_and_vars, *args, **kwargs)
232227

233-
if amp_enabled() and Env.get().config.amp.loss_scale == "dynamic":
228+
if apply_opt and amp_enabled() and Env.get().config.amp.loss_scale == "dynamic":
234229
return amp_update(grads_and_vars, apply_fn, name)
235230
return apply_fn()
236231
return apply_gradients

epl/runtime/gradient_accumulation.py

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -63,17 +63,14 @@ def apply_accmulation(optimizer, apply_gradients_fn,
6363
tf_logging.info("Enable gradient accumulation, num_micro_batch: {}".format(niter))
6464
mean_grad_flag = True if Env.get().config.communication.gradients_reduce_method == \
6565
constant.REDUCE_METHOD_MEAN else False
66-
data = []
67-
grads = []
66+
grads_and_vars = []
6867
for g, v in slots_and_vars:
6968
with ops.device(g.device):
7069
g = array_ops.identity(g, name='acc_grads_tensor')
7170
if mean_grad_flag:
7271
g = g / niter
73-
grads.append(g)
74-
data.append((g, v))
75-
grads_and_vars = data
76-
Graph.get().gradients += grads
72+
grads_and_vars.append((g, v))
73+
Graph.get().add_grads_and_vars(grads_and_vars)
7774
update_ops = []
7875
apply_op = apply_grad_group(optimizer, apply_gradients_fn, grads_and_vars,
7976
ngroup, global_step, "epl_apply_grad_ga")

tests/Makefile

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ test:
1414
CUDA_VISIBLE_DEVICES=$(GPU_4) ./launch.sh auto_cluster_test.py
1515
CUDA_VISIBLE_DEVICES=$(GPU_2) ./launch.sh estimator_test.py
1616
CUDA_VISIBLE_DEVICES=$(GPU_2) ./launch.sh estimator_loss_test.py
17+
CUDA_VISIBLE_DEVICES=$(GPU_2) ./launch.sh multi_optimizer_test.py
1718
CUDA_VISIBLE_DEVICES=$(GPU_2) ./launch.sh optimizer_helper_test.py
1819
CUDA_VISIBLE_DEVICES='' ./launch.sh initializers_test.py
1920
CUDA_VISIBLE_DEVICES='' ./launch.sh phase_test.py

tests/graph_test.py

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -221,10 +221,10 @@ def test_outside_strategy_error(self):
221221
self.assertEqual(0, len(w))
222222
with warnings.catch_warnings(record=True) as w:
223223
gvs = optimizer.compute_gradients(loss)
224-
self.assertEqual(1, len(w))
224+
self.assertEqual(0, len(w))
225225
with warnings.catch_warnings(record=True) as w:
226226
optimizer.apply_gradients(gvs)
227-
self.assertEqual(1, len(w))
227+
self.assertEqual(0, len(w))
228228

229229
x3 = tf.constant(1.1, shape=[2, 2])
230230
self.assertEqual(Graph.get()._user_default_taskgraph, None)

tests/multi_optimizer_test.py

Lines changed: 109 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,109 @@
1+
# Copyright 2021 Alibaba Group Holding Limited. All Rights Reserved.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
# =============================================================================
15+
"""Test for multiple optimizer."""
16+
17+
from __future__ import absolute_import
18+
from __future__ import division
19+
from __future__ import print_function
20+
21+
import numpy as np
22+
import tensorflow as tf
23+
from tensorflow.python.platform import test
24+
25+
import epl
26+
from test_utils import fix_randomness
27+
28+
29+
def get_nested_optimizer():
30+
optimizer = tf.train.AdamOptimizer(learning_rate=0.01)
31+
optimizer = tf.contrib.estimator.clip_gradients_by_norm(optimizer, 0.5)
32+
optimizer = tf.contrib.opt.MovingAverageOptimizer(
33+
optimizer,
34+
average_decay=0.1,
35+
num_updates=2)
36+
return optimizer
37+
38+
def _train_fn(bs, lr_decay=False, max_steps=5, optimizer=None, enable_epl=True):
39+
"""Define train function."""
40+
fix_randomness()
41+
num_x = np.random.randint(0, 10, (500, 20)).astype(dtype=np.float32)
42+
num_y = np.random.randint(0, 10, 500).astype(dtype=np.int64)
43+
dataset = tf.data.Dataset.from_tensor_slices((num_x, num_y)) \
44+
.batch(bs).repeat(1)
45+
iterator = dataset.make_initializable_iterator()
46+
tf.add_to_collection(tf.GraphKeys.TABLE_INITIALIZERS, iterator.initializer)
47+
x, labels = iterator.get_next()
48+
for _ in range(10):
49+
x = tf.layers.dense(x, 10, use_bias=False)
50+
logits = x
51+
loss = tf.losses.sparse_softmax_cross_entropy(labels=labels, logits=logits)
52+
if enable_epl:
53+
epl.add_to_collection(loss, epl.GraphKeys.GLOBAL_MEAN_OBJECTS)
54+
tf.summary.scalar('loss', loss)
55+
global_step = tf.train.get_or_create_global_step()
56+
learning_rate = 0.001
57+
if lr_decay:
58+
learning_rate = tf.train.exponential_decay(learning_rate, global_step,
59+
max_steps, 0.96,
60+
staircase=False)
61+
if optimizer is None:
62+
optimizer = tf.train.AdamOptimizer(learning_rate=learning_rate)
63+
else:
64+
optimizer = optimizer
65+
train_op = optimizer.minimize(loss, global_step=global_step)
66+
merged = tf.summary.merge_all()
67+
hooks = [tf.train.StopAtStepHook(last_step=max_steps)]
68+
losses = []
69+
with tf.train.MonitoredTrainingSession(
70+
hooks=hooks) as sess:
71+
while not sess.should_stop():
72+
fetches = [loss, train_op, global_step, merged]
73+
res = sess.run(fetches)
74+
train_loss = res[0]
75+
losses.append(train_loss)
76+
print("step: {} loss: {}".format(res[2], train_loss))
77+
return losses, optimizer
78+
79+
# pylint: disable=missing-docstring,unused-variable
80+
# pylint: disable=protected-access
81+
class MultiOptimizerTest(test.TestCase):
82+
@classmethod
83+
def setUpClass(cls): # pylint: disable=invalid-name
84+
with tf.Graph().as_default():
85+
optimizer = get_nested_optimizer()
86+
cls.base, _ = _train_fn(bs=2, lr_decay=False, max_steps=5, optimizer=optimizer, enable_epl=False)
87+
88+
89+
def test_nested_optimizer(self):
90+
epl.init(epl.Config({"communication.clip_after_allreduce": True}))
91+
epl.set_default_strategy(epl.replicate(1))
92+
with tf.Graph().as_default():
93+
optimizer = get_nested_optimizer()
94+
res2, _ = _train_fn(bs=1, lr_decay=False, max_steps=5, optimizer=optimizer)
95+
self.assertEqual(len(epl.Graph.get().gradients), 10)
96+
97+
def test_nested_optimizer_clip_after_allreduce(self):
98+
epl.init(epl.Config({"communication.clip_after_allreduce": False}))
99+
epl.set_default_strategy(epl.replicate(1))
100+
with tf.Graph().as_default():
101+
optimizer = get_nested_optimizer()
102+
res2, _ = _train_fn(bs=1, lr_decay=False, max_steps=5, optimizer=optimizer)
103+
self.assertEqual(len(epl.Graph.get().gradients), 10)
104+
105+
# pylint: enable=missing-docstring,unused-variable
106+
# pylint: enable=protected-access
107+
108+
if __name__ == "__main__":
109+
test.main()

tests/optimizer_helper_test.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727

2828
# pylint: disable=missing-docstring,unused-variable
2929
# pylint: disable=protected-access
30-
class GradientAccumulationTest(test.TestCase):
30+
class OptimizerHelperTest(test.TestCase):
3131

3232
@classmethod
3333
def setUpClass(cls): # pylint: disable=invalid-name

0 commit comments

Comments
 (0)