Using dask-yarn + EMR encountering InvalidStateError when doing cluster.close()

Thank you for reporting an issue.

When reporting an issue, please include the following:

- **A description of the bug**

I created a EMR cluster using the provided bootstrap action. I then ssh-tunnel into the master node. I tried creating a dask cluster on local jupyter notebook. I was able to create a cluster, but when I try to run `cluster.close()`. I get InvalidStateError.

- **Steps to reproduce**
```
from dask_yarn import YarnCluster

cluster = YarnCluster()
client = Client(cluster)

# Scale up to 3 workers
cluster.scale(3)

cluster.close()
```

- **Relevant logs/tracebacks**
```---------------------------------------------------------------------------
CancelledError Traceback (most recent call last)
/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _watch_worker_status(self, comm)
902 try:
--> 903 msgs = await comm.read()
904 except OSError:

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/comm/tcp.py in read(self, deserializers)
185 try:
--> 186 n_frames = await stream.read_bytes(8)
187 n_frames = struct.unpack("Q", n_frames)[0]

CancelledError:

During handling of the above exception, another exception occurred:

InvalidStateError Traceback (most recent call last)
in ()
----> 1 cluster.close()

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in close(self, **kwargs)
740 shutdown
741 """
--> 742 return self.shutdown(**kwargs)
743
744 def __del__(self):

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in shutdown(self, status, diagnostics)
727 return self._stop_internal(status=status, diagnostics=diagnostics)
728 if self.loop.asyncio_loop.is_running() and not sys.is_finalizing():
--> 729 self._sync(self._stop_internal(status=status, diagnostics=diagnostics))
730 else:
731 # Always run this!

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _sync(self, task)
699 future = asyncio.run_coroutine_threadsafe(task, self.loop.asyncio_loop)
700 try:
--> 701 return future.result()
702 except BaseException:
703 future.cancel()

/home/hadoop/miniconda/lib/python3.7/concurrent/futures/_base.py in result(self, timeout)
433 raise CancelledError()
434 elif self._state == FINISHED:
--> 435 return self.__get_result()
436 else:
437 raise TimeoutError()

/home/hadoop/miniconda/lib/python3.7/concurrent/futures/_base.py in __get_result(self)
382 def __get_result(self):
383 if self._exception:
--> 384 raise self._exception
385 else:
386 return self._result

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _stop_internal(self, status, diagnostics)
630 self._stop_async(status=status, diagnostics=diagnostics)
631 )
--> 632 await self._stop_task
633
634 async def _stop_async(self, status="SUCCEEDED", diagnostics=None):

/home/hadoop/miniconda/lib/python3.7/site-packages/IPython/core/interactiveshell.py in run_code(self, code_obj, result)
2876 self.hooks.pre_run_code_hook()
2877 #rprint('Running code', repr(code_obj)) # dbg
-> 2878 exec(code_obj, self.user_global_ns, self.user_ns)
2879 finally:
2880 # Reset our crash handler in place

in ()
----> 1 cluster.close()

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in close(self, **kwargs)
740 shutdown
741 """
--> 742 return self.shutdown(**kwargs)
743
744 def __del__(self):

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in shutdown(self, status, diagnostics)
727 return self._stop_internal(status=status, diagnostics=diagnostics)
728 if self.loop.asyncio_loop.is_running() and not sys.is_finalizing():
--> 729 self._sync(self._stop_internal(status=status, diagnostics=diagnostics))
730 else:
731 # Always run this!

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _sync(self, task)
699 future = asyncio.run_coroutine_threadsafe(task, self.loop.asyncio_loop)
700 try:
--> 701 return future.result()
702 except BaseException:
703 future.cancel()

/home/hadoop/miniconda/lib/python3.7/concurrent/futures/_base.py in result(self, timeout)
433 raise CancelledError()
434 elif self._state == FINISHED:
--> 435 return self.__get_result()
436 else:
437 raise TimeoutError()

/home/hadoop/miniconda/lib/python3.7/concurrent/futures/_base.py in __get_result(self)
382 def __get_result(self):
383 if self._exception:
--> 384 raise self._exception
385 else:
386 return self._result

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _stop_internal(self, status, diagnostics)
630 self._stop_async(status=status, diagnostics=diagnostics)
631 )
--> 632 await self._stop_task
633
634 async def _stop_async(self, status="SUCCEEDED", diagnostics=None):

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _stop_async(self, status, diagnostics)
643
644 if self._watch_worker_status_task is not None:
--> 645 await cancel_task(self._watch_worker_status_task)
646 self._watch_worker_status_task = None
647

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in cancel_task(task)
63 task.cancel()
64 try:
---> 65 await task
66 except asyncio.CancelledError:
67 pass

/home/hadoop/miniconda/lib/python3.7/site-packages/dask_yarn/core.py in _watch_worker_status(self, comm)
922 self = None
923 finally:
--> 924 await comm.close()
925
926 def _widget_status(self):

/home/hadoop/miniconda/lib/python3.7/site-packages/tornado/gen.py in wrapper(*args, **kwargs)
324 try:
325 orig_stack_contexts = stack_context._state.contexts
--> 326 yielded = next(result)
327 if stack_context._state.contexts is not orig_stack_contexts:
328 yielded = _create_future()

/home/hadoop/miniconda/lib/python3.7/site-packages/distributed/comm/tcp.py in close(self)
279 finally:
280 self._finalizer.detach()
--> 281 stream.close()
282
283 def abort(self):

/home/hadoop/miniconda/lib/python3.7/site-packages/tornado/iostream.py in close(self, exc_info)
634 self.close_fd()
635 self._closed = True
--> 636 self._maybe_run_close_callback()
637
638 def _maybe_run_close_callback(self):

/home/hadoop/miniconda/lib/python3.7/site-packages/tornado/iostream.py in _maybe_run_close_callback(self)
653 self._ssl_connect_future = None
654 for future in futures:
--> 655 future.set_exception(StreamClosedError(real_error=self.error))
656 future.exception()
657 if self._close_callback is not None:

InvalidStateError: invalid state```

- **Version information**
- Python version 3.7.7
- Dask-Yarn version 0.8.1
https://github.com/dask/dask-yarn/blob/master/deployment_resources/aws-emr/bootstrap-dask