[squashme] Review feedback, typos and code formatting

Author: mutax

networking_nsxv3/common/synchronization.py

@@ -109,7 +109,7 @@ def __init__(self, fnparams: CBPARAMS, fn: Callable[[CBPARAMS], None], priority=
 
     @property
     def identifier(self) -> tuple:
-        return (self.idn, self._fn.__name__)
+        return self.idn, self._fn.__name__
 
     def debugid(self)->str:
         return str((self.idn, self._fn.__name__, str(self._fnparams)))
@@ -183,12 +183,12 @@ def __eq__(self, other):
         existing element, in case it was lower than the item that was about to be added.
         """
         if isinstance(other, Runnable):
-            return (self._fnparams  == other._fnparams  and self._fn == other._fn)
+            return self._fnparams  == other._fnparams  and self._fn == other._fn
         else:
             return False
 
     def __ne__(self, other):
-        return (not self.__eq__(other))
+        return not self.__eq__(other)
 
     def __lt__(self, other):
         """ Order Runnable by their priority
@@ -198,7 +198,7 @@ def __lt__(self, other):
         # if the priority is equal, we want to order
         # by creation time to handle oldest jobs first
         if self.priority == other.priority:
-            return self._created < other._created
+            return self._created < other._created # noqa
 
         return self.priority < other.priority
 
@@ -281,6 +281,9 @@ class JobList():
 
         There can be multiple similar jobs (same id, same method) but with different parameters, then parameter
         is a dict and not a string with only the OpenStack ID.
+
+        These calls get a dict instead of a str: {enable, disable, update]_policy_logging, address_group_update
+
         In these cases this JobList will keep track of the jobs, merging jobs that are identical (same parameters)
         into one, to prevent unneccessary re-executions, but keeping them separate otherwise to not drop them.
 
@@ -309,12 +312,14 @@ class JobList():
         done will then choose a job from the list, that is supposed to run again,
         remove it from the list and return it to the JobRerunner.
 
-        This might be the same job that was just finished or it could be a different one.
+        This might be exactly the same job that was just finished (if it needs rerunning) or
+        it could be a different one, that is: same openstack id, different parameter dict.
+
         For now we will choose the oldest one based on age, which should be the same job that
         was just done, but we might want to change that so we use a helper function for that
-        for now in the POC.
-
+        for now in the POC. The current helper uses pop() so our list basically is a FiFo.
     """
+
     def __init__(self):
         self._job_identifier: Optional[str] = None
         self._runnables: List[tuple[int, Runnable]] = []
@@ -324,7 +329,7 @@ def __len__(self):
 
     @property
     def size(self):
-        return sum( count for count, _ in self._runnables)
+        return sum(count for count, _ in self._runnables)
 
     def empty(self):
         return self.size == 0
@@ -342,7 +347,7 @@ def add(self, job:Runnable)->bool:
 
         See Class documentation for more details.
 
-        @return: True if the same job was not present yet, False if one already existed
+        returns True if the same job was not present yet, False if one already existed
         """
 
         if self._job_identifier is not None:
@@ -354,14 +359,18 @@ def add(self, job:Runnable)->bool:
         # search through our list and update the counter or append the job:
         for index, (count, existing_job) in enumerate(self._runnables):
             if job == existing_job:
+                if count < 1:
+                    # fix the list, otherwise we would never run that job.
+                    LOG.error("Joblist counter for job %s is %d, indicating job should have been removed.", job, count)
+                    count = 0
                 count += 1
                 self._runnables[index] = (count, existing_job)
-                if count > 1:
-                    # in case the list was broken and we had a job with count == 0
-                    # in it, we do not want to block and prevent the job from getting
-                    # started!
-                    return False
-                return True
+                if count == 1:
+                    # failsafe:
+                    # return True if this is the first time the job has been added,
+                    # only happens at this point if our bookkeeping was off.
+                    return True
+                return False
 
         # No match found, this is the first of its kind, we can run it.
         # note that a job that gets re-executed will be removed from
@@ -372,7 +381,7 @@ def add(self, job:Runnable)->bool:
         return True
 
     def _runnable_is_done(self, job:Runnable):
-        # search through our list and update the counter or remove the job:
+        """ search through our list and update the counter or remove the job """
         for index, (count, existing_job) in enumerate(self._runnables):
             if job == existing_job:
                 # we do not need this job with this parameters again, it is done,
@@ -385,6 +394,8 @@ def _runnable_is_done(self, job:Runnable):
                         LOG.warning("Job count in JobList was %d for %s", count, job.debugid())
                     del self._runnables[index]
                     return
+                # leave job in the list, so it will get retrieved again later.
+                # when retrieving we use pop() so the list gets cleared then.
                 self._runnables[index] = (count, existing_job)
                 return
 
@@ -394,7 +405,7 @@ def _runnable_is_done(self, job:Runnable):
 
 
     def _runnable_pop_next(self) -> Optional[Runnable]:
-        """ find the next job to run and remove it from the list,
+        """ Find the next job to run and remove it from the list,
             or return None if there is None to run.
         """
         if not self._runnables:
@@ -409,11 +420,11 @@ def _runnable_pop_next(self) -> Optional[Runnable]:
         return job
 
     def done(self, job:Runnable)->Optional[Runnable]:
-        """ mark a job as done
+        """ Mark a job as done and return next job to run.
 
         Removes the job from the list, if there are jobs left in the list, will return the next one to run.
 
-        @return: A job to run next, or None if there are no jobs left to run.
+        returns a job to run next, or None if there are no jobs left to run.
         """
         if self._job_identifier is not None:
             if job.identifier != self._job_identifier:
@@ -423,7 +434,7 @@ def done(self, job:Runnable)->Optional[Runnable]:
         return self._runnable_pop_next()
 
     def get_count(self, job):
-        """ get how often this specific job had been requested to run while already running"""
+        """ Returns how often this specific job had been requested to run while already running """
 
         if self._job_identifier is not None:
             if job.identifier != self._job_identifier:
@@ -452,16 +463,16 @@ class JobRerunner():
     In this class we keep track of all jobs currently running in the workers.
 
     When taking a new job from the active queue, we check if this job is
-    already running. If this is not the case, we will add it to our book keeping
-    here and it will be sent to a worker, wrapped in a function that will call
-    the JobRunner on job completion to update the book keeping.
+    already running. If this is not the case, we will add it to our bookkeeping
+    here, and it will be sent to a worker, wrapped in a function that will call
+    the JobRunner on job completion to update the bookkeeping.
 
     If we find the job already running, the new job will be dropped, but a counter
     for the job will be increased to mark the job for re-execution once its done,
     in case a change to the object had occured while the worker was running.
 
-    When a worker is done with a job, it will notify us and we can either
-    remove the id from our book keeping or, if a new job was added in the meantime,
+    When a worker is done with a job, it will notify us, and we can either
+    remove the id from our bookkeeping or, if a new job was added in the meantime,
     mark it as ready for re-execution. Note that a job will intentionally only be
     re-executed once, independently of how often it was added while a worker was
     already running it.
@@ -495,6 +506,7 @@ def get_rerunnable(self) -> Runnable:
         return job
 
     def job_done(self, job: Runnable):
+        """ Marks job as done, update bookkeeping and if needed add job to the queue for re-execution """
         LOG.debug("JobRerunner job_done called for %s", job)
 
         with LockManager.get_lock(self._lname_running):
@@ -576,7 +588,7 @@ class Runner(object):
     def __init__(self, active_size=INFINITY, passive_size=INFINITY,
                  workers_size=1):
         # if queue_size is < 0, the queue size is infinite.
-        self._active = UniqFiFoQueue(active_size)
+        self._active = UniqFiFoQueue(maxsize=active_size)
         self._passive = UniqPriorityQueue(maxsize=passive_size)
         self._workers = eventlet.greenpool.GreenPool(size=workers_size)
         self._idle = workers_size