[Kimchi-devel] [kimchi-devel][PATCHv3 3/6] Add lock facility for storage volume upload

[lvroyce at linux.vnet.ibm.com]

From: Royce Lv <lvroyce at linux.vnet.ibm.com>

This lock facility guarentees 5 concurrent volume upload
and make sure locks are created and reclaimed when needed.

Signed-off-by: Royce Lv <lvroyce at linux.vnet.ibm.com>
---
 src/kimchi/i18n.py        |  1 +
 src/kimchi/model/utils.py | 46 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 47 insertions(+)

diff --git a/src/kimchi/i18n.py b/src/kimchi/i18n.py
index 0a31cd5..af912bd 100644
--- a/src/kimchi/i18n.py
+++ b/src/kimchi/i18n.py
@@ -211,6 +211,7 @@ messages = {
     "KCHVOL0022E": _("Unable to access file %(url)s. Please, check it."),
     "KCHVOL0023E": _("Unable to clone storage volume '%(name)s' in pool '%(pool)s'. Details: %(err)s"),
     "KCHVOL0024E": _("Upload volume chunk index, size and total size must be integer"),
+    "KCHVOL0026E": _("Inconsistent upload count"),
 
     "KCHIFACE0001E": _("Interface %(name)s does not exist"),
 
diff --git a/src/kimchi/model/utils.py b/src/kimchi/model/utils.py
index 9896289..010966a 100644
--- a/src/kimchi/model/utils.py
+++ b/src/kimchi/model/utils.py
@@ -19,6 +19,7 @@
 
 import libvirt
 import socket
+import threading
 import urlparse
 from lxml import etree, objectify
 from lxml.builder import E, ElementMaker
@@ -28,6 +29,11 @@ from kimchi.model.featuretests import FeatureTests
 
 KIMCHI_META_URL = "https://github.com/kimchi-project/kimchi"
 KIMCHI_NAMESPACE = "kimchi"
+UPLOAD_THREADS = 5  # Concurrent upload volume counts at the same time
+
+upload_semaphore = threading.BoundedSemaphore(value=UPLOAD_THREADS)
+upload_lock_pool = dict()
+pool_lock = threading.Lock()
 
 
 def get_vm_name(vm_name, t_name, name_list):
@@ -161,3 +167,43 @@ def get_metadata_node(dom, tag, metadata_support, mode="current"):
         if node is not None:
             return etree.tostring(node)
     return ""
+
+
+class UpdateLock(object):
+    def __init__(self):
+        self.ref_cnt = 0
+        self.lock = threading.Lock()
+
+    def get_lock(self):
+        self.ref_cnt += 1
+        return self.lock
+
+    def release_lock(self):
+        delete = False
+        self.ref_cnt -= 1
+        if (self.ref_cnt == 0):
+            delete = True
+        return delete
+
+
+def get_vol_update_lock(vol_path):
+    # upload_semaphore controls the max upload count
+    upload_semaphore.acquire()
+
+    # pool lock make sure lock list get/store action is atomic
+    with pool_lock:
+        vol_lock = upload_lock_pool.get(vol_path)
+        if vol_lock:
+            return vol_lock.get_lock()
+        if len(upload_lock_pool.keys()) > (UPLOAD_THREADS - 1):
+            raise OperationFailed("KCHVOL0026E")
+        lock = upload_lock_pool[vol_path] = UpdateLock()
+        return lock.get_lock()
+
+
+def release_vol_update_lock(vol_path):
+    with pool_lock:
+        vol_lock = upload_lock_pool.get(vol_path)
+        if vol_lock.release_lock():
+            upload_lock_pool.pop(vol_path, None)
+    upload_semaphore.release()
-- 
2.1.0