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

Question

По мотивам известного check_blas.py скрипт, я написал этот, чтобы проверить, что 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,863 с против 71,292 с за один прогон.

Итак, я ожидаю, что 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)

Этот скрипт использует только 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 из кераса использовать только 1 ядро ​​для такой же настройки? Это check_blas.py скрипт на самом деле представитель для обучения нейронной сети расчетов?

FYI:

(venv3)herbert@machine:~/ $ 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)herbert@machine:~/ $

РЕДАКТИРОВАТЬ

Я также создал реализацию 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:])

Answer 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 процесса начинают работу по обучению, а затем все они заканчиваются примерно в одно и то же время. Если бы это было единичной обработкой, вам пришлось бы дождаться завершения каждого, прежде чем начинать следующий (зевайте!).