From aa3b6847ad5dc5048ba13666faf179fcaaea3471 Mon Sep 17 00:00:00 2001
From: "Krzysztof (Chris) Bernat" <chris.bernat@telespazio.com>
Date: Mon, 2 Oct 2017 19:02:40 +0100
Subject: [PATCH] #393 #352 local ds, better error handling, handling lock file

---
 cate/ds/local.py | 30 +++++++++++++++++++-----------
 1 file changed, 19 insertions(+), 11 deletions(-)

diff --git a/cate/ds/local.py b/cate/ds/local.py
index 430c35ff8..80112feda 100644
--- a/cate/ds/local.py
+++ b/cate/ds/local.py
@@ -743,6 +743,8 @@ def create_data_source(self, data_source_id: str, region: PolygonLike.TYPE = Non
 
         lock_filename = '{}.lock'.format(data_source_id)
         lock_filepath = os.path.join(self._store_dir, lock_filename)
+        pid = os.getpid()
+        create_time = int(psutil.Process(pid).create_time() * 1_000_000)
 
         data_source = None
         for ds in self._data_sources:
@@ -750,15 +752,22 @@ def create_data_source(self, data_source_id: str, region: PolygonLike.TYPE = Non
                 if lock_file and os.path.isfile(lock_filepath):
                     with open(lock_filepath, 'r') as lock_file:
                         writer_pid = lock_file.readline()
-                        if psutil.pid_exists(int(writer_pid)):
-                            raise DataAccessError(self, "Cannot access data source {}, another process is using it "
-                                                        "(pid:{}". format(ds.id, writer_pid))
-                        # ds.temporal_coverage() == time_range and
-                        if ds.spatial_coverage() == region \
-                                and ds.variables_info == var_names:
-                            data_source = ds
-                            data_source.set_completed(False)
-                            break
+                        if writer_pid:
+                            writer_create_time = -1
+                            writer_pid, writer_timestamp = [(int(val) for val in writer_pid.split(":"))
+                                                            if ":" in writer_pid else writer_pid, writer_create_time]
+                            if psutil.pid_exists(writer_pid) and writer_pid != pid:
+                                if writer_timestamp > writer_create_time:
+                                    writer_create_time = int(psutil.Process(writer_pid).create_time() * 1_000_000)
+                                if writer_create_time == writer_timestamp:
+                                    raise DataAccessError(self, "Data source '{}' is currently being created by other "
+                                                                "process (pid:{})". format(ds.id, writer_pid))
+                            # ds.temporal_coverage() == time_range and
+                            if ds.spatial_coverage() == region \
+                                    and ds.variables_info == var_names:
+                                data_source = ds
+                                data_source.set_completed(False)
+                                break
                 raise DataAccessError(self, "Data source '{}' already exists.". format(data_source_id))
         if not data_source:
             data_source = LocalDataSource(data_source_id, files=[], data_store=self, spatial_coverage=region,
@@ -767,9 +776,8 @@ def create_data_source(self, data_source_id: str, region: PolygonLike.TYPE = Non
             self._save_data_source(data_source)
 
         if lock_file:
-            pid = os.getpid()
             with open(lock_filepath, 'w') as lock_file:
-                lock_file.write(str(pid))
+                lock_file.write("{}:{}".format(pid, create_time))
 
         return data_source