TFRecordReader có vẻ cực kỳ chậm và đa luồng đọc không hoạt động

Question

Quá trình đào tạo của tôi sử dụng định dạng tfrecord cho đào tạo & eval dataset.

Tôi kiểm tra điểm chuẩn của trình đọc, chỉ 8000records/giây. và tốc độ io (xem từ lệnh iotop) chỉ 400KB-500KB/s.

Tôi đang sử dụng phiên bản cpp của protobuf đây

https://github.com/tensorflow/tensorflow/blob/master/tensorflow/g3doc/get_started/os_setup.md#protobuf-library-related-issues

Nếu có thể, cung cấp một ví dụ tái sản xuất tối thiểu (Chúng tôi thường không có thời gian để đọc hàng trăm dòng mã của bạn)

def read_and_decode(filename_queue): 
    reader = tf.TFRecordReader() 
    _, serialized_example = reader.read(filename_queue) 
    return serialized_example 
    serialized_example = read_and_decode(filename_queue) 
    batch_serialized_example = tf.train.shuffle_batch(
     [serialized_example], 
     batch_size=batch_size, 
     num_threads=thread_number, 
     capacity=capacity, 
     min_after_dequeue=min_after_dequeue) 
    features = tf.parse_example(
     batch_serialized_example, 
     features={ 
      "label": tf.FixedLenFeature([], tf.float32), 
      "ids": tf.VarLenFeature(tf.int64), 
      "values": tf.VarLenFeature(tf.float32), 
     }) 

Bạn đã thử các giải pháp đã thử nào khác?

Tôi cố gắng đặt num_threads bằng tf.train.shuffle_batch nhưng không hoạt động.

Dường như khi đặt thành 2 chuỗi, nó hoạt động ở 8000records/s, khi phóng to số chuỗi, nó sẽ chậm hơn. (Tôi loại bỏ tất cả các ops mà chi phí cpus. Chỉ cần đọc dữ liệu.)

Máy chủ của tôi là 24 lõi CPU.

Answer 1

Vấn đề ở đây là có chi phí cố định trên mỗi session.run và điền hàng đợi với nhiều ví dụ nhỏ vào hàng đợi sẽ chậm.

Cụ thể, mỗi phiên.run là khoảng 100-200 usec, vì vậy bạn chỉ có thể thực hiện khoảng 5k-10k session.run cuộc gọi mỗi giây.

Vấn đề này là hiển nhiên nếu làm hồ sơ Python (python -m cProfile), nhưng khó có thể bắt đầu từ hồ sơ dòng thời gian hay cấu hình CPU.

Công việc xung quanh là sử dụng enqueue_many để thêm mọi thứ vào hàng đợi của bạn theo lô. Tôi đã lấy điểm chuẩn của bạn từ https://gist.github.com/ericyue/7705407a88e643f7ab380c6658f641e8 và sửa đổi nó để enqueue nhiều mặt hàng cho mỗi cuộc gọi .run, và cho phép tăng tốc 10x.

Việc sửa đổi là để sửa đổi cuộc gọi như sau:

if enqueue_many: 
    reader = tf.TFRecordReader(options = tf.python_io.TFRecordOptions(tf.python_io.TFRecordCompressionType.ZLIB)) 
    queue_batch = [] 
    for i in range(enqueue_many_size): 
     _, serialized_example = reader.read(filename_queue) 
     queue_batch.append(serialized_example) 
    batch_serialized_example = tf.train.shuffle_batch(
     [queue_batch], 
     batch_size=batch_size, 
     num_threads=thread_number, 
     capacity=capacity, 
     min_after_dequeue=min_after_dequeue, 
     enqueue_many=True) 

Đối với nguồn hoàn chỉnh, kiểm tra ở đây: https://github.com/yaroslavvb/stuff/blob/master/ericyue-slowreader/benchmark.py

Thật khó để tối ưu hóa nó để đi nhanh hơn nhiều kể từ bây giờ hầu hết thời gian là chi tiêu trong các hoạt động xếp hàng. Nhìn vào phiên bản stripped down mà chỉ thêm số nguyên vào hàng đợi, bạn cũng nhận được tốc độ tương tự, và nhìn vào dòng thời gian, thời gian được sử dụng trong các opque dequeue.

Mỗi op dequeue mất khoảng 60 usec, nhưng có trung bình 5 runnning song song, vì vậy bạn sẽ có được 12 usec mỗi dequeue. Vì vậy, điều đó có nghĩa là bạn sẽ nhận được < 200 nghìn ví dụ mỗi giây trong trường hợp tốt nhất.

Answer 2

Dưới đây là một tòa nhà tăng tốc đơn giản về câu trả lời của Yaroslav:

Tensorflow đã tích hợp sẵn chức năng, tf.TFRecordReader.read_up_to, mà đọc nhiều bản ghi trong mỗi session.run() cuộc gọi, do đó loại bỏ các chi phí dư thừa gây ra bởi nhiều cuộc gọi.

enqueue_many_size = SOME_ENQUEUE_MANY_SIZE 
reader = tf.TFRecordReader(options = tf.python_io.TFRecordOptions(tf.python_io.TFRecordCompressionType.ZLIB)) 
_, queue_batch = reader.read_up_to(filename_queue, enqueue_many_size) 
batch_serialized_example = tf.train.shuffle_batch(
    [queue_batch], 
    batch_size=batch_size, 
    num_threads=thread_number, 
    capacity=capacity, 
    min_after_dequeue=min_after_dequeue, 
    enqueue_many=True) 

Như với câu trả lời của Yaroslav, bạn cần đặt enqueue_many=True để chức năng hàng loạt biết rằng nó đang chấp nhận nhiều bản ghi.

Điều này rất nhanh trong trường hợp sử dụng của tôi.

Answer 3

Một phụ lục để trả lời Yaroslav của: Bạn có thể sử dụng tf.python_io.tf_record_iterator để lặp qua các ví dụ để thêm chúng vào một danh sách mà bạn có thể vượt qua để tf.train.shuffle_batch với enqueue_many=true:

queue_batch = [] 
for serialized_example in tf.python_io.tf_record_iterator(filename,options = tf.python_io.TFRecordOptions(tf.python_io.TFRecordCompressionType.ZLIB)): 
    queue_batch.append(serialized_example) 
batch_serialized_example = tf.train.shuffle_batch(
    [queue_batch], 
    batch_size=batch_size, 
    num_threads=thread_number, 
    capacity=capacity, 
    min_after_dequeue=min_after_dequeue, 
    enqueue_many=True) 

Dường như cố gắng để lặp qua ví dụ bằng cách sử dụng reader.read() sẽ cho kết quả một lần đọc mỗi lô. tức là hàng thứ n sẽ là batch_num bản sao của bản ghi thứ n thay vì batch_num nhiều bản ghi duy nhất.