From 572ac4a17be8ad59f3d9c0e6bfc465b2dac9c88e Mon Sep 17 00:00:00 2001
From: blitzmann <holmes.ryan.90@gmail.com>
Date: Thu, 27 Aug 2020 21:44:28 -0400
Subject: [PATCH] Fix for deleting old, completed tasks

(cherry picked from commit e066d7a4b0f47d3327696a11795cdc923ff0f6f3)
---
 cps/helper.py          |  2 +-
 cps/services/worker.py | 51 ++++++++++++++++++++++++++++++------------
 2 files changed, 38 insertions(+), 15 deletions(-)

diff --git a/cps/helper.py b/cps/helper.py
index 4f05c48a..624a3b54 100644
--- a/cps/helper.py
+++ b/cps/helper.py
@@ -730,7 +730,7 @@ def format_runtime(runtime):
 # helper function to apply localize status information in tasklist entries
 def render_task_status(tasklist):
     renderedtasklist = list()
-    for user, added, task in tasklist:
+    for num, user, added, task in tasklist:
         if user == current_user.nickname or current_user.role_admin():
             ret = {}
             if task.start_time:
diff --git a/cps/services/worker.py b/cps/services/worker.py
index bb3e3b07..a22b9a71 100644
--- a/cps/services/worker.py
+++ b/cps/services/worker.py
@@ -8,7 +8,7 @@ try:
     import queue
 except ImportError:
     import Queue as queue
-from datetime import datetime
+from datetime import datetime, timedelta
 from collections import namedtuple
 
 from cps import calibre_db
@@ -23,9 +23,9 @@ STAT_STARTED = 2
 STAT_FINISH_SUCCESS = 3
 
 # Only retain this many tasks in dequeued list
-TASK_CLEANUP_TRIGGER = 20
+TASK_CLEANUP_TRIGGER = 5
 
-QueuedTask = namedtuple('QueuedTask', 'user, added, task')
+QueuedTask = namedtuple('QueuedTask', 'num, user, added, task')
 
 
 def _get_main_thread():
@@ -62,13 +62,15 @@ class WorkerThread(threading.Thread):
 
         self.doLock = threading.Lock()
         self.queue = ImprovedQueue()
-
+        self.num = 0
         self.start()
 
     @classmethod
     def add(cls, user, task):
         ins = cls.getInstance()
+        ins.num += 1
         ins.queue.put(QueuedTask(
+            num=ins.num,
             user=user,
             added=datetime.now(),
             task=task,
@@ -77,8 +79,9 @@ class WorkerThread(threading.Thread):
     @property
     def tasks(self):
         with self.doLock:
-            tasks = list(self.queue.to_list()) + self.dequeued
-        return sorted(tasks, key=lambda x: x.added)
+            tasks = self.queue.to_list() + self.dequeued
+            return sorted(tasks, key=lambda x: x.num)
+
 
     # Main thread loop starting the different tasks
     def run(self):
@@ -86,23 +89,34 @@ class WorkerThread(threading.Thread):
         while main_thread.is_alive():
             item = self.queue.get()
 
-            # add to list so that in-progress tasks show up
             with self.doLock:
-                # Remove completed tasks if needed
+                # once we hit our trigger, start cleaning up dead tasks
                 if len(self.dequeued) > TASK_CLEANUP_TRIGGER:
-                    # sort first (just to be certain), then lob off the extra
-                    self.dequeued = sorted(self.dequeued, key=lambda x: x.added)[-1 * TASK_CLEANUP_TRIGGER:]
+                    dead = []
+                    alive = []
+                    for x in self.dequeued:
+                        (dead if x.task.dead else alive).append(x)
+
+                    # if the ones that we need to keep are within the trigger, do nothing else
+                    delta = len(self.dequeued) - len(dead)
+                    if delta > TASK_CLEANUP_TRIGGER:
+                        ret = alive
+                    else:
+                        # otherwise, lop off the oldest dead tasks until we hit the target trigger
+                        ret = sorted(dead, key=lambda x: x.task.end_time)[-TASK_CLEANUP_TRIGGER:] + alive
+
+                    self.dequeued = sorted(ret, key=lambda x: x.num)
+                # add to list so that in-progress tasks show up
                 self.dequeued.append(item)
 
-            user, added, task = item
-
             # sometimes tasks (like Upload) don't actually have work to do and are created as already finished
-            if task.stat is STAT_WAITING:
+            if item.task.stat is STAT_WAITING:
                 # CalibreTask.start() should wrap all exceptions in it's own error handling
-                task.start(self)
+                item.task.start(self)
 
             self.queue.task_done()
 
+
 class CalibreTask:
     __metaclass__ = abc.ABCMeta
 
@@ -168,6 +182,15 @@ class CalibreTask:
     def runtime(self):
         return (self.end_time or datetime.now()) - self.start_time
 
+    @property
+    def dead(self):
+        """Determines whether or not this task can be garbage collected
+
+        We have a separate dictating this because there may be certain tasks that want to override this
+        """
+        # By default, we're good to clean a task if it's "Done"
+        return self.stat in (STAT_FINISH_SUCCESS, STAT_FAIL)
+
     @progress.setter
     def progress(self, x):
         # todo: throw error if outside of [0,1]