hiro/hopsflow
1
0
Fork 0
mirror of https://github.com/vale981/hopsflow synced 2025-03-05 16:51:39 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

do not wait for gc if there is no gc sleep

Browse source
This commit is contained in:
Valentin Boettcher 2022-08-22 17:47:36 +02:00
parent b6512624cb
commit ff4391dcbe
1 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
hopsflow/util.py
View file

@ -96,7 +96,7 @@ class EnsembleValue:
N, val, σ = self.final_aggregate N, val, σ = self.final_aggregate
return EnsembleValue( return EnsembleValue(
[(N, val.mean().copy(), np.sqrt((σ.copy() ** 2).sum() / val.size**2))] [(N, val.mean().copy(), np.sqrt((σ.copy() ** 2).sum() / val.size ** 2))]
) )
@property @property
@ -605,7 +605,7 @@ def integrate_array(
if err is not None: if err is not None:
dt = t[1:] - t[:-1] dt = t[1:] - t[:-1]
err_sum = [ err_sum = [
np.concatenate(([0], np.cumsum(((e[1:] ** 2 + e[:-1] ** 2) / 4) * dt**2))) np.concatenate(([0], np.cumsum(((e[1:] ** 2 + e[:-1] ** 2) / 4) * dt ** 2)))
for e in err for e in err
] ]
err_integral = np.sqrt(err_sum).real err_integral = np.sqrt(err_sum).real
@ -718,7 +718,7 @@ def ensemble_mean(
overwrite_cache: bool = False, overwrite_cache: bool = False,
chunk_size: Optional[int] = None, chunk_size: Optional[int] = None,
in_flight: Optional[int] = None, in_flight: Optional[int] = None,
gc_sleep: float = 0.1, gc_sleep: float = 0,
) -> EnsembleValue: ) -> EnsembleValue:
results = [] results = []
@ -775,7 +775,6 @@ def ensemble_mean(
chunks = {} chunks = {}
in_flight = in_flight or int(ray.available_resources().get("CPU", 0)) * 2 in_flight = in_flight or int(ray.available_resources().get("CPU", 0)) * 2
function_on_store = ray.put(function) function_on_store = ray.put(function)
highest_index = 0 highest_index = 0
@ -828,8 +827,9 @@ def ensemble_mean(
highest_index += 1 highest_index += 1
gc.collect() if gc_sleep and gc_sleep > 0:
time.sleep(gc_sleep) # wait for the ray store to catch on gc.collect()
time.sleep(gc_sleep) # wait for the ray store to catch on
if next_val: if next_val:
chunk_ref = ray.put(next_val[0]) chunk_ref = ray.put(next_val[0])