def wait_on_children(self):
"""Wait on children exit."""
while self.running:
try:
pid, status = os.wait()
if os.WIFEXITED(status) or os.WIFSIGNALED(status):
self._remove_children(pid)
self._verify_and_respawn_children(pid, status)
except OSError as err:
if err.errno not in (errno.EINTR, errno.ECHILD):
raise
except KeyboardInterrupt:
LOG.info(_LI('Caught keyboard interrupt. Exiting.'))
os.killpg(0, signal.SIGTERM)
break
except exception.SIGHUPInterrupt:
self.reload()
continue
eventlet.greenio.shutdown_safe(self.sock)
self.sock.close()
LOG.debug('Exited')
python类wait()的实例源码
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, '')
self.assertEqual(err, '')
def _run_and_join(self, script):
script = """if 1:
import sys, os, time, threading
# a thread, which waits for the main program to terminate
def joiningfunc(mainthread):
mainthread.join()
print 'end of thread'
\n""" + script
p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
rc = p.wait()
data = p.stdout.read().replace('\r', '')
p.stdout.close()
self.assertEqual(data, "end of main\nend of thread\n")
self.assertFalse(rc == 2, "interpreter was blocked")
self.assertTrue(rc == 0, "Unexpected error")
def test_reinit_tls_after_fork(self):
# Issue #13817: fork() would deadlock in a multithreaded program with
# the ad-hoc TLS implementation.
def do_fork_and_wait():
# just fork a child process and wait it
pid = os.fork()
if pid > 0:
os.waitpid(pid, 0)
else:
os._exit(0)
# start a bunch of threads that will fork() child processes
threads = []
for i in range(16):
t = threading.Thread(target=do_fork_and_wait)
threads.append(t)
t.start()
for t in threads:
t.join()
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, '')
self.assertEqual(err, '')
def _run_and_join(self, script):
script = """if 1:
import sys, os, time, threading
# a thread, which waits for the main program to terminate
def joiningfunc(mainthread):
mainthread.join()
print 'end of thread'
\n""" + script
p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
rc = p.wait()
data = p.stdout.read().replace('\r', '')
p.stdout.close()
self.assertEqual(data, "end of main\nend of thread\n")
self.assertFalse(rc == 2, "interpreter was blocked")
self.assertTrue(rc == 0, "Unexpected error")
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import _thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
_thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, b'')
self.assertEqual(err, b'')
def test_reinit_tls_after_fork(self):
# Issue #13817: fork() would deadlock in a multithreaded program with
# the ad-hoc TLS implementation.
def do_fork_and_wait():
# just fork a child process and wait it
pid = os.fork()
if pid > 0:
os.waitpid(pid, 0)
else:
os._exit(0)
# start a bunch of threads that will fork() child processes
threads = []
for i in range(16):
t = threading.Thread(target=do_fork_and_wait)
threads.append(t)
t.start()
for t in threads:
t.join()
def write_multi_process(self, child_count):
log.info("Generating pages using %d child processes", child_count)
pages = list(self.site.pages.values())
# From http://code.activestate.com/recipes/576785-partition-an-iterable-into-n-lists/
chunks = [pages[i::child_count] for i in range(child_count)]
print(len(pages))
for c in chunks:
print(len(c))
import sys
pids = set()
for chunk in chunks:
pid = os.fork()
if pid == 0:
self.write_pages(chunk)
sys.exit(0)
else:
pids.add(pid)
while pids:
(pid, status) = os.wait()
pids.discard(pid)
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, '')
self.assertEqual(err, '')
def _run_and_join(self, script):
script = """if 1:
import sys, os, time, threading
# a thread, which waits for the main program to terminate
def joiningfunc(mainthread):
mainthread.join()
print 'end of thread'
# stdout is fully buffered because not a tty, we have to flush
# before exit.
sys.stdout.flush()
\n""" + script
p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
rc = p.wait()
data = p.stdout.read().replace('\r', '')
p.stdout.close()
self.assertEqual(data, "end of main\nend of thread\n")
self.assertFalse(rc == 2, "interpreter was blocked")
self.assertTrue(rc == 0, "Unexpected error")
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import _thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
_thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, b'')
self.assertEqual(err, b'')
def test_reinit_tls_after_fork(self):
# Issue #13817: fork() would deadlock in a multithreaded program with
# the ad-hoc TLS implementation.
def do_fork_and_wait():
# just fork a child process and wait it
pid = os.fork()
if pid > 0:
os.waitpid(pid, 0)
else:
os._exit(0)
# start a bunch of threads that will fork() child processes
threads = []
for i in range(16):
t = threading.Thread(target=do_fork_and_wait)
threads.append(t)
t.start()
for t in threads:
t.join()
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, '')
self.assertEqual(err, '')
def _run_and_join(self, script):
script = """if 1:
import sys, os, time, threading
# a thread, which waits for the main program to terminate
def joiningfunc(mainthread):
mainthread.join()
print 'end of thread'
\n""" + script
p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)
rc = p.wait()
data = p.stdout.read().replace('\r', '')
p.stdout.close()
self.assertEqual(data, "end of main\nend of thread\n")
self.assertFalse(rc == 2, "interpreter was blocked")
self.assertTrue(rc == 0, "Unexpected error")
def test_arp_ping(self):
"""test ARP ping - compare to arping utilite"""
arp_ping = Ping(IFACE, ARP_NAME, TIMEOUT, False)
for ip in list(ipaddress.ip_network(TEST_NETWORK).hosts())[:5]:
try:
# need arping installed
with os.popen('arping -c {} -t {} {}'.format(COUNT,
TIMEOUT,
str(ip)), 'r'):
# get exit code
ec = os.wait()[1] & 0xFF00
res = arp_ping.ping_host(str(ip))
except PermissionException:
print('Need root previlegies')
if res[STATUS_INDEX] == ONLINE:
self.assertTrue(ec == 0)
else:
self.assertFalse(ec == 0)
def test_icmp_ping(self):
"""test icmp ping - compare to icmping utilite"""
icmp_ping = Ping(IFACE, ICMP_NAME, TIMEOUT, False)
for ip in list(ipaddress.ip_network(TEST_NETWORK).hosts())[:5]:
try:
# need arping installed
with os.popen('ping -c {} -t {} {}'.format(COUNT,
TIMEOUT,
str(ip)), 'r'):
# get exit code
ec = os.wait()[1] & 0xFF00
res = icmp_ping.ping_host(str(ip))
except PermissionException:
print('Need root previlegies')
if res[STATUS_INDEX] == ONLINE:
self.assertTrue(ec == 0)
else:
self.assertFalse(ec == 0)
def test_dummy_thread_after_fork(self):
# Issue #14308: a dummy thread in the active list doesn't mess up
# the after-fork mechanism.
code = """if 1:
import _thread, threading, os, time
def background_thread(evt):
# Creates and registers the _DummyThread instance
threading.current_thread()
evt.set()
time.sleep(10)
evt = threading.Event()
_thread.start_new_thread(background_thread, (evt,))
evt.wait()
assert threading.active_count() == 2, threading.active_count()
if os.fork() == 0:
assert threading.active_count() == 1, threading.active_count()
os._exit(0)
else:
os.wait()
"""
_, out, err = assert_python_ok("-c", code)
self.assertEqual(out, b'')
self.assertEqual(err, b'')
def test_reinit_tls_after_fork(self):
# Issue #13817: fork() would deadlock in a multithreaded program with
# the ad-hoc TLS implementation.
def do_fork_and_wait():
# just fork a child process and wait it
pid = os.fork()
if pid > 0:
os.waitpid(pid, 0)
else:
os._exit(0)
# start a bunch of threads that will fork() child processes
threads = []
for i in range(16):
t = threading.Thread(target=do_fork_and_wait)
threads.append(t)
t.start()
for t in threads:
t.join()
def grim_reaper(signum, frame):
while True:
try:
#??????os.wait()???
#os.wait()??????????????????????????????????????.
#waitpid()???WNOHANG??????????????????????????????pid=0???????.
#????????waitpid?????????????????????????.
pid, status = os.waitpid(
-1,
os.WNOHANG
)
except OSError:
return
if pid == 0:
return
def workers(master_host, master_port, relay_socket_path, num_workers):
# Start the relay
master_redis_cfg = {'host': master_host, 'port': master_port}
relay_redis_cfg = {'unix_socket_path': relay_socket_path}
if os.fork() == 0:
RelayClient(master_redis_cfg, relay_redis_cfg).run()
return
# Start the workers
noise = SharedNoiseTable() # Workers share the same noise
num_workers = num_workers if num_workers else os.cpu_count()
logging.info('Spawning {} workers'.format(num_workers))
for _ in range(num_workers):
if os.fork() == 0:
run_worker(relay_redis_cfg, noise=noise)
return
os.wait()
def _join_daemon(self):
try:
try:
# Mac, UNIX
os.wait()
except AttributeError:
# Windows
try:
pid = self.get_pid()
except IOError:
# Assume the subprocess deleted the pidfile on shutdown.
pass
else:
os.waitpid(pid, 0)
except OSError:
x = sys.exc_info()[1]
if x.args != (10, 'No child processes'):
raise
def get_asynchronous_eventlet_pool(size=1000):
"""Return eventlet pool to caller.
Also store pools created in global list, to wait on
it after getting signal for graceful shutdown.
:param size: eventlet pool size
:returns: eventlet pool
"""
global ASYNC_EVENTLET_THREAD_POOL_LIST
pool = eventlet.GreenPool(size=size)
# Add pool to global ASYNC_EVENTLET_THREAD_POOL_LIST
ASYNC_EVENTLET_THREAD_POOL_LIST.append(pool)
return pool
def wait_on_children(self):
while self.running:
try:
pid, status = os.wait()
if os.WIFEXITED(status) or os.WIFSIGNALED(status):
self._remove_children(pid)
self._verify_and_respawn_children(pid, status)
except OSError as err:
if err.errno not in (errno.EINTR, errno.ECHILD):
raise
except KeyboardInterrupt:
LOG.info('Caught keyboard interrupt. Exiting.')
break
except glare_exc.SIGHUPInterrupt:
self.reload()
continue
eventlet.greenio.shutdown_safe(self.sock)
self.sock.close()
LOG.debug('Exited')
def start_server():
run_daemon()
try:
pid = os.fork()
except OSError, e:
logging.error('start http server is error.')
os._exit(1)
if pid == 0:
start_server_in_subprocess()
return
while pid:
# wait ?????????????????
ret = os.wait()
logging.error('Child process already stop, try to start')
try:
pid = os.fork()
if pid == 0:
start_server_in_subprocess()
return
except OSError, e:
logging.error('start http server is error.')
os._exit(1)
# ??????
def run():
# we don't care to be notified of our childrens' exit statuses.
# this prevents zombie processes from cluttering up the process
# table when zopectl start/stop is used interactively.
# DM 2004-11-26: from the Linux "execve(2)" manual page:
# Any signals set to be caught by the calling process are reset
# to their default behaviour.
# The SIGCHLD signal (when set to SIG_IGN) may or may not be reset
# to SIG_DFL.
# If it is not reset, 'os.wait[pid]' can non-deterministically fail.
# Thus, use a way such that "SIGCHLD" is definitely reset in children.
# signal.signal(signal.SIGCHLD, signal.SIG_IGN)
if not WIN and os.uname()[0] != 'Darwin':
# On Windows the os.uname method does not exist.
# On Mac OS X, setting up a signal handler causes waitpid to
# raise EINTR, which is not preventable via the Python signal
# handler API and can't be dealt with properly as we can't pass
# the SA_RESTART to the signal API. Since Mac OS X doesn't
# appear to clutter up the process table with zombies if
# SIGCHILD is unset, just don't bother registering a SIGCHILD
# signal handler at all.
signal.signal(signal.SIGCHLD, _ignoreSIGCHLD)
exitstatus = main()
sys.exit(exitstatus)
libmilter.py 文件源码
项目：sipxecs-voicemail-transcription
作者: andrewsauder
项目源码
文件源码
阅读 31
收藏 0
点赞 0
评论 0
def _handleChildren(self):
while not self._close.isSet():
try:
os.wait()
except:
time.sleep(0.5)
def watch(self):
try:
os.wait()
except KeyboardInterrupt:
# I put the capital B in KeyBoardInterrupt so I can
# tell when the Watcher gets the SIGINT
print 'KeyBoardInterrupt'
self.kill()
sys.exit()
def f(path):
"one file object + forking"
with lockpath.keeping(path) as file:
if os.fork():
os.wait()
else:
lockfile(file)
def F(path):
"separate file objects + forking"
with lockpath.keeping(path):
if os.fork():
os.wait()
else:
lockpath(path)
