Skip to content

Commit 30118cd

Browse files
ASvyatkovskiyASvyatkovskiy
committed
Merge branch 'master' of https://github.com/PPPLDeepLearning/plasma-python into keras-2.0_migration
2 parents fa78e70 + 23933a2 commit 30118cd

7 files changed

Lines changed: 247 additions & 121 deletions

File tree

‎examples/compare_batch_iterators.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ def train_epochs(self,M):
4242

4343
try:
4444
batch = batch_generator_func.next()
45-
except:
45+
except StopIteration:
4646
batch_generator_func = self.batch_iterator()
4747
batch = batch_generator_func.next()
4848
print ("Next batch id: {}".format(batch))

‎examples/conf.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,7 @@ model:
8484
dropout_prob: 0.3
8585
#only relevant if we want to do mpi training. The number of steps with a single replica
8686
warmup_steps: 0
87+
backend: 'theano'
8788

8889
training:
8990
as_array_of_shots: True

‎plasma/models/loader.py‎

Lines changed: 80 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -14,16 +14,40 @@
1414
from plasma.primitives.shots import Shot
1515

1616
class Loader(object):
17+
'''
18+
A Python class to ...
19+
20+
The length of shots in e.g. JET data varies by orders of magnitude. For data parallel
21+
synchronous training it is essential that amounds of train data passed to the model replica is about the same size.
22+
Therefore, a patching technique is introduced.
23+
24+
A patch is a subset of shot's time/signal profile having a fixed length, equal among all patches.
25+
Patch size is approximately equal to the minimum shot length. More precisely: it is equal
26+
to the max(1, min_len//rnn_length)*rnn_length - the largest number less or equal to the minimum shot length divisible by the LSTM model length. If minimum shot length is less than the rnn_length, then the patch length is equal to the rnn_length
27+
'''
28+
1729
def __init__(self,conf,normalizer=None):
1830
self.conf = conf
1931
self.stateful = conf['model']['stateful']
2032
self.normalizer = normalizer
2133
self.verbose = True
2234

2335
def training_batch_generator(self,shot_list):
24-
"""Iterates indefinitely over the data set and returns one batch of data at a time.
25-
Can be inefficient during distributed training because one process loading data will
26-
cause all other processes to stall."""
36+
"""
37+
The method implements a training batch generator as a Python generator with a while-loop.
38+
It iterates indefinitely over the data set and returns one mini-batch of data at a time.
39+
40+
NOTE: Can be inefficient during distributed training because one process loading data will
41+
cause all other processes to stall.
42+
43+
Argument list:
44+
- shot_list:
45+
46+
Returns:
47+
- One mini-batch of data and label as a Numpy array: X[start:end],y[start:end]
48+
- reset_states_now: boolean flag indicating when to reset state during stateful RNN training
49+
- num_so_far,num_total: number of samples generated so far and the total dataset size as per shot_list
50+
"""
2751
batch_size = self.conf['training']['batch_size']
2852
num_at_once = self.conf['training']['num_shots_at_once']
2953
epoch = 0
@@ -42,11 +66,13 @@ def training_batch_generator(self,shot_list):
4266
num_examples = X.shape[0]
4367
assert(num_examples % batch_size == 0)
4468
num_chunks = num_examples/batch_size
45-
"""produce batch-sized data X,y to feed during training.
46-
dimensions are (num_examples, num_timesteps, num_dimensions_of_data)
47-
also num_examples is divisible by the batch_size. The ith example and the
48-
(batchsize + 1)th example are consecutive in time, so we do not reset the
49-
RNN internal state unless we start a new chunk."""
69+
"""
70+
The method produces batch-sized training data X and labels y as Numpy arrays to feed during training.
71+
Mini-batch dimensions are (num_examples, num_timesteps, num_dimensions_of_data)
72+
also num_examples has to be divisible by the batch_size. The i-th example and the
73+
(batchsize + 1)-th example are consecutive in time, so we do not reset the
74+
RNN internal state unless we start a new chunk.
75+
"""
5076
for k in range(num_chunks):
5177
#epoch_end = (i == len(shot_sublists) - 1 and j == len(X_list) -1 and k == num_chunks - 1)
5278
reset_states_now = (k == 0)
@@ -62,8 +88,23 @@ def training_batch_generator(self,shot_list):
6288

6389

6490
def load_as_X_y_list(self,shot_list,verbose=False,prediction_mode=False):
65-
"""Turn a list of shots into a set of equal-sized patches which contain a number of examples
66-
that is a multiple of the batch size."""
91+
"""
92+
The method turns a ShotList into a set of equal-sized patches which contain a number of examples
93+
that is a multiple of the batch size.
94+
Initially, shots are "light" meaning signal amd disruption related attributes are not filled.
95+
By invoking Loader.get_signals_results_from_shotlist the shot information is filled and stored in
96+
the object in memory. Next, patches are made, finally patches are arranged into batch input shape expected by RNN model.
97+
98+
Performs calls to: get_signals_results_from_shotlist, make_patches, arange_patches
99+
100+
Argument list:
101+
- shot_list: a ShotList
102+
- verbose: TO BE DEPRECATED, self.verbose data member is used instead
103+
- prediction_mode: unused
104+
105+
Returns:
106+
- X_list,y_list: lists of Numpy arrays of batch input shape
107+
"""
67108
signals,results,total_length = self.get_signals_results_from_shotlist(shot_list)
68109
sig_patches, res_patches = self.make_patches(signals,results)
69110

@@ -192,13 +233,41 @@ def get_max_len(self,arrs,length):
192233
return max_len
193234

194235
def make_patches(self,signals,results):
236+
"""
237+
A patch is a subset of shot's time/signal profile having a fixed length, equal among all patches.
238+
Patch size is approximately equal to the minimum shot length. More precisely: it is equal
239+
to the max(1, min_len//rnn_length)*rnn_length - the largest number less or equal to the minimum shot length divisible by the LSTM model length. If minimum shot length is less than the rnn_length, then the patch length is equal to the rnn_length
240+
241+
Since shot lengthes are not multiples of the minimum shot length in general,
242+
some non-deterministic fraction of patches is created. See:
243+
244+
Deterministic patching:
245+
246+
Random patching:
247+
248+
249+
Argument list:
250+
- signals: a list of 1D Numpy array of doubles containing signal values (a plasma property).
251+
Numpy arrays are shot-sized
252+
- results: a list of 1D Numpy array of doubles containing disruption times or -1 if a shot
253+
is non-disruptive. Numpy arrays are shot-sized
254+
255+
NOTE: signals and results are parallel lists. Since Arrays are shot-sized, the shape veries across the list
256+
257+
Returns:
258+
- sig_patches_det + sig_patches_rand: (concatenated) list of 1D Numpy arrays of doubles containing signal values.
259+
Numpy arrays are patch-sized
260+
- res_patches_det + res_patches_rand: (concatenated) a list of 1D Numpy array of doubles containing disruption times
261+
or -1 if a shot is non-disruptive. Numpy arrays are patch-sized
262+
NOTE: sig_patches_det + sig_patches_rand and res_patches_det + res_patches_rand are prallel lists
263+
All arrays in the list have identical shapes.
264+
"""
195265
total_num = self.conf['training']['batch_size']
196266
sig_patches_det,res_patches_det = self.make_deterministic_patches(signals,results)
197267
num_already = len(sig_patches_det)
198268

199269
total_num = int(np.ceil(1.0 * num_already / total_num)) * total_num
200270

201-
202271
num_additional = total_num - num_already
203272
assert(num_additional >= 0)
204273
sig_patches_rand,res_patches_rand = self.make_random_patches(signals,results,num_additional)
@@ -207,7 +276,6 @@ def make_patches(self,signals,results):
207276
return sig_patches_det + sig_patches_rand,res_patches_det + res_patches_rand
208277

209278

210-
211279
def make_prediction_patches(self,signals,results):
212280
#total_num = self.conf['training']['batch_size']
213281
num_timesteps = self.conf['model']['pred_length']
@@ -394,80 +462,3 @@ def load_shotlists(self,conf):
394462
shot_list_validate = data['shot_list_validate'][()]
395463
shot_list_test = data['shot_list_test'][()]
396464
return shot_list_train,shot_list_validate,shot_list_test
397-
398-
# def produce_indices(signals_list):
399-
# indices_list = []
400-
# for xx in signals_list:
401-
# indices_list.append(arange(len(xx)))
402-
# return indices_list
403-
404-
405-
406-
# def array_to_path_and_external_pred(self,arr,res,return_sequences=False):
407-
# length = self.conf['model']['length']
408-
# skip = self.conf['model']['skip']
409-
# assert(shape(arr)[0] == shape(res)[0])
410-
# X = []
411-
# y = []
412-
# i = 0
413-
# while True:
414-
# pred = i+length
415-
# if pred > len(arr):
416-
# break
417-
# X.append(arr[i:i+length,:])
418-
# if return_sequences:
419-
# y.append(res[i:i+length])
420-
# else:
421-
# y.append(res[i+length-1])
422-
# i += skip
423-
# X = array(X)
424-
# y = array(y)
425-
# if len(shape(X)) == 1:
426-
# X = np.expand_dims(X,axis=len(shape(X)))
427-
# if return_sequences and len(shape(y)) == 1:
428-
# y = np.expand_dims(y,axis=len(shape(y)))
429-
# return X,y
430-
431-
432-
# def array_to_path_and_next(self,arr):
433-
# length = self.conf['model']['length']
434-
# skip = self.conf['model']['skip']
435-
# X = []
436-
# y = []
437-
# i = 0
438-
# while True:
439-
# pred = i+length
440-
# if pred >= len(arr):
441-
# break
442-
# X.append(arr[i:i+length])
443-
# y.append(arr[i+length])
444-
# i += skip
445-
# X = array(X)
446-
# X = np.expand_dims(X,axis=len(shape(X)))
447-
# return X,array(y)
448-
449-
450-
# def array_to_path(self,arr):
451-
# length = self.conf['model']['length']
452-
# skip = self.conf['model']['skip']
453-
# X = []
454-
# i = 0
455-
# while True:
456-
# pred = i+length
457-
# if pred > len(arr):
458-
# break
459-
# X.append(arr[i:i+length,:])
460-
# i += skip
461-
# X = array(X)
462-
# if len(shape(X)) == 1:
463-
# X = np.expand_dims(X,axis=len(shape(X)))
464-
# return X
465-
466-
467-
#unused: handling sequences of shots
468-
# def load_shots_as_X_y(conf,shots,verbose=False,stateful=True,prediction_mode=False):
469-
# X,y = zip(*[load_shot_as_X_y(conf,shot,verbose,stateful,prediction_mode) for shot in shots])
470-
# return vstack(X),hstack(y)
471-
472-
# def load_shots_as_X_y_list(conf,shots,verbose=False,stateful=True,prediction_mode=False):
473-
# return [load_shot_as_X_y(conf,shot,verbose,stateful,prediction_mode) for shot in shots]

‎plasma/models/mpi_runner.py‎

Lines changed: 86 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -34,16 +34,22 @@
3434
num_workers = comm.Get_size()
3535
NUM_GPUS = 4
3636
MY_GPU = task_index % NUM_GPUS
37-
backend = 'theano'
3837

3938
from pprint import pprint
4039
from plasma.conf import conf
4140

41+
backend = conf['model']['backend']
42+
4243
if backend == 'tf' or backend == 'tensorflow':
4344
os.environ['CUDA_VISIBLE_DEVICES'] = '{}'.format(MY_GPU)#,mode=NanGuardMode'
4445
os.environ['KERAS_BACKEND'] = 'tensorflow'
45-
import tensorflow
46+
import tensorflow as tf
47+
from keras.backend.tensorflow_backend import set_session
48+
gpu_options = tf.GPUOptions(per_process_gpu_memory_fraction=0.95, allow_growth=True)
49+
config = tf.ConfigProto(gpu_options=gpu_options)
50+
set_session(tf.Session(config=config))
4651
else:
52+
os.environ['KERAS_BACKEND'] = 'theano'
4753
base_compile_dir = '{}/tmp/{}-{}'.format(conf['paths']['output_path'],socket.gethostname(),task_index)
4854
os.environ['THEANO_FLAGS'] = 'device=gpu{},floatX=float32,base_compiledir={}'.format(MY_GPU,base_compile_dir)#,mode=NanGuardMode'
4955
import theano
@@ -173,6 +179,22 @@ def compile(self,loss='mse'):
173179

174180

175181
def get_deltas(self,X_batch,Y_batch,verbose=False):
182+
'''
183+
The purpose of the method is to perform a single gradient update over one mini-batch for one model replica.
184+
Given a mini-batch, it first accesses the current model weights, performs single gradient update over one mini-batch,
185+
gets new model weights, calculates weight updates (deltas) by subtracting weight scalars, applies the learning rate.
186+
187+
It performs calls to: subtract_params, multiply_params
188+
189+
Argument list:
190+
- X_batch: input data for one mini-batch as a Numpy array
191+
- Y_batch: labels for one mini-batch as a Numpy array
192+
- verbose: set verbosity level (currently unused)
193+
194+
Returns:
195+
- deltas: a list of model weight updates
196+
- loss: scalar training loss
197+
'''
176198
weights_before_update = self.model.get_weights()
177199

178200
loss = self.model.train_on_batch(X_batch,Y_batch)
@@ -202,12 +224,34 @@ def mpi_average_gradients(self,arr,num_replicas=None):
202224

203225

204226
def mpi_average_scalars(self,val,num_replicas=None):
227+
'''
228+
The purpose of the method is to calculate a simple scalar arithmetic mean over num_replicas.
229+
230+
It performs calls to: MPIModel.mpi_sum_scalars
231+
232+
Argument list:
233+
- val: value averaged, scalar
234+
- num_replicas: the size of the ensemble an average is perfromed over
235+
236+
Returns:
237+
- val_global: scalar arithmetic mean over num_replicas
238+
'''
205239
val_global = self.mpi_sum_scalars(val,num_replicas)
206240
val_global /= num_replicas
207241
return val_global
208242

209243

210244
def mpi_sum_scalars(self,val,num_replicas=None):
245+
'''
246+
The purpose of the method is to calculate a simple scalar arithmetic mean over num_replicas using MPI allreduce action with fixed op=MPI.SIM
247+
248+
Argument list:
249+
- val: value averaged, scalar
250+
- num_replicas: the size of the ensemble an average is perfromed over
251+
252+
Returns:
253+
- val_global: scalar arithmetic mean over num_replicas
254+
'''
211255
if num_replicas == None:
212256
num_replicas = self.num_workers
213257
if self.task_index >= num_replicas:
@@ -240,11 +284,24 @@ def set_new_weights(self,deltas,num_replicas=None):
240284
self.model.set_weights(new_weights)
241285

242286
def build_callbacks(self,conf,callbacks_list):
243-
#prepare callbacks to pass here
244-
#other possible Callbacks to add: RemoteMonitor, LearningRateScheduler
245-
#https://github.com/fchollet/keras/blob/fbc9a18f0abc5784607cd4a2a3886558efa3f794/keras/callbacks.py
287+
'''
288+
The purpose of the method is to set up logging and history. It is based on Keras Callbacks
289+
https://github.com/fchollet/keras/blob/fbc9a18f0abc5784607cd4a2a3886558efa3f794/keras/callbacks.py
290+
291+
Currently used callbacks include: BaseLogger, CSVLogger, EarlyStopping.
292+
Other possible callbacks to add in future: RemoteMonitor, LearningRateScheduler
293+
294+
Argument list:
295+
- conf: There is a "callbacks" section in conf.yaml file. Relevant parameters are:
296+
list: Parameter specifying additional callbacks, read in the driver script and passed as an argument of type list (see next arg)
297+
metrics: List of quantities monitored during training and validation
298+
mode: one of {auto, min, max}. The decision to overwrite the current save file is made based on either the maximization or the minimization of the monitored quantity. For val_acc, this should be max, for val_loss this should be min, etc. In auto mode, the direction is automatically inferred from the name of the monitored quantity.
299+
monitor: Quantity used for early stopping, has to be from the list of metrics
300+
patience: Number of epochs used to decide on whether to apply early stopping or continue training
301+
- callbacks_list: uses callbacks.list configuration parameter, specifies the list of additional callbacks
302+
Returns: modified list of callbacks
303+
'''
246304

247-
#potentially move to conf.yaml
248305
mode = conf['callbacks']['mode']
249306
monitor = conf['callbacks']['monitor']
250307
patience = conf['callbacks']['patience']
@@ -264,6 +321,29 @@ def build_callbacks(self,conf,callbacks_list):
264321

265322

266323
def train_epoch(self):
324+
'''
325+
The purpose of the method is to perform distributed mini-batch SGD for one epoch.
326+
It takes the batch iterator function and a NN model from MPIModel object, fetches mini-batches
327+
in a while-loop until number of samples seen by the ensemble of workers (num_so_far) exceeds the
328+
training dataset size (num_total).
329+
330+
During each iteration, the gradient updates (deltas) and the loss are calculated for each model replica
331+
in the ensemble, weights are averaged over ensemble, and the new weights are set.
332+
333+
It performs calls to: MPIModel.get_deltas, MPIModel.set_new_weights methods
334+
335+
Argument list: Empty
336+
337+
Returns:
338+
- step: epoch number
339+
- ave_loss: training loss averaged over replicas
340+
- curr_loss:
341+
- num_so_far: the number of samples seen by ensemble of replicas to a current epoch (step)
342+
343+
Intermediate outputs and logging: debug printout of task_index (MPI), epoch number, number of samples seen to
344+
a current epoch, average training loss
345+
'''
346+
267347
verbose = False
268348
step = 0
269349
loss_averager = Averager()

0 commit comments

Comments
 (0)