pool/python-distributed
SHA256
4
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
python-distributed/distributed-pr5952-py310.patch

376 lines
12 KiB
Diff
Raw Blame History

From 9c6a4c905c75c5e64ca460ea17bb2bdf0f2782fa Mon Sep 17 00:00:00 2001
From: James Bourbeau <jrbourbeau@gmail.com>
Date: Thu, 3 Feb 2022 12:58:32 -0600
Subject: [PATCH 01/12] Add Python 3.10 build to CI
---
.github/workflows/tests.yaml | 2 +-
continuous_integration/environment-3.10.yaml | 56 ++++++++++++++++++++
2 files changed, 57 insertions(+), 1 deletion(-)
create mode 100644 continuous_integration/environment-3.10.yaml
Index: distributed-2022.03.0/.github/workflows/tests.yaml
===================================================================
--- distributed-2022.03.0.orig/.github/workflows/tests.yaml
+++ distributed-2022.03.0/.github/workflows/tests.yaml
@@ -23,7 +23,7 @@ jobs:
fail-fast: false
matrix:
os: [ubuntu-latest, windows-latest, macos-latest]
- python-version: ["3.8", "3.9"]
+ python-version: ["3.8", "3.9", "3.10"]
# Cherry-pick test modules to split the overall runtime roughly in half
partition: [ci1, not ci1]
include:
@@ -65,12 +65,6 @@ jobs:
shell: bash -l {0}
run: conda config --show
- - name: Install stacktrace
- shell: bash -l {0}
- # stacktrace for Python 3.8 has not been released at the moment of writing
- if: ${{ matrix.os == 'ubuntu-latest' && matrix.python-version < '3.8' }}
- run: mamba install -c conda-forge -c defaults -c numba libunwind stacktrace
-
- name: Hack around https://github.com/ipython/ipython/issues/12197
# This upstream issue causes an interpreter crash when running
# distributed/protocol/tests/test_serialize.py::test_profile_nested_sizeof
Index: distributed-2022.03.0/continuous_integration/environment-3.10.yaml
===================================================================
--- /dev/null
+++ distributed-2022.03.0/continuous_integration/environment-3.10.yaml
@@ -0,0 +1,56 @@
+name: dask-distributed
+channels:
+ - conda-forge
+ - defaults
+dependencies:
+ - python=3.10
+ - packaging
+ - pip
+ - asyncssh
+ - bokeh
+ - click
+ - cloudpickle
+ - coverage<6.3 # https://github.com/nedbat/coveragepy/issues/1310
+ - dask # overridden by git tip below
+ - filesystem-spec # overridden by git tip below
+ - h5py
+ - ipykernel
+ - ipywidgets
+ - jinja2
+ - joblib # overridden by git tip below
+ - jupyter_client
+ - lz4 # Only tested here
+ - msgpack-python
+ - netcdf4
+ - paramiko
+ - pre-commit
+ - prometheus_client
+ - psutil
+ - pynvml # Only tested here
+ - pytest
+ - pytest-cov
+ - pytest-faulthandler
+ - pytest-repeat
+ - pytest-rerunfailures
+ - pytest-timeout
+ - python-blosc # Only tested here
+ - python-snappy # Only tested here
+ - requests
+ - s3fs # overridden by git tip below
+ - scikit-learn
+ - scipy
+ - sortedcollections
+ - tblib
+ - toolz
+ - tornado=6
+ - zict # overridden by git tip below
+ - zstandard
+ - pip:
+ - git+https://github.com/dask/dask
+ - git+https://github.com/dask/s3fs
+ - git+https://github.com/dask/zict
+ # FIXME https://github.com/dask/distributed/issues/5345
+ # - git+https://github.com/intake/filesystem_spec
+ - git+https://github.com/joblib/joblib
+ - keras
+ - pytest-asyncio<0.14.0 # `pytest-asyncio<0.14.0` isn't available on conda-forge for Python 3.10
Index: distributed-2022.03.0/distributed/tests/test_client.py
===================================================================
--- distributed-2022.03.0.orig/distributed/tests/test_client.py
+++ distributed-2022.03.0/distributed/tests/test_client.py
@@ -6461,6 +6461,10 @@ async def test_performance_report(c, s,
assert "cdn.bokeh.org" in data
+@pytest.mark.skipif(
+ sys.version_info >= (3, 10),
+ reason="On Py3.10+ semaphore._loop is not bound until .acquire() blocks",
+)
@gen_cluster(nthreads=[])
async def test_client_gather_semaphore_loop(s):
async with Client(s.address, asynchronous=True) as c:
@@ -6471,9 +6475,16 @@ async def test_client_gather_semaphore_l
async def test_as_completed_condition_loop(c, s, a, b):
seq = c.map(inc, range(5))
ac = as_completed(seq)
+ # consume the ac so that the ac.condition is bound to the loop on py3.10+
+ async for _ in ac:
+ pass
assert ac.condition._loop == c.loop.asyncio_loop
+@pytest.mark.skipif(
+ sys.version_info >= (3, 10),
+ reason="On Py3.10+ semaphore._loop is not bound until .acquire() blocks",
+)
def test_client_connectionpool_semaphore_loop(s, a, b):
with Client(s["address"]) as c:
assert c.rpc.semaphore._loop is c.loop.asyncio_loop
Index: distributed-2022.03.0/distributed/node.py
===================================================================
--- distributed-2022.03.0.orig/distributed/node.py
+++ distributed-2022.03.0/distributed/node.py
@@ -131,12 +131,9 @@ class ServerNode(Server):
import ssl
ssl_options = ssl.create_default_context(
- cafile=tls_ca_file, purpose=ssl.Purpose.SERVER_AUTH
+ cafile=tls_ca_file, purpose=ssl.Purpose.CLIENT_AUTH
)
ssl_options.load_cert_chain(tls_cert, keyfile=tls_key)
- # We don't care about auth here, just encryption
- ssl_options.check_hostname = False
- ssl_options.verify_mode = ssl.CERT_NONE
self.http_server = HTTPServer(self.http_application, ssl_options=ssl_options)
Index: distributed-2022.03.0/distributed/profile.py
===================================================================
--- distributed-2022.03.0.orig/distributed/profile.py
+++ distributed-2022.03.0/distributed/profile.py
@@ -27,6 +27,7 @@ We represent this tree as a nested dicti
from __future__ import annotations
import bisect
+import dis
import linecache
import sys
import threading
@@ -59,21 +60,41 @@ def identifier(frame):
)
+# work around some frames lacking an f_lineo eg: https://bugs.python.org/issue47085
+def _f_lineno(frame):
+ f_lineno = frame.f_lineno
+ if f_lineno is not None:
+ return f_lineno
+
+ f_lasti = frame.f_lasti
+ code = frame.f_code
+ prev_line = code.co_firstlineno
+
+ for start, next_line in dis.findlinestarts(code):
+ if f_lasti < start:
+ return prev_line
+ prev_line = next_line
+
+ return prev_line
+
+
def repr_frame(frame):
"""Render a frame as a line for inclusion into a text traceback"""
co = frame.f_code
- text = f' File "{co.co_filename}", line {frame.f_lineno}, in {co.co_name}'
- line = linecache.getline(co.co_filename, frame.f_lineno, frame.f_globals).lstrip()
+ f_lineno = _f_lineno(frame)
+ text = f' File "{co.co_filename}", line {f_lineno}, in {co.co_name}'
+ line = linecache.getline(co.co_filename, f_lineno, frame.f_globals).lstrip()
return text + "\n\t" + line
def info_frame(frame):
co = frame.f_code
- line = linecache.getline(co.co_filename, frame.f_lineno, frame.f_globals).lstrip()
+ f_lineno = _f_lineno(frame)
+ line = linecache.getline(co.co_filename, f_lineno, frame.f_globals).lstrip()
return {
"filename": co.co_filename,
"name": co.co_name,
- "line_number": frame.f_lineno,
+ "line_number": f_lineno,
"line": line,
}
Index: distributed-2022.03.0/distributed/tests/test_profile.py
===================================================================
--- distributed-2022.03.0.orig/distributed/tests/test_profile.py
+++ distributed-2022.03.0/distributed/tests/test_profile.py
@@ -1,5 +1,9 @@
+from __future__ import annotations
+
+import dataclasses
import sys
import threading
+from collections.abc import Iterator, Sequence
from time import sleep
import pytest
@@ -11,6 +15,7 @@ from distributed.profile import (
call_stack,
create,
identifier,
+ info_frame,
ll_get_stack,
llprocess,
merge,
@@ -200,3 +205,102 @@ def test_watch():
while threading.active_count() > start_threads:
assert time() < start + 2
sleep(0.01)
+
+
+@dataclasses.dataclass(frozen=True)
+class FakeCode:
+ co_filename: str
+ co_name: str
+ co_firstlineno: int
+ co_lnotab: bytes
+ co_lines_seq: Sequence[tuple[int, int, int | None]]
+ co_code: bytes
+
+ def co_lines(self) -> Iterator[tuple[int, int, int | None]]:
+ yield from self.co_lines_seq
+
+
+FAKE_CODE = FakeCode(
+ co_filename="<stdin>",
+ co_name="example",
+ co_firstlineno=1,
+ # https://github.com/python/cpython/blob/b68431fadb3150134ac6ccbf501cdfeaf4c75678/Objects/lnotab_notes.txt#L84
+ # generated from:
+ # def example():
+ # for i in range(1):
+ # if i >= 0:
+ # pass
+ # example.__code__.co_lnotab
+ co_lnotab=b"\x00\x01\x0c\x01\x08\x01\x04\xfe",
+ # generated with list(example.__code__.co_lines())
+ co_lines_seq=[
+ (0, 12, 2),
+ (12, 20, 3),
+ (20, 22, 4),
+ (22, 24, None),
+ (24, 28, 2),
+ ],
+ # used in dis.findlinestarts as bytecode_len = len(code.co_code)
+ # https://github.com/python/cpython/blob/6f345d363308e3e6ecf0ad518ea0fcc30afde2a8/Lib/dis.py#L457
+ co_code=bytes(28),
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class FakeFrame:
+ f_lasti: int
+ f_code: FakeCode
+ f_lineno: int | None = None
+ f_back: FakeFrame | None = None
+ f_globals: dict[str, object] = dataclasses.field(default_factory=dict)
+
+
+@pytest.mark.parametrize(
+ "f_lasti,f_lineno",
+ [
+ (-1, 1),
+ (0, 2),
+ (1, 2),
+ (11, 2),
+ (12, 3),
+ (21, 4),
+ (22, 4),
+ (23, 4),
+ (24, 2),
+ (25, 2),
+ (26, 2),
+ (27, 2),
+ (100, 2),
+ ],
+)
+def test_info_frame_f_lineno(f_lasti: int, f_lineno: int) -> None:
+ assert info_frame(FakeFrame(f_lasti=f_lasti, f_code=FAKE_CODE)) == {
+ "filename": "<stdin>",
+ "name": "example",
+ "line_number": f_lineno,
+ "line": "",
+ }
+
+
+@pytest.mark.parametrize(
+ "f_lasti,f_lineno",
+ [
+ (-1, 1),
+ (0, 2),
+ (1, 2),
+ (11, 2),
+ (12, 3),
+ (21, 4),
+ (22, 4),
+ (23, 4),
+ (24, 2),
+ (25, 2),
+ (26, 2),
+ (27, 2),
+ (100, 2),
+ ],
+)
+def test_call_stack_f_lineno(f_lasti: int, f_lineno: int) -> None:
+ assert call_stack(FakeFrame(f_lasti=f_lasti, f_code=FAKE_CODE)) == [
+ f' File "<stdin>", line {f_lineno}, in example\n\t'
+ ]
Index: distributed-2022.03.0/distributed/utils_test.py
===================================================================
--- distributed-2022.03.0.orig/distributed/utils_test.py
+++ distributed-2022.03.0/distributed/utils_test.py
@@ -706,13 +706,16 @@ def cluster(
except KeyError:
rpc_kwargs = {}
- with rpc(saddr, **rpc_kwargs) as s:
- while True:
- nthreads = loop.run_sync(s.ncores)
- if len(nthreads) == nworkers:
- break
- if time() - start > 5:
- raise Exception("Timeout on cluster creation")
+ async def wait_for_workers():
+ async with rpc(saddr, **rpc_kwargs) as s:
+ while True:
+ nthreads = await s.ncores()
+ if len(nthreads) == nworkers:
+ break
+ if time() - start > 5:
+ raise Exception("Timeout on cluster creation")
+
+ loop.run_sync(wait_for_workers)
# avoid sending processes down to function
yield {"address": saddr}, [
Index: distributed-2022.03.0/setup.py
===================================================================
--- distributed-2022.03.0.orig/setup.py
+++ distributed-2022.03.0/setup.py
@@ -98,8 +98,11 @@ setup(
"License :: OSI Approved :: BSD License",
"Operating System :: OS Independent",
"Programming Language :: Python",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3 :: Only",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
+ "Programming Language :: Python :: 3.10",
"Topic :: Scientific/Engineering",
"Topic :: System :: Distributed Computing",
],
Reference in New Issue View Git Blame Copy Permalink