diff options
Diffstat (limited to 'test/git/async')
| -rw-r--r-- | test/git/async/__init__.py | 0 | ||||
| -rw-r--r-- | test/git/async/task.py | 202 | ||||
| -rw-r--r-- | test/git/async/test_channel.py | 87 | ||||
| -rw-r--r-- | test/git/async/test_graph.py | 80 | ||||
| -rw-r--r-- | test/git/async/test_performance.py | 51 | ||||
| -rw-r--r-- | test/git/async/test_pool.py | 476 | ||||
| -rw-r--r-- | test/git/async/test_task.py | 15 | ||||
| -rw-r--r-- | test/git/async/test_thread.py | 44 |
8 files changed, 955 insertions, 0 deletions
diff --git a/test/git/async/__init__.py b/test/git/async/__init__.py new file mode 100644 index 00000000..e69de29b --- /dev/null +++ b/test/git/async/__init__.py diff --git a/test/git/async/task.py b/test/git/async/task.py new file mode 100644 index 00000000..583cb1f8 --- /dev/null +++ b/test/git/async/task.py @@ -0,0 +1,202 @@ +"""Module containing task implementations useful for testing them""" +from git.async.task import * + +import threading +import weakref + +class _TestTaskBase(object): + """Note: causes great slowdown due to the required locking of task variables""" + def __init__(self, *args, **kwargs): + super(_TestTaskBase, self).__init__(*args, **kwargs) + self.should_fail = False + self.lock = threading.Lock() # yes, can't safely do x = x + 1 :) + self.plock = threading.Lock() + self.item_count = 0 + self.process_count = 0 + + def do_fun(self, item): + self.lock.acquire() + self.item_count += 1 + self.lock.release() + if self.should_fail: + raise AssertionError("I am failing just for the fun of it") + return item + + def process(self, count=1): + # must do it first, otherwise we might read and check results before + # the thread gets here :). Its a lesson ! + self.plock.acquire() + self.process_count += 1 + self.plock.release() + super(_TestTaskBase, self).process(count) + + def _assert(self, pc, fc, check_scheduled=False): + """Assert for num process counts (pc) and num function counts (fc) + :return: self""" + self.lock.acquire() + if self.item_count != fc: + print self.item_count, fc + assert self.item_count == fc + self.lock.release() + + # NOTE: asserting num-writers fails every now and then, implying a thread is + # still processing (an empty chunk) when we are checking it. This can + # only be prevented by checking the scheduled items, which requires locking + # and causes slowdows, so we don't do that. If the num_writers + # counter wouldn't be maintained properly, more tests would fail, so + # we can safely refrain from checking this here + # self._wlock.acquire() + # assert self._num_writers == 0 + # self._wlock.release() + return self + + +class TestThreadTask(_TestTaskBase, IteratorThreadTask): + pass + + +class TestFailureThreadTask(TestThreadTask): + """Fails after X items""" + def __init__(self, *args, **kwargs): + self.fail_after = kwargs.pop('fail_after') + super(TestFailureThreadTask, self).__init__(*args, **kwargs) + + def do_fun(self, item): + item = TestThreadTask.do_fun(self, item) + + self.lock.acquire() + try: + if self.item_count > self.fail_after: + raise AssertionError("Simulated failure after processing %i items" % self.fail_after) + finally: + self.lock.release() + # END handle fail after + return item + + +class TestChannelThreadTask(_TestTaskBase, ChannelThreadTask): + """Apply a transformation on items read from an input channel""" + def __init__(self, *args, **kwargs): + self.fail_after = kwargs.pop('fail_after', 0) + super(TestChannelThreadTask, self).__init__(*args, **kwargs) + + def do_fun(self, item): + """return tuple(i, i*2)""" + item = super(TestChannelThreadTask, self).do_fun(item) + + # fail after support + if self.fail_after: + self.lock.acquire() + try: + if self.item_count > self.fail_after: + raise AssertionError("Simulated failure after processing %i items" % self.fail_after) + finally: + self.lock.release() + # END handle fail-after + + if isinstance(item, tuple): + i = item[0] + return item + (i * self.id, ) + else: + return (item, item * self.id) + # END handle tuple + + +class TestPerformanceThreadTask(ChannelThreadTask): + """Applies no operation to the item, and does not lock, measuring + the actual throughput of the system""" + + def do_fun(self, item): + return item + + +class TestVerifyChannelThreadTask(_TestTaskBase, ChannelThreadTask): + """An input channel task, which verifies the result of its input channels, + should be last in the chain. + Id must be int""" + + def do_fun(self, item): + """return tuple(i, i*2)""" + item = super(TestVerifyChannelThreadTask, self).do_fun(item) + + # make sure the computation order matches + assert isinstance(item, tuple), "input was no tuple: %s" % item + + base = item[0] + for id, num in enumerate(item[1:]): + assert num == base * id, "%i != %i, orig = %s" % (num, base * id, str(item)) + # END verify order + + return item + +#{ Utilities + +def make_proxy_method(t): + """required to prevent binding self into the method we call""" + wt = weakref.proxy(t) + return lambda item: wt.do_fun(item) + +def add_task_chain(p, ni, count=1, fail_setup=list(), feeder_channel=None, id_offset=0, + feedercls=TestThreadTask, transformercls=TestChannelThreadTask, + include_verifier=True): + """Create a task chain of feeder, count transformers and order verifcator + to the pool p, like t1 -> t2 -> t3 + :param fail_setup: a list of pairs, task_id, fail_after, i.e. [(2, 20)] would + make the third transformer fail after 20 items + :param feeder_channel: if set to a channel, it will be used as input of the + first transformation task. The respective first task in the return value + will be None. + :param id_offset: defines the id of the first transformation task, all subsequent + ones will add one + :return: tuple(list(task1, taskN, ...), list(rc1, rcN, ...))""" + nt = p.num_tasks() + + feeder = None + frc = feeder_channel + if feeder_channel is None: + feeder = make_iterator_task(ni, taskcls=feedercls) + frc = p.add_task(feeder) + # END handle specific feeder + + rcs = [frc] + tasks = [feeder] + + inrc = frc + for tc in xrange(count): + t = transformercls(inrc, tc+id_offset, None) + + t.fun = make_proxy_method(t) + #t.fun = t.do_fun + inrc = p.add_task(t) + + tasks.append(t) + rcs.append(inrc) + # END create count transformers + + # setup failure + for id, fail_after in fail_setup: + tasks[1+id].fail_after = fail_after + # END setup failure + + if include_verifier: + verifier = TestVerifyChannelThreadTask(inrc, 'verifier', None) + #verifier.fun = verifier.do_fun + verifier.fun = make_proxy_method(verifier) + vrc = p.add_task(verifier) + + + tasks.append(verifier) + rcs.append(vrc) + # END handle include verifier + return tasks, rcs + +def make_iterator_task(ni, taskcls=TestThreadTask, **kwargs): + """:return: task which yields ni items + :param taskcls: the actual iterator type to use + :param **kwargs: additional kwargs to be passed to the task""" + t = taskcls(iter(range(ni)), 'iterator', None, **kwargs) + if isinstance(t, _TestTaskBase): + t.fun = make_proxy_method(t) + return t + +#} END utilities diff --git a/test/git/async/test_channel.py b/test/git/async/test_channel.py new file mode 100644 index 00000000..e9e1b64c --- /dev/null +++ b/test/git/async/test_channel.py @@ -0,0 +1,87 @@ +"""Channel testing""" +from test.testlib import * +from git.async.channel import * + +import time + +class TestChannels(TestBase): + + def test_base(self): + # creating channel yields a write and a read channal + wc, rc = mkchannel() + assert isinstance(wc, ChannelWriter) # default args + assert isinstance(rc, ChannelReader) + + + # TEST UNLIMITED SIZE CHANNEL - writing+reading is FIFO + item = 1 + item2 = 2 + wc.write(item) + wc.write(item2) + + # read all - it blocks as its still open for writing + to = 0.2 + st = time.time() + assert rc.read(timeout=to) == [item, item2] + assert time.time() - st >= to + + # next read blocks. it waits a second + st = time.time() + assert len(rc.read(1, True, to)) == 0 + assert time.time() - st >= to + + # writing to a closed channel raises + assert not wc.closed() + wc.close() + assert wc.closed() + wc.close() # fine + assert wc.closed() + + self.failUnlessRaises(ReadOnly, wc.write, 1) + + # reading from a closed channel never blocks + assert len(rc.read()) == 0 + assert len(rc.read(5)) == 0 + assert len(rc.read(1)) == 0 + + + # test callback channels + wc, rc = mkchannel(wtype = CallbackChannelWriter, rtype = CallbackChannelReader) + + cb = [0, 0] # set slots to one if called + def pre_write(item): + cb[0] = 1 + return item + 1 + def pre_read(count): + cb[1] = 1 + + # set, verify it returns previous one + assert wc.set_pre_cb(pre_write) is None + assert rc.set_pre_cb(pre_read) is None + assert wc.set_pre_cb(pre_write) is pre_write + assert rc.set_pre_cb(pre_read) is pre_read + + # writer transforms input + val = 5 + wc.write(val) + assert cb[0] == 1 and cb[1] == 0 + + rval = rc.read(1)[0] # read one item, must not block + assert cb[0] == 1 and cb[1] == 1 + assert rval == val + 1 + + + + # ITERATOR READER + reader = IteratorReader(iter(range(10))) + assert len(reader.read(2)) == 2 + assert len(reader.read(0)) == 8 + # its empty now + assert len(reader.read(0)) == 0 + assert len(reader.read(5)) == 0 + + # doesn't work if item is not an iterator + self.failUnlessRaises(ValueError, IteratorReader, list()) + + # NOTE: its thread-safety is tested by the pool + diff --git a/test/git/async/test_graph.py b/test/git/async/test_graph.py new file mode 100644 index 00000000..7630226b --- /dev/null +++ b/test/git/async/test_graph.py @@ -0,0 +1,80 @@ +"""Channel testing""" +from test.testlib import * +from git.async.graph import * + +import time +import sys + +class TestGraph(TestBase): + + def test_base(self): + g = Graph() + nn = 10 + assert nn > 2, "need at least 3 nodes" + + # add unconnected nodes + for i in range(nn): + assert isinstance(g.add_node(Node()), Node) + # END add nodes + assert len(g.nodes) == nn + + # delete unconnected nodes + for n in g.nodes[:]: + g.remove_node(n) + # END del nodes + + # add a chain of connected nodes + last = None + for i in range(nn): + n = g.add_node(Node(i)) + if last: + assert not last.out_nodes + assert not n.in_nodes + assert g.add_edge(last, n) is g + assert last.out_nodes[0] is n + assert n.in_nodes[0] is last + last = n + # END for each node to connect + + # try to connect a node with itself + self.failUnlessRaises(ValueError, g.add_edge, last, last) + + # try to create a cycle + self.failUnlessRaises(ValueError, g.add_edge, g.nodes[0], g.nodes[-1]) + self.failUnlessRaises(ValueError, g.add_edge, g.nodes[-1], g.nodes[0]) + + # we have undirected edges, readding the same edge, but the other way + # around does not change anything + n1, n2, n3 = g.nodes[0], g.nodes[1], g.nodes[2] + g.add_edge(n1, n2) # already connected + g.add_edge(n2, n1) # same thing + assert len(n1.out_nodes) == 1 + assert len(n1.in_nodes) == 0 + assert len(n2.in_nodes) == 1 + assert len(n2.out_nodes) == 1 + + # deleting a connected node clears its neighbour connections + assert n3.in_nodes[0] is n2 + assert g.remove_node(n2) is g + assert g.remove_node(n2) is g # multi-deletion okay + assert len(g.nodes) == nn - 1 + assert len(n3.in_nodes) == 0 + assert len(n1.out_nodes) == 0 + + # check the history from the last node + end = g.nodes[-1] + dfirst_nodes = g.input_inclusive_dfirst_reversed(end) + num_nodes_seen = nn - 2 # deleted second, which leaves first one disconnected + assert len(dfirst_nodes) == num_nodes_seen + assert dfirst_nodes[-1] == end and dfirst_nodes[-2].id == end.id-1 + + + # test cleanup + # its at least kept by its graph + assert sys.getrefcount(end) > 3 + del(g) + del(n1); del(n2); del(n3) + del(dfirst_nodes) + del(last) + del(n) + assert sys.getrefcount(end) == 2 diff --git a/test/git/async/test_performance.py b/test/git/async/test_performance.py new file mode 100644 index 00000000..703c8593 --- /dev/null +++ b/test/git/async/test_performance.py @@ -0,0 +1,51 @@ +"""Channel testing""" +from test.testlib import * +from task import * + +from git.async.pool import * +from git.async.thread import terminate_threads +from git.async.util import cpu_count + +import time +import sys + + + +class TestThreadPoolPerformance(TestBase): + + max_threads = cpu_count() + + def test_base(self): + # create a dependency network, and see how the performance changes + # when adjusting the amount of threads + pool = ThreadPool(0) + ni = 1000 # number of items to process + print self.max_threads + for num_threads in range(self.max_threads*2 + 1): + pool.set_size(num_threads) + for num_transformers in (1, 5, 10): + for read_mode in range(2): + ts, rcs = add_task_chain(pool, ni, count=num_transformers, + feedercls=IteratorThreadTask, + transformercls=TestPerformanceThreadTask, + include_verifier=False) + + mode_info = "read(0)" + if read_mode == 1: + mode_info = "read(1) * %i" % ni + # END mode info + fmt = "Threadcount=%%i: Produced %%i items using %s in %%i transformations in %%f s (%%f items / s)" % mode_info + reader = rcs[-1] + st = time.time() + if read_mode == 1: + for i in xrange(ni): + assert len(reader.read(1)) == 1 + # END for each item to read + else: + assert len(reader.read(0)) == ni + # END handle read mode + elapsed = time.time() - st + print >> sys.stderr, fmt % (num_threads, ni, num_transformers, elapsed, ni / elapsed) + # END for each read-mode + # END for each amount of processors + # END for each thread count diff --git a/test/git/async/test_pool.py b/test/git/async/test_pool.py new file mode 100644 index 00000000..aab618aa --- /dev/null +++ b/test/git/async/test_pool.py @@ -0,0 +1,476 @@ +"""Channel testing""" +from test.testlib import * +from task import * + +from git.async.pool import * +from git.async.thread import terminate_threads +from git.async.util import cpu_count + +import threading +import weakref +import time +import sys + + + +class TestThreadPool(TestBase): + + max_threads = cpu_count() + + def _assert_single_task(self, p, async=False): + """Performs testing in a synchronized environment""" + print >> sys.stderr, "Threadpool: Starting single task (async = %i) with %i threads" % (async, p.size()) + null_tasks = p.num_tasks() # in case we had some before + + # add a simple task + # it iterates n items + ni = 1000 + assert ni % 2 == 0, "ni needs to be dividable by 2" + assert ni % 4 == 0, "ni needs to be dividable by 4" + + make_task = lambda *args, **kwargs: make_iterator_task(ni, *args, **kwargs) + + task = make_task() + + assert p.num_tasks() == null_tasks + rc = p.add_task(task) + assert p.num_tasks() == 1 + null_tasks + assert isinstance(rc, PoolReader) + assert task._out_writer is not None + + # pull the result completely - we should get one task, which calls its + # function once. In sync mode, the order matches + print "read(0)" + items = rc.read() + assert len(items) == ni + task._assert(1, ni) + if not async: + assert items[0] == 0 and items[-1] == ni-1 + + # as the task is done, it should have been removed - we have read everything + assert task.is_done() + del(rc) + assert p.num_tasks() == null_tasks + task = make_task() + + # pull individual items + rc = p.add_task(task) + assert p.num_tasks() == 1 + null_tasks + st = time.time() + print "read(1) * %i" % ni + for i in range(ni): + items = rc.read(1) + assert len(items) == 1 + + # can't assert order in async mode + if not async: + assert i == items[0] + # END for each item + elapsed = time.time() - st + print >> sys.stderr, "Threadpool: processed %i individual items, with %i threads, one at a time, in %f s ( %f items / s )" % (ni, p.size(), elapsed, ni / elapsed) + + # it couldn't yet notice that the input is depleted as we pulled exaclty + # ni items - the next one would remove it. Instead, we delete our channel + # which triggers orphan handling + assert not task.is_done() + assert p.num_tasks() == 1 + null_tasks + del(rc) + assert p.num_tasks() == null_tasks + + # test min count + # if we query 1 item, it will prepare ni / 2 + task = make_task() + task.min_count = ni / 2 + rc = p.add_task(task) + print "read(1)" + items = rc.read(1) + assert len(items) == 1 and items[0] == 0 # processes ni / 2 + print "read(1)" + items = rc.read(1) + assert len(items) == 1 and items[0] == 1 # processes nothing + # rest - it has ni/2 - 2 on the queue, and pulls ni-2 + # It wants too much, so the task realizes its done. The task + # doesn't care about the items in its output channel + nri = ni-2 + print "read(%i)" % nri + items = rc.read(nri) + assert len(items) == nri + p.remove_task(task) + assert p.num_tasks() == null_tasks + task._assert(2, ni) # two chunks, ni calls + + # its already done, gives us no more, its still okay to use it though + # as a task doesn't have to be in the graph to allow reading its produced + # items + print "read(0) on closed" + # it can happen that a thread closes the channel just a tiny fraction of time + # after we check this, so the test fails, although it is nearly closed. + # When we start reading, we should wake up once it sends its signal + # assert task.is_closed() + assert len(rc.read()) == 0 + + # test chunking + # we always want 4 chunks, these could go to individual nodes + task = make_task() + task.min_count = ni / 2 # restore previous value + task.max_chunksize = ni / 4 # 4 chunks + rc = p.add_task(task) + + # must read a specific item count + # count is still at ni / 2 - here we want more than that + # 2 steps with n / 4 items, + 1 step with n/4 items to get + 2 + nri = ni / 2 + 2 + print "read(%i) chunksize set" % nri + items = rc.read(nri) + assert len(items) == nri + # have n / 4 - 2 items on queue, want n / 4 in first chunk, cause 1 processing + # ( 4 in total ). Still want n / 4 - 2 in second chunk, causing another processing + nri = ni / 2 - 2 + print "read(%i) chunksize set" % nri + items = rc.read(nri) + assert len(items) == nri + + task._assert( 5, ni) + + # delete the handle first, causing the task to be removed and to be set + # done. We check for the set-done state later. Depending on the timing, + # The task is not yet set done when we are checking it because we were + # scheduled in before the flag could be set. + del(rc) + assert task.is_done() + assert p.num_tasks() == null_tasks # depleted + + # but this only hits if we want too many items, if we want less, it could + # still do too much - hence we set the min_count to the same number to enforce + # at least ni / 4 items to be preocessed, no matter what we request + task = make_task() + task.min_count = None + task.max_chunksize = ni / 4 # match previous setup + rc = p.add_task(task) + st = time.time() + print "read(1) * %i, chunksize set" % ni + for i in range(ni): + if async: + assert len(rc.read(1)) == 1 + else: + assert rc.read(1)[0] == i + # END handle async mode + # END pull individual items + # too many processing counts ;) + elapsed = time.time() - st + print >> sys.stderr, "Threadpool: processed %i individual items in chunks of %i, with %i threads, one at a time, in %f s ( %f items / s )" % (ni, ni/4, p.size(), elapsed, ni / elapsed) + + task._assert(ni, ni) + assert p.num_tasks() == 1 + null_tasks + assert p.remove_task(task) is p # del manually this time + assert p.num_tasks() == null_tasks + + # now with we set the minimum count to reduce the number of processing counts + task = make_task() + task.min_count = ni / 4 + task.max_chunksize = ni / 4 # match previous setup + rc = p.add_task(task) + print "read(1) * %i, min_count%i + chunksize" % (ni, task.min_count) + for i in range(ni): + items = rc.read(1) + assert len(items) == 1 + if not async: + assert items[0] == i + # END for each item + task._assert(ni / task.min_count, ni) + del(rc) + assert p.num_tasks() == null_tasks + + # test failure + # on failure, the processing stops and the task is finished, keeping + # his error for later + task = make_task() + task.should_fail = True + rc = p.add_task(task) + print "read(0) with failure" + assert len(rc.read()) == 0 # failure on first item + + assert isinstance(task.error(), AssertionError) + assert task.is_done() # on error, its marked done as well + del(rc) + assert p.num_tasks() == null_tasks + + # test failure after ni / 2 items + # This makes sure it correctly closes the channel on failure to prevent blocking + nri = ni/2 + task = make_task(TestFailureThreadTask, fail_after=ni/2) + rc = p.add_task(task) + assert len(rc.read()) == nri + assert task.is_done() + assert isinstance(task.error(), AssertionError) + + print >> sys.stderr, "done with everything" + + + + def _assert_async_dependent_tasks(self, pool): + # includes failure in center task, 'recursive' orphan cleanup + # This will also verify that the channel-close mechanism works + # t1 -> t2 -> t3 + + print >> sys.stderr, "Threadpool: starting async dependency test in %i threads" % pool.size() + null_tasks = pool.num_tasks() + ni = 1000 + count = 3 + aic = count + 2 + make_task = lambda *args, **kwargs: add_task_chain(pool, ni, count, *args, **kwargs) + + ts, rcs = make_task() + assert len(ts) == aic + assert len(rcs) == aic + assert pool.num_tasks() == null_tasks + len(ts) + + # read(0) + ######### + st = time.time() + items = rcs[-1].read() + elapsed = time.time() - st + print len(items), ni + assert len(items) == ni + del(rcs) + assert pool.num_tasks() == 0 # tasks depleted, all done, no handles + # wait a tiny moment - there could still be something unprocessed on the + # queue, increasing the refcount + time.sleep(0.15) + assert sys.getrefcount(ts[-1]) == 2 # ts + call + assert sys.getrefcount(ts[0]) == 2 # ts + call + print >> sys.stderr, "Dependent Tasks: evaluated %i items of %i dependent in %f s ( %i items / s )" % (ni, aic, elapsed, ni / elapsed) + + + # read(1) + ######### + ts, rcs = make_task() + st = time.time() + for i in xrange(ni): + items = rcs[-1].read(1) + assert len(items) == 1 + # END for each item to pull + elapsed_single = time.time() - st + # another read yields nothing, its empty + assert len(rcs[-1].read()) == 0 + print >> sys.stderr, "Dependent Tasks: evaluated %i items with read(1) of %i dependent in %f s ( %i items / s )" % (ni, aic, elapsed_single, ni / elapsed_single) + + + # read with min-count size + ########################### + # must be faster, as it will read ni / 4 chunks + # Its enough to set one task, as it will force all others in the chain + # to min_size as well. + ts, rcs = make_task() + assert pool.num_tasks() == len(ts) + nri = ni / 4 + ts[-1].min_count = nri + st = time.time() + for i in xrange(ni): + items = rcs[-1].read(1) + assert len(items) == 1 + # END for each item to read + elapsed_minsize = time.time() - st + # its empty + assert len(rcs[-1].read()) == 0 + print >> sys.stderr, "Dependent Tasks: evaluated %i items with read(1), min_size=%i, of %i dependent in %f s ( %i items / s )" % (ni, nri, aic, elapsed_minsize, ni / elapsed_minsize) + + # it should have been a bit faster at least, and most of the time it is + # Sometimes, its not, mainly because: + # * The test tasks lock a lot, hence they slow down the system + # * Each read will still trigger the pool to evaluate, causing some overhead + # even though there are enough items on the queue in that case. Keeping + # track of the scheduled items helped there, but it caused further inacceptable + # slowdown + # assert elapsed_minsize < elapsed_single + + + # read with failure + ################### + # it should recover and give at least fail_after items + # t1 -> x -> t3 + fail_after = ni/2 + ts, rcs = make_task(fail_setup=[(0, fail_after)]) + items = rcs[-1].read() + assert len(items) == fail_after + + + # MULTI-POOL + # If two pools are connected, this shold work as well. + # The second one has just one more thread + ts, rcs = make_task() + + # connect verifier channel as feeder of the second pool + p2 = ThreadPool(0) # don't spawn new threads, they have the tendency not to wake up on mutexes + assert p2.size() == 0 + p2ts, p2rcs = add_task_chain(p2, ni, count, feeder_channel=rcs[-1], id_offset=count) + assert p2ts[0] is None # we have no feeder task + assert rcs[-1].pool_ref()() is pool # it didnt change the pool + assert rcs[-1] is p2ts[1].reader() + assert p2.num_tasks() == len(p2ts)-1 # first is None + + # reading from the last one will evaluate all pools correctly + print "read(0) multi-pool" + st = time.time() + items = p2rcs[-1].read() + elapsed = time.time() - st + assert len(items) == ni + + print >> sys.stderr, "Dependent Tasks: evaluated 2 connected pools and %i items with read(0), of %i dependent tasks in %f s ( %i items / s )" % (ni, aic + aic-1, elapsed, ni / elapsed) + + + # loose the handles of the second pool to allow others to go as well + del(p2rcs); del(p2ts) + assert p2.num_tasks() == 0 + + # now we lost our old handles as well, and the tasks go away + ts, rcs = make_task() + assert pool.num_tasks() == len(ts) + + p2ts, p2rcs = add_task_chain(p2, ni, count, feeder_channel=rcs[-1], id_offset=count) + assert p2.num_tasks() == len(p2ts) - 1 + + # Test multi-read(1) + print "read(1) * %i" % ni + reader = rcs[-1] + st = time.time() + for i in xrange(ni): + items = reader.read(1) + assert len(items) == 1 + # END for each item to get + elapsed = time.time() - st + del(reader) # decrement refcount + + print >> sys.stderr, "Dependent Tasks: evaluated 2 connected pools and %i items with read(1), of %i dependent tasks in %f s ( %i items / s )" % (ni, aic + aic-1, elapsed, ni / elapsed) + + # another read is empty + assert len(rcs[-1].read()) == 0 + + # now that both are connected, I can drop my handle to the reader + # without affecting the task-count, but whats more important: + # They remove their tasks correctly once we drop our references in the + # right order + del(p2ts) + assert p2rcs[0] is rcs[-1] + del(p2rcs) + assert p2.num_tasks() == 0 + del(p2) + + assert pool.num_tasks() == null_tasks + len(ts) + + + del(ts) + del(rcs) + + assert pool.num_tasks() == null_tasks + + + # ASSERTION: We already tested that one pool behaves correctly when an error + # occours - if two pools handle their ref-counts correctly, which they + # do if we are here, then they should handle errors happening during + # the task processing as expected as well. Hence we can safe this here + + + + @terminate_threads + def test_base(self): + max_wait_attempts = 3 + sleep_time = 0.1 + for mc in range(max_wait_attempts): + # wait for threads to die + if len(threading.enumerate()) != 1: + time.sleep(sleep_time) + # END for each attempt + assert len(threading.enumerate()) == 1, "Waited %f s for threads to die, its still alive" % (max_wait_attempts, sleep_time) + + p = ThreadPool() + + # default pools have no workers + assert p.size() == 0 + + # increase and decrease the size + num_threads = len(threading.enumerate()) + for i in range(self.max_threads): + p.set_size(i) + assert p.size() == i + assert len(threading.enumerate()) == num_threads + i + + for i in range(self.max_threads, -1, -1): + p.set_size(i) + assert p.size() == i + + assert p.size() == 0 + # threads should be killed already, but we let them a tiny amount of time + # just to be sure + time.sleep(0.05) + assert len(threading.enumerate()) == num_threads + + # SINGLE TASK SERIAL SYNC MODE + ############################## + # put a few unrelated tasks that we forget about - check ref counts and cleanup + t1, t2 = TestThreadTask(iter(list()), "nothing1", None), TestThreadTask(iter(list()), "nothing2", None) + urc1 = p.add_task(t1) + urc2 = p.add_task(t2) + assert p.num_tasks() == 2 + + ## SINGLE TASK ################# + self._assert_single_task(p, False) + assert p.num_tasks() == 2 + del(urc1) + assert p.num_tasks() == 1 + + p.remove_task(t2) + assert p.num_tasks() == 0 + assert sys.getrefcount(t2) == 2 + + t3 = TestChannelThreadTask(urc2, "channel", None) + urc3 = p.add_task(t3) + assert p.num_tasks() == 1 + del(urc3) + assert p.num_tasks() == 0 + assert sys.getrefcount(t3) == 2 + + + # DEPENDENT TASKS SYNC MODE + ########################### + self._assert_async_dependent_tasks(p) + + + # SINGLE TASK THREADED ASYNC MODE ( 1 thread ) + ############################################## + # step one gear up - just one thread for now. + p.set_size(1) + assert p.size() == 1 + assert len(threading.enumerate()) == num_threads + 1 + # deleting the pool stops its threads - just to be sure ;) + # Its not synchronized, hence we wait a moment + del(p) + time.sleep(0.05) + assert len(threading.enumerate()) == num_threads + + p = ThreadPool(1) + assert len(threading.enumerate()) == num_threads + 1 + + # here we go + self._assert_single_task(p, True) + + + + # SINGLE TASK ASYNC MODE ( 2 threads ) + ###################################### + # two threads to compete for a single task + p.set_size(2) + self._assert_single_task(p, True) + + # real stress test- should be native on every dual-core cpu with 2 hardware + # threads per core + p.set_size(4) + self._assert_single_task(p, True) + + + # DEPENDENT TASK ASYNC MODE + ########################### + self._assert_async_dependent_tasks(p) + + print >> sys.stderr, "Done with everything" + diff --git a/test/git/async/test_task.py b/test/git/async/test_task.py new file mode 100644 index 00000000..c6a796e9 --- /dev/null +++ b/test/git/async/test_task.py @@ -0,0 +1,15 @@ +"""Channel testing""" +from test.testlib import * +from git.async.util import * +from git.async.task import * + +import time + +class TestTask(TestBase): + + max_threads = cpu_count() + + def test_iterator_task(self): + # tested via test_pool + pass + diff --git a/test/git/async/test_thread.py b/test/git/async/test_thread.py new file mode 100644 index 00000000..a08c1dc7 --- /dev/null +++ b/test/git/async/test_thread.py @@ -0,0 +1,44 @@ +# -*- coding: utf-8 -*- +""" Test thead classes and functions""" +from test.testlib import * +from git.async.thread import * +from Queue import Queue +import time + +class TestWorker(WorkerThread): + def __init__(self, *args, **kwargs): + super(TestWorker, self).__init__(*args, **kwargs) + self.reset() + + def fun(self, arg): + self.called = True + self.arg = arg + return True + + def make_assertion(self): + assert self.called + assert self.arg + self.reset() + + def reset(self): + self.called = False + self.arg = None + + +class TestThreads( TestCase ): + + @terminate_threads + def test_worker_thread(self): + worker = TestWorker() + assert isinstance(worker.start(), WorkerThread) + + # test different method types + standalone_func = lambda *args, **kwargs: worker.fun(*args, **kwargs) + for function in (TestWorker.fun, worker.fun, standalone_func): + worker.inq.put((function, 1)) + time.sleep(0.01) + worker.make_assertion() + # END for each function type + + worker.stop_and_join() + |