mirror of
https://github.com/frappe/gunicorn.git
synced 2026-01-14 11:09:11 +08:00
Merge pull request #1270 from benoitc/improve-arbiter-promotion
remove file locking
This commit is contained in:
Benoit Chesneau
commit
c62cf2f500
@ -14,7 +14,6 @@ import time
|
||||
import traceback
|
||||
|
||||
from gunicorn.errors import HaltServer, AppImportError
|
||||
from gunicorn.lockfile import LockFile
|
||||
from gunicorn.pidfile import Pidfile
|
||||
from gunicorn.sock import create_sockets
|
||||
from gunicorn import util
|
||||
@ -40,7 +39,6 @@ class Arbiter(object):
|
||||
START_CTX = {}
|
||||
|
||||
LISTENERS = []
|
||||
LOCK_FILE = None
|
||||
WORKERS = {}
|
||||
PIPE = []
|
||||
|
||||
@ -65,6 +63,7 @@ class Arbiter(object):
|
||||
self.pidfile = None
|
||||
self.worker_age = 0
|
||||
self.reexec_pid = 0
|
||||
self.master_pid = 0
|
||||
self.master_name = "Master"
|
||||
|
||||
cwd = util.getcwd()
|
||||
@ -126,28 +125,23 @@ class Arbiter(object):
|
||||
"""
|
||||
self.log.info("Starting gunicorn %s", __version__)
|
||||
|
||||
if 'GUNICORN_PID' in os.environ:
|
||||
self.master_pid = int(os.environ.get('GUNICORN_PID'))
|
||||
self.proc_name = self.proc_name + ".2"
|
||||
self.master_name = "Master.2"
|
||||
|
||||
self.pid = os.getpid()
|
||||
if self.cfg.pidfile is not None:
|
||||
self.pidfile = Pidfile(self.cfg.pidfile)
|
||||
pidname = self.cfg.pidfile
|
||||
if self.master_pid != 0:
|
||||
pidname += ".2"
|
||||
self.pidfile = Pidfile(pidname)
|
||||
self.pidfile.create(self.pid)
|
||||
self.cfg.on_starting(self)
|
||||
|
||||
self.init_signals()
|
||||
need_lock = False
|
||||
if not self.LISTENERS:
|
||||
self.LISTENERS, need_lock = create_sockets(self.cfg, self.log)
|
||||
|
||||
if need_lock:
|
||||
if not self.LOCK_FILE:
|
||||
# reuse the lockfile if already set
|
||||
if 'GUNICORN_LOCK' in os.environ:
|
||||
lock_path = os.environ.get('GUNICORN_LOCK')
|
||||
else:
|
||||
lock_path = self.cfg.lockfile
|
||||
|
||||
self.LOCK_FILE = LockFile(lock_path)
|
||||
# add us to the shared lock
|
||||
self.LOCK_FILE.lock()
|
||||
self.LISTENERS = create_sockets(self.cfg, self.log)
|
||||
|
||||
listeners_str = ",".join([str(l) for l in self.LISTENERS])
|
||||
self.log.debug("Arbiter booted")
|
||||
@ -193,7 +187,10 @@ class Arbiter(object):
|
||||
|
||||
try:
|
||||
self.manage_workers()
|
||||
|
||||
while True:
|
||||
self.maybe_promote_master()
|
||||
|
||||
sig = self.SIG_QUEUE.pop(0) if len(self.SIG_QUEUE) else None
|
||||
if sig is None:
|
||||
self.sleep()
|
||||
@ -302,6 +299,23 @@ class Arbiter(object):
|
||||
else:
|
||||
self.log.debug("SIGWINCH ignored. Not daemonized")
|
||||
|
||||
def maybe_promote_master(self):
|
||||
if self.master_pid == 0:
|
||||
return
|
||||
|
||||
if self.master_pid != os.getppid():
|
||||
self.log.info("Master has been promoted.")
|
||||
# reset master infos
|
||||
self.master_name = "Master"
|
||||
self.master_pid = 0
|
||||
self.proc_name = self.cfg.proc_name
|
||||
del os.environ['GUNICORN_PID']
|
||||
# rename the pidfile
|
||||
if self.pidfile is not None:
|
||||
self.pidfile.rename(self.cfg.pidfile)
|
||||
# reset proctitle
|
||||
util._setproctitle("master [%s]" % self.proc_name)
|
||||
|
||||
def wakeup(self):
|
||||
"""\
|
||||
Wake up the arbiter by writing to the PIPE
|
||||
@ -350,17 +364,11 @@ class Arbiter(object):
|
||||
:attr graceful: boolean, If True (the default) workers will be
|
||||
killed gracefully (ie. trying to wait for the current connection)
|
||||
"""
|
||||
locked = False
|
||||
if self.LOCK_FILE:
|
||||
self.LOCK_FILE.unlock()
|
||||
locked = self.LOCK_FILE.locked()
|
||||
|
||||
# delete the lock file if needed
|
||||
if not locked and 'GUNICORN_LOCK' in os.environ:
|
||||
del os.environ['GUNICORN_LOCK']
|
||||
if self.reexec_pid == 0 and self.master_pid == 0:
|
||||
for l in self.LISTENERS:
|
||||
l.close()
|
||||
|
||||
for l in self.LISTENERS:
|
||||
l.close(locked)
|
||||
self.LISTENERS = []
|
||||
sig = signal.SIGTERM
|
||||
if not graceful:
|
||||
@ -378,20 +386,23 @@ class Arbiter(object):
|
||||
"""\
|
||||
Relaunch the master and workers.
|
||||
"""
|
||||
if self.pidfile is not None:
|
||||
self.pidfile.rename("%s.oldbin" % self.pidfile.fname)
|
||||
if self.reexec_pid != 0:
|
||||
self.log.warning("USR2 signal ignored. Child exists.")
|
||||
return
|
||||
|
||||
if self.master_pid != 0:
|
||||
self.log.warning("USR2 signal ignored. Parent exists")
|
||||
return
|
||||
|
||||
master_pid = os.getpid()
|
||||
self.reexec_pid = os.fork()
|
||||
if self.reexec_pid != 0:
|
||||
self.master_name = "Old Master"
|
||||
return
|
||||
|
||||
environ = self.cfg.env_orig.copy()
|
||||
fds = [l.fileno() for l in self.LISTENERS]
|
||||
environ['GUNICORN_FD'] = ",".join([str(fd) for fd in fds])
|
||||
|
||||
if self.LOCK_FILE:
|
||||
environ['GUNICORN_LOCK'] = self.LOCK_FILE.name()
|
||||
environ['GUNICORN_PID'] = str(master_pid)
|
||||
|
||||
os.chdir(self.START_CTX['cwd'])
|
||||
self.cfg.pre_exec(self)
|
||||
|
||||
@ -904,20 +904,6 @@ class Pidfile(Setting):
|
||||
If not set, no PID file will be written.
|
||||
"""
|
||||
|
||||
class LockFile(Setting):
|
||||
name = "lockfile"
|
||||
section = "Server Mechanics"
|
||||
cli = ["--lock-file"]
|
||||
meta = "FILE"
|
||||
validator = validate_string
|
||||
default = util.tmpfile(suffix=".lock", prefix="gunicorn-")
|
||||
|
||||
desc = """\
|
||||
A filename to use for the lock file. A lock file is created when using unix sockets.
|
||||
If not set, the default file 'gunicorn-<RANDOMSTRING>.lock' will be created in the
|
||||
temporary directory.
|
||||
"""
|
||||
|
||||
class WorkerTmpDir(Setting):
|
||||
name = "worker_tmp_dir"
|
||||
section = "Server Mechanics"
|
||||
|
||||
@ -1,72 +0,0 @@
|
||||
# -*- coding: utf-8 -
|
||||
#
|
||||
# This file is part of gunicorn released under the MIT license.
|
||||
# See the NOTICE for more information.
|
||||
|
||||
import os
|
||||
|
||||
from gunicorn import util
|
||||
|
||||
if os.name == 'nt':
|
||||
import msvcrt
|
||||
|
||||
def _lock(fd):
|
||||
msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
|
||||
|
||||
|
||||
def _unlock(fd):
|
||||
try:
|
||||
msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
|
||||
except OSError:
|
||||
return False
|
||||
return True
|
||||
|
||||
else:
|
||||
import fcntl
|
||||
|
||||
def _lock(fd):
|
||||
fcntl.lockf(fd, fcntl.LOCK_SH | fcntl.LOCK_NB)
|
||||
|
||||
|
||||
def _unlock(fd):
|
||||
try:
|
||||
fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
||||
except:
|
||||
print("no unlock")
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
class LockFile(object):
|
||||
"""Manage a LOCK file"""
|
||||
|
||||
def __init__(self, fname):
|
||||
self.fname = fname
|
||||
fdir = os.path.dirname(self.fname)
|
||||
if fdir and not os.path.isdir(fdir):
|
||||
raise RuntimeError("%s doesn't exist. Can't create lock file." % fdir)
|
||||
self._lockfile = open(self.fname, 'w+b')
|
||||
# set permissions to -rw-r--r--
|
||||
os.chmod(self.fname, 420)
|
||||
self._locked = False
|
||||
|
||||
def lock(self):
|
||||
_lock(self._lockfile.fileno())
|
||||
self._locked = True
|
||||
|
||||
def unlock(self):
|
||||
if not self.locked():
|
||||
return
|
||||
|
||||
if _unlock(self._lockfile.fileno()):
|
||||
self._lockfile.close()
|
||||
util.unlink(self.fname)
|
||||
self._lockfile = None
|
||||
self._locked = False
|
||||
|
||||
def locked(self):
|
||||
return self._lockfile is not None and self._locked
|
||||
|
||||
def name(self):
|
||||
return self.fname
|
||||
@ -53,7 +53,7 @@ class BaseSocket(object):
|
||||
def bind(self, sock):
|
||||
sock.bind(self.cfg_addr)
|
||||
|
||||
def close(self, locked=False):
|
||||
def close(self):
|
||||
if self.sock is None:
|
||||
return
|
||||
|
||||
@ -108,7 +108,6 @@ class UnixSocket(BaseSocket):
|
||||
os.remove(addr)
|
||||
else:
|
||||
raise ValueError("%r is not a socket" % addr)
|
||||
self.parent = os.getpid()
|
||||
super(UnixSocket, self).__init__(addr, conf, log, fd=fd)
|
||||
|
||||
def __str__(self):
|
||||
@ -120,9 +119,8 @@ class UnixSocket(BaseSocket):
|
||||
util.chown(self.cfg_addr, self.conf.uid, self.conf.gid)
|
||||
os.umask(old_umask)
|
||||
|
||||
def close(self, locked=False):
|
||||
if self.parent == os.getpid() and not locked:
|
||||
os.unlink(self.cfg_addr)
|
||||
def close(self):
|
||||
os.unlink(self.cfg_addr)
|
||||
super(UnixSocket, self).close()
|
||||
|
||||
|
||||
@ -151,7 +149,6 @@ def create_sockets(conf, log):
|
||||
# gunicorn.
|
||||
# http://www.freedesktop.org/software/systemd/man/systemd.socket.html
|
||||
listeners = []
|
||||
need_lock = False
|
||||
if ('LISTEN_PID' in os.environ
|
||||
and int(os.environ.get('LISTEN_PID')) == os.getpid()):
|
||||
for i in range(int(os.environ.get('LISTEN_FDS', 0))):
|
||||
@ -160,7 +157,6 @@ def create_sockets(conf, log):
|
||||
sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
sockname = sock.getsockname()
|
||||
if isinstance(sockname, str) and sockname.startswith('/'):
|
||||
need_lock = True
|
||||
listeners.append(UnixSocket(sockname, conf, log, fd=fd))
|
||||
elif len(sockname) == 2 and '.' in sockname[0]:
|
||||
listeners.append(TCPSocket("%s:%s" % sockname, conf, log,
|
||||
@ -175,7 +171,7 @@ def create_sockets(conf, log):
|
||||
if listeners:
|
||||
log.debug('Socket activation sockets: %s',
|
||||
",".join([str(l) for l in listeners]))
|
||||
return listeners, need_lock
|
||||
return listeners
|
||||
|
||||
# get it only once
|
||||
laddr = conf.address
|
||||
@ -196,9 +192,6 @@ def create_sockets(conf, log):
|
||||
addr = laddr[i]
|
||||
sock_type = _sock_type(addr)
|
||||
|
||||
if sock_type == UnixSocket:
|
||||
need_lock = True
|
||||
|
||||
try:
|
||||
listeners.append(sock_type(addr, conf, log, fd=fd))
|
||||
except socket.error as e:
|
||||
@ -206,14 +199,11 @@ def create_sockets(conf, log):
|
||||
log.error("GUNICORN_FD should refer to an open socket.")
|
||||
else:
|
||||
raise
|
||||
return listeners, need_lock
|
||||
return listeners
|
||||
|
||||
# no sockets is bound, first initialization of gunicorn in this env.
|
||||
for addr in laddr:
|
||||
sock_type = _sock_type(addr)
|
||||
if sock_type == UnixSocket:
|
||||
need_lock = True
|
||||
|
||||
# If we fail to create a socket from GUNICORN_FD
|
||||
# we fall through and try and open the socket
|
||||
# normally.
|
||||
@ -240,4 +230,4 @@ def create_sockets(conf, log):
|
||||
|
||||
listeners.append(sock)
|
||||
|
||||
return listeners, need_lock
|
||||
return listeners
|
||||
|
||||
@ -17,7 +17,6 @@ import resource
|
||||
import socket
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
import textwrap
|
||||
import time
|
||||
import traceback
|
||||
@ -39,8 +38,6 @@ CHUNK_SIZE = (16 * 1024)
|
||||
|
||||
MAX_BODY = 1024 * 132
|
||||
|
||||
normcase = os.path.normcase
|
||||
|
||||
# Server and Date aren't technically hop-by-hop
|
||||
# headers, but they are in the purview of the
|
||||
# origin server which the WSGI spec says we should
|
||||
@ -548,15 +545,4 @@ def make_fail_app(msg):
|
||||
])
|
||||
return [msg]
|
||||
|
||||
return app
|
||||
|
||||
|
||||
characters = ("abcdefghijklmnopqrstuvwxyz" +
|
||||
"ABCDEFGHIJKLMNOPQRSTUVWXYZ" +
|
||||
"0123456789_")
|
||||
|
||||
def tmpfile(suffix="", prefix="tmp"):
|
||||
c = characters
|
||||
rand_str = normcase("".join([random.choice(c) for _ in "123456"]))
|
||||
fname = "".join([prefix, rand_str, suffix])
|
||||
return os.path.join(tempfile.gettempdir(), fname)
|
||||
return app
|
||||
@ -35,8 +35,8 @@ def test_arbiter_shutdown_closes_listeners():
|
||||
listener2 = mock.Mock()
|
||||
arbiter.LISTENERS = [listener1, listener2]
|
||||
arbiter.stop()
|
||||
listener1.close.assert_called_with(False)
|
||||
listener2.close.assert_called_with(False)
|
||||
listener1.close.assert_called_with()
|
||||
listener2.close.assert_called_with()
|
||||
|
||||
|
||||
class PreloadedAppWithEnvSettings(DummyApplication):
|
||||
|
||||
@ -1,15 +0,0 @@
|
||||
import os
|
||||
|
||||
from gunicorn.lockfile import LockFile
|
||||
from gunicorn.util import tmpfile
|
||||
|
||||
def test_lockfile():
|
||||
lockname = tmpfile(prefix="gunicorn-tests", suffix=".lock")
|
||||
lock_file = LockFile(lockname)
|
||||
assert lock_file.locked() == False
|
||||
assert os.path.exists(lockname)
|
||||
lock_file.lock()
|
||||
assert lock_file.locked() == True
|
||||
lock_file.unlock()
|
||||
assert lock_file.locked() == False
|
||||
assert os.path.exists(lockname) == False
|
||||
@ -9,28 +9,7 @@ from gunicorn import sock
|
||||
@mock.patch('os.getpid')
|
||||
@mock.patch('os.unlink')
|
||||
@mock.patch('socket.fromfd')
|
||||
def test_unix_socket_close_delete_if_exlock(fromfd, unlink, getpid):
|
||||
def test_unix_socket_close_unlink(fromfd, unlink, getpid):
|
||||
gsock = sock.UnixSocket('test.sock', mock.Mock(), mock.Mock(), mock.Mock())
|
||||
gsock.close(False)
|
||||
unlink.assert_called_with('test.sock')
|
||||
|
||||
|
||||
@mock.patch('os.getpid')
|
||||
@mock.patch('os.unlink')
|
||||
@mock.patch('socket.fromfd')
|
||||
def test_unix_socket_close_keep_if_no_exlock(fromfd, unlink, getpid):
|
||||
gsock = sock.UnixSocket('test.sock', mock.Mock(), mock.Mock(), mock.Mock())
|
||||
gsock.close(True)
|
||||
unlink.assert_not_called()
|
||||
|
||||
|
||||
@mock.patch('os.getpid')
|
||||
@mock.patch('os.unlink')
|
||||
@mock.patch('socket.fromfd')
|
||||
def test_unix_socket_not_deleted_by_worker(fromfd, unlink, getpid):
|
||||
fd = mock.Mock()
|
||||
gsock = sock.UnixSocket('test.sock', mock.Mock(), mock.Mock(), fd)
|
||||
getpid.reset_mock()
|
||||
getpid.return_value = "fake" # fake a pid change
|
||||
gsock.close(False)
|
||||
unlink.assert_not_called()
|
||||
gsock.close()
|
||||
unlink.assert_called_with("test.sock")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user