unix sockets/ lighttpd

#!/usr/bin/env python
"""
A SCGI handler that uses Quixote to publish dynamic content.
"""

import sys
import time
import os
import getopt
import signal
from quixote import enable_ptl, publish
import scgi_server

pidfilename = None # set by main()

def debug(msg):
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S",
                              time.localtime(time.time()))
    sys.stderr.write("[%s] %s\n" % (timestamp, msg))


class QuixoteHandler(scgi_server.SCGIHandler):

    # override in subclass
    publisher_class = publish.Publisher
    root_namespace = None
    prefix = ""

    def __init__(self, *args, **kwargs):
        debug("%s created" % self.__class__.__name__)
        scgi_server.SCGIHandler.__init__(self, *args, **kwargs)
        assert self.root_namespace, "You must provide a namespace to publish"
        self.publisher = self.publisher_class(self.root_namespace)

    def handle_connection (self, conn):
        input = conn.makefile("r")
        output = conn.makefile("w")

        env = self.read_env(input)

        # mod_scgi never passes PATH_INFO, fake it
        prefix = self.prefix
        path = env['SCRIPT_NAME']
        assert path[:len(prefix)] == prefix, (
                "path %r doesn't start with prefix %r" % (path, prefix))
        env['SCRIPT_NAME'] = prefix
        env['PATH_INFO'] = path[len(prefix):] + env.get('PATH_INFO', '')

        self.publisher.publish(input, output, sys.stderr, env)

        try:
            input.close()
            output.close()
            conn.close()
        except IOError, err:
            debug("IOError while closing connection ignored: %s" % err)

        if self.publisher.config.run_once:
            sys.exit(0)


class DemoHandler(QuixoteHandler):

    root_namespace = "quixote.demo"
    prefix = "/dynamic" # must match Location directive

    def __init__(self, *args, **kwargs):
        enable_ptl()
        QuixoteHandler.__init__(self, *args, **kwargs)


def change_uid_gid(uid, gid=None):
    "Try to change UID and GID to the provided values"
    # This will only work if this script is run by root.

    # Try to convert uid and gid to integers, in case they're numeric
    import pwd, grp
    try:
        uid = int(uid)
        default_grp = pwd.getpwuid(uid)[3]
    except ValueError:
        uid, default_grp = pwd.getpwnam(uid)[2:4]

    if gid is None:
        gid = default_grp
    else:
        try:
            gid = int(gid)
        except ValueError:
            gid = grp.getgrnam(gid)[2]

    os.setgid(gid)
    os.setuid(uid)


def term_signal(signum, frame):
    global pidfilename
    try:
        os.unlink(pidfilename)
    except OSError:
        pass
    sys.exit()

def main(handler=DemoHandler):
    usage = """Usage: %s [options]

    -F -- stay in foreground (don't fork)
    -P -- PID filename
    -l -- log filename
    -m -- max children
    -p -- TCP port to listen on
    -s -- Unix Socket to listen on
    -u -- user id to run under
    """ % sys.argv[0]
    nofork = 0
    global pidfilename
    pidfilename = "/var/tmp/quixote-scgi.pid"
    logfilename = "/var/tmp/quixote-scgi.log"
    max_children = 5    # scgi default
    uid = "nobody"
    port = 4000
    host = "127.0.0.1"
    sockfilename = ""
    unix_socket = False
    
    try:
        opts, args = getopt.getopt(sys.argv[1:], 'FP:l:m:p:u:s:')
    except getopt.GetoptError, exc:
        print >>sys.stderr, exc
        print >>sys.stderr, usage
        sys.exit(1)
    for o, v in opts:
        if o == "-F":
            nofork = 1
        elif o == "-P":
            pidfilename = v
        elif o == "-l":
            logfilename = v
        elif o == "-m":
            max_children = int(v)
        elif o == "-p":
            port = int(v)
        elif o == "-u":
            uid = v
        elif o == "-s":
            sockfilename = str(v)
            unix_socket = True

    log = open(logfilename, "a", 1)
    os.dup2(log.fileno(), 1)
    os.dup2(log.fileno(), 2)
    os.close(0)

    if os.getuid() == 0:
        change_uid_gid(uid)
    
    if nofork:
        scgi_server.SCGIServer(handler, host=host, port=port,
                               max_children=max_children,
                               socketfile=sockfilename,
                               unix_socket=unix_socket).serve()
    else:
        pid = os.fork()
        if pid == 0:
            pid = os.getpid()
            pidfile = open(pidfilename, 'w')
            pidfile.write(str(pid))
            pidfile.close()
            signal.signal(signal.SIGTERM, term_signal)
            try:
                scgi_server.SCGIServer(handler, host=host, port=port,
                                       max_children=max_children,
                                       socketfile=sockfilename,
                                       unix_socket=unix_socket).serve()
            finally:
                # grandchildren get here too, don't let them unlink the pid
                if pid == os.getpid():
                    try:
                        os.unlink(pidfilename)
                    except OSError:
                        pass

if __name__ == '__main__':
    main()

#!/usr/bin/env python
"""
A pre-forking SCGI server that uses file descriptor passing to off-load
requests to child worker processes.
"""

import sys
import socket
import os
import select
import errno
import fcntl
import signal
from scgi import passfd

# netstring utility functions
def ns_read_size(input):
    size = ""
    while 1:
        c = input.read(1)
        if c == ':':
            break
        elif not c:
            raise IOError, 'short netstring read'
        size = size + c
    return long(size)

def ns_reads(input):
    size = ns_read_size(input)
    data = ""
    while size > 0:
        s = input.read(size)
        if not s:
            raise IOError, 'short netstring read'
        data = data + s
        size -= len(s)
    if input.read(1) != ',':
        raise IOError, 'missing netstring terminator'
    return data

def read_env(input):
    headers = ns_reads(input)
    items = headers.split("\0")
    items = items[:-1]
    assert len(items) % 2 == 0, "malformed headers"
    env = {}
    for i in range(0, len(items), 2):
        env[items[i]] = items[i+1]
    return env

class SCGIHandler:

    # Subclasses should override the handle_connection method.

    def __init__(self, parent_fd):
        self.parent_fd = parent_fd

    def serve(self):
        while 1:
            try:
                os.write(self.parent_fd, "1") # indicates that child is ready
                fd = passfd.recvfd(self.parent_fd)
            except (IOError, OSError):
                # parent probably exited  (EPIPE comes thru as OSError)
                raise SystemExit
            conn = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
            # Make sure the socket is blocking.  Apparently, on FreeBSD the
            # socket is non-blocking.  I think that's an OS bug but I don't
            # have the resources to track it down.
            conn.setblocking(1)
            os.close(fd)
            self.handle_connection(conn)


    def read_env(self, input):
        return read_env(input)

    def handle_connection(self, conn):
        input = conn.makefile("r")
        output = conn.makefile("w")

        env = self.read_env(input)
        output.write("Content-Type: text/plain\r\n")
        output.write("\r\n")
        for k, v in env.items():
            output.write("%s: %r\n" % (k, v))

        output.close()
        input.close()
        conn.close()


class SCGIServer:

    DEFAULT_PORT = 4000
    DEFAULT_SOCKET_FILE = "/tmp/scgi.socket"
    
    def __init__(self, handler_class=SCGIHandler, host="", port=DEFAULT_PORT,
                 max_children=5,socketfile=DEFAULT_SOCKET_FILE,unix_socket=False):
        self.handler_class = handler_class
        self.host = host
        self.port = port
        self.max_children = max_children
        self.children = {} # { pid : fd }
        self.spawn_child()
        self.restart = 0
        self.unix_socket = unix_socket
        self.unix_socket_filename = socketfile
        
    #
    # Deal with a hangup signal.  All we can really do here is
    # note that it happened.
    #
    def hup_signal(self, signum, frame):
        self.restart = 1

    def spawn_child(self, conn=None):
        parent_fd, child_fd = passfd.socketpair(socket.AF_UNIX,
                                                socket.SOCK_STREAM)
        # make child fd non-blocking
        flags = fcntl.fcntl(child_fd, fcntl.F_GETFL, 0)
        fcntl.fcntl(child_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
        pid = os.fork()
        if pid == 0:
            if conn:
                conn.close() # in the midst of handling a request, close
                             # the connection in the child 
            os.close(child_fd)
            self.handler_class(parent_fd).serve()
            sys.exit(0)
        else:
            os.close(parent_fd)
            self.children[pid] = child_fd

    def reap_children(self):
        while self.children:
            (pid, status) = os.waitpid(-1, os.WNOHANG)
            if pid <= 0:
                break
            os.close(self.children[pid])
            del self.children[pid]

    def do_restart(self):
        #
        # First close connections to the children, which will cause them
        # to exit after finishing what they are doing.
        #
        for fd in self.children.values():
            os.close(fd)
        #
        # Then do a blocking wait on each until we have cleared the
        # slate.
        #
        for pid in self.children.keys():
            (pid, status) = os.waitpid(pid, 0)
        self.children = {}
        #
        # Fire off a new child, we'll be wanting it soon.
        #
        self.spawn_child()
        self.restart = 0
        

    def delegate_request(self, conn):
        """Pass a request fd to a child process to handle.  This method
        blocks if all the children are busy and we have reached the
        max_children limit."""

        # There lots of subtleties here.  First, we can't use the write
        # status of the pipes to the child since select will return true
        # if the buffer is not filled.  Instead, each child writes one
        # byte of data when it is ready for a request.  The normal case
        # is that a child is ready for a request.  We want that case to
        # be fast.  Also, we want to pass requests to the same child if
        # possible.  Finally, we need to gracefully handle children
        # dying at any time.

        # If no children are ready and we haven't reached max_children
        # then we want another child to be started without delay.
        timeout = 0

        while 1:
            try:
                r, w, e = select.select(self.children.values(), [], [], timeout)
            except select.error, e:
                if e[0] == errno.EINTR:  # got a signal, try again
                    continue
                raise
            if r:
                # One or more children look like they are ready.  Sort
                # the file descriptions so that we keep preferring the
                # same child.
                r.sort()
                child_fd = r[0]

                # Try to read the single byte written by the child.
                # This can fail if the child died or the pipe really
                # wasn't ready (select returns a hint only).  The fd has
                # been made non-blocking by spawn_child.  If this fails
                # we fall through to the "reap_children" logic and will
                # retry the select call.
                try:
                    ready_byte = os.read(child_fd, 1)
                    if not ready_byte:
                        raise IOError # child died?
                    assert ready_byte == "1", repr(ready_byte)
                except socket.error, exc:
                    if exc[0]  == errno.EWOULDBLOCK:
                        pass # select was wrong
                    else:
                        raise
                except (OSError, IOError):
                    pass # child died?
                else:
                    # The byte was read okay, now we need to pass the fd
                    # of the request to the child.  This can also fail
                    # if the child died.  Again, if this fails we fall
                    # through to the "reap_children" logic and will
                    # retry the select call.
                    try:
                        passfd.sendfd(child_fd, conn.fileno())
                    except IOError, exc:
                        if exc.errno == errno.EPIPE:
                            pass # broken pipe, child died?
                        else:
                            raise
                    else:
                        # fd was apparently passed okay to the child.
                        # The child could die before completing the
                        # request but that's not our problem anymore.
                        return

            # didn't find any child, check if any died
            self.reap_children()
            
            # start more children if we haven't met max_children limit
            if len(self.children) < self.max_children:
                self.spawn_child(conn)
            
            # Start blocking inside select.  We might have reached
            # max_children limit and they are all busy.
            timeout = 2


    def serve(self):
        #jd
        if self.unix_socket:
            s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind((self.unix_socket_filename))
        else:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind((self.host, self.port))
        s.listen(40)
        signal.signal(signal.SIGHUP, self.hup_signal)
        while 1:
            try:
                conn, addr = s.accept()
                self.delegate_request(conn)
                conn.close()
            except socket.error, e:
                if e[0] != errno.EINTR:
                    raise  # something weird
            if self.restart:
                self.do_restart()


def main():
    if len(sys.argv) == 2:
        unix_socket = False
        try:
            port = int(sys.argv[1])
        except ValueError:
            try:
                port = str(sys.argv[1])
            except:
                pass
            else:
                unix_socket = True
    else:
        port = SCGIServer.DEFAULT_PORT
        
    if unix_socket:    
        SCGIServer(socketfile=port,unix_socket=unix_socket).serve()
    else:
        SCGIServer(port=port,unix_socket=unix_socket).serve()
if __name__ == "__main__":
    main()