magicbus / magicbus / plugins / servers.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
"""
Multiple servers/ports
======================

If you need to start more than one HTTP server (to serve on multiple ports, or
protocols, etc.), you can manually register each one and then start them all
with bus.start::

    s1 = ServerPlugin(bus, MyWSGIServer(host='0.0.0.0', port=80))
    s2 = ServerPlugin(bus, another.HTTPServer(host='127.0.0.1', SSL=True))
    s1.subscribe()
    s2.subscribe()
    bus.start()

.. index:: SCGI

FastCGI/SCGI
============

There are also Flup\ **F**\ CGIServer and Flup\ **S**\ CGIServer classes in
:mod:`magicbus.plugins.servers`. To start an fcgi server, for example,
wrap an instance of it in a ServerPlugin::

    addr = ('0.0.0.0', 4000)
    f = servers.FlupFCGIServer(application=mywsgiapp, bindAddress=addr)
    s = servers.ServerPlugin(bus, httpserver=f, bind_addr=addr)
    s.subscribe()

Note that you need to download and install `flup <http://trac.saddi.com/flup>`_
yourself.

.. _fastcgi:
.. index:: FastCGI

FastCGI
-------

A very simple setup lets your server run with FastCGI.
You just need the flup library,
plus a running Apache server (with ``mod_fastcgi``) or lighttpd server.

Apache
^^^^^^

At the top level in httpd.conf::

    FastCgiIpcDir /tmp
    FastCgiServer /path/to/myapp.fcgi -idle-timeout 120 -processes 4

And inside the relevant VirtualHost section::

    # FastCGI config
    AddHandler fastcgi-script .fcgi
    ScriptAliasMatch (.*$) /path/to/myapp.fcgi$1

Lighttpd
^^^^^^^^

For `Lighttpd <http://www.lighttpd.net/>`_ you can follow these
instructions. Within ``lighttpd.conf`` make sure ``mod_fastcgi`` is
active within ``server.modules``. Then, within your ``$HTTP["host"]``
directive, configure your fastcgi script like the following::

    $HTTP["url"] =~ "" {
      fastcgi.server = (
        "/" => (
          "script.fcgi" => (
            "bin-path" => "/path/to/your/script.fcgi",
            "socket"          => "/tmp/script.sock",
            "check-local"     => "disable",
            "disable-time"    => 1,
            "min-procs"       => 1,
            "max-procs"       => 1, # adjust as needed
          ),
        ),
      )
    } # end of $HTTP["url"] =~ "^/"

Please see `Lighttpd FastCGI Docs
<http://redmine.lighttpd.net/wiki/lighttpd/Docs:ModFastCGI>`_ for an explanation 
of the possible configuration options.
"""

import socket
import sys
import threading
import time


class ServerPlugin(object):
    """Bus plugin for an HTTP server.
    
    You don't have to use this plugin; you can make your own that listens on
    the appropriate bus channels. This one is designed to:

        * wrap HTTP servers whose accept loop blocks by running it in a
          separate thread; any exceptions in it exit the bus
        * wait until the server is truly ready to receive requests before
          returning from the bus.start listener
        * wait until the server has finished processing requestss before
          returning from the bus.stop listener
        * log server start/stop via the bus

    The httpserver argument MUST possess 'start' and 'stop' methods,
    and a 'ready' boolean attribute which is True when the HTTP server
    is ready to receive requests on its socket.
    
    If you need to start more than one HTTP server (to serve on multiple
    ports, or protocols, etc.), you can manually register each one and then
    start them all with bus.start:
    
        s1 = ServerPlugin(bus, MyWSGIServer(host='0.0.0.0', port=80))
        s2 = ServerPlugin(bus, another.HTTPServer(host='127.0.0.1', SSL=True))
        s1.subscribe()
        s2.subscribe()
        bus.start()
    """
    
    def __init__(self, bus, httpserver=None, bind_addr=None):
        self.bus = bus
        self.httpserver = httpserver
        self.bind_addr = bind_addr
        self.interrupt = None
        self.running = False
    
    def subscribe(self):
        self.bus.subscribe('start', self.start)
        self.bus.subscribe('stop', self.stop)
    
    def unsubscribe(self):
        self.bus.unsubscribe('start', self.start)
        self.bus.unsubscribe('stop', self.stop)
    
    def start(self):
        """Start the HTTP server."""
        if self.bind_addr is None:
            on_what = "unknown interface (dynamic?)"
        elif isinstance(self.bind_addr, tuple):
            host, port = self.bind_addr
            on_what = "%s:%s" % (host, port)
        else:
            on_what = "socket file: %s" % self.bind_addr
        
        if self.running:
            self.bus.log("Already serving on %s" % on_what)
            return
        
        self.interrupt = None
        if not self.httpserver:
            raise ValueError("No HTTP server has been created.")
        
        # Start the httpserver in a new thread.
        if isinstance(self.bind_addr, tuple):
            wait_for_free_port(*self.bind_addr)
        
        t = threading.Thread(target=self._start_http_thread)
        t.setName("HTTPServer " + t.getName())
        t.start()
        
        self.wait()
        self.running = True
        self.bus.log("Serving on %s" % on_what)
    start.priority = 75
    
    def _start_http_thread(self):
        """HTTP servers MUST be running in new threads, so that the
        main thread persists to receive KeyboardInterrupt's. If an
        exception is raised in the httpserver's thread then it's
        trapped here, and the bus (and therefore our httpserver)
        are shut down.
        """
        try:
            self.httpserver.start()
        except KeyboardInterrupt:
            self.bus.log("<Ctrl-C> hit: shutting down HTTP server")
            self.interrupt = sys.exc_info()[1]
            self.bus.exit()
        except SystemExit:
            self.bus.log("SystemExit raised: shutting down HTTP server")
            self.interrupt = sys.exc_info()[1]
            self.bus.exit()
            raise
        except:
            self.interrupt = sys.exc_info()[1]
            self.bus.log("Error in HTTP server: shutting down",
                         traceback=True, level=40)
            self.bus.exit()
            raise
    
    def wait(self):
        """Wait until the HTTP server is ready to receive requests."""
        while not getattr(self.httpserver, "ready", False):
            if self.interrupt:
                raise self.interrupt
            time.sleep(.1)
        
        # Wait for port to be occupied
        if isinstance(self.bind_addr, tuple):
            host, port = self.bind_addr
            wait_for_occupied_port(host, port)
    
    def stop(self):
        """Stop the HTTP server."""
        if self.running:
            # stop() MUST block until the server is *truly* stopped.
            self.httpserver.stop()
            # Wait for the socket to be truly freed.
            if isinstance(self.bind_addr, tuple):
                wait_for_free_port(*self.bind_addr)
            self.running = False
            self.bus.log("HTTP Server %s shut down" % self.httpserver)
        else:
            self.bus.log("HTTP Server %s already shut down" % self.httpserver)
    stop.priority = 25
    
    def restart(self):
        """Restart the HTTP server."""
        self.stop()
        self.start()



# ------- Wrappers for various HTTP servers for use with ServerPlugin ------- #


class FlupCGIServer(object):
    """Adapter for a flup.server.cgi.WSGIServer."""
   
    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
        self.ready = False
    
    def start(self):
        """Start the CGI server."""
        # We have to instantiate the server class here because its __init__
        # starts a threadpool. If we do it too early, daemonize won't work.
        from flup.server.cgi import WSGIServer
       
        self.cgiserver = WSGIServer(*self.args, **self.kwargs)
        self.ready = True
        self.cgiserver.run()
    
    def stop(self):
        """Stop the HTTP server."""
        self.ready = False


class FlupFCGIServer(object):
    """Adapter for a flup.server.fcgi.WSGIServer."""
    
    def __init__(self, *args, **kwargs):
        if kwargs.get('bindAddress', None) is None:
            if not hasattr(socket, 'fromfd'):
                raise ValueError(
                    'Dynamic FCGI server not available on this platform. '
                    'You must use a static or external one by providing a '
                    'legal bindAddress.')
        self.args = args
        self.kwargs = kwargs
        self.ready = False
    
    def start(self):
        """Start the FCGI server."""
        # We have to instantiate the server class here because its __init__
        # starts a threadpool. If we do it too early, daemonize won't work.
        from flup.server.fcgi import WSGIServer
        self.fcgiserver = WSGIServer(*self.args, **self.kwargs)
        # TODO: report this bug upstream to flup.
        # If we don't set _oldSIGs on Windows, we get:
        #   File "C:\Python24\Lib\site-packages\flup\server\threadedserver.py",
        #   line 108, in run
        #     self._restoreSignalHandlers()
        #   File "C:\Python24\Lib\site-packages\flup\server\threadedserver.py",
        #   line 156, in _restoreSignalHandlers
        #     for signum,handler in self._oldSIGs:
        #   AttributeError: 'WSGIServer' object has no attribute '_oldSIGs'
        self.fcgiserver._installSignalHandlers = lambda: None
        self.fcgiserver._oldSIGs = []
        self.ready = True
        self.fcgiserver.run()
    
    def stop(self):
        """Stop the HTTP server."""
        # Forcibly stop the fcgi server main event loop.
        self.fcgiserver._keepGoing = False
        # Force all worker threads to die off.
        self.fcgiserver._threadPool.maxSpare = self.fcgiserver._threadPool._idleCount
        self.ready = False


class FlupSCGIServer(object):
    """Adapter for a flup.server.scgi.WSGIServer."""
    
    def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
        self.ready = False
    
    def start(self):
        """Start the SCGI server."""
        # We have to instantiate the server class here because its __init__
        # starts a threadpool. If we do it too early, daemonize won't work.
        from flup.server.scgi import WSGIServer
        self.scgiserver = WSGIServer(*self.args, **self.kwargs)
        # TODO: report this bug upstream to flup.
        # If we don't set _oldSIGs on Windows, we get:
        #   File "C:\Python24\Lib\site-packages\flup\server\threadedserver.py",
        #   line 108, in run
        #     self._restoreSignalHandlers()
        #   File "C:\Python24\Lib\site-packages\flup\server\threadedserver.py",
        #   line 156, in _restoreSignalHandlers
        #     for signum,handler in self._oldSIGs:
        #   AttributeError: 'WSGIServer' object has no attribute '_oldSIGs'
        self.scgiserver._installSignalHandlers = lambda: None
        self.scgiserver._oldSIGs = []
        self.ready = True
        self.scgiserver.run()
    
    def stop(self):
        """Stop the HTTP server."""
        self.ready = False
        # Forcibly stop the scgi server main event loop.
        self.scgiserver._keepGoing = False
        # Force all worker threads to die off.
        self.scgiserver._threadPool.maxSpare = 0



# ---------------------------- Utility functions ---------------------------- #


def client_host(server_host):
    """Return the host on which a client can connect to the given listener."""
    if server_host == '0.0.0.0':
        # 0.0.0.0 is INADDR_ANY, which should answer on localhost.
        return '127.0.0.1'
    if server_host in ('::', '::0', '::0.0.0.0'):
        # :: is IN6ADDR_ANY, which should answer on localhost.
        # ::0 and ::0.0.0.0 are non-canonical but common ways to write IN6ADDR_ANY.
        return '::1'
    return server_host

def check_port(host, port, timeout=1.0):
    """Raise an error if the given port is not free on the given host."""
    if not host:
        raise ValueError("Host values of '' or None are not allowed.")
    host = client_host(host)
    port = int(port)
    
    # AF_INET or AF_INET6 socket
    # Get the correct address family for our host (allows IPv6 addresses)
    try:
        info = socket.getaddrinfo(host, port, socket.AF_UNSPEC,
                                  socket.SOCK_STREAM)
    except socket.gaierror:
        if ':' in host:
            info = [(socket.AF_INET6, socket.SOCK_STREAM, 0, "", (host, port, 0, 0))]
        else:
            info = [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (host, port))]
    
    for res in info:
        af, socktype, proto, canonname, sa = res
        s = None
        try:
            s = socket.socket(af, socktype, proto)
            # See http://groups.google.com/group/cherrypy-users/
            #        browse_frm/thread/bbfe5eb39c904fe0
            s.settimeout(timeout)
            s.connect((host, port))
            s.close()
            raise IOError("Port %s is in use on %s; perhaps the previous "
                          "httpserver did not shut down properly." %
                          (repr(port), repr(host)))
        except socket.error:
            if s:
                s.close()


# Feel free to increase these defaults on slow systems:
free_port_timeout = 0.1
occupied_port_timeout = 1.0

def wait_for_free_port(host, port, timeout=None):
    """Wait for the specified port to become free (drop requests)."""
    if not host:
        raise ValueError("Host values of '' or None are not allowed.")
    if timeout is None:
        timeout = free_port_timeout
    
    for trial in range(50):
        try:
            # we are expecting a free port, so reduce the timeout
            check_port(host, port, timeout=timeout)
        except IOError:
            # Give the old server thread time to free the port.
            time.sleep(timeout)
        else:
            return
    
    raise IOError("Port %r not free on %r" % (port, host))

def wait_for_occupied_port(host, port, timeout=None):
    """Wait for the specified port to become active (receive requests)."""
    if not host:
        raise ValueError("Host values of '' or None are not allowed.")
    if timeout is None:
        timeout = occupied_port_timeout
    
    for trial in range(50):
        try:
            check_port(host, port, timeout=timeout)
        except IOError:
            return
        else:
            time.sleep(timeout)
    
    raise IOError("Port %r not bound on %r" % (port, host))
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.