Bound usage sharing cleanup at five seconds and take fifty as the minimum entries per message

pipeline/fiftyone_pipeline_engines_fiftyone/src/fiftyone_pipeline_engines_fiftyone/share_usage.py

@@ -58,10 +58,22 @@
 # The default number of seconds to wait for the server connection before failing the request when sending usage data
 REQUEST_TIMEOUT = 60
 
+# The default number of entries that must be queued before a message is
+# started. A message holds from this to twice this number.
+REQUESTED_PACKAGE_SIZE_DEFAULT = 50
+
 # The default maximum number of entries waiting to be sent, unless ten
 # times the requested package size is larger.
 MAX_QUEUE_SIZE_DEFAULT = 1000
 
+# The time in seconds that closing may spend sending the entries still
+# queued. Closing stops after this, so it does not hold up the process that
+# closes it. It is not a user setting.
+CLEANUP_TIMEOUT_SECONDS = 5
+
+# How long closing waits between looks at a send that is already running.
+CLEANUP_POLL_SECONDS = 0.01
+
 # While the queue is full, a message is logged for the first entry discarded
 # and then once for every this many more, never one for each entry.
 QUEUE_FULL_LOG_EVERY_ENTRIES = 1000
@@ -87,7 +99,7 @@ class ShareUsage(Engine):
     def __init__(
         self,
         interval = 1200,
-        requested_package_size = 10,
+        requested_package_size = REQUESTED_PACKAGE_SIZE_DEFAULT,
         cookie = None,
         query_whitelist = [],
         header_blacklist = [],
@@ -110,10 +122,11 @@ def __init__(
         multiple times within this time limit (in seconds) then they 
         will only be shared once.
         @type requested_package_size: int
-        @param requested_package_size: The usage element will group data into 
-        single requests before sending it. This setting controls the minimum 
-        number of entries before data is sent. If you are sharing large 
-        amounts of data, increasing this value is recommended in order to 
+        @param requested_package_size: The usage element will group data into
+        single requests before sending it. This setting controls the minimum
+        number of entries before data is sent, and a message holds from this
+        to twice this number. 50 by default. If you are sharing large
+        amounts of data, increasing this value is recommended in order to
         reduce the overhead of sending HTTP messages.
         For example, the 51Degrees cloud service uses a value of 2500.
         @type cookie: string
@@ -238,6 +251,9 @@ def __init__(
         self._resume_timer = None
         # Set when the element closes.
         self._closing = threading.Event()
+        # The time closing may spend sending what is still queued. Not a
+        # user setting, and here so that a test can shorten it.
+        self.cleanup_timeout_seconds = CLEANUP_TIMEOUT_SECONDS
 
         self.constant_xml = None
 
@@ -252,7 +268,11 @@ def __init__(
     def __del__(self):
         thread_pool = getattr(self, "thread_pool", None)
         if thread_pool is not None:
-            thread_pool.shutdown(wait=True)
+            closing = getattr(self, "_closing", None)
+            # Closing has already given sending all the time it is allowed,
+            # so waiting again here would undo that bound.
+            thread_pool.shutdown(
+                wait=closing is None or not closing.is_set())
 
     def close(self):
 
@@ -262,6 +282,11 @@ def close(self):
         when it is closed. Closing does not wait for a pause after failed
         sends, and sends nothing while paused.
 
+        Closing gives up after cleanup_timeout_seconds, discards the entries
+        still queued and logs a warning, so that it does not hold up the
+        process that closes it. A message already going out when that time
+        passes is left to the request timeout on its own thread.
+
         """
 
         super(ShareUsage, self).close()
@@ -270,36 +295,84 @@ def close(self):
             if self._resume_timer is not None:
                 self._resume_timer.cancel()
                 self._resume_timer = None
-        self.thread_pool.shutdown(wait=True)
-        self._send_remaining()
+        # Sending runs on its own thread, so that a service which does not
+        # answer cannot hold closing up for longer than the bound.
+        deadline = time.monotonic() + self.cleanup_timeout_seconds
+        # Set when sending stops for a reason of its own rather than for
+        # want of time, so that a thread which stops exactly at the
+        # deadline is still counted as having run out of time.
+        done = threading.Event()
+        flush = threading.Thread(
+            target=self._send_remaining, args=(deadline, done),
+            name="ShareUsageClose")
+        # The thread does not keep the process running.
+        flush.daemon = True
+        flush.start()
+        flush.join(self.cleanup_timeout_seconds)
+        if not done.is_set():
+            with self._lock:
+                queued = len(self.share_data)
+            self._log(
+                "warning",
+                "Usage sharing stopped sending when closing down, because "
+                "it did not finish within "
+                f"{self.cleanup_timeout_seconds} seconds. {queued} entries "
+                "were left in the queue and discarded.")
+        self.thread_pool.shutdown(wait=False)
 
-    def _send_remaining(self):
+    def _send_remaining(self, deadline, done):
         """!
         Send the entries still queued when closing, even if there are fewer
-        than the requested package size. Nothing is sent while paused, or
-        after a message fails, and the entries left are discarded and
-        counted in the log.
+        than the requested package size. A send already running finishes
+        first, so only one message goes out at a time. Nothing is sent while
+        paused, after a message fails, or once deadline has passed, and the
+        entries left are discarded. Closing logs the ones left when the time
+        ran out, and this logs the ones left for the other two reasons.
+
+        @type deadline: float
+        @param deadline: The monotonic time sending has to stop at.
+        @type done: threading.Event
+        @param done: Set when sending stops for a reason other than the
+        deadline.
         """
 
+        while time.monotonic() < deadline:
+            with self._lock:
+                if not self._sending:
+                    break
+            time.sleep(CLEANUP_POLL_SECONDS)
         while True:
+            # None while sending is paused, so nothing is sent.
+            share_data = None
             with self._lock:
                 queued = len(self.share_data)
+                if time.monotonic() >= deadline:
+                    return
                 if queued == 0:
+                    done.set()
                     return
-                if self.backoff.remaining(time.monotonic()) > 0:
-                    break
-                share_data = self.share_data[:self._message_size]
-                del self.share_data[:self._message_size]
-            data = devices_message(share_data)
-            if not self.send_thread(self.endpoint, data):
+                if self.backoff.remaining(time.monotonic()) == 0:
+                    share_data = self.share_data[:self._message_size]
+                    del self.share_data[:self._message_size]
+            if share_data is None:
+                self._log(
+                    "info",
+                    f"{queued} usage sharing entries were not sent, as "
+                    "sending was paused after failed sends when usage "
+                    "sharing closed.")
+                done.set()
+                return
+            if not self.send_thread(self.endpoint, devices_message(
+                    share_data)):
                 with self._lock:
                     queued = len(self.share_data)
-                break
-        if queued > 0:
-            self._log(
-                "info",
-                f"{queued} usage sharing entries were not sent, as sending "
-                "had failed when usage sharing closed.")
+                if queued > 0:
+                    self._log(
+                        "info",
+                        f"{queued} usage sharing entries were not sent, as "
+                        "sending had failed when usage sharing closed.")
+                done.set()
+                return
 
     def get_constant_xml(self):
         if self.constant_xml is None:
@@ -462,7 +535,7 @@ def _send_messages(self):
                     pause = self.backoff.failed(time.monotonic())
                     if pause > 0:
                         self._schedule_resume(pause)
-                done = pause > 0 or \
+                done = pause > 0 or self._closing.is_set() or \
                     len(self.share_data) < self.requested_package_size
                 if done:
                     self._sending = False

pipeline/fiftyone_pipeline_engines_fiftyone/tests/test_usage.py

@@ -43,6 +43,12 @@
 
 shareUsageVersion = "1.1"
 
+# The numbers the specification gives, written out here rather than read
+# from the element, so that a test fails when the element does not keep to
+# them.
+CLEANUP_BOUND_SECONDS = 5
+MINIMUM_ENTRIES_PER_MESSAGE = 50
+
 
 class ReceiverThread(threading.Thread):
 
@@ -69,10 +75,16 @@ def get_received(self):
 app.status = 200
 # The content the receiver answers with.
 app.body = ""
+# How long in seconds the receiver holds a request before answering.
+app.delay = 0
+# Set to let a held request go at once.
+app.delay_release = threading.Event()
 
 @app.route('/', methods=['POST'])
 def server():
     app.attempts.append(time.monotonic())
+    if app.delay:
+        app.delay_release.wait(app.delay)
     data = gzip.decompress(request.data).decode("utf8")
     if app.status == 200:
         app.received.append(data)
@@ -103,6 +115,8 @@ def setUp(self):
         app.attempts = []
         app.status = 200
         app.body = ""
+        app.delay = 0
+        app.delay_release.clear()
         self.requests = 0
         self.logs = []
 
@@ -393,6 +407,88 @@ def test_close_sends_fewer_entries_than_the_package_size(self):
 
         self.assertEqual(ips, client_ips(app.received))
 
+    def test_close_stops_sending_after_the_cleanup_bound(self):
+        """!
+        Closing stops after the five seconds the specification allows,
+        rather than waiting for a service that does not answer, and logs a
+        warning counting the entries it discarded.
+        """
+
+        app.delay = CLEANUP_BOUND_SECONDS + 5
+        # Let the held request go as soon as the test is done with it.
+        self.addCleanup(app.delay_release.set)
+        usage = ShareUsage(
+            requested_package_size=1,
+            endpoint=testEndpoint)
+        usage._log = lambda level, message: self.logs.append(
+            (level, message))
+        pipeline = PipelineBuilder().add(usage).build()
+        self.process_unique(pipeline)
+        self.wait_for(
+            lambda: len(app.attempts) == 1,
+            lambda: "The first send was not made, so nothing was holding "
+            "sending up when closing.")
+        for i in range(5):
+            self.process_unique(pipeline)
+
+        start = time.monotonic()
+        usage.close()
+        took = time.monotonic() - start
+
+        self.assertLess(
+            took, CLEANUP_BOUND_SECONDS + 2,
+            f"Closing took {took:.2f} seconds, so it waited for the service "
+            "rather than stopping at the bound.")
+        self.assertGreaterEqual(
+            took, CLEANUP_BOUND_SECONDS - 0.5,
+            f"Closing took {took:.2f} seconds, so it gave up before the "
+            "bound.")
+        warnings = [log for log in self.logs if "did not finish" in log[1]]
+        self.assertEqual(
+            1, len(warnings),
+            f"The entries discarded were not logged once. Logged: "
+            f"{self.logs}")
+        self.assertEqual(
+            ("warning",
+             "Usage sharing stopped sending when closing down, because it "
+             f"did not finish within {CLEANUP_BOUND_SECONDS} seconds. 5 "
+             "entries were left in the queue and discarded."),
+            warnings[0])
+
+        # Let the held message go and wait for it to arrive, so that
+        # nothing from this element lands in another test.
+        app.delay_release.set()
+        self.wait_for(
+            lambda: len(app.received) == 1,
+            lambda: "The held message did not arrive once it was let go.")
+
+    def test_minimum_entries_per_message_default(self):
+        """!
+        The default minimum entries per message is the 50 the specification
+        gives, so no message is started before 50 entries are queued.
+        """
+
+        usage = ShareUsage(endpoint=testEndpoint)
+        pipeline = PipelineBuilder().add(usage).build()
+        self.addCleanup(usage.close)
+        self.assertEqual(
+            MINIMUM_ENTRIES_PER_MESSAGE, usage.requested_package_size,
+            "The default is not the 50 the specification gives.")
+
+        for i in range(MINIMUM_ENTRIES_PER_MESSAGE - 1):
+            self.process_unique(pipeline)
+        sleep(0.5)
+        self.assertEqual(
+            0, len(app.attempts),
+            "A message was started before 50 entries were queued.")
+
+        self.process_unique(pipeline)
+        self.wait_for(
+            lambda: len(client_ips(app.received))
+            == MINIMUM_ENTRIES_PER_MESSAGE,
+            lambda: "The fiftieth entry did not start a message. Sent: "
+            f"{len(client_ips(app.received))}")
+
     def test_back_off_invalid_settings(self):
         """!
         Settings the pause cannot work with are rejected.