3:02 a.m.
From: Royce Lv <lvroyce(a)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(a)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 c012a62..b712c16 100644
--- a/src/kimchi/i18n.py
+++ b/src/kimchi/i18n.py
@@ -216,6 +216,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 b2739b2..b089be5 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):
@@ -162,3 +168,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