theano - 加载小批量的有效方法 < GPU 内存

Question

我有以下情况:

• 我的数据集 >> GPU 内存
• 我的小批量 < GPU 内存...这样，根据大小，我可以一次在内存中容纳最多 10 个小批量，同时仍然可以毫无问题地进行训练。

我的数据集的大小意味着我不会重新访问数据点，所以我想共享它们没有意义？或者有吗？我在想，也许拥有最多 10 个 size=mini-batch 的共享初始化变量会很有好处，这样我就可以一次交换 10 个变量，而不是一次只交换一个变量。另外，是否可以并行预加载小批量？

Answer 1

如果您不重新访问数据点，那么使用共享变量可能没有任何值(value)。

可以修改以下代码并用于评估将数据获取到特定计算中的不同方法。

当您不需要重新访问数据时，“输入”方法可能是最好的方法。 “shared_all”方法可能会优于其他方法，但前提是您可以将整个数据集放入 GPU 内存中。 “shared_batched”允许您评估分层批处理数据是否有帮助。

在“shared_batched”方法中，数据集被分为许多宏批处理，每个宏批处理又被分为许多微批处理。单个共享变量用于保存单个宏批处理。该代码评估当前宏批处理中的所有微批处理。一旦处理完一个完整的宏批处理，下一个宏批处理就会加载到共享变量中，并且代码会再次迭代其中的微批处理。

一般来说，可以预期少量的大内存传输会比大量的小传输运行得更快(其中每个传输的总传输量是相同的)。但这需要经过测试(例如使用下面的代码)才能确定； YMMV。

使用“借用”参数也可能对性能产生重大影响，但要注意the implications使用前。

import math
import timeit
import numpy
import theano
import theano.tensor as tt


def test_input(data, batch_size):
    assert data.shape[0] % batch_size == 0
    batch_count = data.shape[0] / batch_size
    x = tt.tensor4()
    f = theano.function([x], outputs=x.sum())
    total = 0.
    start = timeit.default_timer()
    for batch_index in xrange(batch_count):
        total += f(data[batch_index * batch_size: (batch_index + 1) * batch_size])
    print 'IN\tNA\t%s\t%s\t%s\t%s' % (batch_size, batch_size, timeit.default_timer() - start, total)


def test_shared_all(data, batch_size):
    batch_count = data.shape[0] / batch_size
    for borrow in (True, False):
        start = timeit.default_timer()
        all = theano.shared(data, borrow=borrow)
        load_time = timeit.default_timer() - start
        x = tt.tensor4()
        i = tt.lscalar()
        f = theano.function([i], outputs=x.sum(), givens={x: all[i * batch_size:(i + 1) * batch_size]})
        total = 0.
        start = timeit.default_timer()
        for batch_index in xrange(batch_count):
            total += f(batch_index)
        print 'SA\t%s\t%s\t%s\t%s\t%s' % (
            borrow, batch_size, batch_size, load_time + timeit.default_timer() - start, total)


def test_shared_batched(data, macro_batch_size, micro_batch_size):
    assert data.shape[0] % macro_batch_size == 0
    assert macro_batch_size % micro_batch_size == 0
    macro_batch_count = data.shape[0] / macro_batch_size
    micro_batch_count = macro_batch_size / micro_batch_size
    macro_batch = theano.shared(numpy.empty((macro_batch_size,) + data.shape[1:], dtype=theano.config.floatX),
                                borrow=True)
    x = tt.tensor4()
    i = tt.lscalar()
    f = theano.function([i], outputs=x.sum(), givens={x: macro_batch[i * micro_batch_size:(i + 1) * micro_batch_size]})
    for borrow in (True, False):
        total = 0.
        start = timeit.default_timer()
        for macro_batch_index in xrange(macro_batch_count):
            macro_batch.set_value(
                data[macro_batch_index * macro_batch_size: (macro_batch_index + 1) * macro_batch_size], borrow=borrow)
            for micro_batch_index in xrange(micro_batch_count):
                total += f(micro_batch_index)
        print 'SB\t%s\t%s\t%s\t%s\t%s' % (
            borrow, macro_batch_size, micro_batch_size, timeit.default_timer() - start, total)


def main():
    numpy.random.seed(1)

    shape = (20000, 3, 32, 32)

    print 'Creating random data with shape', shape
    data = numpy.random.standard_normal(size=shape).astype(theano.config.floatX)

    print 'Running tests'
    for macro_batch_size in (shape[0] / pow(10, i) for i in xrange(int(math.log(shape[0], 10)))):
        test_shared_all(data, macro_batch_size)
        test_input(data, macro_batch_size)
        for micro_batch_size in (macro_batch_size / pow(10, i) for i in
                                 xrange(int(math.log(macro_batch_size, 10)) + 1)):
            test_shared_batched(data, macro_batch_size, micro_batch_size)


main()