

def testMultiTask(self):
    local = TaskLocal()

    def t():
        local.piet = []
        for i in range(10):
            local.piet.append(i)
            Tasklet.yield_()
        self.assertEquals(range(10), local.piet)
    t1 = Tasklet.new(t)()
    t2 = Tasklet.new(t)()
    Tasklet.join_all([t1, t2])
    self.assertEquals(2, len(local._d.keys()))
    del t1
    del t2
    Tasklet.yield_()
    Tasklet.yield_()
    self.assertEquals([], local._d.keys())

dataset/ETHPy150Open concurrence/concurrence/test/testlocal.py TestTaskLocal.testMultiTask/original