test load balancing with map

distributed/client.py

@@ -1320,8 +1320,10 @@ def map(self, func, *iterables, **kwargs):
         user_priority = kwargs.pop('priority', 0)
         allow_other_workers = kwargs.pop('allow_other_workers', False)
         fifo_timeout = kwargs.pop('fifo_timeout', '100ms')
+        actor = kwargs.pop('actor', False)
         actors = kwargs.pop('actors', False)
-        pure = kwargs.pop('pure', not actors)
+        actor = actor or actors
+        pure = kwargs.pop('pure', not actor)
 
         if allow_other_workers and workers is None:
             raise ValueError("Only use allow_other_workers= if using workers=")
@@ -1380,7 +1382,7 @@ def map(self, func, *iterables, **kwargs):
                                          retries=retries,
                                          user_priority=user_priority,
                                          fifo_timeout=fifo_timeout,
-                                         actors=actors)
+                                         actors=actor)
         logger.debug("map(%s, ...)", funcname(func))
 
         return [futures[tokey(k)] for k in keys]

distributed/tests/test_actor.py

@@ -390,3 +390,18 @@ def __init__(self, x):
 
     assert s.tasks[x.key].who_has == {ws}  # first went to best match
     assert s.tasks[x.key].who_has != s.tasks[y.key].who_has  # second load balanced
+
+
+@gen_cluster(client=True, ncores=[('127.0.0.1', 1)] * 5)
+def test_load_balance_map(c, s, *workers):
+    class Foo(object):
+        def __init__(self, x, y=None):
+            pass
+
+    b = c.submit(operator.mul, 'b', 1000000)
+    yield wait(b)
+
+    actors = c.map(Foo, range(10), y=b, actor=True)
+    yield wait(actors)
+
+    assert all(len(w.actors) == 2 for w in workers)