(self, ident)
| 384 | verbose_print("tasks done") |
| 385 | |
| 386 | def task2(self, ident): |
| 387 | for i in range(NUMTRIPS): |
| 388 | if ident == 0: |
| 389 | # give it a good chance to enter the next |
| 390 | # barrier before the others are all out |
| 391 | # of the current one |
| 392 | delay = 0 |
| 393 | else: |
| 394 | with self.random_mutex: |
| 395 | delay = random.random() / 10000.0 |
| 396 | verbose_print("task %s will run for %sus" % |
| 397 | (ident, round(delay * 1e6))) |
| 398 | time.sleep(delay) |
| 399 | verbose_print("task %s entering %s" % (ident, i)) |
| 400 | self.bar.enter() |
| 401 | verbose_print("task %s leaving barrier" % ident) |
| 402 | with self.running_mutex: |
| 403 | self.running -= 1 |
| 404 | # Must release mutex before releasing done, else the main thread can |
| 405 | # exit and set mutex to None as part of global teardown; then |
| 406 | # mutex.release() raises AttributeError. |
| 407 | finished = self.running == 0 |
| 408 | if finished: |
| 409 | self.done_mutex.release() |
| 410 | |
| 411 | class LockTests(lock_tests.LockTests): |
| 412 | locktype = thread.allocate_lock |
nothing calls this directly
no test coverage detected