Dask Update (#1577)

Co-authored-by: Paul Yang <[email protected]>

runhouse/resources/hardware/cluster.py

@@ -2074,16 +2074,16 @@ def connect_dask(
         )
 
         # Note: We need to do this on the head node too, because this creates all the worker processes
-        for node in self.ips:
+        for idx, node in enumerate(self.ips):
             logger.info(f"Starting Dask worker on {node}.")
             # Connect to localhost if on the head node, otherwise use the internal ip of head node
             scheduler = (
                 local_scheduler_address
                 if node == self.head_ip
                 else remote_scheduler_address
             )
-            self.run_bash_over_ssh(
-                f"nohup dask worker {scheduler} {worker_options_str} > dask_worker.out 2>&1 &",
+            self.run(
+                f"nohup dask worker {scheduler} --host {self.internal_ips[idx]} {worker_options_str} > dask_worker.out 2>&1 &",
                 node=node,
             )