title: How to pass futures around to different endpoints in dask
body: <p>I have a dask cluster (not asynchronous) setup in FastAPI, and I have created two endpoints. One submits the function to the dask cluster and the other gets polled by vuejs every so often and once the status is &quot;finished&quot; it stops polling. However dask has the issue that it forgets about futures if it thinks you do not need them. So I have a dictionary to hold onto them (similar to <a href="https://stackoverflow.com/questions/62164283/why-do-my-dask-futures-get-stuck-in-pending-and-never-finish">here</a>. So I used that in my code and in my local container it worked. However deployed in pods on kubernetes I am able to print out the future key and future status to the logs, but the future is not there. The code for this is like as follows:</p>
  <pre><code>futures = {}
  
  @router.post('/simulate')
  async def simulate(request: Request, background_tasks: BackgroundTasks) -&gt; object:
  
      data = await request.json()
      if data is None or not data:
          raise HTTPException(status_code=422, detail={
              &quot;message&quot;: &quot;Json body can't be empty&quot;})
  
      try:
          data_is_cached = await get_cache(key=json.dumps(data, cls=CustomJSONEncoder))
          if data_is_cached:
              return {
                  &quot;is_cached&quot;: True,
                  &quot;result&quot;: data_is_cached
              }
          else:
              # logic to create inputs_dict using data here
              future = client.submit(run_simulation, inputs_dict)
              futures[future.key] = [future, data]
              return {
                  &quot;is_cached&quot;: False,
                  &quot;key&quot;: future.key
              }
      except Exception as e:
          traceback.print_exc()
          raise HTTPException(status_code=500, detail={
              &quot;message&quot;: str(e)})
  
  
  @router.get('/get_status/{key_id}')
  async def check_status(key_id: str, request: Request, background_tasks: BackgroundTasks) -&gt; object:
  
      try:
          future = futures[key_id][0]
          data = futures[key_id][1]
  
          if 'finished' in future.status:
              result = future.result()
              background_tasks.add_task(set_cache, key=json.dumps(data, cls=CustomJSONEncoder), data=result)
              return {
                  &quot;statusFinished&quot;: True,
                  &quot;result&quot;: result
              }
          elif &quot;pending&quot; in future.status:
              return {
                  &quot;statusFinished&quot;: False,
                  &quot;result&quot;: &quot;pending&quot;
              }
          else:
              return {
                  &quot;statusFinished&quot;: False,
                  &quot;result&quot;: &quot;error&quot;
              }
      except Exception as e:
          traceback.print_exc()
          raise HTTPException(status_code=500, detail={
              &quot;message&quot;: str(e)})
  </code></pre>
  <p>I then looked at <a href="https://distributed.dask.org/en/stable/publish.html#" rel="nofollow noreferrer">dask datasets</a> as a dataset can be shared across dask workers. But this always leave the state in &quot;pending&quot; I have to explicitly call <code>future.result()</code> why defies the purpose of what I am trying to do. This was in my container in localhost.</p>
  <p>Here the code for this is:</p>
  <pre><code>@router.post('/simulate')
  async def simulate(request: Request, background_tasks: BackgroundTasks) -&gt; object:
  
      data = await request.json()
      if data is None or not data:
          raise HTTPException(status_code=422, detail={
              &quot;message&quot;: &quot;Json body can't be empty&quot;})
  
      try:
          data_is_cached = await get_cache(key=json.dumps(data, cls=CustomJSONEncoder))
          if data_is_cached:
              return {
                  &quot;is_cached&quot;: True,
                  &quot;result&quot;: data_is_cached
              }
          else:
              # logic to create inputs_dict using data here
              future = client.submit(run_simulation, inputs_dict)
              client.datasets[future.key] = future
              client.datasets[f&quot;{future.key}-data&quot;] = data
              return {
                  &quot;is_cached&quot;: False,
                  &quot;key&quot;: future.key
              }
      except Exception as e:
          traceback.print_exc()
          raise HTTPException(status_code=500, detail={
              &quot;message&quot;: str(e)})
  
  
  @router.get('/get_status/{key_id}')
  async def check_status(key_id: str, request: Request, background_tasks: BackgroundTasks) -&gt; object:
  
      try:
          future = client.get_dataset(key_id)
          data = client.get_dataset(f&quot;{key_id}-data&quot;)
  
          if 'finished' in future.status:
              result = future.result()
              background_tasks.add_task(set_cache, key=json.dumps(data, cls=CustomJSONEncoder), data=result)
              return {
                  &quot;statusFinished&quot;: True,
                  &quot;result&quot;: result
              }
          elif &quot;pending&quot; in future.status:
              return {
                  &quot;statusFinished&quot;: False,
                  &quot;result&quot;: &quot;pending&quot;
              }
          else:
              return {
                  &quot;statusFinished&quot;: False,
                  &quot;result&quot;: &quot;error&quot;
              }
      except Exception as e:
          traceback.print_exc()
          raise HTTPException(status_code=500, detail={
              &quot;message&quot;: str(e)})
  </code></pre>
  <p>I have the <code>dask-worker</code> to have 1 worker and 1 thread. Is there not a way to hold on to these futures properly?</p>