I finally got some time this past weekend do the integration, complete with
lots of unit tests. I've got a patch based on the latest svn revision,
758. The configuration for FastCGI programs is the same as regular programs
except an additional "socket" parameter. Substitution happens on the socket
parameter with the 'here' and 'program_name' variables.
[fcgi-program:fcgi_test]
;socket=tcp://localhost:8002
socket=unix:///path/to/fcgi/socket
...
One caveat with this first implementation is that FastCGI programs must be
homogeneous groups. It may not be too difficult to lift this restriction if
you guys think it's a good idea.
I appreciate whatever feedback you have on this code. If you're willing to
accept the patch, I can help with updating the documentation or whatever
needs to be done.
Thanks,
Roger
On Sun, Mar 2, 2008 at 1:09 PM, Roger Hoover <[EMAIL PROTECTED]> wrote:
>
> > Thanks, Chris. I understand and was planning on writing thorough unit
> > tests.
> >
> >
> > On Sun, Mar 2, 2008 at 11:58 AM, Chris McDonough <[EMAIL PROTECTED]>
> > wrote:
> >
> > > Roger Hoover wrote:
> > > > Hi Mike and Chris,
> > > >
> > > > Thanks for the tips. I wrote a python FastCGI spawner and am ready
> > > to
> > > > integrate it into supervisor. After looking through the code and
> > > Mike's
> > > > tips, I think it will be mostly straightforward.
> > > >
> > > > Here's what I think I need to do:
> > > > - Add PNullDispatcher class to throw away stdin messages (FastCGI
> > > programs
> > > > expect the FastCGI socket to be file descriptor 0 so supervisor pipe
> > > to
> > > > stdin)
> > > > - Add FastCGIConfig class
> > > > - Add FastCGISubProcess class
> > > > - refactor _spawn method in existing Subprocess class to use a
> > > > _prepare_child_fds() method
> > > > - override _prepare_child_fds() to do FastCGI stuff
> > > > - Add FCGIGroupConfig to parse fcgi-program section
> > > >
> > > > I'm new to python so I'm wondering how to get setup to run the
> > > existing
> > > > tests. I checked the code out of svn but can't get the tests to
> > > run. Do I
> > > > need to run an easy_install command of some kind?
> > >
> > > In the checkout, you can do "python setup.py test" to run the tests
> > > (at least on
> > > any Internet-connected system). Note that supervisor is meant to be
> > > compatible
> > > with all of Python 2.3, 2.4, and 2.5, so using any of those versions
> > > of python
> > > to do that, the tests should pass.
> > >
> > > FTR, before you do much work on the integration, I'll offer a warning:
> > > this
> > > stuff will need to have good test coverage before we can consider
> > > putting it in.
> > > Often writing the tests is far more time-consuming than writing the
> > > code (or
> > > it is for me), so please add this into your time calculations when
> > > coming to a
> > > decision about whether you want to go ahead with the integration or
> > > not.
> > >
> > > Thanks!
> > >
> > > - C
> > >
> >
> >
>
Index: src/supervisor/options.py
===================================================================
--- src/supervisor/options.py (revision 758)
+++ src/supervisor/options.py (working copy)
@@ -50,12 +50,16 @@
from supervisor.datatypes import logging_level
from supervisor.datatypes import colon_separated_user_group
from supervisor.datatypes import inet_address
+from supervisor.datatypes import InetStreamSocketConfig
+from supervisor.datatypes import UnixStreamSocketConfig
from supervisor.datatypes import url
from supervisor.datatypes import Automatic
from supervisor.datatypes import auto_restart
from supervisor.datatypes import profile_options
from supervisor.datatypes import set_here
+from supervisor.socket_manager import SocketManager
+
from supervisor import loggers
from supervisor import states
from supervisor import xmlrpc
@@ -637,9 +641,56 @@
result_handler)
)
+ # process fastcgi homogeneous groups
+ for section in all_sections:
+ if ( (not section.startswith('fcgi-program:') )
+ or section in homogeneous_exclude ):
+ continue
+ program_name = section.split(':', 1)[1]
+ priority = integer(get(section, 'priority', 999))
+ socket = get(section, 'socket', None)
+ if not socket:
+ raise ValueError('[%s] section requires a "socket" line' %
+ section)
+
+ expansions = {'here':self.here,
+ 'program_name':program_name}
+ socket = expand(socket, expansions, 'socket')
+ try:
+ socket_config = self.parse_fcgi_socket(socket)
+ except ValueError, e:
+ raise ValueError('%s in [%s] socket' % (str(e), section))
+
+ processes=self.processes_from_section(parser, section,
program_name,
+ FastCGIProcessConfig)
+ groups.append(
+ FastCGIGroupConfig(self, program_name, priority, processes,
+ SocketManager(socket_config))
+ )
+
+
groups.sort()
return groups
+ def parse_fcgi_socket(self, sock):
+ if sock.startswith('unix://'):
+ path = sock[7:]
+ #Check it's an absolute path
+ if not os.path.isabs(path):
+ raise ValueError("Unix socket path %s is not an absolute path",
+ path)
+ path = normalize_path(path)
+ return UnixStreamSocketConfig(path)
+
+ tcp_re = re.compile(r'^tcp://([^\s:]+):(\d+)$')
+ m = tcp_re.match(sock)
+ if m:
+ host = m.group(1)
+ port = int(m.group(2))
+ return InetStreamSocketConfig(host, port)
+
+ raise ValueError("Bad socket format %s", sock)
+
def processes_from_section(self, parser, section, group_name,
klass=None):
if klass is None:
@@ -1495,6 +1546,25 @@
dispatchers[stdin_fd] = PInputDispatcher(proc, 'stdin', stdin_fd)
return dispatchers, p
+class FastCGIProcessConfig(ProcessConfig):
+ def make_process(self, group=None):
+ if group is None:
+ raise NotImplementedError('FastCGI programs require a group')
+ from supervisor.process import FastCGISubprocess
+ process = FastCGISubprocess(self)
+ process.group = group
+ return process
+
+ def make_dispatchers(self, proc):
+ dispatchers, p = ProcessConfig.make_dispatchers(self, proc)
+ #FastCGI child processes expect the FastCGI socket set to
+ #file descriptor 0, so supervisord cannot use stdin
+ #to communicate with the child process
+ stdin_fd = p['stdin']
+ if stdin_fd is not None:
+ dispatchers[stdin_fd].close()
+ return dispatchers, p
+
class ProcessGroupConfig(Config):
def __init__(self, options, name, priority, process_configs):
self.options = options
@@ -1529,6 +1599,19 @@
from supervisor.process import EventListenerPool
return EventListenerPool(self)
+class FastCGIGroupConfig(ProcessGroupConfig):
+ def __init__(self, options, name, priority, process_configs,
+ socket_manager):
+ self.options = options
+ self.name = name
+ self.priority = priority
+ self.process_configs = process_configs
+ self.socket_manager = socket_manager
+
+ def after_setuid(self):
+ ProcessGroupConfig.after_setuid(self)
+ self.socket_manager.prepare_socket()
+
def readFile(filename, offset, length):
""" Read length bytes from the file named by filename starting at
offset """
Index: src/supervisor/datatypes.py
===================================================================
--- src/supervisor/datatypes.py (revision 758)
+++ src/supervisor/datatypes.py (working copy)
@@ -14,6 +14,7 @@
import os
import sys
+import socket
from supervisor.loggers import getLevelNumByDescription
# I dont know why we bother, this doesn't run on Windows, but just
@@ -144,6 +145,60 @@
self.family = socket.AF_INET
self.address = inet_address(s)
+class SocketConfig:
+ """ Abstract base class which provides a uniform abstraction
+ for TCP vs Unix sockets """
+ url = '' # socket url
+ addr = None #socket addr
+
+ def __repr__(self):
+ return '<%s at %s for %s>' % (self.__class__,
+ id(self),
+ self.url)
+
+ def addr(self):
+ raise NotImplementedError
+
+ def create(self):
+ raise NotImplementedError
+
+class InetStreamSocketConfig(SocketConfig):
+ """ TCP socket config helper """
+
+ host = None # host name or ip to bind to
+ port = None # integer port to bind to
+
+ def __init__(self, host, port):
+ self.host = host.lower()
+ self.port = port_number(port)
+ self.url = 'tcp://%s:%d' % (self.host, self.port)
+
+ def addr(self):
+ return (self.host, self.port)
+
+ def create(self):
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ return sock
+
+class UnixStreamSocketConfig(SocketConfig):
+ """ Unix domain socket config helper """
+
+ path = None # Unix domain socket path
+
+ def __init__(self, path):
+ self.path = path
+ self.url = 'unix://%s' % (path)
+
+ def addr(self):
+ return self.path
+
+ def create(self):
+ if os.path.exists(self.path):
+ os.unlink(self.path)
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ return sock
+
def colon_separated_user_group(arg):
try:
result = arg.split(':', 1)
Index: src/supervisor/tests/test_options.py
===================================================================
--- src/supervisor/tests/test_options.py (revision 758)
+++ src/supervisor/tests/test_options.py (working copy)
@@ -12,6 +12,7 @@
from supervisor.tests.base import DummyOptions
from supervisor.tests.base import DummyPConfig
from supervisor.tests.base import DummyProcess
+from supervisor.tests.base import DummySocketManager
from supervisor.tests.base import lstrip
class ClientOptionsTests(unittest.TestCase):
@@ -620,6 +621,120 @@
instance = self._makeOne()
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+ def test_fcgi_programs_from_parser(self):
+ from supervisor.options import FastCGIGroupConfig
+ from supervisor.options import FastCGIProcessConfig
+ text = lstrip("""\
+ [fcgi-program:foo]
+ socket=unix:///tmp/%(program_name)s.sock
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+
+ [fcgi-program:bar]
+ socket=tcp://localhost:6000
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/bar
+ numprocs = 3
+ """)
+ from supervisor.options import UnhosedConfigParser
+ from supervisor.dispatchers import default_handler
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+ gconfigs = instance.process_groups_from_parser(config)
+ self.assertEqual(len(gconfigs), 2)
+
+ gconfig0 = gconfigs[0]
+ self.assertEqual(gconfig0.__class__, FastCGIGroupConfig)
+ self.assertEqual(gconfig0.name, 'foo')
+ self.assertEqual(gconfig0.priority, 1)
+ self.assertEqual(gconfig0.socket_manager.config().url,
'unix:///tmp/foo.sock')
+ self.assertEqual(len(gconfig0.process_configs), 2)
+ self.assertEqual(gconfig0.process_configs[0].__class__,
FastCGIProcessConfig)
+ self.assertEqual(gconfig0.process_configs[1].__class__,
FastCGIProcessConfig)
+
+ gconfig1 = gconfigs[1]
+ self.assertEqual(gconfig1.name, 'bar')
+ self.assertEqual(gconfig1.priority, 999)
+ self.assertEqual(gconfig1.socket_manager.config().url,
'tcp://localhost:6000')
+ self.assertEqual(len(gconfig1.process_configs), 3)
+
+ def test_fcgi_program_no_socket(self):
+ text = lstrip("""\
+ [fcgi-program:foo]
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+ """)
+ from supervisor.options import UnhosedConfigParser
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
+ def test_fcgi_program_unknown_socket_protocol(self):
+ text = lstrip("""\
+ [fcgi-program:foo]
+ socket=junk://blah
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+ """)
+ from supervisor.options import UnhosedConfigParser
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
+ def test_fcgi_program_rel_unix_sock_path(self):
+ text = lstrip("""\
+ [fcgi-program:foo]
+ socket=unix://relative/path
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+ """)
+ from supervisor.options import UnhosedConfigParser
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
+ def test_fcgi_program_bad_tcp_sock_format(self):
+ text = lstrip("""\
+ [fcgi-program:foo]
+ socket=tcp://missingport
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+ """)
+ from supervisor.options import UnhosedConfigParser
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
+ def test_fcgi_program_bad_expansion_proc_num(self):
+ text = lstrip("""\
+ [fcgi-program:foo]
+ socket=unix:///tmp/%(process_num)s.sock
+ process_name = %(program_name)s_%(process_num)s
+ command = /bin/foo
+ numprocs = 2
+ priority = 1
+ """)
+ from supervisor.options import UnhosedConfigParser
+ config = UnhosedConfigParser()
+ config.read_string(text)
+ instance = self._makeOne()
+
self.assertRaises(ValueError,instance.process_groups_from_parser,config)
+
def test_heterogeneous_process_groups_from_parser(self):
text = lstrip("""\
[program:one]
@@ -840,6 +955,58 @@
self.assertEqual(pipes['stdout'], 5)
self.assertEqual(pipes['stderr'], None)
+class FastCGIProcessConfigTest(unittest.TestCase):
+ def _getTargetClass(self):
+ from supervisor.options import FastCGIProcessConfig
+ return FastCGIProcessConfig
+
+ def _makeOne(self, *arg, **kw):
+ defaults = {}
+ for name in ('name', 'command', 'directory', 'umask',
+ 'priority', 'autostart', 'autorestart',
+ 'startsecs', 'startretries', 'uid',
+ 'stdout_logfile', 'stdout_capture_maxbytes',
+ 'stdout_logfile_backups', 'stdout_logfile_maxbytes',
+ 'stderr_logfile', 'stderr_capture_maxbytes',
+ 'stderr_logfile_backups', 'stderr_logfile_maxbytes',
+ 'stopsignal', 'stopwaitsecs', 'exitcodes',
+ 'redirect_stderr', 'environment'):
+ defaults[name] = name
+ defaults.update(kw)
+ return self._getTargetClass()(*arg, **defaults)
+
+ def test_make_process(self):
+ options = DummyOptions()
+ instance = self._makeOne(options)
+ self.assertRaises(NotImplementedError, instance.make_process)
+
+ def test_make_process_with_group(self):
+ options = DummyOptions()
+ instance = self._makeOne(options)
+ process = instance.make_process('abc')
+ from supervisor.process import FastCGISubprocess
+ self.assertEqual(process.__class__, FastCGISubprocess)
+ self.assertEqual(process.group, 'abc')
+
+ def test_make_dispatchers(self):
+ options = DummyOptions()
+ instance = self._makeOne(options)
+ instance.redirect_stderr = False
+ process1 = DummyProcess(instance)
+ dispatchers, pipes = instance.make_dispatchers(process1)
+ self.assertEqual(dispatchers[4].channel, 'stdin')
+ self.assertEqual(dispatchers[4].closed, True)
+ self.assertEqual(dispatchers[5].channel, 'stdout')
+ from supervisor.events import ProcessCommunicationStdoutEvent
+ self.assertEqual(dispatchers[5].event_type,
+ ProcessCommunicationStdoutEvent)
+ self.assertEqual(pipes['stdout'], 5)
+ self.assertEqual(dispatchers[7].channel, 'stderr')
+ from supervisor.events import ProcessCommunicationStderrEvent
+ self.assertEqual(dispatchers[7].event_type,
+ ProcessCommunicationStderrEvent)
+ self.assertEqual(pipes['stderr'], 7)
+
class ProcessGroupConfigTests(unittest.TestCase):
def _getTargetClass(self):
from supervisor.options import ProcessGroupConfig
@@ -871,6 +1038,33 @@
from supervisor.process import ProcessGroup
self.assertEqual(group.__class__, ProcessGroup)
+class FastCGIGroupConfigTests(unittest.TestCase):
+ def _getTargetClass(self):
+ from supervisor.options import FastCGIGroupConfig
+ return FastCGIGroupConfig
+
+ def _makeOne(self, *args, **kw):
+ return self._getTargetClass()(*args, **kw)
+
+ def test_ctor(self):
+ options = DummyOptions()
+ sock_manager = DummySocketManager(6)
+ instance = self._makeOne(options, 'whatever', 999, [], sock_manager)
+ self.assertEqual(instance.options, options)
+ self.assertEqual(instance.name, 'whatever')
+ self.assertEqual(instance.priority, 999)
+ self.assertEqual(instance.process_configs, [])
+ self.assertEqual(instance.socket_manager, sock_manager)
+
+ def test_after_setuid(self):
+ options = DummyOptions()
+ sock_manager = DummySocketManager(6)
+ pconfigs = [DummyPConfig(options, 'process1', '/bin/process1')]
+ instance = self._makeOne(options, 'whatever', 999, pconfigs,
sock_manager)
+ instance.after_setuid()
+ self.assertTrue(pconfigs[0].autochildlogs_created)
+ self.assertTrue(instance.socket_manager.prepare_socket_called)
+
class UtilFunctionsTests(unittest.TestCase):
def test_make_namespec(self):
from supervisor.options import make_namespec
Index: src/supervisor/tests/test_datatypes.py
===================================================================
--- src/supervisor/tests/test_datatypes.py (revision 0)
+++ src/supervisor/tests/test_datatypes.py (revision 0)
@@ -0,0 +1,84 @@
+"""Test suite for supervisor.datatypes"""
+
+import sys
+import os
+import unittest
+import socket
+import tempfile
+
+from supervisor.tests.base import DummySocket
+from supervisor.tests.base import DummySocketConfig
+from supervisor.datatypes import UnixStreamSocketConfig
+from supervisor.datatypes import InetStreamSocketConfig
+
+class InetStreamSocketConfigTests(unittest.TestCase):
+ def _getTargetClass(self):
+ return InetStreamSocketConfig
+
+ def _makeOne(self, *args, **kw):
+ return self._getTargetClass()(*args, **kw)
+
+ def test_url(self):
+ conf = self._makeOne('127.0.0.1', 8675)
+ self.assertEqual(conf.url, 'tcp://127.0.0.1:8675')
+
+ def test_repr(self):
+ conf = self._makeOne('127.0.0.1', 8675)
+ s = repr(conf)
+ self.assertTrue(s.startswith(
+ '<supervisor.datatypes.InetStreamSocketConfig at'), s)
+ self.assertTrue(s.endswith('for tcp://127.0.0.1:8675>'), s)
+
+ def test_addr(self):
+ conf = self._makeOne('127.0.0.1', 8675)
+ addr = conf.addr()
+ self.assertEqual(addr, ('127.0.0.1', 8675))
+
+ def test_port_as_string(self):
+ conf = self._makeOne('localhost', '5001')
+ addr = conf.addr()
+ self.assertEqual(addr, ('localhost', 5001))
+
+ def test_create(self):
+ conf = self._makeOne('127.0.0.1', 8675)
+ sock = conf.create()
+ reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
+ self.assertTrue(reuse)
+ sock.close
+
+class UnixStreamSocketConfigTests(unittest.TestCase):
+ def _getTargetClass(self):
+ return UnixStreamSocketConfig
+
+ def _makeOne(self, *args, **kw):
+ return self._getTargetClass()(*args, **kw)
+
+ def test_url(self):
+ conf = self._makeOne('/tmp/foo.sock')
+ self.assertEqual(conf.url, 'unix:///tmp/foo.sock')
+
+ def test_repr(self):
+ conf = self._makeOne('/tmp/foo.sock')
+ s = repr(conf)
+ self.assertTrue(s.startswith(
+ '<supervisor.datatypes.UnixStreamSocketConfig at'), s)
+ self.assertTrue(s.endswith('for unix:///tmp/foo.sock>'), s)
+
+ def test_get_addr(self):
+ conf = self._makeOne('/tmp/foo.sock')
+ addr = conf.addr()
+ self.assertEqual(addr, '/tmp/foo.sock')
+
+ def test_create(self):
+ (tf_fd, tf_name) = tempfile.mkstemp()
+ conf = self._makeOne(tf_name)
+ os.close(tf_fd)
+ sock = conf.create()
+ self.assertFalse(os.path.exists(tf_name))
+ sock.close
+
+def test_suite():
+ return unittest.findTestCases(sys.modules[__name__])
+
+if __name__ == '__main__':
+ unittest.main(defaultTest='test_suite')
\ No newline at end of file
Index: src/supervisor/tests/base.py
===================================================================
--- src/supervisor/tests/base.py (revision 758)
+++ src/supervisor/tests/base.py (working copy)
@@ -821,6 +821,11 @@
def make_group(self):
return DummyProcessGroup(self)
+class DummyFCGIGroupConfig(DummyPGroupConfig):
+ def __init__(self, options, name, priority, pconfigs, socket_manager):
+ DummyPGroupConfig.__init__(self, options, name, priority, pconfigs)
+ self.socket_manager = socket_manager
+
class DummyProcessGroup:
def __init__(self, config):
self.config = config
@@ -934,6 +939,54 @@
def __str__(self):
return 'dummy event'
+class DummySocket:
+ bind_called = False
+ bind_addr = None
+ listen_called = False
+ listen_backlog = None
+ close_called = False
+
+ def __init__(self, fd):
+ self.fd = fd
+
+ def fileno(self):
+ return self.fd
+
+ def bind(self, addr):
+ self.bind_called = True
+ self.bind_addr = addr
+
+ def listen(self, backlog):
+ self.listen_called = True
+ self.listen_backlog = backlog
+
+ def close(self):
+ self.close_called = True
+
+ def __str__(self):
+ return 'dummy socket'
+
+class DummySocketConfig:
+ def __init__(self, fd):
+ self.fd = fd
+
+ def addr(self):
+ return 'dummy addr'
+
+ def create(self):
+ return DummySocket(self.fd)
+
+class DummySocketManager:
+ def __init__(self, sock_fd):
+ self.sock_fd = sock_fd
+ self.prepare_socket_called = False
+
+ def prepare_socket(self):
+ self.prepare_socket_called = True
+
+ def get_socket(self):
+ return DummySocket(self.sock_fd)
+
def dummy_handler(event, result):
pass
@@ -947,4 +1000,3 @@
def lstrip(s):
strings = [x.strip() for x in s.split('\n')]
return '\n'.join(strings)
-
Index: src/supervisor/tests/test_process.py
===================================================================
--- src/supervisor/tests/test_process.py (revision 758)
+++ src/supervisor/tests/test_process.py (working copy)
@@ -11,6 +11,9 @@
from supervisor.tests.base import DummyPGroupConfig
from supervisor.tests.base import DummyDispatcher
from supervisor.tests.base import DummyEvent
+from supervisor.tests.base import DummyFCGIGroupConfig
+from supervisor.tests.base import DummySocketManager
+from supervisor.tests.base import DummyProcessGroup
class SubprocessTests(unittest.TestCase):
def _getTargetClass(self):
@@ -1150,6 +1153,66 @@
self.assertEqual(instance.backoff, 1)
self.failUnless(instance.delay > 0)
+class FastCGISubprocessTests(unittest.TestCase):
+ def _getTargetClass(self):
+ from supervisor.process import FastCGISubprocess
+ return FastCGISubprocess
+
+ def _makeOne(self, *arg, **kw):
+ return self._getTargetClass()(*arg, **kw)
+
+ def tearDown(self):
+ from supervisor.events import clear
+ clear()
+
+ def test_no_group(self):
+ options = DummyOptions()
+ options.forkpid = 0
+ config = DummyPConfig(options, 'good', '/good/filename', uid=1)
+ instance = self._makeOne(config)
+ self.assertRaises(NotImplementedError, instance.spawn)
+
+ def test_no_socket_manager(self):
+ options = DummyOptions()
+ options.forkpid = 0
+ config = DummyPConfig(options, 'good', '/good/filename', uid=1)
+ instance = self._makeOne(config)
+ instance.group = DummyProcessGroup(DummyPGroupConfig(options))
+ self.assertRaises(NotImplementedError, instance.spawn)
+
+ def test_prepare_child_fds(self):
+ options = DummyOptions()
+ options.forkpid = 0
+ config = DummyPConfig(options, 'good', '/good/filename', uid=1)
+ instance = self._makeOne(config)
+ sock_manager = DummySocketManager(7)
+ gconfig = DummyFCGIGroupConfig(options, 'whatever', 999, None,
+ sock_manager)
+ instance.group = DummyProcessGroup(gconfig)
+ result = instance.spawn()
+ self.assertEqual(result, None)
+ self.assertEqual(len(options.duped), 3)
+ self.assertEqual(options.duped[7], 0)
+ self.assertEqual(options.duped[instance.pipes['child_stdout']], 1)
+ self.assertEqual(options.duped[instance.pipes['child_stderr']], 2)
+ self.assertEqual(len(options.fds_closed), options.minfds - 3)
+
+ def test_prepare_child_fds_stderr_redirected(self):
+ options = DummyOptions()
+ options.forkpid = 0
+ config = DummyPConfig(options, 'good', '/good/filename', uid=1)
+ config.redirect_stderr = True
+ instance = self._makeOne(config)
+ sock_manager = DummySocketManager(13)
+ gconfig = DummyFCGIGroupConfig(options, 'whatever', 999, None,
+ sock_manager)
+ instance.group = DummyProcessGroup(gconfig)
+ result = instance.spawn()
+ self.assertEqual(result, None)
+ self.assertEqual(len(options.duped), 2)
+ self.assertEqual(options.duped[13], 0)
+ self.assertEqual(len(options.fds_closed), options.minfds - 3)
+
class ProcessGroupBaseTests(unittest.TestCase):
def _getTargetClass(self):
from supervisor.process import ProcessGroupBase
Index: src/supervisor/tests/test_socket_manager.py
===================================================================
--- src/supervisor/tests/test_socket_manager.py (revision 0)
+++ src/supervisor/tests/test_socket_manager.py (revision 0)
@@ -0,0 +1,102 @@
+"""Test suite for supervisor.socket_manager"""
+
+import sys
+import os
+import unittest
+import socket
+import tempfile
+
+from supervisor.tests.base import DummySocket
+from supervisor.tests.base import DummySocketConfig
+from supervisor.datatypes import UnixStreamSocketConfig
+from supervisor.datatypes import InetStreamSocketConfig
+
+class SocketManagerTest(unittest.TestCase):
+ def _getTargetClass(self):
+ from supervisor.socket_manager import SocketManager
+ return SocketManager
+
+ def _makeOne(self, *args, **kw):
+ return self._getTargetClass()(*args, **kw)
+
+ def test_get_config(self):
+ conf = DummySocketConfig(2)
+ sock_manager = self._makeOne(conf)
+ self.assertEqual(conf, sock_manager.config())
+
+ def test_tcp_w_hostname(self):
+ conf = InetStreamSocketConfig('localhost', 12345)
+ sock_manager = self._makeOne(conf)
+ self.assertEqual(sock_manager.socket_config, conf)
+ sock = sock_manager.get_socket()
+ self.assertEqual(sock.getsockname(), ('127.0.0.1', 12345))
+ sock_manager.close()
+
+ def test_tcp_w_ip(self):
+ conf = InetStreamSocketConfig('127.0.0.1', 12345)
+ sock_manager = self._makeOne(conf)
+ self.assertEqual(sock_manager.socket_config, conf)
+ sock = sock_manager.get_socket()
+ self.assertEqual(sock.getsockname(), ('127.0.0.1', 12345))
+ sock_manager.close()
+
+ def test_unix(self):
+ (tf_fd, tf_name) = tempfile.mkstemp();
+ conf = UnixStreamSocketConfig(tf_name)
+ sock_manager = self._makeOne(conf)
+ self.assertEqual(sock_manager.socket_config, conf)
+ sock = sock_manager.get_socket()
+ self.assertEqual(sock.getsockname(), tf_name)
+ sock_manager.close()
+ os.close(tf_fd)
+
+ def test_get_socket(self):
+ conf = DummySocketConfig(2)
+ sock_manager = self._makeOne(conf)
+ sock = sock_manager.get_socket()
+ sock2 = sock_manager.get_socket()
+ self.assertEqual(sock, sock2)
+ sock_manager.close()
+ sock3 = sock_manager.get_socket()
+ self.assertNotEqual(sock, sock3)
+
+ def test_prepare_socket(self):
+ conf = DummySocketConfig(1)
+ sock_manager = self._makeOne(conf)
+ sock = sock_manager.get_socket()
+ self.assertTrue(sock_manager.prepared)
+ self.assertTrue(sock.bind_called)
+ self.assertEqual(sock.bind_addr, 'dummy addr')
+ self.assertTrue(sock.listen_called)
+ self.assertEqual(sock.listen_backlog, socket.SOMAXCONN)
+ self.assertFalse(sock.close_called)
+
+ def test_close(self):
+ conf = DummySocketConfig(6)
+ sock_manager = self._makeOne(conf)
+ sock = sock_manager.get_socket()
+ self.assertFalse(sock.close_called)
+ self.assertTrue(sock_manager.prepared)
+ sock_manager.close()
+ self.assertFalse(sock_manager.prepared)
+ self.assertTrue(sock.close_called)
+
+ def test_tcp_socket_already_taken(self):
+ conf = InetStreamSocketConfig('127.0.0.1', 12345)
+ sock_manager = self._makeOne(conf)
+ sock_manager.get_socket()
+ sock_manager2 = self._makeOne(conf)
+ self.assertRaises(socket.error, sock_manager2.prepare_socket)
+ sock_manager.close()
+
+ def test_unix_bad_sock(self):
+ conf = UnixStreamSocketConfig('/notthere/foo.sock')
+ sock_manager = self._makeOne(conf)
+ self.assertRaises(socket.error, sock_manager.get_socket)
+ sock_manager.close()
+
+def test_suite():
+ return unittest.findTestCases(sys.modules[__name__])
+
+if __name__ == '__main__':
+ unittest.main(defaultTest='test_suite')
\ No newline at end of file
Index: src/supervisor/process.py
===================================================================
--- src/supervisor/process.py (revision 758)
+++ src/supervisor/process.py (working copy)
@@ -268,6 +268,17 @@
options.pidhistory[pid] = self
return pid
+ def _prepare_child_fds(self):
+ options = self.config.options
+ options.dup2(self.pipes['child_stdin'], 0)
+ options.dup2(self.pipes['child_stdout'], 1)
+ if self.config.redirect_stderr:
+ options.dup2(self.pipes['child_stdout'], 2)
+ else:
+ options.dup2(self.pipes['child_stderr'], 2)
+ for i in range(3, options.minfds):
+ options.close_fd(i)
+
def _spawn_as_child(self, filename, argv):
options = self.config.options
try:
@@ -280,14 +291,7 @@
# Presumably it also prevents HUP, etc received by
# supervisord from being sent to children.
options.setpgrp()
- options.dup2(self.pipes['child_stdin'], 0)
- options.dup2(self.pipes['child_stdout'], 1)
- if self.config.redirect_stderr:
- options.dup2(self.pipes['child_stdout'], 2)
- else:
- options.dup2(self.pipes['child_stderr'], 2)
- for i in range(3, options.minfds):
- options.close_fd(i)
+ self._prepare_child_fds()
# sending to fd 2 will put this output in the stderr log
msg = self.set_uid()
if msg:
@@ -549,6 +553,29 @@
self.pid))
self.kill(signal.SIGKILL)
+class FastCGISubprocess(Subprocess):
+ """Extends Subprocess class to handle FastCGI subprocesses"""
+
+ def _prepare_child_fds(self):
+ if self.group is None:
+ raise NotImplementedError('No group set for FastCGISubprocess')
+ if not hasattr(self.group, 'config'):
+ raise NotImplementedError('No config found for group on
FastCGISubprocess')
+ if not hasattr(self.group.config, 'socket_manager'):
+ raise NotImplementedError('No SocketManager set for
FastCGISubprocess group')
+ sock = self.group.config.socket_manager.get_socket()
+ sock_fd = sock.fileno()
+
+ options = self.config.options
+ options.dup2(sock_fd, 0)
+ options.dup2(self.pipes['child_stdout'], 1)
+ if self.config.redirect_stderr:
+ options.dup2(self.pipes['child_stdout'], 2)
+ else:
+ options.dup2(self.pipes['child_stderr'], 2)
+ for i in range(3, options.minfds):
+ options.close_fd(i)
+
class ProcessGroupBase:
def __init__(self, config):
self.config = config
Index: src/supervisor/socket_manager.py
===================================================================
--- src/supervisor/socket_manager.py (revision 0)
+++ src/supervisor/socket_manager.py (revision 0)
@@ -0,0 +1,53 @@
+##############################################################################
+#
+# Copyright (c) 2007 Agendaless Consulting and Contributors.
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the BSD-like license at
+# http://www.repoze.org/LICENSE.txt. A copy of the license should accompany
+# this distribution. THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL
+# EXPRESS OR IMPLIED WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO,
+# THE IMPLIED WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND
+# FITNESS FOR A PARTICULAR PURPOSE
+#
+##############################################################################
+
+import socket
+import os
+
+from supervisor.datatypes import SocketAddress
+from supervisor.datatypes import port_number
+
+class SocketManager:
+ """ Class for managing sockets in servers that create/bind/listen
+ before forking multiple child processes to accept() """
+
+ socket_config = None #SocketConfig object
+ socket = None #Socket being managed
+ prepared = False
+
+ def __init__(self, socket_config):
+ self.socket_config = socket_config
+
+ def __repr__(self):
+ return '<%s at %s for %s>' % (self.__class__,
+ id(self),
+ self.socket_config.url)
+
+ def config(self):
+ return self.socket_config
+
+ def prepare_socket(self):
+ self.socket = self.socket_config.create()
+ self.socket.bind(self.socket_config.addr())
+ self.socket.listen(socket.SOMAXCONN)
+ self.prepared = True
+
+ def get_socket(self):
+ if not self.prepared:
+ self.prepare_socket()
+ return self.socket
+
+ def close(self):
+ self.socket.close()
+ self.prepared = False
_______________________________________________
Supervisor-users mailing list
Supervisor-users@lists.supervisord.org
http://lists.supervisord.org/mailman/listinfo/supervisor-users
