diff --git a/src/damaris/data/DataPool.py b/src/damaris/data/DataPool.py index 8bf6a39..c4e9b87 100644 --- a/src/damaris/data/DataPool.py +++ b/src/damaris/data/DataPool.py @@ -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)): diff --git a/src/damaris/gui/DamarisGUI.py b/src/damaris/gui/DamarisGUI.py index b81c3a1..415528d 100644 --- a/src/damaris/gui/DamarisGUI.py +++ b/src/damaris/gui/DamarisGUI.py @@ -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" )