commit 8f92a27705e78ec6751b6974f6f31d772137b026 Author: Damian Johnson <atagar@torproject.org> Date: Wed Feb 8 20:24:10 2012 -0800 Putting all locks under a 'with' clause At first I was dubious of the usefulness of 'with' keyword gsathya showed me. However, now that I've discovered that locking can be done under it I take that all back - it's a wonderful, wonderful thing and I don't know how I got by with manual locking/releasing before. ... and then they ate Sir Robin's minstrels and there was much rejoicing. --- stem/socket.py | 113 +++++++++++++++----------------------- stem/util/conf.py | 127 +++++++++++++++++++------------------------ test/runner.py | 155 ++++++++++++++++++++++++----------------------------- 3 files changed, 171 insertions(+), 224 deletions(-) diff --git a/stem/socket.py b/stem/socket.py index 5ae4418..9379884 100644 --- a/stem/socket.py +++ b/stem/socket.py @@ -110,18 +110,15 @@ class ControlSocket: stem.socket.SocketClosed if the socket is known to be shut down """ - self._send_lock.acquire() - - try: - if not self.is_alive(): raise SocketClosed() - send_message(self._socket_file, message, raw) - except SocketClosed, exc: - # if send_message raises a SocketClosed then we should properly shut - # everything down - if self.is_alive(): self.close() - raise exc - finally: - self._send_lock.release() + with self._send_lock: + try: + if not self.is_alive(): raise SocketClosed() + send_message(self._socket_file, message, raw) + except SocketClosed, exc: + # if send_message raises a SocketClosed then we should properly shut + # everything down + if self.is_alive(): self.close() + raise exc def recv(self): """ @@ -137,18 +134,15 @@ class ControlSocket: complete message """ - self._recv_lock.acquire() - - try: - if not self.is_alive(): raise SocketClosed() - return recv_message(self._socket_file) - except SocketClosed, exc: - # if recv_message raises a SocketClosed then we should properly shut - # everything down - if self.is_alive(): self.close() - raise exc - finally: - self._recv_lock.release() + with self._recv_lock: + try: + if not self.is_alive(): raise SocketClosed() + return recv_message(self._socket_file) + except SocketClosed, exc: + # if recv_message raises a SocketClosed then we should properly shut + # everything down + if self.is_alive(): self.close() + raise exc def is_alive(self): """ @@ -178,54 +172,41 @@ class ControlSocket: stem.socket.SocketError if unable to make a socket """ - # we need both locks for this - self._send_lock.acquire() - self._recv_lock.acquire() - - # close the socket if we're currently attached to one - if self.is_alive(): self.close() - - try: + with self._send_lock, self._recv_lock: + # close the socket if we're currently attached to one + if self.is_alive(): self.close() + self._socket = self._make_socket() self._socket_file = self._socket.makefile() self._is_alive = True - finally: - self._send_lock.release() - self._recv_lock.release() def close(self): """ Shuts down the socket. If it's already closed then this is a no-op. """ - # we need both locks for this - self._send_lock.acquire() - self._recv_lock.acquire() - - if self._socket: - # if we haven't yet established a connection then this raises an error - # socket.error: [Errno 107] Transport endpoint is not connected - try: self._socket.shutdown(socket.SHUT_RDWR) - except socket.error: pass + with self._send_lock, self._recv_lock: + if self._socket: + # if we haven't yet established a connection then this raises an error + # socket.error: [Errno 107] Transport endpoint is not connected + try: self._socket.shutdown(socket.SHUT_RDWR) + except socket.error: pass + + # Suppressing unexpected exceptions from close. For instance, if the + # socket's file has already been closed then with python 2.7 that raises + # with... + # error: [Errno 32] Broken pipe + + try: self._socket.close() + except: pass - # Suppressing unexpected exceptions from close. For instance, if the - # socket's file has already been closed then with python 2.7 that raises - # with... - # error: [Errno 32] Broken pipe + if self._socket_file: + try: self._socket_file.close() + except: pass - try: self._socket.close() - except: pass - - if self._socket_file: - try: self._socket_file.close() - except: pass - - self._socket = None - self._socket_file = None - self._is_alive = False - - self._send_lock.release() - self._recv_lock.release() + self._socket = None + self._socket_file = None + self._is_alive = False def __enter__(self): return self @@ -545,13 +526,10 @@ class ControlLine(str): IndexError if we don't have any remaining content left to parse """ - try: - self._remainder_lock.acquire() + with self._remainder_lock: next_entry, remainder = _parse_entry(self._remainder, quoted, escaped) self._remainder = remainder return next_entry - finally: - self._remainder_lock.release() def pop_mapping(self, quoted = False, escaped = False): """ @@ -570,8 +548,7 @@ class ControlLine(str): the value being quoted """ - try: - self._remainder_lock.acquire() + with self._remainder_lock: if self.is_empty(): raise IndexError("no remaining content to parse") key_match = KEY_ARG.match(self._remainder) @@ -585,8 +562,6 @@ class ControlLine(str): next_entry, remainder = _parse_entry(remainder, quoted, escaped) self._remainder = remainder return (key, next_entry) - finally: - self._remainder_lock.release() def _parse_entry(line, quoted, escaped): """ diff --git a/stem/util/conf.py b/stem/util/conf.py index 7d086f4..05e98f8 100644 --- a/stem/util/conf.py +++ b/stem/util/conf.py @@ -30,7 +30,7 @@ three things... There are many ways of using the Config class but the most common ones are... -- Call config_dict to get a dictionary that's always synced with with a Config. +- Call config_dict to get a dictionary that's always synced with a Config. - Make a dictionary and call synchronize() to bring it into sync with the Config. This does not keep it in sync as the Config changes. See the Config @@ -216,43 +216,41 @@ class Config(): with open(self._path, "r") as config_file: read_contents = config_file.readlines() - self._contents_lock.acquire() - self._raw_contents = read_contents - remainder = list(self._raw_contents) - - while remainder: - line = remainder.pop(0) - - # strips any commenting or excess whitespace - comment_start = line.find("#") - if comment_start != -1: line = line[:comment_start] - line = line.strip() + with self._contents_lock: + self._raw_contents = read_contents + remainder = list(self._raw_contents) - # parse the key/value pair - if line: - try: - key, value = line.split(" ", 1) - value = value.strip() - except ValueError: - log.debug("Config entry '%s' is expected to be of the format 'Key Value', defaulting to '%s' -> ''" % (line, line)) - key, value = line, "" + while remainder: + line = remainder.pop(0) + + # strips any commenting or excess whitespace + comment_start = line.find("#") + if comment_start != -1: line = line[:comment_start] + line = line.strip() - if not value: - # this might be a multi-line entry, try processing it as such - multiline_buffer = [] + # parse the key/value pair + if line: + try: + key, value = line.split(" ", 1) + value = value.strip() + except ValueError: + log.debug("Config entry '%s' is expected to be of the format 'Key Value', defaulting to '%s' -> ''" % (line, line)) + key, value = line, "" - while remainder and remainder[0].lstrip().startswith("|"): - content = remainder.pop(0).lstrip()[1:] # removes '\s+|' prefix - content = content.rstrip("\n") # trailing newline - multiline_buffer.append(content) + if not value: + # this might be a multi-line entry, try processing it as such + multiline_buffer = [] + + while remainder and remainder[0].lstrip().startswith("|"): + content = remainder.pop(0).lstrip()[1:] # removes '\s+|' prefix + content = content.rstrip("\n") # trailing newline + multiline_buffer.append(content) + + if multiline_buffer: + self.set(key, "\n".join(multiline_buffer), False) + continue - if multiline_buffer: - self.set(key, "\n".join(multiline_buffer), False) - continue - - self.set(key, value, False) - - self._contents_lock.release() + self.set(key, value, False) def save(self, path = None): """ @@ -272,17 +270,13 @@ class Config(): elif not self._path: raise ValueError("Unable to save configuration: no path provided") - self._contents_lock.acquire() - - with open(self._path, 'w') as output_file: + with self._contents_lock, open(self._path, 'w') as output_file: for entry_key in sorted(self.keys()): for entry_value in self.get_value(entry_key, multiple = True): # check for multi line entries if "\n" in entry_value: entry_value = "\n|" + entry_value.replace("\n", "\n|") output_file.write('%s %s\n' % (entry_key, entry_value)) - - self._contents_lock.release() def clear(self): """ @@ -290,11 +284,10 @@ class Config(): state. """ - self._contents_lock.acquire() - self._contents.clear() - self._raw_contents = [] - self._requested_keys = set() - self._contents_lock.release() + with self._contents_lock: + self._contents.clear() + self._raw_contents = [] + self._requested_keys = set() def synchronize(self, conf_mappings, limits = None): """ @@ -344,14 +337,12 @@ class Config(): backfill (bool) - calls the function with our current values if true """ - self._contents_lock.acquire() - self._listeners.append(listener) - - if backfill: - for key in self.keys(): - listener(self, key) - - self._contents_lock.release() + with self._contents_lock: + self._listeners.append(listener) + + if backfill: + for key in self.keys(): + listener(self, key) def clear_listeners(self): """ @@ -393,9 +384,7 @@ class Config(): the values are appended """ - try: - self._contents_lock.acquire() - + with self._contents_lock: if isinstance(value, str): if not overwrite and key in self._contents: self._contents[key].append(value) else: self._contents[key] = [value] @@ -409,8 +398,6 @@ class Config(): for listener in self._listeners: listener(self, key) else: raise ValueError("Config.set() only accepts str, list, or tuple. Provided value was a '%s'" % type(value)) - finally: - self._contents_lock.release() def get(self, key, default = None): """ @@ -498,20 +485,18 @@ class Config(): key, providing the default if no such key exists """ - self._contents_lock.acquire() - - if key in self._contents: - val = self._contents[key] - if not multiple: val = val[-1] - self._requested_keys.add(key) - else: - message_id = "stem.util.conf.missing_config_key_%s" % key - log.log_once(message_id, log.TRACE, "config entry '%s' not found, defaulting to '%s'" % (key, default)) - val = default - - self._contents_lock.release() - - return val + with self._contents_lock: + if key in self._contents: + self._requested_keys.add(key) + + if multiple: + return self._contents[key] + else: + return self._contents[key][-1] + else: + message_id = "stem.util.conf.missing_config_key_%s" % key + log.log_once(message_id, log.TRACE, "config entry '%s' not found, defaulting to '%s'" % (key, default)) + return default def get_str_csv(self, key, default = None, count = None, sub_key = None): """ diff --git a/test/runner.py b/test/runner.py index 07a54fd..cf596d8 100644 --- a/test/runner.py +++ b/test/runner.py @@ -160,79 +160,75 @@ class Runner: OSError if unable to run test preparations or start tor """ - self._runner_lock.acquire() - - # if we're holding on to a tor process (running or not) then clean up after - # it so we can start a fresh instance - if self._tor_process: self.stop() - - test.output.print_line("Setting up a test instance...", *STATUS_ATTR) - - # if 'test_directory' is unset then we make a new data directory in /tmp - # and clean it up when we're done - - config_test_dir = CONFIG["integ.test_directory"] - - if config_test_dir: - self._test_dir = stem.util.system.expand_path(config_test_dir, STEM_BASE) - else: - self._test_dir = tempfile.mktemp("-stem-integ") - - original_cwd, data_dir_path = os.getcwd(), self._test_dir - - if CONFIG["test.target.relative_data_dir"]: - tor_cwd = os.path.dirname(self._test_dir) - if not os.path.exists(tor_cwd): os.makedirs(tor_cwd) + with self._runner_lock: + # if we're holding on to a tor process (running or not) then clean up after + # it so we can start a fresh instance + if self._tor_process: self.stop() - os.chdir(tor_cwd) - data_dir_path = "./%s" % os.path.basename(self._test_dir) - - self._tor_cmd = tor_cmd - self._custom_opts = extra_torrc_opts - self._torrc_contents = BASE_TORRC % data_dir_path - - if extra_torrc_opts: - self._torrc_contents += "\n".join(extra_torrc_opts) + "\n" - - try: - self._tor_cwd = os.getcwd() - self._run_setup() - self._start_tor(tor_cmd) + test.output.print_line("Setting up a test instance...", *STATUS_ATTR) + + # if 'test_directory' is unset then we make a new data directory in /tmp + # and clean it up when we're done + + config_test_dir = CONFIG["integ.test_directory"] + + if config_test_dir: + self._test_dir = stem.util.system.expand_path(config_test_dir, STEM_BASE) + else: + self._test_dir = tempfile.mktemp("-stem-integ") + + original_cwd, data_dir_path = os.getcwd(), self._test_dir - # revert our cwd back to normal if CONFIG["test.target.relative_data_dir"]: - os.chdir(original_cwd) - except OSError, exc: - self.stop() - raise exc - finally: - self._runner_lock.release() + tor_cwd = os.path.dirname(self._test_dir) + if not os.path.exists(tor_cwd): os.makedirs(tor_cwd) + + os.chdir(tor_cwd) + data_dir_path = "./%s" % os.path.basename(self._test_dir) + + self._tor_cmd = tor_cmd + self._custom_opts = extra_torrc_opts + self._torrc_contents = BASE_TORRC % data_dir_path + + if extra_torrc_opts: + self._torrc_contents += "\n".join(extra_torrc_opts) + "\n" + + try: + self._tor_cwd = os.getcwd() + self._run_setup() + self._start_tor(tor_cmd) + + # revert our cwd back to normal + if CONFIG["test.target.relative_data_dir"]: + os.chdir(original_cwd) + except OSError, exc: + self.stop() + raise exc def stop(self): """ Stops our tor test instance and cleans up any temporary resources. """ - self._runner_lock.acquire() - test.output.print_noline("Shutting down tor... ", *STATUS_ATTR) - - if self._tor_process: - self._tor_process.kill() - self._tor_process.communicate() # blocks until the process is done - - # if we've made a temporary data directory then clean it up - if self._test_dir and CONFIG["integ.test_directory"] == "": - shutil.rmtree(self._test_dir, ignore_errors = True) - - self._test_dir = "" - self._tor_cmd = None - self._tor_cwd = "" - self._torrc_contents = "" - self._custom_opts = None - self._tor_process = None - - test.output.print_line("done", *STATUS_ATTR) - self._runner_lock.release() + with self._runner_lock: + test.output.print_noline("Shutting down tor... ", *STATUS_ATTR) + + if self._tor_process: + self._tor_process.kill() + self._tor_process.communicate() # blocks until the process is done + + # if we've made a temporary data directory then clean it up + if self._test_dir and CONFIG["integ.test_directory"] == "": + shutil.rmtree(self._test_dir, ignore_errors = True) + + self._test_dir = "" + self._tor_cmd = None + self._tor_cwd = "" + self._torrc_contents = "" + self._custom_opts = None + self._tor_process = None + + test.output.print_line("done", *STATUS_ATTR) def is_running(self): """ @@ -242,21 +238,16 @@ class Runner: True if we have a running tor test instance, False otherwise """ - # subprocess.Popen.poll() checks the return code, returning None if it's - # still going - - self._runner_lock.acquire() - is_running = self._tor_process and self._tor_process.poll() == None - - # If the tor process closed unexpectedly then this is probably the first - # place that we're realizing it. Clean up the temporary resources now since - # we might not end up calling stop() as normal. - - if not is_running: self.stop(True) - - self._runner_lock.release() - - return is_running + with self._runner_lock: + # Check for an unexpected shutdown by calling subprocess.Popen.poll(), + # which returns the exit code or None if we're still running. + + if self._tor_process and self._tor_process.poll() != None: + # clean up the temporary resources and note the unexpected shutdown + self.stop() + test.output.print_line("tor shut down unexpectedly", *ERROR_ATTR) + + return bool(self._tor_process) def is_accessible(self): """ @@ -439,14 +430,10 @@ class Runner: RunnerStopped if we aren't running """ - try: - self._runner_lock.acquire() - + with self._runner_lock: if self.is_running(): return self.__dict__[attr] else: raise RunnerStopped() - finally: - self._runner_lock.release() def _run_setup(self): """