Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Next Generation Assets #432

Merged
merged 14 commits into from
Apr 23, 2020
Merged
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
assets: Add test for DelayedThreadExecutor
AstraLuma committed Apr 18, 2020
commit 44eef0e78755b1e3b5d8adb0f97be95967ba6298
16 changes: 15 additions & 1 deletion ppb/assetlib.py
Original file line number Diff line number Diff line change
@@ -6,6 +6,7 @@
import concurrent.futures
from functools import partial
import logging
import sys
import threading
import weakref

@@ -36,7 +37,20 @@ def __enter__(self):
return self

def __exit__(self, *exc):
self.shutdown(cancel_futures=True, wait=False)
if sys.version_info >= (3, 9):
self.shutdown(wait=False, cancel_futures=True)
else:
import queue
# Backport of 3.9 future cancelling code
while True:
try:
work_item = self._work_queue.get_nowait()
except queue.Empty:
break
if work_item is not None:
work_item.future.cancel()

self.shutdown(wait=False)


class AbstractAsset(abc.ABC):
22 changes: 21 additions & 1 deletion tests/test_assets.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
import gc
import time

import pytest

from ppb import GameEngine, BaseScene
import ppb.events
import ppb.assetlib
from ppb.assetlib import Asset, AssetLoadingSystem
from ppb.assetlib import DelayedThreadExecutor, Asset, AssetLoadingSystem
from ppb.testutils import Failer


@@ -24,6 +25,25 @@ def on_asset_loaded(self, event, signal):
signal(ppb.events.Quit())


def test_executor():
# Can't easily test the cancellation, since jobs in progress can't be cancelled.
def work():
return "spam"

pool = DelayedThreadExecutor()
assert not pool._threads

fut = pool.submit(work)
time.sleep(0.01) # Let any hypothetical threads do work
assert not pool._threads
assert not (fut.done() or fut.running())

with pool:
assert fut.result() == "spam"

assert pool._shutdown


def test_loading(clean_assets):
a = Asset('ppb/engine.py')
engine = GameEngine(