#189 Parcel calculation mutex

* adds cache based mutex (valid e.g. for celery use cases)
* drops atomic parcel bulk creation in favor of proper mutex implementation

konova/models/geometry.py

@@ -6,14 +6,17 @@ Created on: 15.11.21
 
 """
 import json
+from time import sleep
 
 from django.contrib.gis.db.models import MultiPolygonField
 from django.contrib.gis.geos import Polygon
-from django.db import models, transaction
+from django.core.exceptions import MultipleObjectsReturned
+from django.db import models
 from django.utils import timezone
 
 from konova.models import BaseResource, UuidModel
 from konova.sub_settings.lanis_settings import DEFAULT_SRID_RLP
+from konova.utils.mutex import cache_lock
 from konova.utils.wfs.spatial import ParcelWFSFetcher
 
 
@@ -99,7 +102,6 @@ class Geometry(BaseResource):
             objs += set_objs
         return objs
 
-    @transaction.atomic
     def update_parcels(self):
         """ Updates underlying parcel information
 
@@ -122,38 +124,53 @@ class Geometry(BaseResource):
             # which needs to be deleted and just keep the numerical values
             ## THIS CAN BE REMOVED IN THE FUTURE, WHEN 'Flur' WON'T OCCUR ANYMORE!
             flr_val = parcel_properties["flur"].replace("Flur ", "")
-            district = District.objects.get_or_create(
+
-                key=parcel_properties["kreisschl"],
+            # Run possible race-condition snippet mutexed
-                name=parcel_properties["kreis"],
+            # Use Flurstückkennzeichen as identifier to prevent the same calculation runs parallel, leading to
-            )[0]
+            # a race condition
-            municipal = Municipal.objects.get_or_create(
+            flr_id = parcel_properties['flstkennz']
-                key=parcel_properties["gmdschl"],
+            lock_id = f"parcel_calc-lock-{flr_id}"
-                name=parcel_properties["gemeinde"],
+            with cache_lock(lock_id) as acquired:
-                district=district,
+                while not acquired:
-            )[0]
+                    print(f"Am locked. Need to rest. Calculating: {parcel_properties}")
-            parcel_group = ParcelGroup.objects.get_or_create(
+                    sleep(0.5)
-                key=parcel_properties["gemaschl"],
+                    acquired = cache_lock(lock_id)
-                name=parcel_properties["gemarkung"],
+                district = District.objects.get_or_create(
-                municipal=municipal,
+                    key=parcel_properties["kreisschl"],
-            )[0]
+                    name=parcel_properties["kreis"],
-            flrstck_nnr = parcel_properties['flstnrnen']
+                )[0]
-            if not flrstck_nnr:
+                municipal = Municipal.objects.get_or_create(
-                flrstck_nnr = None
+                    key=parcel_properties["gmdschl"],
-            flrstck_zhlr = parcel_properties['flstnrzae']
+                    name=parcel_properties["gemeinde"],
-            if not flrstck_zhlr:
+                    district=district,
-                flrstck_zhlr = None
+                )[0]
-            parcel_obj = Parcel.objects.get_or_create(
+                parcel_group = ParcelGroup.objects.get_or_create(
-                district=district,
+                    key=parcel_properties["gemaschl"],
-                municipal=municipal,
+                    name=parcel_properties["gemarkung"],
-                parcel_group=parcel_group,
+                    municipal=municipal,
-                flr=flr_val,
+                )[0]
-                flrstck_nnr=flrstck_nnr,
+                flrstck_nnr = parcel_properties['flstnrnen']
-                flrstck_zhlr=flrstck_zhlr,
+                if not flrstck_nnr:
-            )[0]
+                    flrstck_nnr = None
-            parcel_obj.district = district
+                flrstck_zhlr = parcel_properties['flstnrzae']
-            parcel_obj.updated_on = _now
+                if not flrstck_zhlr:
-            parcel_obj.save()
+                    flrstck_zhlr = None
-            underlying_parcels.append(parcel_obj)
+
+                try:
+                    parcel_obj = Parcel.objects.get_or_create(
+                        district=district,
+                        municipal=municipal,
+                        parcel_group=parcel_group,
+                        flr=flr_val,
+                        flrstck_nnr=flrstck_nnr,
+                        flrstck_zhlr=flrstck_zhlr,
+                    )[0]
+                except MultipleObjectsReturned as e:
+                    raise MultipleObjectsReturned(f"{e}: {flr_id}")
+                parcel_obj.district = district
+                parcel_obj.updated_on = _now
+                parcel_obj.save()
+                underlying_parcels.append(parcel_obj)
 
         # Update the linked parcels
         self.parcels.clear()

konova/sub_settings/django_settings.py

@@ -132,6 +132,17 @@ DATABASES = {
     }
 }
 
+CACHES = {
+    "default": {
+        "BACKEND": "django_redis.cache.RedisCache",
+        "LOCATION": "redis://localhost:6379/1",
+        "OPTIONS": {
+            "CLIENT_CLASS": "django_redis.client.DefaultClient"
+        },
+        "KEY_PREFIX": "konova_cache"
+    }
+}
+
 
 # Password validation
 # https://docs.djangoproject.com/en/3.1/ref/settings/#auth-password-validators

konova/utils/mutex.py

@@ -0,0 +1,35 @@
+from contextlib import contextmanager
+
+from django.core.cache import cache
+import time
+
+LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes
+
+
+@contextmanager
+def cache_lock(lock_id):
+    """ Cache based lock e.g. for parallel celery processing
+
+    Derived from https://docs.celeryq.dev/en/latest/tutorials/task-cookbook.html
+
+    Use it like
+    ...
+    with cache_lock(lock_id, self.app.oid) as acquired:
+        if acquired:
+            do_mutexed_stuff()
+    ...
+
+    """
+    timeout_at = time.monotonic() + LOCK_EXPIRE - 3
+    # cache.add fails if the key already exists
+    status = cache.add(lock_id, "LOCKED", LOCK_EXPIRE)
+    try:
+        yield status
+    finally:
+        # advantage of using add() for atomic locking
+        if time.monotonic() < timeout_at and status:
+            # don't release the lock if we exceeded the timeout
+            # to lessen the chance of releasing an expired lock
+            # owned by someone else
+            # also don't release the lock if we didn't acquire it
+            cache.delete(lock_id)

requirements.txt

@@ -18,6 +18,7 @@ django-bootstrap4==3.0.1
 django-debug-toolbar==3.1.1
 django-filter==2.4.0
 django-fontawesome-5==1.0.18
+django-redis==5.2.0
 django-simple-sso==1.1.0
 django-tables2==2.3.4
 et-xmlfile==1.1.0
@@ -48,4 +49,4 @@ wcwidth==0.2.5
 webservices==0.7
 wrapt==1.13.3
 xmltodict==0.12.0
-zipp==3.4.1
+zipp==3.4.1