Timeout on write_item?

[google-deepmind]

Timeout on write_item?

[acassirer]

Hey,

Sorry for the slow response!

I'm afraid that there currently isn't any timeout support in the writer. We have a backlog entry for the feature but haven't gotten around to implementing it yet as this isn't much of an issue for our own experiments where failing tasks are automatically restarted (and thus the blocking makes sense as the call eventually succeeds).

I'll bump up the priority on this feature but if this is causing you a lot of pain then maybe you could try offloading the call to a threadpool? You could then apply a timeout on the future instead.

Something along these lines maybe:

import concurrent.futures
import reverb

_TIMEOUT = 10  # Seconds.

def main():
    client = reverb.Client('localhost:8000')

    executor = futures.ThreadPoolExecutor(max_workers=1)
    with client.writer(max_sequence_length=2) as writer:
      for i in range(1000):
        try:
          # In general writer.append can also block due to communication.
          executor.submit(writer.append, tf.constant([1.]).result(_TIMEOUT)
          if i > 0:
            print(f'creating item for i={i}')
            executor.submit(writer.create_item, 'my_table', 2, 1.0).result(_TIMEOUT)
            print(f'wrote item for i={i}')
            time.sleep(0.5)
         except futures.TimeoutError:
           print('closing')

[jmacglashan]

Okay, thanks! That might be a reasonable work around for now

[ebrevdo]

This is fixed in the new TrajectoryWriter flush call (create_item is now asynchronous in the new API).