reduce idle cpu usage

Signed-off-by: Praneeth Bedapudi <[email protected]>

fastdeploy/_loop.py

@@ -85,6 +85,7 @@ def fetch_batch(
     input_batch = []
     current_batch_length = 0
     batch_collection_started_at = time.time()
+    last_input_received_at = time.time()
 
     while current_batch_length < optimal_batch_size:
         to_process = main_index.search(
@@ -106,15 +107,17 @@ def fetch_batch(
 
         for unique_id, data in to_process.items():
             outputs = data[f"{predictor_sequence - 1}.outputs"]
-
             input_count = len(outputs)
-
             unique_id_wise_input_count[unique_id] = input_count
             input_batch.extend(outputs)
             current_batch_length += input_count
+            last_input_received_at = time.time()
 
         if current_batch_length == 0:
-            time.sleep(max_wait_time_for_batch_collection / 2)
+            if time.time() - last_input_received_at > 5:
+                time.sleep(0.05)
+            else:
+                time.sleep(max_wait_time_for_batch_collection / 2)
             continue
 
         elif (

fastdeploy/_rest.py

@@ -63,6 +63,7 @@ def deregister_request(self, unique_id):
             self.pending_requests.pop(unique_id, None)
 
     def _response_checker(self):
+        last_input_received_at = time.time()
         while True:
             try:
                 unique_ids = []
@@ -73,6 +74,11 @@ def _response_checker(self):
                         unique_ids.append(uid)
                         is_compresseds.append(data["is_compressed"])
                         input_types.append(data["input_type"])
+                        last_input_received_at = data["timestamp"]
+
+                if not unique_ids and (time.time() - last_input_received_at) > 5:
+                    time.sleep(0.05)
+                    continue
 
                 if unique_ids:
                     _utils.logger.debug(

fastdeploy/_utils.py

@@ -114,7 +114,9 @@
 MAIN_INDEX.optimize_for_query(["timedout_in_queue"])
 
 
-GLOBAL_METRICS_INDEX = KVIndex(os.path.join("fastdeploy_dbs", f"global_metrics_index.db"))
+GLOBAL_METRICS_INDEX = KVIndex(
+    os.path.join("fastdeploy_dbs", f"global_metrics_index.db")
+)
 GLOBAL_METRICS_INDEX["total_predictor_run_for_hours"] = 0
 GLOBAL_METRICS_INDEX["total_predictor_up_for_hours"] = 0
 

setup.py

@@ -18,7 +18,7 @@
 EMAIL = "[email protected]"
 AUTHOR = "BEDAPUDI PRANEETH"
 REQUIRES_PYTHON = ">=3.6.0"
-VERSION = "3.0.27"
+VERSION = "3.0.28"
 
 # What packages are required for this module to be executed?
 REQUIRED = ["falcon", "liteindex==0.0.3.2.dev4", "zstandard", "gunicorn[gevent]", "msgpack"]