gp2Scale immrovements

fvgp/gp2Scale.py

@@ -120,8 +120,8 @@ def compute_covariance(self, hyperparameters,client):
 
                 ####collect finished workers but only if actor is not busy, otherwise do it later
                 if len(finished_futures) >= 2000:
-                    #actor_futures.append(self.SparsePriorCovariance.get_future_results(set(finished_futures), info = self.info))
                     actor_futures.append(self.SparsePriorCovariance.get_future_results(finished_futures.copy(), info = self.info))
+                    actor_futures = self.clean_actor_futures(actor_futures)
                     finished_futures = set()
 
                 #get idle worker and submit work
@@ -141,7 +141,6 @@ def compute_covariance(self, hyperparameters,client):
 
         actor_futures.append(self.SparsePriorCovariance.get_future_results(finished_futures.union(futures), info = self.info))
         client.gather(actor_futures)
-        #actor_futures[-1].result()
 
         #########
         if self.info:
@@ -162,6 +161,13 @@ def free_workers(self, futures, finished_futures):
         del futures
         return free_workers, remaining_futures, finished_futures
 
+    def clean_actor_futures(self,actor_futures):
+        for entry in reversed(actor_futures):
+            if entry.status == "finished":
+                actor_futures.remove(entry)
+        return actor_futures
+
+
     def assign_future_2_worker(self, future_key, worker_address):
         self.future_worker_assignments[future_key] = worker_address
 

fvgp/sparse_matrix.py

@@ -51,6 +51,7 @@ def get_future_results(self, futures, info = False):
 
         self.insert_many(res)
         if info: print("    Size of the current covariance matrix: ", self.K.count_nonzero(), flush = True)
+        del futures
         return 0
 
     """