我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用threading._Event()。
def __init__(self, *towatch): """MessageTracker(*towatch) Create a message tracker to track a set of mesages. Parameters ---------- *towatch : tuple of Event, MessageTracker, Message instances. This list of objects to track. This class can track the low-level Events used by the Message class, other MessageTrackers or actual Messages. """ self.events = set() self.peers = set() for obj in towatch: if isinstance(obj, Event): self.events.add(obj) elif isinstance(obj, MessageTracker): self.peers.add(obj) elif isinstance(obj, Frame): if not obj.tracker: raise ValueError("Not a tracked message") self.peers.add(obj.tracker) else: raise TypeError("Require Events or Message Frames, not %s"%type(obj))
def test_event(self): event = self.Event() wait = TimingWrapper(event.wait) # Removed temporarily, due to API shear, this does not # work with threading._Event objects. is_set == isSet self.assertEqual(event.is_set(), False) # Removed, threading.Event.wait() will return the value of the __flag # instead of None. API Shear with the semaphore backed mp.Event self.assertEqual(wait(0.0), False) self.assertTimingAlmostEqual(wait.elapsed, 0.0) self.assertEqual(wait(TIMEOUT1), False) self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) event.set() # See note above on the API differences self.assertEqual(event.is_set(), True) self.assertEqual(wait(), True) self.assertTimingAlmostEqual(wait.elapsed, 0.0) self.assertEqual(wait(TIMEOUT1), True) self.assertTimingAlmostEqual(wait.elapsed, 0.0) # self.assertEqual(event.is_set(), True) event.clear() #self.assertEqual(event.is_set(), False) p = self.Process(target=self._test_event, args=(event,)) p.daemon = True p.start() self.assertEqual(wait(), True) # # #
def test_get_event_camera_ready(self): assert_is_instance(self.videoThread.get_event_camera_ready(), threading._Event)
def test_get_event_camera_chosen(self): assert_is_instance(self.videoThread.get_event_camera_chosen(), threading._Event)
def __init__(self, work_queue=None): _th.Thread.__init__(self) self.ev = _th._Event() self._event_status = False; # ???? self.var_lock = True; self.work_queue = work_queue self.mutex = _th.Lock() # ??? self.start();
def wait(self, timeout): threading._Event.wait(self, timeout) return self.isSet()
def event_loop_with_quit_event(self, quit_event): ''' in kernel debugging session IDebugControl.WaitForEvent() must be called with an 'infinite' timeout value. this is why we have to create thread that checks the given given quit event. once set it will force a debugger break. ''' if (self.is_deleted): raise DebuggerException("called when object is deleted") # sanity check on quit_event #if (not isinstance(quit_event, threading._Event)): # raise DebuggerException("invalid type for quit event") # is already set? if (quit_event.is_set()): # no job for us return # create abort event abort_quit_waiter_event = Event() # start quit thread quit_waiter = KernelAttacher.QuitEventWaiter(quit_event = quit_event, abort_event = abort_quit_waiter_event, top = self) # event loop self.__event_loop_with_forced_break_check(quit_event) # stop quit waiter thread self.dbg_eng_log("KernelAttacher.event_loop: waiting for quit_waiter to end") abort_quit_waiter_event.set() quit_waiter.join() # dont wait for garbage collection, and force delete on self self.__del__() ###########################################################
def event_loop_with_quit_event(self, quit_event): #if (not isinstance(quit_event, threading._Event)): # raise DebuggerException("UserModeSession.event_loop_with_quit_event(): invalid quit_event") while (not quit_event.is_set()): if (self.wait_for_event(200) == False): self.dbg_eng_log("UserModeSession.event_loop_with_quit_event: wait_for_event() done. breaking loop.") break ########################################################### # handle functions ###########################################################
def test_event(self): event = self.Event() wait = TimingWrapper(event.wait) # Removed temporarily, due to API shear, this does not # work with threading._Event objects. is_set == isSet self.assertEqual(event.is_set(), False) # Removed, threading.Event.wait() will return the value of the __flag # instead of None. API Shear with the semaphore backed mp.Event self.assertEqual(wait(0.0), False) self.assertTimingAlmostEqual(wait.elapsed, 0.0) self.assertEqual(wait(TIMEOUT1), False) self.assertTimingAlmostEqual(wait.elapsed, TIMEOUT1) event.set() # See note above on the API differences self.assertEqual(event.is_set(), True) self.assertEqual(wait(), True) self.assertTimingAlmostEqual(wait.elapsed, 0.0) self.assertEqual(wait(TIMEOUT1), True) self.assertTimingAlmostEqual(wait.elapsed, 0.0) # self.assertEqual(event.is_set(), True) event.clear() #self.assertEqual(event.is_set(), False) p = self.Process(target=self._test_event, args=(event,)) p.daemon = True p.start() self.assertEqual(wait(), True) # # Tests for Barrier - adapted from tests in test/lock_tests.py # # Many of the tests for threading.Barrier use a list as an atomic # counter: a value is appended to increment the counter, and the # length of the list gives the value. We use the class DummyList # for the same purpose.
