Parallelize LUCreateInstance

[ganeti-local] / lib / mcpu.py

diff --git a/lib/mcpu.py b/lib/mcpu.py
index 6bd0542..7bb4866 100644 (file)
--- a/lib/mcpu.py
+++ b/lib/mcpu.py
@@ -34,7 +34,6 @@ from ganeti import constants
 from ganeti import errors
 from ganeti import rpc
 from ganeti import cmdlib
-from ganeti import config
 from ganeti import ssconf
 from ganeti import logger
 from ganeti import locking
@@ -47,7 +46,6 @@ class Processor(object):
     opcodes.OpDestroyCluster: cmdlib.LUDestroyCluster,
     opcodes.OpQueryClusterInfo: cmdlib.LUQueryClusterInfo,
     opcodes.OpVerifyCluster: cmdlib.LUVerifyCluster,
-    opcodes.OpMasterFailover: cmdlib.LUMasterFailover,
     opcodes.OpDumpClusterConfig: cmdlib.LUDumpClusterConfig,
     opcodes.OpRenameCluster: cmdlib.LURenameCluster,
     opcodes.OpVerifyDisks: cmdlib.LUVerifyDisks,
@@ -90,7 +88,7 @@ class Processor(object):
     opcodes.OpTestAllocator: cmdlib.LUTestAllocator,
     }
 
-  def __init__(self, context, feedback=None):
+  def __init__(self, context):
     """Constructor for Processor
 
     Args:
@@ -98,7 +96,7 @@ class Processor(object):
                     interesting events are happening
     """
     self.context = context
-    self._feedback_fn = feedback
+    self._feedback_fn = None
     self.exclusive_BGL = False
 
   def _ExecLU(self, lu):
@@ -123,7 +121,57 @@ class Processor(object):
 
     return result
 
-  def ExecOpCode(self, op):
+  def _LockAndExecLU(self, lu, level):
+    """Execute a Logical Unit, with the needed locks.
+
+    This is a recursive function that starts locking the given level, and
+    proceeds up, till there are no more locks to acquire. Then it executes the
+    given LU and its opcodes.
+
+    """
+    adding_locks = level in lu.add_locks
+    acquiring_locks = level in lu.needed_locks
+    if level not in locking.LEVELS:
+      result = self._ExecLU(lu)
+    elif adding_locks and acquiring_locks:
+      # We could both acquire and add locks at the same level, but for now we
+      # don't need this, so we'll avoid the complicated code needed.
+      raise NotImplementedError(
+        "Can't declare locks to acquire when adding others")
+    elif adding_locks or acquiring_locks:
+      lu.DeclareLocks(level)
+      share = lu.share_locks[level]
+      if acquiring_locks:
+        needed_locks = lu.needed_locks[level]
+        lu.acquired_locks[level] = self.context.glm.acquire(level,
+                                                            needed_locks,
+                                                            shared=share)
+      else: # adding_locks
+        add_locks = lu.add_locks[level]
+        lu.remove_locks[level] = add_locks
+        try:
+          self.context.glm.add(level, add_locks, acquired=1, shared=share)
+        except errors.LockError:
+          raise errors.OpPrereqError(
+            "Coudn't add locks (%s), probably because of a race condition"
+            " with another job, who added them first" % add_locks)
+      try:
+        try:
+          if adding_locks:
+            lu.acquired_locks[level] = add_locks
+          result = self._LockAndExecLU(lu, level + 1)
+        finally:
+          if level in lu.remove_locks:
+            self.context.glm.remove(level, lu.remove_locks[level])
+      finally:
+        if self.context.glm.is_owned(level):
+          self.context.glm.release(level)
+    else:
+      result = self._LockAndExecLU(lu, level + 1)
+
+    return result
+
+  def ExecOpCode(self, op, feedback_fn):
     """Execute an opcode.
 
     Args:
@@ -134,6 +182,7 @@ class Processor(object):
       raise errors.ProgrammerError("Non-opcode instance passed"
                                    " to ExecOpcode")
 
+    self._feedback_fn = feedback_fn
     lu_class = self.DISPATCH_TABLE.get(op.__class__, None)
     if lu_class is None:
       raise errors.OpCodeUnknown("Unknown opcode")
@@ -150,54 +199,15 @@ class Processor(object):
     try:
       self.exclusive_BGL = lu_class.REQ_BGL
       lu = lu_class(self, op, self.context, sstore)
-      result = self._ExecLU(lu)
+      lu.ExpandNames()
+      assert lu.needed_locks is not None, "needed_locks not set by LU"
+      result = self._LockAndExecLU(lu, locking.LEVEL_INSTANCE)
     finally:
       self.context.glm.release(locking.LEVEL_CLUSTER)
       self.exclusive_BGL = False
 
     return result
 
-  def ChainOpCode(self, op):
-    """Chain and execute an opcode.
-
-    This is used by LUs when they need to execute a child LU.
-
-    Args:
-     - opcode: the opcode to be executed
-
-    """
-    if not isinstance(op, opcodes.OpCode):
-      raise errors.ProgrammerError("Non-opcode instance passed"
-                                   " to ExecOpcode")
-
-    lu_class = self.DISPATCH_TABLE.get(op.__class__, None)
-    if lu_class is None:
-      raise errors.OpCodeUnknown("Unknown opcode")
-
-    if lu_class.REQ_BGL and not self.exclusive_BGL:
-      raise errors.ProgrammerError("LUs which require the BGL cannot"
-                                   " be chained to granular ones.")
-
-    if lu_class.REQ_WSSTORE:
-      sstore = ssconf.WritableSimpleStore()
-    else:
-      sstore = ssconf.SimpleStore()
-
-    #do_hooks = lu_class.HPATH is not None
-    lu = lu_class(self, op, self.context, sstore)
-    lu.CheckPrereq()
-    #if do_hooks:
-    #  hm = HooksMaster(rpc.call_hooks_runner, self, lu)
-    #  h_results = hm.RunPhase(constants.HOOKS_PHASE_PRE)
-    #  lu.HooksCallBack(constants.HOOKS_PHASE_PRE,
-    #                   h_results, self._feedback_fn, None)
-    result = lu.Exec(self._feedback_fn)
-    #if do_hooks:
-    #  h_results = hm.RunPhase(constants.HOOKS_PHASE_POST)
-    #  result = lu.HooksCallBack(constants.HOOKS_PHASE_POST,
-    #                   h_results, self._feedback_fn, result)
-    return result
-
   def LogStep(self, current, total, message):
     """Log a change in LU execution progress.