Fix multiprocessing starmap to allow passing in zip (#13664)

python/ray/tests/test_multiprocessing.py

@@ -340,6 +340,7 @@ def test_starmap(pool):
 
     args = [tuple(range(i)) for i in range(100)]
     assert pool.starmap(f, args) == args
+    assert pool.starmap(lambda x, y: x + y, zip([1, 2], [3, 4])) == [4, 6]
 
 
 def test_callbacks(pool_4_processes):

python/ray/util/multiprocessing/pool.py

@@ -494,7 +494,7 @@ class Pool:
     def _chunk_and_run(self, func, iterable, chunksize=None,
                        unpack_args=False):
         if not hasattr(iterable, "__len__"):
-            iterable = [iterable]
+            iterable = list(iterable)
 
         if chunksize is None:
             chunksize = self._calculate_chunksize(iterable)