title: Dask function that sleeps without occupying a worker?
body: <p>In certain classes of data pipelines it is useful to wait for some external process to finish by (say) watching if a file is written.<br>
  Implementing this naively in dask causes a long running task that blocks a worker for its entire duration. </p>
  
  <pre><code>def wait_for_file(filename='some_filename', max_wait_time=600):
      start_time = time.time()
      while True:
          if time.time() - start_time &gt; max_wait_time:
              raise Exception('Timeout')
          if exists(filename):
              return filename
          time.sleep(0.1)
  
  file_exists = delayed(wait_for_file)()    
  res = delayed(process_file)(file_exists)
  </code></pre>
  
  <p>How do i make this code not block a worker</p>