Public Access
improved performance a lot, incremental writing of HDF5
This commit is contained in:
@@ -148,10 +148,7 @@ class DataPool(collections.abc.MutableMapping):
|
||||
If False, write all keys. Defaults to False.
|
||||
|
||||
write_interval : int, optional
|
||||
Time interval in seconds between periodic writes. If 0, no periodic writes are performed.
|
||||
This parameter affects the writing strategy:
|
||||
- If write_interval is 0: Always use incremental writing for better performance
|
||||
- If write_interval > 0: Only write dirty keys that haven't been written yet
|
||||
Time interval in seconds between periodic writes.
|
||||
Defaults to 0.
|
||||
|
||||
Raises:
|
||||
@@ -187,9 +184,6 @@ class DataPool(collections.abc.MutableMapping):
|
||||
else:
|
||||
# Full sync: current keys + any other dirty keys (e.g. deletions)
|
||||
dict_keys=list(set(self.__mydict.keys()) | self.__dirty_keys)
|
||||
# Optimization: for periodic writes, we only write dirty keys
|
||||
if write_interval > 0:
|
||||
dict_keys = list(self.__dirty_keys)
|
||||
|
||||
self._dictlock.release()
|
||||
|
||||
@@ -219,22 +213,26 @@ class DataPool(collections.abc.MutableMapping):
|
||||
group_keyname="dict_"+str(namelist[-1]).translate(DataPool.translation_table)
|
||||
|
||||
# find existing node for this key
|
||||
node_found = False
|
||||
if group_keyname in dump_dir:
|
||||
if dump_dir._v_children[group_keyname]._v_title == key:
|
||||
dump_file.remove_node(dump_dir, group_keyname, recursive=True)
|
||||
node_found = True
|
||||
else:
|
||||
extension_count = 0
|
||||
while group_keyname + "_%03d" % extension_count in dump_dir:
|
||||
suffix_name = group_keyname + "_%03d" % extension_count
|
||||
if dump_dir._v_children[suffix_name]._v_title == key:
|
||||
dump_file.remove_node(dump_dir, suffix_name, recursive=True)
|
||||
group_keyname = suffix_name
|
||||
node_found = True
|
||||
break
|
||||
extension_count += 1
|
||||
else:
|
||||
# truly new key with collision
|
||||
group_keyname += "_%03d" % extension_count
|
||||
|
||||
if node_found:
|
||||
dump_file.remove_node(dump_dir, group_keyname, recursive=True)
|
||||
|
||||
self._dictlock.acquire()
|
||||
if key not in self.__mydict:
|
||||
# Key was deleted
|
||||
@@ -243,10 +241,12 @@ class DataPool(collections.abc.MutableMapping):
|
||||
self._dictlock.release()
|
||||
continue
|
||||
value=self.__mydict[key]
|
||||
if key in self.__dirty_keys:
|
||||
self.__dirty_keys.remove(key)
|
||||
# We will clear the dirty flag only after successful write
|
||||
# to handle potential transient failures or re-dirtied items
|
||||
self._dictlock.release()
|
||||
|
||||
# now write data, assuming, the object is constant during write operation
|
||||
write_success = False
|
||||
if hasattr(value, "write_to_hdf"):
|
||||
try:
|
||||
value.write_to_hdf(hdffile=dump_file,
|
||||
@@ -255,6 +255,7 @@ class DataPool(collections.abc.MutableMapping):
|
||||
title=key,
|
||||
complib=complib,
|
||||
complevel=complevel)
|
||||
write_success = True
|
||||
except Exception as e:
|
||||
print("failed to write data_pool[\"%s\"]: %s"%(key,str(e)))
|
||||
traceback_file=io.StringIO()
|
||||
@@ -270,12 +271,24 @@ class DataPool(collections.abc.MutableMapping):
|
||||
name=group_keyname,
|
||||
obj=obj_to_save,
|
||||
title=key)
|
||||
write_success = True
|
||||
except Exception as e:
|
||||
print("failed to write data_pool[\"%s\"]: %s"%(key,str(e)))
|
||||
else:
|
||||
print("don't know how to store data_pool[\"%s\"] (type: %s)"%(key, type(value)))
|
||||
|
||||
if write_success:
|
||||
self._dictlock.acquire()
|
||||
if key in self.__dirty_keys:
|
||||
# Only remove if it hasn't been modified again in the meantime
|
||||
# (Checking identity is a simple way for objects)
|
||||
if self.__mydict.get(key) is value:
|
||||
self.__dirty_keys.remove(key)
|
||||
self._dictlock.release()
|
||||
value=None
|
||||
|
||||
dump_file.flush()
|
||||
|
||||
finally:
|
||||
dump_group=None
|
||||
if isinstance(hdffile, (bytes, str)):
|
||||
|
||||
@@ -506,9 +506,9 @@ class DamarisGUI:
|
||||
print(f"DEBUG: start_experiment called at {overall_start}")
|
||||
|
||||
# something running?
|
||||
if self.si is not None:
|
||||
print("Last Experiment is not clearly stopped!")
|
||||
self.si = None
|
||||
if self.si is not None or self.state != DamarisGUI.Edit_State:
|
||||
print("Experiment is still running or cleaning up!")
|
||||
return
|
||||
|
||||
# get config values:
|
||||
config_start = time.time()
|
||||
@@ -744,6 +744,9 @@ class DamarisGUI:
|
||||
"""
|
||||
periodically look at running threads
|
||||
"""
|
||||
if self.si is None:
|
||||
return False
|
||||
|
||||
# look at components and update them
|
||||
# test whether backend and scripts are done
|
||||
r = self.si.data.get( "__recentresult", -1 ) + 1
|
||||
@@ -888,6 +891,11 @@ class DamarisGUI:
|
||||
dump_size = os.stat( self.dump_filename ).st_size / 1e6
|
||||
print("done (%.1f s, %.1f MB)" % (time.time( ) - self.dump_start_time, dump_size))
|
||||
|
||||
# keep data to display but throw away everything else
|
||||
self.si = None
|
||||
# delete lock file so that other experiment can start
|
||||
self.lockfile.del_experiment(self.id)
|
||||
|
||||
# now everything is stopped
|
||||
self.state = DamarisGUI.Edit_State
|
||||
gdk.threads_enter( )
|
||||
@@ -896,16 +904,12 @@ class DamarisGUI:
|
||||
self.toolbar_stop_button.set_sensitive( False )
|
||||
self.toolbar_pause_button.set_sensitive( False )
|
||||
gdk.threads_leave( )
|
||||
|
||||
# keep data to display but throw away everything else
|
||||
self.si = None
|
||||
# delete lock file so that other experiment can start
|
||||
self.lockfile.del_experiment(self.id)
|
||||
return False
|
||||
|
||||
# dump states?
|
||||
dump_interval = self.dump_timeinterval if self.dump_timeinterval > 0 else 2.0
|
||||
if self.dump_thread is None and self.dump_filename != "" and \
|
||||
self.dump_timeinterval != 0 and self.last_dumped + self.dump_timeinterval < time.time( ):
|
||||
self.last_dumped + dump_interval < time.time( ):
|
||||
print("dumping data pool")
|
||||
self.dump_start_time = time.time( )
|
||||
self.dump_dots = 0
|
||||
@@ -1062,19 +1066,25 @@ class DamarisGUI:
|
||||
dump_file.disable_undo( )
|
||||
|
||||
# save the data!
|
||||
self.data.write_hdf5( dump_file, where="/", name="data_pool",
|
||||
complib=self.dump_complib, complevel=self.dump_complevel,
|
||||
incremental=not init, write_interval=self.dump_timeinterval )
|
||||
if self.data is not None:
|
||||
self.data.write_hdf5( dump_file, where="/", name="data_pool",
|
||||
complib=self.dump_complib, complevel=self.dump_complevel,
|
||||
incremental=not init, write_interval=self.dump_timeinterval )
|
||||
|
||||
# Re-enable run button if this was called from save thread
|
||||
if not init and threading.current_thread().name == "dump states":
|
||||
gdk.threads_enter()
|
||||
self.toolbar_run_button.set_sensitive(True)
|
||||
gdk.threads_leave()
|
||||
# NO: this is dangerous as it allows starting a new experiment before cleanup is finished.
|
||||
# it will be re-enabled in observe_running_experiment after joining this thread.
|
||||
# if not init and threading.current_thread().name == "dump states":
|
||||
# gdk.threads_enter()
|
||||
# self.toolbar_run_button.set_sensitive(True)
|
||||
# gdk.threads_leave()
|
||||
|
||||
# now save additional information
|
||||
e = self.si.data.get( "__recentexperiment", -1 ) + 1
|
||||
r = self.si.data.get( "__recentresult", -1 ) + 1
|
||||
e = 0
|
||||
r = 0
|
||||
if self.si is not None:
|
||||
e = self.si.data.get( "__recentexperiment", -1 ) + 1
|
||||
r = self.si.data.get( "__recentresult", -1 ) + 1
|
||||
timeline_table = dump_file.root.timeline
|
||||
timeline_row = timeline_table.row
|
||||
timeline_row[ "time" ] = time.strftime( "%Y%m%d %H:%M:%S" )
|
||||
|
||||
Reference in New Issue
Block a user