nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: dask/dask
Pull request #4294 (2018-12-12 18:07:14)

Fixes for SerializableLock
locked() should be a method (this version was both broken and untested).

acquire() now supports keyword arguments, e.g., lock.acquire(blocking=True).

- [x] Tests added / passed
- [x] Passes `flake8 dask`



Patch:
diff --git a/dask/utils.py b/dask/utils.py
index 5c5f3851a..295d92a92 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -777,11 +777,11 @@ class SerializableLock(object):
             self.lock = Lock()
             SerializableLock._locks[self.token] = self.lock
 
-    def acquire(self, *args):
-        return self.lock.acquire(*args)
+    def acquire(self, *args, **kwargs):
+        return self.lock.acquire(*args, **kwargs)
 
-    def release(self, *args):
-        return self.lock.release(*args)
+    def release(self, *args, **kwargs):
+        return self.lock.release(*args, **kwargs)
 
     def __enter__(self):
         self.lock.__enter__()
@@ -789,9 +789,8 @@ class SerializableLock(object):
     def __exit__(self, *args):
         self.lock.__exit__(*args)
 
-    @property
     def locked(self):
-        return self.locked
+        return self.lock.locked()
 
     def __getstate__(self):
         return self.token


Test patch:
diff --git a/dask/tests/test_utils.py b/dask/tests/test_utils.py
index 42fcd3450..7a3a8c827 100644
--- a/dask/tests/test_utils.py
+++ b/dask/tests/test_utils.py
@@ -5,6 +5,7 @@ import pickle
 import numpy as np
 import pytest
 
+from dask.compatibility import PY2
 from dask.utils import (takes_multiple_arguments, Dispatch, random_state_data,
                         memory_repr, methodcaller, M, skip_doctest,
                         SerializableLock, funcname, ndeepmap, ensure_dict,
@@ -257,6 +258,22 @@ def test_SerializableLock_name_collision():
     assert d.lock not in (a.lock, b.lock, c.lock)
 
 
+def test_SerializableLock_locked():
+    a = SerializableLock('a')
+    assert not a.locked()
+    with a:
+        assert a.locked()
+    assert not a.locked()
+
+
+@pytest.mark.skipif(PY2, reason="no blocking= keyword in Python 2")
+def test_SerializableLock_acquire_blocking():
+    a = SerializableLock('a')
+    assert a.acquire(blocking=True)
+    assert not a.acquire(blocking=False)
+    a.release()
+
+
 def test_funcname():
     def foo(a, b, c):
         pass
Repository: dask/dask
Pull request #4351 (2019-01-04 19:04:08)

Support pool= keyword argument in threaded/multiprocessing get functions
- [x] Tests added / passed
- [x] Passes `flake8 dask`


jakirkham: LGTM. Thoughts @jcrist?
jcrist: Fine by me. Merging.

Patch:
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index 3a377b54c..7a3276489 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -139,7 +139,7 @@ def get_context():
 
 
 def get(dsk, keys, num_workers=None, func_loads=None, func_dumps=None,
-        optimize_graph=True, **kwargs):
+        optimize_graph=True, pool=None, **kwargs):
     """ Multiprocessed get function appropriate for Bags
 
     Parameters
@@ -159,7 +159,7 @@ def get(dsk, keys, num_workers=None, func_loads=None, func_dumps=None,
     optimize_graph : bool
         If True [default], `fuse` is applied to the graph before computation.
     """
-    pool = config.get('pool', None)
+    pool = pool or config.get('pool', None)
     num_workers = num_workers or config.get('num_workers', None)
     if pool is None:
         context = get_context()
diff --git a/dask/threaded.py b/dask/threaded.py
index 84a382e9c..46bfffcce 100644
--- a/dask/threaded.py
+++ b/dask/threaded.py
@@ -30,7 +30,7 @@ def pack_exception(e, dumps):
     return e, sys.exc_info()[2]
 
 
-def get(dsk, result, cache=None, num_workers=None, **kwargs):
+def get(dsk, result, cache=None, num_workers=None, pool=None, **kwargs):
     """ Threaded cached implementation of dask.get
 
     Parameters
@@ -55,7 +55,7 @@ def get(dsk, result, cache=None, num_workers=None, **kwargs):
     (4, 2)
     """
     global default_pool
-    pool = config.get('pool', None)
+    pool = pool or config.get('pool', None)
     num_workers = num_workers or config.get('num_workers', None)
     thread = current_thread()
 


Test patch:
diff --git a/dask/tests/test_threaded.py b/dask/tests/test_threaded.py
index 49e523d9b..13c69a440 100644
--- a/dask/tests/test_threaded.py
+++ b/dask/tests/test_threaded.py
@@ -45,6 +45,19 @@ def test_reuse_pool():
         assert get({'x': (inc, 1)}, 'x') == 2
 
 
+@pytest.mark.skipif(PY2, reason="threading API changed")
+def test_pool_kwarg():
+    def f():
+        sleep(0.01)
+        return threading.get_ident()
+
+    dsk = {('x', i): (f,) for i in range(30)}
+    dsk['x'] = (len, (set, [('x', i) for i in range(len(dsk))]))
+
+    with ThreadPool(3) as pool:
+        assert get(dsk, 'x', pool=pool) == 3
+
+
 def test_threaded_within_thread():
     L = []