TFRecordReader scheint extrem langsam, und Multi-Thread-Lesen funktioniert nicht

Question

Mein Trainingsprozess verwenden tfrecord Format für Zug & Evaluierungsdatensatz.

Ich teste den Benchmark des Lesers, nur 8000 Datensätze/Sekunde. und io Geschwindigkeit (siehe von iotop Befehl) nur 400KB-500KB/s.

Ich bin mit der CPP-Version von protobuf hier

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

Wenn möglich, ein minimales reproduzierbaren Beispiel bieten (wir haben in der Regel keine Zeit Hunderte von Zeilen des Codes zu lesen)

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), 
     }) 

Welche anderen Lösungsversuche haben Sie versucht?

Ich versuche, num_threads in tf.train.shuffle_batch zu setzen, aber nicht funktioniert.

Es scheint, dass wenn es auf 2 Threads eingestellt ist, es bei 8000records/s arbeitet, wenn die Thread-Nummer vergrößert wird, wird es langsamer. (Ich entferne alle Operationen, die cpus kosten. Einfach Daten lesen.)

Meine Server sind 24 core cpus.

Answer 1

Das Problem hier ist, dass es einen festen Kostenaufwand für jede session.run, und das Füllen der Warteschlange mit vielen kleinen Beispielen auf die Warteschlange wird langsam sein.

Insbesondere ist jeder session.run etwa 100-200 Usec, so dass Sie nur etwa 5k-10k session.run Anrufe pro Sekunde tun können.

Dieses Problem ist offensichtlich, wenn Sie Python-Profiling (Python-m cProfile) tun, aber schwer zu sehen, wenn Sie von Timeline-Profil oder CPU-Profil starten.

Die Problemumgehung ist, enqueue_many zu verwenden, um Dinge in Ihrer Warteschlange in Batches hinzuzufügen. Ich nahm Ihren Benchmark von https://gist.github.com/ericyue/7705407a88e643f7ab380c6658f641e8 und änderte es, um viele Artikel per .run Anruf in die Warteschlange zu setzen, und das gibt 10x beschleunigen.

Die Änderung ist Aufruf wie folgt zu ändern:

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) 

Für eine vollständige Quelle, hier überprüfen: https://github.com/yaroslavvb/stuff/blob/master/ericyue-slowreader/benchmark.py

Es ist schwer, es zu optimieren, da viel schneller geht jetzt die meiste Zeit ist in Warteschlangenoperationen verbracht. Betrachtet man die stripped down Version, die einfach ganze Zahlen zu einer Warteschlange hinzufügt, erhält man auch eine ähnliche Geschwindigkeit, und wenn man sich die Zeitleiste anschaut, wird die Zeit in Warteschlangenoperationen verbracht.

Jedes dequeue op dauert etwa 60 & mgr; s, aber es gibt im Durchschnitt 5 parallel runnning, so dass Sie 12 erhalten usec pro dequeue. Das bedeutet, dass Sie im besten Fall < 200k Beispiele pro Sekunde erhalten.

Answer 2

Hier ist ein einfaches Speedup Gebäude auf Yaroslav Antwort:

Tensorflow in Funktion eine gebaut hat, tf.TFRecordReader.read_up_to, die mehrere Datensätze in jedem session.run() Aufruf liest, wodurch die überschüssige Overhead durch mehrere Anrufe verursacht zu entfernen.

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) 

Wie bei Yaroslav Antwort, müssen Sie enqueue_many=True so einzustellen, dass die Batch-Funktion weiß, dass es mehrere Datensätze akzeptiert.

Das war sehr schnell in meinem Anwendungsfall.

Answer 3

Ein Nachtrag zu Yaroslav Antwort: Sie tf.python_io.tf_record_iterator können durch die Beispiele, um hängen Sie sie auf eine Liste zu durchlaufen, die Sie tf.train.shuffle_batch mit enqueue_many=true passieren kann:

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) 

Es scheint, dass der Versuch, zu durchlaufen Beispiele unter Verwendung von reader.read() führen zu einem gelesenen pro Batch. h. der n-te Stapel wird batch_num Kopien des n-ten Datensatzes sein, anstatt batch_num viele einzigartige Datensätze.