FIX: Garbage collection for parallel workload.

src/radclss/core/radclss_core.py

@@ -497,7 +497,7 @@ def _get_nexrad_wrapper(time_str):
         if verbose:
             print("  Merging NEXRAD data into combined dataset...")
         ds_concat = xr.merge([ds_concat, nexrad_columns])
-
+    
     if verbose:
         print(f"  Total variables in merged dataset: {len(ds_concat.data_vars)}")
         print("\n" + "=" * 80)
@@ -508,7 +508,6 @@ def _get_nexrad_wrapper(time_str):
         print("Variables in merged dataset:")
         for vars in ds_concat.data_vars:
             print(vars)
-
     ds = act.io.create_ds_from_arm_dod(
         f"{output_platform}.{output_level}",
         {
@@ -601,8 +600,19 @@ def _get_nexrad_wrapper(time_str):
     # Drop duplicate latitude and longitude
     if verbose:
         print("\n  Freeing memory: deleting intermediate datasets...")
+    ds_concat.close()
+    if serial is not True:
+        if current_client is None:
+            try:
+                current_client = Client.current()
+            except ValueError:
+                raise RuntimeError(
+                    "No Dask client found. Please start a Dask client before running in parallel mode."
+                )
+        current_client.restart()
     del ds_concat
-
+
+
     # Free up Memory
     del columns
 

src/radclss/util/column_utils.py

@@ -442,6 +442,7 @@ def match_datasets_act(
                 column[k] = (
                     column[k].fillna(column[k].attrs["missing_value"]).astype(float)
                 )
+    grd_ds.close()
     return column