Make sure we're releasing all the locks, even if there's an exception

-autopull

Author: makyen

bodyfetcher.py

@@ -60,31 +60,30 @@ def add_to_queue(self, post, should_check_site=False):
         post_id = d["id"]
         if (post_id == 3122 or post_id == 51812) and site_base == "meta.stackexchange.com":
             return  # don't check meta sandbox, it's full of weird posts
-        self.queue_modify_lock.acquire()
-        if site_base not in self.queue:
-            self.queue[site_base] = {}
-
-        # Something about how the queue is being filled is storing Post IDs in a list.
-        # So, if we get here we need to make sure that the correct types are paseed.
-        #
-        # If the item in self.queue[site_base] is a dict, do nothing.
-        # If the item in self.queue[site_base] is not a dict but is a list or a tuple, then convert to dict and
-        # then replace the list or tuple with the dict.
-        # If the item in self.queue[site_base] is neither a dict or a list, then explode.
-        if type(self.queue[site_base]) is dict:
-            pass
-        elif type(self.queue[site_base]) is not dict and type(self.queue[site_base]) in [list, tuple]:
-            post_list_dict = {}
-            for post_list_id in self.queue[site_base]:
-                post_list_dict[post_list_id] = None
-            self.queue[site_base] = post_list_dict
-        else:
-            raise TypeError("A non-iterable is in the queue item for a given site, this will cause errors!")
+        with self.queue_modify_lock:
+            if site_base not in self.queue:
+                self.queue[site_base] = {}
+
+            # Something about how the queue is being filled is storing Post IDs in a list.
+            # So, if we get here we need to make sure that the correct types are paseed.
+            #
+            # If the item in self.queue[site_base] is a dict, do nothing.
+            # If the item in self.queue[site_base] is not a dict but is a list or a tuple, then convert to dict and
+            # then replace the list or tuple with the dict.
+            # If the item in self.queue[site_base] is neither a dict or a list, then explode.
+            if type(self.queue[site_base]) is dict:
+                pass
+            elif type(self.queue[site_base]) is not dict and type(self.queue[site_base]) in [list, tuple]:
+                post_list_dict = {}
+                for post_list_id in self.queue[site_base]:
+                    post_list_dict[post_list_id] = None
+                self.queue[site_base] = post_list_dict
+            else:
+                raise TypeError("A non-iterable is in the queue item for a given site, this will cause errors!")
 
-        # This line only works if we are using a dict in the self.queue[site_base] object, which we should be with
-        # the previous conversion code.
-        self.queue[site_base][str(post_id)] = datetime.utcnow()
-        self.queue_modify_lock.release()
+            # This line only works if we are using a dict in the self.queue[site_base] object, which we should be with
+            # the previous conversion code.
+            self.queue[site_base][str(post_id)] = datetime.utcnow()
 
         if GlobalVars.flovis is not None:
             GlobalVars.flovis.stage('bodyfetcher/enqueued', site_base, post_id,
@@ -114,9 +113,8 @@ def check_queue(self):
                 return
 
         # We're not making an API request, so explicitly store the queue
-        self.queue_modify_lock.acquire()
-        store_bodyfetcher_queue()
-        self.queue_modify_lock.release()
+        with self.queue_modify_lock:
+            store_bodyfetcher_queue()
 
     def print_queue(self):
         return '\n'.join(["{0}: {1}".format(key, str(len(values))) for (key, values) in self.queue.items()])
@@ -125,10 +123,9 @@ def make_api_call_for_site(self, site):
         if site not in self.queue:
             return
 
-        self.queue_modify_lock.acquire()
-        new_posts = self.queue.pop(site)
-        store_bodyfetcher_queue()
-        self.queue_modify_lock.release()
+        with self.queue_modify_lock:
+            new_posts = self.queue.pop(site)
+            store_bodyfetcher_queue()
 
         new_post_ids = [int(k) for k in new_posts.keys()]
 
@@ -137,51 +134,47 @@ def make_api_call_for_site(self, site):
                 GlobalVars.flovis.stage('bodyfetcher/api_request', site, post_id,
                                         {'site': site, 'posts': list(new_posts.keys())})
 
-        self.queue_timing_modify_lock.acquire()
-        post_add_times = [v for k, v in new_posts.items()]
-        pop_time = datetime.utcnow()
-
-        for add_time in post_add_times:
-            try:
-                seconds_in_queue = (pop_time - add_time).total_seconds()
-                if site in self.queue_timings:
-                    self.queue_timings[site].append(seconds_in_queue)
-                else:
-                    self.queue_timings[site] = [seconds_in_queue]
-            except KeyError:  # XXX: Any other possible exception?
-                continue  # Skip to next item if we've got invalid data or missing values.
+        with self.queue_timing_modify_lock:
+            post_add_times = [v for k, v in new_posts.items()]
+            pop_time = datetime.utcnow()
 
-        store_queue_timings()
+            for add_time in post_add_times:
+                try:
+                    seconds_in_queue = (pop_time - add_time).total_seconds()
+                    if site in self.queue_timings:
+                        self.queue_timings[site].append(seconds_in_queue)
+                    else:
+                        self.queue_timings[site] = [seconds_in_queue]
+                except KeyError:  # XXX: Any other possible exception?
+                    continue  # Skip to next item if we've got invalid data or missing values.
 
-        self.queue_timing_modify_lock.release()
-        self.max_ids_modify_lock.acquire()
+            store_queue_timings()
 
-        if site in self.previous_max_ids and max(new_post_ids) > self.previous_max_ids[site]:
-            previous_max_id = self.previous_max_ids[site]
-            intermediate_posts = range(previous_max_id + 1, max(new_post_ids))
+        with self.max_ids_modify_lock:
+            if site in self.previous_max_ids and max(new_post_ids) > self.previous_max_ids[site]:
+                previous_max_id = self.previous_max_ids[site]
+                intermediate_posts = range(previous_max_id + 1, max(new_post_ids))
 
-            # We don't want to go over the 100-post API cutoff, so take the last
-            # (100-len(new_post_ids)) from intermediate_posts
+                # We don't want to go over the 100-post API cutoff, so take the last
+                # (100-len(new_post_ids)) from intermediate_posts
 
-            intermediate_posts = intermediate_posts[(100 - len(new_post_ids)):]
+                intermediate_posts = intermediate_posts[(100 - len(new_post_ids)):]
 
-            # new_post_ids could contain edited posts, so merge it back in
-            combined = chain(intermediate_posts, new_post_ids)
+                # new_post_ids could contain edited posts, so merge it back in
+                combined = chain(intermediate_posts, new_post_ids)
 
-            # Could be duplicates, so uniquify
-            posts = list(set(combined))
-        else:
-            posts = new_post_ids
+                # Could be duplicates, so uniquify
+                posts = list(set(combined))
+            else:
+                posts = new_post_ids
 
-        try:
-            if max(new_post_ids) > self.previous_max_ids[site]:
+            try:
+                if max(new_post_ids) > self.previous_max_ids[site]:
+                    self.previous_max_ids[site] = max(new_post_ids)
+                    store_bodyfetcher_max_ids()
+            except KeyError:
                 self.previous_max_ids[site] = max(new_post_ids)
                 store_bodyfetcher_max_ids()
-        except KeyError:
-            self.previous_max_ids[site] = max(new_post_ids)
-            store_bodyfetcher_max_ids()
-
-        self.max_ids_modify_lock.release()
 
         log('debug', "New IDs / Hybrid Intermediate IDs for {}:".format(site))
         if len(new_post_ids) > 30:
@@ -224,8 +217,7 @@ def make_api_call_for_site(self, site):
         # wait to make sure API has/updates post data
         time.sleep(3)
 
-        GlobalVars.api_request_lock.acquire()
-        try:
+        with GlobalVars.api_request_lock:
             # Respect backoff, if we were given one
             if GlobalVars.api_backoff_time > time.time():
                 time.sleep(GlobalVars.api_backoff_time - time.time() + 2)
@@ -235,22 +227,18 @@ def make_api_call_for_site(self, site):
             except (requests.exceptions.Timeout, requests.ConnectionError, Exception):
                 # Any failure in the request being made (timeout or otherwise) should be added back to
                 # the queue.
-                self.queue_modify_lock.acquire()
-                if site in self.queue:
-                    self.queue[site].update(new_posts)
-                else:
-                    self.queue[site] = new_posts
-                self.queue_modify_lock.release()
-                GlobalVars.api_request_lock.release()
+                with self.queue_modify_lock:
+                    if site in self.queue:
+                        self.queue[site].update(new_posts)
+                    else:
+                        self.queue[site] = new_posts
                 return
 
-            self.api_data_lock.acquire()
-            add_or_update_api_data(site)
-            self.api_data_lock.release()
+            with self.api_data_lock:
+                add_or_update_api_data(site)
 
             message_hq = ""
-            GlobalVars.apiquota_rw_lock.acquire()
-            try:
+            with GlobalVars.apiquota_rw_lock:
                 if "quota_remaining" in response:
                     if response["quota_remaining"] - GlobalVars.apiquota >= 5000 and GlobalVars.apiquota >= 0:
                         tell_rooms_with("debug", "API quota rolled over with {0} requests remaining. "
@@ -276,8 +264,6 @@ def make_api_call_for_site(self, site):
                     GlobalVars.apiquota = response["quota_remaining"]
                 else:
                     message_hq = "The quota_remaining property was not in the API response."
-            finally:
-                GlobalVars.apiquota_rw_lock.release()
 
             if "error_message" in response:
                 message_hq += " Error: {} at {} UTC.".format(response["error_message"], time_request_made)
@@ -291,9 +277,6 @@ def make_api_call_for_site(self, site):
                 if GlobalVars.api_backoff_time < time.time() + response["backoff"]:
                     GlobalVars.api_backoff_time = time.time() + response["backoff"]
 
-        finally:
-            GlobalVars.api_request_lock.release()
-
         if len(message_hq) > 0 and "site is required" not in message_hq:
             tell_rooms_with("debug", message_hq.strip())
 
@@ -390,8 +373,7 @@ def make_api_call_for_site(self, site):
                 log('error', "Exception handling answers:", e)
 
         end_time = time.time()
-        GlobalVars.posts_scan_stats_lock.acquire()
-        GlobalVars.num_posts_scanned += num_scanned
-        GlobalVars.post_scan_time += end_time - start_time
-        GlobalVars.posts_scan_stats_lock.release()
+        with GlobalVars.posts_scan_stats_lock:
+            GlobalVars.num_posts_scanned += num_scanned
+            GlobalVars.post_scan_time += end_time - start_time
         return