Подтвердить что ты не робот

Keras не использует несколько ядер

Основываясь на знаменитом check_blas.py script, я написал это, чтобы проверить, что theano может фактически использовать несколько ядер:

import os
os.environ['MKL_NUM_THREADS'] = '8'
os.environ['GOTO_NUM_THREADS'] = '8'
os.environ['OMP_NUM_THREADS'] = '8'
os.environ['THEANO_FLAGS'] = 'device=cpu,blas.ldflags=-lblas -lgfortran'

import numpy
import theano
import theano.tensor as T

M=2000
N=2000
K=2000
iters=100
order='C'

a = theano.shared(numpy.ones((M, N), dtype=theano.config.floatX, order=order))
b = theano.shared(numpy.ones((N, K), dtype=theano.config.floatX, order=order))
c = theano.shared(numpy.ones((M, K), dtype=theano.config.floatX, order=order))
f = theano.function([], updates=[(c, 0.4 * c + .8 * T.dot(a, b))])

for i in range(iters):
    f(y)

Выполнение этого как python3 check_theano.py показывает, что используются 8 потоков. И что более важно, код работает примерно в 9 раз быстрее, чем без параметров os.environ, которые применяют только 1 ядро: 7.863s против 71.292s за один прогон.

Итак, я ожидал бы, что Keras теперь также использует несколько ядер при вызове fit (или predict, если на то пошло). Однако это не относится к следующему коду:

import os
os.environ['MKL_NUM_THREADS'] = '8'
os.environ['GOTO_NUM_THREADS'] = '8'
os.environ['OMP_NUM_THREADS'] = '8'
os.environ['THEANO_FLAGS'] = 'device=cpu,blas.ldflags=-lblas -lgfortran'

import numpy
from keras.models import Sequential
from keras.layers import Dense

coeffs = numpy.random.randn(100)

x = numpy.random.randn(100000, 100);
y = numpy.dot(x, coeffs) + numpy.random.randn(100000) * 0.01

model = Sequential()
model.add(Dense(20, input_shape=(100,)))
model.add(Dense(1, input_shape=(20,)))
model.compile(optimizer='rmsprop', loss='categorical_crossentropy')

model.fit(x, y, verbose=0, nb_epoch=10)

Этот script использует только 1 ядро ​​с этим выходом:

Using Theano backend.
/home/herbert/venv3/lib/python3.4/site-packages/theano/tensor/signal/downsample.py:5: UserWarning: downsample module has been moved to the pool module.
warnings.warn("downsample module has been moved to the pool module.")

Почему fit Keras использует только 1 ядро ​​для одной и той же настройки? Является ли фактический представитель check_blas.py script для расчетов обучения нейронной сети?

FYI:

(venv3)[email protected]:~/ $ python3 -c 'import numpy, theano, keras; print(numpy.__version__); print(theano.__version__); print(keras.__version__);'
ERROR (theano.sandbox.cuda): nvcc compiler not found on $PATH. Check your nvcc installation and try again.
1.11.0
0.8.0rc1.dev-e6e88ce21df4fbb21c76e68da342e276548d4afd
0.3.2
(venv3)[email protected]:~/ $

ИЗМЕНИТЬ

Я создал приложение Theano для простого MLP, которое также не запускает многоядерные:

import os
os.environ['MKL_NUM_THREADS'] = '8'
os.environ['GOTO_NUM_THREADS'] = '8'
os.environ['OMP_NUM_THREADS'] = '8'
os.environ['THEANO_FLAGS'] = 'device=cpu,blas.ldflags=-lblas -lgfortran'

import numpy
import theano
import theano.tensor as T

M=2000
N=2000
K=2000
iters=100
order='C'

coeffs = numpy.random.randn(100)
x = numpy.random.randn(100000, 100).astype(theano.config.floatX)
y = (numpy.dot(x, coeffs) + numpy.random.randn(100000) * 0.01).astype(theano.config.floatX).reshape(100000, 1)

x_shared = theano.shared(x)
y_shared = theano.shared(y)

x_tensor = T.matrix('x')
y_tensor = T.matrix('y')

W0_values = numpy.asarray(
    numpy.random.uniform(
        low=-numpy.sqrt(6. / 120),
        high=numpy.sqrt(6. / 120),
        size=(100, 20)
    ),
    dtype=theano.config.floatX
)
W0 = theano.shared(value=W0_values, name='W0', borrow=True)

b0_values = numpy.zeros((20,), dtype=theano.config.floatX)
b0 = theano.shared(value=b0_values, name='b0', borrow=True)

output0 = T.dot(x_tensor, W0) + b0

W1_values = numpy.asarray(
    numpy.random.uniform(
        low=-numpy.sqrt(6. / 120),
        high=numpy.sqrt(6. / 120),
        size=(20, 1)
    ),
    dtype=theano.config.floatX
)
W1 = theano.shared(value=W1_values, name='W1', borrow=True)

b1_values = numpy.zeros((1,), dtype=theano.config.floatX)
b1 = theano.shared(value=b1_values, name='b1', borrow=True)

output1 = T.dot(output0, W1) + b1

params = [W0, b0, W1, b1]
cost = ((output1 - y_tensor) ** 2).sum()

gradients = [T.grad(cost, param) for param in params]

learning_rate = 0.0000001

updates = [
    (param, param - learning_rate * gradient)
    for param, gradient in zip(params, gradients)
]

train_model = theano.function(
    inputs=[],#x_tensor, y_tensor],
    outputs=cost,
    updates=updates,
    givens={
        x_tensor: x_shared,
        y_tensor: y_shared
    }
)

errors = []
for i in range(1000):
    errors.append(train_model())

print(errors[0:50:])
4b9b3361

ОТВЕТЫ

Ответ 1

Сами Keras и TF не используют целые ядра и емкость процессора! Если вы заинтересованы в использовании всех 100% вашего ЦП, тогда multiprocessing.Pool основном создает пул заданий, которые необходимо выполнить. Процессы подберут эти задания и запустят их. Когда работа закончена, процесс выберет другую работу из пула.

NB. Если вы хотите просто ускорить эту модель, посмотрите на графические процессоры или измените гиперпараметры, такие как размер пакета и количество нейронов (размер слоя).

Вот как вы можете использовать multiprocessing для одновременного обучения нескольких моделей (используя процессы, выполняющиеся параллельно на каждом отдельном ядре ЦП вашей машины).

Этот ответ вдохновлен @repploved

import time
import signal
import multiprocessing

def init_worker():
    ''' Add KeyboardInterrupt exception to mutliprocessing workers '''
    signal.signal(signal.SIGINT, signal.SIG_IGN)


def train_model(layer_size):
    '''
    This code is parallelized and runs on each process
    It trains a model with different layer sizes (hyperparameters)
    It saves the model and returns the score (error)
    '''
    import keras
    from keras.models import Sequential
    from keras.layers import Dense

    print(f'Training a model with layer size {layer_size}')

    # build your model here
    model_RNN = Sequential()
    model_RNN.add(Dense(layer_size))

    # fit the model (the bit that takes time!)
    model_RNN.fit(...)

    # lets demonstrate with a sleep timer
    time.sleep(5)

    # save trained model to a file
    model_RNN.save(...)

    # you can also return values eg. the eval score
    return model_RNN.evaluate(...)


num_workers = 4
hyperparams = [800, 960, 1100]

pool = multiprocessing.Pool(num_workers, init_worker)

scores = pool.map(train_model, hyperparams)

print(scores)

Выход:

Training a model with layer size 800
Training a model with layer size 960
Training a model with layer size 1100
[{'size':960,'score':1.0}, {'size':800,'score':1.2}, {'size':1100,'score':0.7}]

Это легко продемонстрировать с помощью time.sleep в коде. Вы увидите, что все 3 процесса начинают учебную работу, а затем все они заканчиваются примерно в одно и то же время. Если бы это было обработано один раз, вам пришлось бы ждать завершения каждого из них, прежде чем начинать следующее (зевать!).