Merge pull request #1370 from quartata/master

Use asyncio Tasks

Author: ArtOfCode-

chatcommands.py

@@ -15,8 +15,8 @@
 from parsing import *
 from spamhandling import handle_spam
 from gitmanager import GitManager
+from tasks import Tasks
 import threading
-from threading import Thread
 import random
 import requests
 import os
@@ -687,10 +687,7 @@ def status():
 # noinspection PyIncorrectDocstring
 @command(privileged=True)
 def stopflagging():
-    t_metasmoke = Thread(name="stop_autoflagging", target=Metasmoke.stop_autoflagging,
-                         args=())
-    t_metasmoke.start()
-
+    Tasks.do(Metasmoke.stop_autoflagging)
     return "Request sent..."
 
 

classes/feedback.py

@@ -1,5 +1,5 @@
-import threading
 from metasmoke import Metasmoke
+from tasks import Tasks
 
 
 class Feedback:
@@ -18,9 +18,7 @@ def send(self, url, msg):
 
     @staticmethod
     def send_custom(type, url, msg):
-        threading.Thread(name="metasmoke feedback send on " + url,
-                         target=Metasmoke.send_feedback_for_post,
-                         args=(url, type, msg.owner.name, msg.owner.id, msg._client.host,)).start()
+        Tasks.do(Metasmoke.send_feedback_for_post, url, type, msg.owner.name, msg.owner.id, msg._client.host)
 
 
 TRUE_FEEDBACKS = {

deletionwatcher.py

@@ -6,13 +6,13 @@
 import websocket
 # noinspection PyPackageRequirements
 from bs4 import BeautifulSoup
-from threading import Thread
 from urllib.parse import urlparse
 import chatcommunicate
 import metasmoke
 from globalvars import GlobalVars
 import datahandling
 from parsing import fetch_post_id_and_site_from_url
+from tasks import Tasks
 
 
 # noinspection PyClassHasNoInit,PyBroadException,PyMethodParameters
@@ -52,9 +52,7 @@ def check_websocket_for_deletion(self, post_site_id, post_url, timeout):
             try:
                 a = ws.recv()
             except websocket.WebSocketTimeoutException:
-                t_metasmoke = Thread(name="metasmoke send deletion stats",
-                                     target=metasmoke.Metasmoke.send_deletion_stats_for_post, args=(post_url, False))
-                t_metasmoke.start()
+                Tasks.do(metasmoke.Metasmoke.send_deletion_stats_for_post, post_url, False)
                 return False
             if a is not None and a != "":
                 try:
@@ -66,18 +64,12 @@ def check_websocket_for_deletion(self, post_site_id, post_url, timeout):
                         d = json.loads(json.loads(a)["data"])
                 except:
                     continue
-                if d["a"] == "post-deleted" and str(d["qId"]) == question_id \
-                        and ((post_type == "answer" and "aId" in d and str(d["aId"]) == post_id) or
-                             post_type == "question"):
+                if d["a"] == "post-deleted" and str(d["qId"]) == question_id:
+                    if (post_type == "answer" and "aId" in d and str(d["aId"]) == post_id) or post_type == "question":
+                        Tasks.do(metasmoke.Metasmoke.send_deletion_stats_for_post, post_url, True)
+                        return True
 
-                    t_metasmoke = Thread(name="metasmoke send deletion stats",
-                                         target=metasmoke.Metasmoke.send_deletion_stats_for_post, args=(post_url, True))
-                    t_metasmoke.start()
-                    return True
-
-        t_metasmoke = Thread(name="metasmoke send deletion stats",
-                             target=metasmoke.Metasmoke.send_deletion_stats_for_post, args=(post_url, False))
-        t_metasmoke.start()
+        Tasks.do(metasmoke.Metasmoke.send_deletion_stats_for_post, post_url, False)
         return False
 
     @classmethod

metasmoke.py

@@ -67,7 +67,6 @@ def init_websocket():
 
     @staticmethod
     def check_last_pingtime():
-        threading.Timer(30, Metasmoke.check_last_pingtime).start()
         now = datetime.utcnow()
         errlog = open('errorLogs.txt', 'a', encoding="utf-8")
         if GlobalVars.metasmoke_last_ping_time is None:
@@ -265,7 +264,6 @@ def send_status_ping():
             log('info', "Metasmoke location not defined; not sending status ping")
             return
 
-        threading.Timer(60, Metasmoke.send_status_ping).start()
         metasmoke_key = GlobalVars.metasmoke_key
 
         try:
@@ -347,7 +345,7 @@ def stop_autoflagging():
                       data=json.dumps(payload), headers=headers)
 
     @staticmethod
-    def send_statistics(should_repeat=True):
+    def send_statistics():
         GlobalVars.posts_scan_stats_lock.acquire()
         if GlobalVars.post_scan_time != 0:
             posts_per_second = GlobalVars.num_posts_scanned / GlobalVars.post_scan_time
@@ -367,6 +365,3 @@ def send_statistics(should_repeat=True):
         if GlobalVars.metasmoke_host is not None:
             requests.post(GlobalVars.metasmoke_host + "/statistics.json",
                           data=json.dumps(payload), headers=headers)
-
-        if should_repeat:
-            threading.Timer(600, Metasmoke.send_statistics).start()

spamhandling.py

@@ -13,6 +13,7 @@
 import regex
 from classes import Post, PostParseError
 from helpers import log
+from tasks import Tasks
 
 
 # noinspection PyMissingTypeHints
@@ -127,12 +128,10 @@ def handle_spam(post, reasons, why):
                                                             post.user_name.strip(), poster_url, shortened_site)
             username = post.user_name.strip()
 
-        t_metasmoke = Thread(name="metasmoke send post",
-                             target=metasmoke.Metasmoke.send_stats_on_post,
-                             args=(post.title_ignore_type, post_url, reasons, post.body, username,
-                                   post.user_link, why, post.owner_rep, post.post_score,
-                                   post.up_vote_count, post.down_vote_count))
-        t_metasmoke.start()
+        Tasks.do(metasmoke.Metasmoke.send_stats_on_post,
+                 post.title_ignore_type, post_url, reasons, post.body, username,
+                 post.user_link, why, post.owner_rep, post.post_score,
+                 post.up_vote_count, post.down_vote_count)
 
         log('debug', GlobalVars.parser.unescape(s).encode('ascii', errors='replace'))
         datahandling.append_to_latest_questions(post.post_site, post.post_id, post.title)

tasks.py

@@ -0,0 +1,37 @@
+import asyncio
+import threading
+
+
+class Tasks:
+    loop = asyncio.new_event_loop()
+
+    @classmethod
+    def _run(cls):
+        try:
+            cls.loop.run_forever()
+        finally:
+            cls.loop.close()
+
+    @classmethod
+    def do(cls, func, *args, **kwargs):
+        cls.loop.call_soon(lambda: func(*args, **kwargs))
+        cls.loop._write_to_self()
+
+    @classmethod
+    def later(cls, func, *args, after=None, **kwargs):
+        cls.loop.call_later(after, lambda: func(*args, **kwargs))
+        cls.loop._write_to_self()
+
+    @classmethod
+    def periodic(cls, func, *args, interval=None, **kwargs):
+        @asyncio.coroutine
+        def f():
+            while True:
+                yield from asyncio.sleep(interval)
+                func(*args, **kwargs)
+
+        cls.loop.create_task(f())
+        cls.loop._write_to_self()
+
+
+threading.Thread(name="tasks", target=Tasks._run, daemon=True).start()

ws.py

@@ -16,7 +16,6 @@
 # noinspection PyPackageRequirements
 import websocket
 import getpass
-import threading
 from threading import Thread
 import traceback
 from bodyfetcher import BodyFetcher
@@ -34,6 +33,7 @@
 # noinspection PyPackageRequirements
 from tld.utils import update_tld_names, TldIOError
 from helpers import log
+from tasks import Tasks
 
 import chatcommands
 
@@ -121,26 +121,19 @@
 
 # noinspection PyProtectedMember
 def check_socket_connections():
-    while True:
-        time.sleep(90)
-
-        for client in chatcommunicate._clients.values():
-            if client.last_activity:
-                if (datetime.utcnow() - client.last_activity).total_seconds() >= 60:
-                    os._exit(10)
-
-
-Thread(name="check socket connections", target=check_socket_connections, daemon=True).start()
+    for client in chatcommunicate._clients.values():
+        if client.last_activity and (datetime.utcnow() - client.last_activity).total_seconds() >= 60:
+            os._exit(10)
 
 
 # noinspection PyProtectedMember
-def restart_automatically(time_in_seconds):
-    time.sleep(time_in_seconds)
-    Metasmoke.send_statistics(False)  # false indicates not to auto-repeat
+def restart_automatically():
+    Metasmoke.send_statistics()
     os._exit(1)
 
 
-Thread(name="auto restart thread", target=restart_automatically, args=(21600,)).start()
+Tasks.periodic(check_socket_connections, interval=90)
+Tasks.later(restart_automatically, after=21600)
 
 log('info', GlobalVars.location)
 log('info', GlobalVars.metasmoke_host)
@@ -155,14 +148,13 @@ def restart_automatically(time_in_seconds):
 elif "first_start" in sys.argv and not GlobalVars.on_master:
     chatcommunicate.tell_rooms_with("debug", GlobalVars.s_reverted)
 
-Metasmoke.send_status_ping()  # This will call itself every minute or so
-threading.Timer(600, Metasmoke.send_statistics).start()
+Tasks.periodic(Metasmoke.send_status_ping, interval=60)
+Tasks.periodic(Metasmoke.send_statistics, interval=600)
+Tasks.periodic(Metasmoke.check_last_pingtime, interval=30)
 
 metasmoke_ws_t = Thread(name="metasmoke websocket", target=Metasmoke.init_websocket)
 metasmoke_ws_t.start()
 
-Metasmoke.check_last_pingtime()  # This will call itself every 10 seconds or so
-
 while True:
     try:
         a = ws.recv()