/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/server.py

  • Committer: John Arbash Meinel
  • Date: 2011-04-20 14:27:19 UTC
  • mto: This revision was merged to the branch mainline in revision 5837.
  • Revision ID: john@arbash-meinel.com-20110420142719-advs1k5vztqzbrgv
https://launchpad.net/bugs/767177
Fix bug #767177. Be more agressive with file.close() calls.

Our test suite gets a number of thread leaks and failures because it happens to get async
SFTPFile.close() calls. (if an SFTPFile closes due to __del__ it is done as an async request,
while if you call SFTPFile.close() it is done as a synchronous request.)
We have a couple other cases, probably. Namely SFTPTransport.get() also does an async
prefetch of the content, so if you don't .read() you'll also leak threads that think they
are doing work that you want.

The biggest change here, though, is using a try/finally in a generator, which is not 
python2.4 compatible.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/server.py
22
22
import sys
23
23
import threading
24
24
 
25
 
from bzrlib.hooks import HookPoint, Hooks
 
25
from bzrlib.hooks import Hooks
26
26
from bzrlib import (
27
27
    errors,
28
28
    trace,
29
 
    transport,
 
29
    transport as _mod_transport,
30
30
)
31
31
from bzrlib.lazy_import import lazy_import
32
32
lazy_import(globals(), """
33
33
from bzrlib.smart import medium
34
34
from bzrlib.transport import (
35
35
    chroot,
36
 
    get_transport,
37
36
    pathfilter,
38
37
    )
39
38
from bzrlib import (
51
50
    hooks: An instance of SmartServerHooks.
52
51
    """
53
52
 
54
 
    def __init__(self, backing_transport, host='127.0.0.1', port=0,
55
 
                 root_client_path='/'):
 
53
    def __init__(self, backing_transport, root_client_path='/'):
56
54
        """Construct a new server.
57
55
 
58
56
        To actually start it running, call either start_background_thread or
59
57
        serve.
60
58
 
61
59
        :param backing_transport: The transport to serve.
 
60
        :param root_client_path: The client path that will correspond to root
 
61
            of backing_transport.
 
62
        """
 
63
        self.backing_transport = backing_transport
 
64
        self.root_client_path = root_client_path
 
65
 
 
66
    def start_server(self, host, port):
 
67
        """Create the server listening socket.
 
68
 
62
69
        :param host: Name of the interface to listen on.
63
70
        :param port: TCP port to listen on, or 0 to allocate a transient port.
64
 
        :param root_client_path: The client path that will correspond to root
65
 
            of backing_transport.
66
71
        """
67
72
        # let connections timeout so that we get a chance to terminate
68
73
        # Keep a reference to the exceptions we want to catch because the socket
89
94
        self.port = self._sockname[1]
90
95
        self._server_socket.listen(1)
91
96
        self._server_socket.settimeout(1)
92
 
        self.backing_transport = backing_transport
93
97
        self._started = threading.Event()
94
98
        self._stopped = threading.Event()
95
 
        self.root_client_path = root_client_path
96
99
 
97
 
    def serve(self, thread_name_suffix=''):
98
 
        self._should_terminate = False
99
 
        # for hooks we are letting code know that a server has started (and
100
 
        # later stopped).
 
100
    def _backing_urls(self):
101
101
        # There are three interesting urls:
102
102
        # The URL the server can be contacted on. (e.g. bzr://host/)
103
103
        # The URL that a commit done on the same machine as the server will
104
104
        # have within the servers space. (e.g. file:///home/user/source)
105
105
        # The URL that will be given to other hooks in the same process -
106
 
        # the URL of the backing transport itself. (e.g. chroot+:///)
 
106
        # the URL of the backing transport itself. (e.g. filtered-36195:///)
107
107
        # We need all three because:
108
108
        #  * other machines see the first
109
109
        #  * local commits on this machine should be able to be mapped to
113
113
        # The latter two urls are different aliases to the servers url,
114
114
        # so we group those in a list - as there might be more aliases
115
115
        # in the future.
116
 
        backing_urls = [self.backing_transport.base]
 
116
        urls = [self.backing_transport.base]
117
117
        try:
118
 
            backing_urls.append(self.backing_transport.external_url())
 
118
            urls.append(self.backing_transport.external_url())
119
119
        except errors.InProcessTransport:
120
120
            pass
 
121
        return urls
 
122
 
 
123
    def run_server_started_hooks(self, backing_urls=None):
 
124
        if backing_urls is None:
 
125
            backing_urls = self._backing_urls()
121
126
        for hook in SmartTCPServer.hooks['server_started']:
122
127
            hook(backing_urls, self.get_url())
123
128
        for hook in SmartTCPServer.hooks['server_started_ex']:
124
129
            hook(backing_urls, self)
 
130
 
 
131
    def run_server_stopped_hooks(self, backing_urls=None):
 
132
        if backing_urls is None:
 
133
            backing_urls = self._backing_urls()
 
134
        for hook in SmartTCPServer.hooks['server_stopped']:
 
135
            hook(backing_urls, self.get_url())
 
136
 
 
137
    def serve(self, thread_name_suffix=''):
 
138
        self._should_terminate = False
 
139
        # for hooks we are letting code know that a server has started (and
 
140
        # later stopped).
 
141
        self.run_server_started_hooks()
125
142
        self._started.set()
126
143
        try:
127
144
            try:
155
172
            except self._socket_error:
156
173
                # ignore errors on close
157
174
                pass
158
 
            for hook in SmartTCPServer.hooks['server_stopped']:
159
 
                hook(backing_urls, self.get_url())
 
175
            self.run_server_stopped_hooks()
160
176
 
161
177
    def get_url(self):
162
178
        """Return the url of the server"""
163
 
        return "bzr://%s:%d/" % self._sockname
 
179
        return "bzr://%s:%s/" % (self._sockname[0], self._sockname[1])
164
180
 
165
181
    def serve_conn(self, conn, thread_name_suffix):
166
182
        # For WIN32, where the timeout value from the listening socket
172
188
        thread_name = 'smart-server-child' + thread_name_suffix
173
189
        connection_thread = threading.Thread(
174
190
            None, handler.serve, name=thread_name)
 
191
        # FIXME: This thread is never joined, it should at least be collected
 
192
        # somewhere so that tests that want to check for leaked threads can get
 
193
        # rid of them -- vila 20100531
175
194
        connection_thread.setDaemon(True)
176
195
        connection_thread.start()
 
196
        return connection_thread
177
197
 
178
198
    def start_background_thread(self, thread_name_suffix=''):
179
199
        self._started.clear()
219
239
        These are all empty initially, because by default nothing should get
220
240
        notified.
221
241
        """
222
 
        Hooks.__init__(self)
223
 
        self.create_hook(HookPoint('server_started',
 
242
        Hooks.__init__(self, "bzrlib.smart.server", "SmartTCPServer.hooks")
 
243
        self.add_hook('server_started',
224
244
            "Called by the bzr server when it starts serving a directory. "
225
245
            "server_started is called with (backing urls, public url), "
226
246
            "where backing_url is a list of URLs giving the "
227
247
            "server-specific directory locations, and public_url is the "
228
 
            "public URL for the directory being served.", (0, 16), None))
229
 
        self.create_hook(HookPoint('server_started_ex',
 
248
            "public URL for the directory being served.", (0, 16))
 
249
        self.add_hook('server_started_ex',
230
250
            "Called by the bzr server when it starts serving a directory. "
231
251
            "server_started is called with (backing_urls, server_obj).",
232
 
            (1, 17), None))
233
 
        self.create_hook(HookPoint('server_stopped',
 
252
            (1, 17))
 
253
        self.add_hook('server_stopped',
234
254
            "Called by the bzr server when it stops serving a directory. "
235
255
            "server_stopped is called with the same parameters as the "
236
 
            "server_started hook: (backing_urls, public_url).", (0, 16), None))
 
256
            "server_started hook: (backing_urls, public_url).", (0, 16))
237
257
 
238
258
SmartTCPServer.hooks = SmartServerHooks()
239
259
 
305
325
        chroot_server = chroot.ChrootServer(transport)
306
326
        chroot_server.start_server()
307
327
        self.cleanups.append(chroot_server.stop_server)
308
 
        transport = get_transport(chroot_server.get_url())
 
328
        transport = _mod_transport.get_transport(chroot_server.get_url())
309
329
        if self.base_path is not None:
310
330
            # Decorate the server's backing transport with a filter that can
311
331
            # expand homedirs.
312
332
            expand_userdirs = self._make_expand_userdirs_filter(transport)
313
333
            expand_userdirs.start_server()
314
334
            self.cleanups.append(expand_userdirs.stop_server)
315
 
            transport = get_transport(expand_userdirs.get_url())
 
335
            transport = _mod_transport.get_transport(expand_userdirs.get_url())
316
336
        self.transport = transport
317
337
 
318
338
    def _make_smart_server(self, host, port, inet):
324
344
                host = medium.BZR_DEFAULT_INTERFACE
325
345
            if port is None:
326
346
                port = medium.BZR_DEFAULT_PORT
327
 
            smart_server = SmartTCPServer(self.transport, host=host, port=port)
 
347
            smart_server = SmartTCPServer(self.transport)
 
348
            smart_server.start_server(host, port)
328
349
            trace.note('listening on port: %s' % smart_server.port)
329
350
        self.smart_server = smart_server
330
351