← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/http_server.py

Merge previous attempt into current trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/http_server.py
30
30
import urllib
31
31
import urlparse
32
32
 
33
 
from bzrlib import transport
 
33
from bzrlib import (
 
34
    tests,
 
35
    transport,
 
36
    )
34
37
from bzrlib.tests import test_server
35
38
from bzrlib.transport import local
36
39
 
78
81
        connection early to avoid polluting the test results.
79
82
        """
80
83
        try:
81
 
            SimpleHTTPServer.SimpleHTTPRequestHandler.handle_one_request(self)
 
84
            self._handle_one_request()
82
85
        except socket.error, e:
83
86
            # Any socket error should close the connection, but some errors are
84
87
            # due to the client closing early and we don't want to pollute test
89
92
                                     errno.ECONNABORTED, errno.EBADF)):
90
93
                raise
91
94
 
 
95
    def _handle_one_request(self):
 
96
        SimpleHTTPServer.SimpleHTTPRequestHandler.handle_one_request(self)
 
97
 
92
98
    _range_regexp = re.compile(r'^(?P<start>\d+)-(?P<end>\d+)$')
93
99
    _tail_regexp = re.compile(r'^-(?P<tail>\d+)$')
94
100
 
318
324
        # the tests cases.
319
325
        self.test_case_server = test_case_server
320
326
        self._home_dir = test_case_server._home_dir
321
 
 
322
 
    def stop_server(self):
323
 
         """Called to clean-up the server.
324
 
 
325
 
         Since the server may be (surely is, even) in a blocking listen, we
326
 
         shutdown its socket before closing it.
327
 
         """
328
 
         # Note that is this executed as part of the implicit tear down in the
329
 
         # main thread while the server runs in its own thread. The clean way
330
 
         # to tear down the server is to instruct him to stop accepting
331
 
         # connections and wait for the current connection(s) to end
332
 
         # naturally. To end the connection naturally, the http transports
333
 
         # should close their socket when they do not need to talk to the
334
 
         # server anymore. This happens naturally during the garbage collection
335
 
         # phase of the test transport objetcs (the server clients), so we
336
 
         # don't have to worry about them.  So, for the server, we must tear
337
 
         # down here, from the main thread, when the test have ended.  Note
338
 
         # that since the server is in a blocking operation and since python
339
 
         # use select internally, shutting down the socket is reliable and
340
 
         # relatively clean.
341
 
         try:
342
 
             self.socket.shutdown(socket.SHUT_RDWR)
343
 
         except socket.error, e:
344
 
             # WSAENOTCONN (10057) 'Socket is not connected' is harmless on
345
 
             # windows (occurs before the first connection attempt
346
 
             # vila--20071230)
347
 
 
348
 
             # 'Socket is not connected' can also occur on OSX, with a
349
 
             # "regular" ENOTCONN (when something went wrong during test case
350
 
             # setup leading to self.setUp() *not* being called but
351
 
             # self.stop_server() still being called -- vila20081106
352
 
             if not len(e.args) or e.args[0] not in (errno.ENOTCONN, 10057):
353
 
                 raise
354
 
         # Let the server properly close the socket
355
 
         self.server_close()
356
 
 
357
 
 
358
 
class TestingHTTPServer(SocketServer.TCPServer, TestingHTTPServerMixin):
 
327
        self.serving = None
 
328
        self.is_shut_down = threading.Event()
 
329
        # We collect the sockets/threads used by the clients so we can
 
330
        # close/join them when shutting down
 
331
        self.clients = []
 
332
 
 
333
    def get_request (self):
 
334
        """Get the request and client address from the socket.
 
335
        """
 
336
        sock, addr = self._get_request()
 
337
        self.clients.append([sock, addr])
 
338
        return sock, addr
 
339
 
 
340
    def verify_request(self, request, client_address):
 
341
        """Verify the request.
 
342
 
 
343
        Return True if we should proceed with this request, False if we should
 
344
        not even touch a single byte in the socket !
 
345
        """
 
346
        return self.serving is not None and self.serving.isSet()
 
347
 
 
348
    def handle_request(self):
 
349
        request, client_address = self.get_request()
 
350
        try:
 
351
            if self.verify_request(request, client_address):
 
352
                self.process_request(request, client_address)
 
353
        except:
 
354
            if self.serving is not None and self.serving.isSet():
 
355
                self.handle_error(request, client_address)
 
356
            else:
 
357
                # Exceptions raised while we shut down are just noise, but feel
 
358
                # free to put a breakpoint here if you suspect something
 
359
                # else. Such an example is the SSL handshake: it's automatic
 
360
                # once we start processing the request but the last connection
 
361
                # will close immediately and will not be able to correctly
 
362
                # reply.
 
363
                pass
 
364
            self.close_request(request)
 
365
 
 
366
    def server_bind(self):
 
367
        # The following has been fixed in 2.5 so we need to provide it for
 
368
        # older python versions.
 
369
        if sys.version < (2, 5):
 
370
            self.server_address = self.socket.getsockname()
 
371
 
 
372
    def serve(self, started):
 
373
        self.serving  = threading.Event()
 
374
        self.serving.set()
 
375
        self.is_shut_down.clear()
 
376
        if 'threads' in tests.selftest_debug_flags:
 
377
            print 'Starting %r' % (self.server_address,)
 
378
        # We are listening and ready to accept connections
 
379
        started.set()
 
380
        while self.serving.isSet():
 
381
            if 'threads' in tests.selftest_debug_flags:
 
382
                print 'Accepting on %r' % (self.server_address,)
 
383
            # Really a connection but the python framework is generic and
 
384
            # call them requests
 
385
            self.handle_request()
 
386
        if 'threads' in tests.selftest_debug_flags:
 
387
            print 'Closing  %r' % (self.server_address,)
 
388
        # Let's close the listening socket
 
389
        self.server_close()
 
390
        if 'threads' in tests.selftest_debug_flags:
 
391
            print 'Closed   %r' % (self.server_address,)
 
392
        self.is_shut_down.set()
 
393
 
 
394
    def connect_socket(self):
 
395
        err = socket.error('getaddrinfo returns an empty list')
 
396
        for res in socket.getaddrinfo(*self.server_address):
 
397
            af, socktype, proto, canonname, sa = res
 
398
            sock = None
 
399
            try:
 
400
                sock = socket.socket(af, socktype, proto)
 
401
                sock.connect(sa)
 
402
                return sock
 
403
 
 
404
            except socket.error, err:
 
405
                # 'err' is now the most recent error
 
406
                if sock is not None:
 
407
                    sock.close()
 
408
        raise err
 
409
 
 
410
    def join_thread(self, thread, timeout=2):
 
411
        thread.join(timeout)
 
412
        if thread.isAlive():
 
413
            # The timeout expired without joining the thread, the thread is
 
414
            # therefore stucked and that's a failure as far as the test is
 
415
            # concerned. We used to hang here.
 
416
            raise AssertionError('thread %s hung' % (thread.name,))
 
417
 
 
418
    def shutdown(self):
 
419
        """Stops the serve() loop.
 
420
 
 
421
        Blocks until the loop has finished. This must be called while serve()
 
422
        is running in another thread, or it will deadlock.
 
423
        """
 
424
        if self.serving is None:
 
425
            # If the server wasn't properly started, there is nothing to
 
426
            # shutdown.
 
427
            return
 
428
        # As soon as we stop serving, no more connection are accepted except
 
429
        # one to get out of the blocking listen.
 
430
        self.serving.clear()
 
431
        # The server is listening for a last connection, let's give it:
 
432
        last_conn = None
 
433
        try:
 
434
            last_conn = self.connect_socket()
 
435
        except socket.error, e:
 
436
            # But ignore connection errors as the point is to unblock the
 
437
            # server thread, it may happen that it's not blocked or even not
 
438
            # started (when something went wrong during test case setup
 
439
            # leading to self.setUp() *not* being called but self.tearDown()
 
440
            # still being called)
 
441
            pass
 
442
        # We don't have to wait for the server to shut down to start shutting
 
443
        # down the clients, so let's start now.
 
444
        for c in self.clients:
 
445
            self.shutdown_client(c)
 
446
        self.clients = []
 
447
        # Now we wait for the thread running serve() to finish
 
448
        self.is_shut_down.wait()
 
449
        if last_conn is not None:
 
450
            # Close the last connection without trying to use it. The server
 
451
            # will not process a single byte on that socket to avoid
 
452
            # complications (SSL starts with a handshake for example).
 
453
            last_conn.close()
 
454
 
 
455
    def shutdown_client(self, client):
 
456
        sock, addr = client[:2]
 
457
        self.shutdown_client_socket(sock)
 
458
 
 
459
    def shutdown_client_socket(self, sock):
 
460
        """Properly shutdown a client socket.
 
461
 
 
462
        Under some circumstances (as in bug #383920), we need to force the
 
463
        shutdown as python delays it until gc occur otherwise and the client
 
464
        may hang.
 
465
 
 
466
        This should be called only when no other thread is trying to use the
 
467
        socket.
 
468
        """
 
469
        try:
 
470
            # The request process has been completed, the thread is about to
 
471
            # die, let's shutdown the socket if we can.
 
472
            sock.shutdown(socket.SHUT_RDWR)
 
473
        except (socket.error, select.error), e:
 
474
            if e[0] in (errno.EBADF, errno.ENOTCONN):
 
475
                # Right, the socket is already down
 
476
                pass
 
477
            else:
 
478
                raise
 
479
 
 
480
 
 
481
class TestingHTTPServer(TestingHTTPServerMixin, SocketServer.TCPServer):
359
482
 
360
483
    def __init__(self, server_address, request_handler_class,
361
484
                 test_case_server):
363
486
        SocketServer.TCPServer.__init__(self, server_address,
364
487
                                        request_handler_class)
365
488
 
366
 
 
367
 
class TestingThreadingHTTPServer(SocketServer.ThreadingTCPServer,
368
 
                                 TestingHTTPServerMixin):
 
489
    def _get_request (self):
 
490
        return SocketServer.TCPServer.get_request(self)
 
491
 
 
492
    def server_bind(self):
 
493
        SocketServer.TCPServer.server_bind(self)
 
494
        TestingHTTPServerMixin.server_bind(self)
 
495
 
 
496
 
 
497
class TestingThreadingHTTPServer(TestingHTTPServerMixin,
 
498
                                 SocketServer.ThreadingTCPServer,
 
499
                                 ):
369
500
    """A threading HTTP test server for HTTP 1.1.
370
501
 
371
502
    Since tests can initiate several concurrent connections to the same http
383
514
        # lying around.
384
515
        self.daemon_threads = True
385
516
 
386
 
    def process_request_thread(self, request, client_address):
 
517
    def _get_request (self):
 
518
        return SocketServer.ThreadingTCPServer.get_request(self)
 
519
 
 
520
    def process_request_thread(self, started, request, client_address):
 
521
        if 'threads' in tests.selftest_debug_flags:
 
522
            print 'Processing: %s' % (threading.currentThread().name,)
 
523
        started.set()
387
524
        SocketServer.ThreadingTCPServer.process_request_thread(
388
525
            self, request, client_address)
389
 
        # Under some circumstances (as in bug #383920), we need to force the
390
 
        # shutdown as python delays it until gc occur otherwise and the client
391
 
        # may hang.
392
 
        try:
393
 
            # The request process has been completed, the thread is about to
394
 
            # die, let's shutdown the socket if we can.
395
 
            request.shutdown(socket.SHUT_RDWR)
396
 
        except (socket.error, select.error), e:
397
 
            if e[0] in (errno.EBADF, errno.ENOTCONN):
398
 
                # Right, the socket is already down
399
 
                pass
400
 
            else:
401
 
                raise
 
526
        # Shutdown the socket as soon as possible, the thread will be joined
 
527
        # later if needed during server shutdown thread.
 
528
        self.shutdown_client_socket(request)
 
529
 
 
530
    def process_request(self, request, client_address):
 
531
        """Start a new thread to process the request."""
 
532
        client = self.clients.pop()
 
533
        started = threading.Event()
 
534
        t = threading.Thread(target = self.process_request_thread,
 
535
                             args = (started, request, client_address))
 
536
        t.name = '%s -> %s' % (client_address, self.server_address)
 
537
        client.append(t)
 
538
        self.clients.append(client)
 
539
        if self.daemon_threads:
 
540
            t.setDaemon (1)
 
541
        t.start()
 
542
        started.wait()
 
543
 
 
544
    def shutdown_client(self, client):
 
545
        TestingHTTPServerMixin.shutdown_client(self, client)
 
546
        if len(client) == 3:
 
547
            # The thread has been created only if the request is processed but
 
548
            # after the connection is inited. This could happne when the server
 
549
            # is shut down.
 
550
            sock, addr, thread = client
 
551
            if 'threads' in tests.selftest_debug_flags:
 
552
                print 'Try    joining: %s' % (thread.name,)
 
553
            self.join_thread(thread)
 
554
 
 
555
    def server_bind(self):
 
556
        SocketServer.ThreadingTCPServer.server_bind(self)
 
557
        TestingHTTPServerMixin.server_bind(self)
402
558
 
403
559
 
404
560
class HttpServer(transport.Server):
463
619
                raise httplib.UnknownProtocol(proto_vers)
464
620
            else:
465
621
                self._httpd = self.create_httpd(serv_cls, rhandler)
466
 
            self.host, self.port = self._httpd.socket.getsockname()
 
622
            # Ensure we get the right port and an updated host if needed
 
623
            self.host, self.port = self._httpd.server_address
467
624
        return self._httpd
468
625
 
469
 
    def _http_start(self):
 
626
    def _http_start(self, started):
470
627
        """Server thread main entry point. """
471
 
        self._http_running = False
 
628
        server = None
472
629
        try:
473
 
            try:
474
 
                httpd = self._get_httpd()
475
 
                self._http_base_url = '%s://%s:%s/' % (self._url_protocol,
476
 
                                                       self.host, self.port)
477
 
                self._http_running = True
478
 
            except:
479
 
                # Whatever goes wrong, we save the exception for the main
480
 
                # thread. Note that since we are running in a thread, no signal
481
 
                # can be received, so we don't care about KeyboardInterrupt.
482
 
                self._http_exception = sys.exc_info()
483
 
        finally:
484
 
            # Release the lock or the main thread will block and the whole
485
 
            # process will hang.
486
 
            self._http_starting.release()
 
630
            server = self._get_httpd()
 
631
            self._http_base_url = '%s://%s:%s/' % (self._url_protocol,
 
632
                                                   self.host, self.port)
 
633
        except:
 
634
            # Whatever goes wrong, we save the exception for the main
 
635
            # thread. Note that since we are running in a thread, no signal
 
636
            # can be received, so we don't care about KeyboardInterrupt.
 
637
            self._http_exception = sys.exc_info()
487
638
 
488
 
        # From now on, exceptions are taken care of by the
489
 
        # SocketServer.BaseServer or the request handler.
490
 
        while self._http_running:
491
 
            try:
492
 
                # Really an HTTP connection but the python framework is generic
493
 
                # and call them requests
494
 
                httpd.handle_request()
495
 
            except socket.timeout:
496
 
                pass
497
 
            except (socket.error, select.error), e:
498
 
                if (e[0] == errno.EBADF
499
 
                    or (sys.platform == 'win32' and e[0] == 10038)):
500
 
                    # Starting with python-2.6, handle_request may raise socket
501
 
                    # or select exceptions when the server is shut down (as we
502
 
                    # do).
503
 
                    # 10038 = WSAENOTSOCK
504
 
                    # http://msdn.microsoft.com/en-us/library/ms740668%28VS.85%29.aspx
505
 
                    pass
506
 
                else:
507
 
                    raise
 
639
        if server is not None:
 
640
            # From now on, exceptions are taken care of by the
 
641
            # SocketServer.BaseServer or the request handler.
 
642
            server.serve(started)
 
643
        if not started.isSet():
 
644
            # Hmm, something went wrong, but we can release the caller anyway
 
645
            started.set()
508
646
 
509
647
    def _get_remote_url(self, path):
510
648
        path_parts = path.split(os.path.sep)
535
673
                or isinstance(backing_transport_server,
536
674
                              test_server.LocalURLServer)):
537
675
            raise AssertionError(
538
 
                "HTTPServer currently assumes local transport, got %s" % \
 
676
                "HTTPServer currently assumes local transport, got %s" %
539
677
                backing_transport_server)
540
678
        self._home_dir = os.getcwdu()
541
679
        self._local_path_parts = self._home_dir.split(os.path.sep)
542
680
        self._http_base_url = None
543
681
 
544
682
        # Create the server thread
545
 
        self._http_starting = threading.Lock()
546
 
        self._http_starting.acquire()
547
 
        self._http_thread = threading.Thread(target=self._http_start)
 
683
        started = threading.Event()
 
684
        self._http_thread = threading.Thread(target=self._http_start,
 
685
                                             args = (started,))
548
686
        self._http_thread.setDaemon(True)
549
687
        self._http_exception = None
550
688
        self._http_thread.start()
551
 
 
552
689
        # Wait for the server thread to start (i.e release the lock)
553
 
        self._http_starting.acquire()
 
690
        started.wait()
 
691
        self._http_thread.name = self._http_base_url
 
692
        if 'threads' in tests.selftest_debug_flags:
 
693
            print 'Thread started: %s' % (self._http_thread.name,)
 
694
 
554
695
 
555
696
        if self._http_exception is not None:
556
697
            # Something went wrong during server start
557
698
            exc_class, exc_value, exc_tb = self._http_exception
558
699
            raise exc_class, exc_value, exc_tb
559
 
        self._http_starting.release()
560
700
        self.logs = []
561
701
 
562
702
    def stop_server(self):
563
 
        self._httpd.stop_server()
564
 
        self._http_running = False
565
 
        # We don't need to 'self._http_thread.join()' here since the thread is
566
 
        # a daemonic one and will be garbage collected anyway. Joining just
567
 
        # slows us down for no added benefit.
 
703
        """See bzrlib.transport.Server.tearDown."""
 
704
        self._httpd.shutdown()
 
705
        if 'threads' in tests.selftest_debug_flags:
 
706
            print 'Try    joining: %s' % (self._http_thread.name,)
 
707
        self._httpd.join_thread(self._http_thread)
 
708
        if 'threads' in tests.selftest_debug_flags:
 
709
            print 'Thread  joined: %s' % (self._http_thread.name,)
568
710
 
569
711
    def get_url(self):
570
712
        """See bzrlib.transport.Server.get_url."""