andras/great-ai
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases 4 Packages Wiki Activity Actions

Try debugging hanging

Browse source
This commit is contained in:
Andras Schmelczer 2022-07-03 13:32:42 +02:00
parent 2d7dfe3b95
commit bbe0b978b5
No known key found for this signature in database
GPG key ID: 0EA1BC97D0AB076E
2 changed files with 194 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

186
tests/utilities/test_parallel_map.py
View file

@ -1,116 +1,116 @@
import unittest # import unittest
import pytest # import pytest
from src.great_ai.utilities import parallel_map # from src.great_ai.utilities import parallel_map
COUNT = int(1e5) + 3 # COUNT = int(1e5) + 3
class TestParallelMap(unittest.TestCase): # class TestParallelMap(unittest.TestCase):
def test_simple_case_with_progress_bar(self) -> None: # def test_simple_case_with_progress_bar(self) -> None:
assert list(parallel_map(lambda v: v**2, range(COUNT), concurrency=4)) == [ # assert list(parallel_map(lambda v: v**2, range(COUNT), concurrency=4)) == [
v**2 for v in range(COUNT) # v**2 for v in range(COUNT)
] # ]
def test_with_iterable(self) -> None: # def test_with_iterable(self) -> None:
from time import sleep # from time import sleep
def my_generator(): # def my_generator():
for i in range(10): # for i in range(10):
yield i # yield i
sleep(0.1) # sleep(0.1)
expected = [v**3 for v in range(10)] # expected = [v**3 for v in range(10)]
assert ( # assert (
list(parallel_map(lambda x: x**3, my_generator(), chunk_size=1)) # list(parallel_map(lambda x: x**3, my_generator(), chunk_size=1))
== expected # == expected
) # )
def test_simple_case_without_progress_bar(self) -> None: # def test_simple_case_without_progress_bar(self) -> None:
assert list(parallel_map(lambda v: v**2, range(COUNT), concurrency=2)) == [ # assert list(parallel_map(lambda v: v**2, range(COUNT), concurrency=2)) == [
v**2 for v in range(COUNT) # v**2 for v in range(COUNT)
] # ]
def test_simple_case_invalid_values(self) -> None: # def test_simple_case_invalid_values(self) -> None:
with pytest.raises(AssertionError): # with pytest.raises(AssertionError):
list(parallel_map(lambda v: v**2, range(COUNT), concurrency=0)) # list(parallel_map(lambda v: v**2, range(COUNT), concurrency=0))
with pytest.raises(AssertionError): # with pytest.raises(AssertionError):
list(parallel_map(lambda v: v**2, range(COUNT), chunk_size=0)) # list(parallel_map(lambda v: v**2, range(COUNT), chunk_size=0))
def test_this_process_exception(self) -> None: # def test_this_process_exception(self) -> None:
def my_generator(): # def my_generator():
yield 1 # yield 1
yield 2 # yield 2
yield 3 # yield 3
assert False # assert False
with pytest.raises(AssertionError): # with pytest.raises(AssertionError):
list( # list(
parallel_map( # parallel_map(
lambda v: v**2, my_generator(), concurrency=2, chunk_size=2 # lambda v: v**2, my_generator(), concurrency=2, chunk_size=2
) # )
) # )
with pytest.raises(AssertionError): # with pytest.raises(AssertionError):
list( # list(
parallel_map( # parallel_map(
lambda v: v**2, my_generator(), concurrency=1, chunk_size=2 # lambda v: v**2, my_generator(), concurrency=1, chunk_size=2
) # )
) # )
def test_ignore_this_process_exception(self) -> None: # def test_ignore_this_process_exception(self) -> None:
def my_generator(): # def my_generator():
yield 1 # yield 1
yield 2 # yield 2
yield 3 # yield 3
yield 1 / 0 # yield 1 / 0
assert list( # assert list(
parallel_map( # parallel_map(
lambda v: v**2, # lambda v: v**2,
my_generator(), # my_generator(),
concurrency=2, # concurrency=2,
chunk_size=2, # chunk_size=2,
ignore_exceptions=True, # ignore_exceptions=True,
) # )
) == [1, 4] # ) == [1, 4]
assert list( # assert list(
parallel_map( # parallel_map(
lambda v: v**2, # lambda v: v**2,
my_generator(), # my_generator(),
concurrency=1, # concurrency=1,
chunk_size=2, # chunk_size=2,
ignore_exceptions=True, # ignore_exceptions=True,
) # )
) == [1, 4, 9] # ) == [1, 4, 9]
def test_worker_process_exception(self) -> None: # def test_worker_process_exception(self) -> None:
def oh_no(_): # def oh_no(_):
raise ValueError("hi") # raise ValueError("hi")
with pytest.raises(ValueError): # with pytest.raises(ValueError):
list(parallel_map(oh_no, range(COUNT), concurrency=2)) # list(parallel_map(oh_no, range(COUNT), concurrency=2))
with pytest.raises(ValueError): # with pytest.raises(ValueError):
list(parallel_map(oh_no, range(COUNT), concurrency=1)) # list(parallel_map(oh_no, range(COUNT), concurrency=1))
def test_ignore_worker_process_exception(self) -> None: # def test_ignore_worker_process_exception(self) -> None:
def oh_no(_): # def oh_no(_):
raise ValueError("hi") # raise ValueError("hi")
assert ( # assert (
list(parallel_map(oh_no, range(3), concurrency=2, ignore_exceptions=True)) # list(parallel_map(oh_no, range(3), concurrency=2, ignore_exceptions=True))
== [None] * 3 # == [None] * 3
) # )
assert ( # assert (
list(parallel_map(oh_no, range(3), concurrency=1, ignore_exceptions=True)) # list(parallel_map(oh_no, range(3), concurrency=1, ignore_exceptions=True))
== [None] * 3 # == [None] * 3
) # )
def test_no_op(self) -> None: # def test_no_op(self) -> None:
assert list(parallel_map(lambda v: v**2, [])) == [] # assert list(parallel_map(lambda v: v**2, [])) == []
assert list(parallel_map(lambda v: v**2, [], chunk_size=100)) == [] # assert list(parallel_map(lambda v: v**2, [], chunk_size=100)) == []
assert list(parallel_map(lambda v: v**2, [], concurrency=100)) == [] # assert list(parallel_map(lambda v: v**2, [], concurrency=100)) == []

202
tests/utilities/test_threaded_parallel_map.py
View file

@ -1,124 +1,124 @@
# import unittest import unittest
# import pytest import pytest
# from src.great_ai.utilities import threaded_parallel_map from src.great_ai.utilities import threaded_parallel_map
# COUNT = int(1e5) + 3 COUNT = int(1e5) + 3
# class TestParallelMap(unittest.TestCase): class TestParallelMap(unittest.TestCase):
# def test_simple_case_with_progress_bar(self) -> None: def test_simple_case_with_progress_bar(self) -> None:
# assert list( assert list(
# threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=4) threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=4)
# ) == [v**2 for v in range(COUNT)] ) == [v**2 for v in range(COUNT)]
# def test_with_iterable(self) -> None: def test_with_iterable(self) -> None:
# from time import sleep from time import sleep
# def my_generator(): def my_generator():
# for i in range(10): for i in range(10):
# yield i yield i
# sleep(0.1) sleep(0.1)
# expected = [v**3 for v in range(10)] expected = [v**3 for v in range(10)]
# assert ( assert (
# list(threaded_parallel_map(lambda x: x**3, my_generator(), chunk_size=1)) list(threaded_parallel_map(lambda x: x**3, my_generator(), chunk_size=1))
# == expected == expected
# ) )
# def test_simple_case_without_progress_bar(self) -> None: def test_simple_case_without_progress_bar(self) -> None:
# assert list( assert list(
# threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=2) threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=2)
# ) == [v**2 for v in range(COUNT)] ) == [v**2 for v in range(COUNT)]
# def test_simple_case_invalid_values(self) -> None: def test_simple_case_invalid_values(self) -> None:
# with pytest.raises(AssertionError): with pytest.raises(AssertionError):
# list(threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=0)) list(threaded_parallel_map(lambda v: v**2, range(COUNT), concurrency=0))
# with pytest.raises(AssertionError): with pytest.raises(AssertionError):
# list(threaded_parallel_map(lambda v: v**2, range(COUNT), chunk_size=0)) list(threaded_parallel_map(lambda v: v**2, range(COUNT), chunk_size=0))
# def test_this_worker_exception(self) -> None: def test_this_worker_exception(self) -> None:
# def my_generator(): def my_generator():
# yield 1 yield 1
# yield 2 yield 2
# yield 3 yield 3
# assert False assert False
# with pytest.raises(AssertionError): with pytest.raises(AssertionError):
# list( list(
# threaded_parallel_map( threaded_parallel_map(
# lambda v: v**2, my_generator(), concurrency=2, chunk_size=2 lambda v: v**2, my_generator(), concurrency=2, chunk_size=2
# ) )
# ) )
# with pytest.raises(AssertionError): with pytest.raises(AssertionError):
# list( list(
# threaded_parallel_map( threaded_parallel_map(
# lambda v: v**2, my_generator(), concurrency=1, chunk_size=2 lambda v: v**2, my_generator(), concurrency=1, chunk_size=2
# ) )
# ) )
# def test_ignore_this_worker_exception(self) -> None: def test_ignore_this_worker_exception(self) -> None:
# def my_generator(): def my_generator():
# yield 1 yield 1
# yield 2 yield 2
# yield 3 yield 3
# yield 1 / 0 yield 1 / 0
# assert list( assert list(
# threaded_parallel_map( threaded_parallel_map(
# lambda v: v**2, lambda v: v**2,
# my_generator(), my_generator(),
# concurrency=2, concurrency=2,
# chunk_size=2, chunk_size=2,
# ignore_exceptions=True, ignore_exceptions=True,
# ) )
# ) == [1, 4] ) == [1, 4]
# assert list( assert list(
# threaded_parallel_map( threaded_parallel_map(
# lambda v: v**2, lambda v: v**2,
# my_generator(), my_generator(),
# concurrency=1, concurrency=1,
# chunk_size=2, chunk_size=2,
# ignore_exceptions=True, ignore_exceptions=True,
# ) )
# ) == [1, 4, 9] ) == [1, 4, 9]
# def test_worker_worker_exception(self) -> None: def test_worker_worker_exception(self) -> None:
# def oh_no(_): def oh_no(_):
# raise ValueError("hi") raise ValueError("hi")
# with pytest.raises(ValueError): with pytest.raises(ValueError):
# list(threaded_parallel_map(oh_no, range(COUNT), concurrency=2)) list(threaded_parallel_map(oh_no, range(COUNT), concurrency=2))
# with pytest.raises(ValueError): with pytest.raises(ValueError):
# list(threaded_parallel_map(oh_no, range(COUNT), concurrency=1)) list(threaded_parallel_map(oh_no, range(COUNT), concurrency=1))
# def test_ignore_worker_worker_exception(self) -> None: def test_ignore_worker_worker_exception(self) -> None:
# def oh_no(_): def oh_no(_):
# raise ValueError("hi") raise ValueError("hi")
# assert ( assert (
# list( list(
# threaded_parallel_map( threaded_parallel_map(
# oh_no, range(3), concurrency=2, ignore_exceptions=True oh_no, range(3), concurrency=2, ignore_exceptions=True
# ) )
# ) )
# == [None] * 3 == [None] * 3
# ) )
# assert ( assert (
# list( list(
# threaded_parallel_map( threaded_parallel_map(
# oh_no, range(3), concurrency=1, ignore_exceptions=True oh_no, range(3), concurrency=1, ignore_exceptions=True
# ) )
# ) )
# == [None] * 3 == [None] * 3
# ) )
# def test_no_op(self) -> None: def test_no_op(self) -> None:
# assert list(threaded_parallel_map(lambda v: v**2, [])) == [] assert list(threaded_parallel_map(lambda v: v**2, [])) == []
# assert list(threaded_parallel_map(lambda v: v**2, [], chunk_size=100)) == [] assert list(threaded_parallel_map(lambda v: v**2, [], chunk_size=100)) == []
# assert list(threaded_parallel_map(lambda v: v**2, [], concurrency=100)) == [] assert list(threaded_parallel_map(lambda v: v**2, [], concurrency=100)) == []