upstream/ipython Files · IPython/zmq/exitpoller.py

Make our temp file mixin unittest-friendly....

Make our temp file mixin unittest-friendly. This does not break any nose compatibility, but it lets us use our mixin with normal unittest.TestCase objects in addition to nose-oriented tests. We're trying to lower our dependency on specifics of the nose api, so that in the future we might be able to run without nose at all (as unittest improves).

Brian Granger - - Load All Authors

File last commit:

r2754:6f2df84b


                r2907:566a32b3

Download file

             exitpoller.py
        
                    41 lines
            
             | 1.2 KiB
            
                | text/x-python
            
             |
                PythonLexer
            
             / IPython / zmq / exitpoller.py
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      import os

      import time

      from threading import Thread

      class ExitPollerUnix(Thread):

          """ A Unix-specific daemon thread that terminates the program immediately 

          when the parent process no longer exists.

          """

          def __init__(self):

              super(ExitPollerUnix, self).__init__()

              self.daemon = True

          def run(self):

              # We cannot use os.waitpid because it works only for child processes.

              from errno import EINTR

              while True:

                  try:

                      if os.getppid() == 1:

                          os._exit(1)

                      time.sleep(1.0)

                  except OSError, e:

                      if e.errno == EINTR:

                          continue

                      raise

      class ExitPollerWindows(Thread):

          """ A Windows-specific daemon thread that terminates the program immediately

          when a Win32 handle is signaled.

          """ 

          def __init__(self, handle):

              super(ExitPollerWindows, self).__init__()

              self.daemon = True

              self.handle = handle

          def run(self):

              from _subprocess import WaitForSingleObject, WAIT_OBJECT_0, INFINITE

              result = WaitForSingleObject(self.handle, INFINITE)

              if result == WAIT_OBJECT_0:

                  os._exit(1)

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

				import os
				import time
				from threading import Thread


				class ExitPollerUnix(Thread):
				""" A Unix-specific daemon thread that terminates the program immediately
				when the parent process no longer exists.
				"""

				def __init__(self):
				super(ExitPollerUnix, self).__init__()
				self.daemon = True

				def run(self):
				# We cannot use os.waitpid because it works only for child processes.
				from errno import EINTR
				while True:
				try:
				if os.getppid() == 1:
				os._exit(1)
				time.sleep(1.0)
				except OSError, e:
				if e.errno == EINTR:
				continue
				raise

				class ExitPollerWindows(Thread):
				""" A Windows-specific daemon thread that terminates the program immediately
				when a Win32 handle is signaled.
				"""

				def __init__(self, handle):
				super(ExitPollerWindows, self).__init__()
				self.daemon = True
				self.handle = handle

				def run(self):
				from _subprocess import WaitForSingleObject, WAIT_OBJECT_0, INFINITE
				result = WaitForSingleObject(self.handle, INFINITE)
				if result == WAIT_OBJECT_0:
				os._exit(1)