<commit_before>import pytest

from parsl.app.app import App
from parsl.dataflow.dflow import DataFlowKernel


def run_checkpointed(checkpoints):
    # set_stream_logger()
    from parsl.tests.configs.local_threads_checkpoint_task_exit import config
    config.checkpoint_files = checkpoints
    dfk = DataFlowKernel(config=config)

    @App('python', dfk, cache=True)
    def cached_rand(x):
        import random
        return random.randint(0, 10000)

    items = []
    for i in range(0, 5):
        x = cached_rand(i)
        items.append(x)

    dfk.cleanup()
    return [i.result() for i in items], dfk.run_dir


def run_race(sleep_dur):

    from parsl.tests.configs.local_threads_checkpoint_dfk_exit import config
    dfk = DataFlowKernel(config=config)

    @App('python', dfk, cache=True)
    def cached_rand(x, sleep_dur=0):
        import random
        import time
        time.sleep(sleep_dur)
        return random.randint(0, 10000)

    items = []
    for i in range(0, 5):
        x = cached_rand(i, sleep_dur=sleep_dur)
        items.append(x)

    dfk.cleanup()
    return [i.result for i in items]


@pytest.mark.local
def test_regress_234():
    """Test task_exit checkpointing with fast tasks"""
    run_race(0)


@pytest.mark.local
def test_slower_apps():
    """Test task_exit tests with slow apps"""
    run_race(0.5)


@pytest.mark.local
@pytest.mark.skip('fails due to likely race in checkpointing; see issue #681')
def test_checkpoint_availability():<commit_msg>Fix race condition in checkpoint test (issue #681) (#1250)<commit_after>import pytest

from parsl.app.app import App
from parsl.dataflow.dflow import DataFlowKernel


def run_checkpointed(checkpoints):
    # set_stream_logger()
    from parsl.tests.configs.local_threads_checkpoint_task_exit import config
    config.checkpoint_files = checkpoints
    dfk = DataFlowKernel(config=config)

    @App('python', dfk, cache=True)
    def cached_rand(x):
        import random
        return random.randint(0, 10000)

    items = []
    for i in range(0, 5):
        x = cached_rand(i)
        items.append(x)

    # wait for all of the results
    results = [i.result() for i in items]

    dfk.cleanup()
    return results, dfk.run_dir


def run_race(sleep_dur):

    from parsl.tests.configs.local_threads_checkpoint_dfk_exit import config
    dfk = DataFlowKernel(config=config)

    @App('python', dfk, cache=True)
    def cached_rand(x, sleep_dur=0):
        import random
        import time
        time.sleep(sleep_dur)
        return random.randint(0, 10000)

    items = []
    for i in range(0, 5):
        x = cached_rand(i, sleep_dur=sleep_dur)
        items.append(x)

    dfk.cleanup()
    return [i.result for i in items]


@pytest.mark.local
def test_regress_234():
    """Test task_exit checkpointing with fast tasks"""
    run_race(0)


@pytest.mark.local
def test_slower_apps():
    """Test task_exit tests with slow apps"""
    run_race(0.5)


@pytest.mark.local
def test_checkpoint_availability():
{'fraction_non_alphanumeric': 0.10282861896838602, 'fraction_numerical': 0.018635607321131446, 'mean_word_length': 3.2577903682719547, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6982656', 'n_tokens_mistral': 1072, 'n_tokens_neox': 1007, 'n_words': 245}