Revision e4e59de8 lib/mcpu.py
b/lib/mcpu.py | ||
---|---|---|
142 | 142 |
|
143 | 143 |
""" |
144 | 144 |
|
145 |
def CheckCancel(self):
|
|
146 |
"""Check whether job has been cancelled.
|
|
145 |
def CurrentPriority(self): # pylint: disable=R0201
|
|
146 |
"""Returns current priority or C{None}.
|
|
147 | 147 |
|
148 | 148 |
""" |
149 |
return None |
|
149 | 150 |
|
150 | 151 |
def SubmitManyJobs(self, jobs): |
151 | 152 |
"""Submits jobs for processing. |
... | ... | |
274 | 275 |
if not self._enable_locks: |
275 | 276 |
raise errors.ProgrammerError("Attempted to use disabled locks") |
276 | 277 |
|
277 |
def _AcquireLocks(self, level, names, shared, timeout, priority):
|
|
278 |
def _AcquireLocks(self, level, names, shared, timeout): |
|
278 | 279 |
"""Acquires locks via the Ganeti lock manager. |
279 | 280 |
|
280 | 281 |
@type level: int |
... | ... | |
292 | 293 |
self._CheckLocksEnabled() |
293 | 294 |
|
294 | 295 |
if self._cbs: |
295 |
self._cbs.CheckCancel() |
|
296 |
priority = self._cbs.CurrentPriority() |
|
297 |
else: |
|
298 |
priority = None |
|
296 | 299 |
|
297 | 300 |
acquired = self.context.glm.acquire(level, names, shared=shared, |
298 | 301 |
timeout=timeout, priority=priority) |
... | ... | |
342 | 345 |
def BuildHooksManager(self, lu): |
343 | 346 |
return self.hmclass.BuildFromLu(lu.rpc.call_hooks_runner, lu) |
344 | 347 |
|
345 |
def _LockAndExecLU(self, lu, level, calc_timeout, priority):
|
|
348 |
def _LockAndExecLU(self, lu, level, calc_timeout): |
|
346 | 349 |
"""Execute a Logical Unit, with the needed locks. |
347 | 350 |
|
348 | 351 |
This is a recursive function that starts locking the given level, and |
... | ... | |
391 | 394 |
needed_locks = lu.needed_locks[level] |
392 | 395 |
|
393 | 396 |
self._AcquireLocks(level, needed_locks, share, |
394 |
calc_timeout(), priority)
|
|
397 |
calc_timeout()) |
|
395 | 398 |
else: |
396 | 399 |
# Adding locks |
397 | 400 |
add_locks = lu.add_locks[level] |
... | ... | |
408 | 411 |
errors.ECODE_NOTUNIQUE) |
409 | 412 |
|
410 | 413 |
try: |
411 |
result = self._LockAndExecLU(lu, level + 1, calc_timeout, priority)
|
|
414 |
result = self._LockAndExecLU(lu, level + 1, calc_timeout) |
|
412 | 415 |
finally: |
413 | 416 |
if level in lu.remove_locks: |
414 | 417 |
self.context.glm.remove(level, lu.remove_locks[level]) |
... | ... | |
417 | 420 |
self.context.glm.release(level) |
418 | 421 |
|
419 | 422 |
else: |
420 |
result = self._LockAndExecLU(lu, level + 1, calc_timeout, priority)
|
|
423 |
result = self._LockAndExecLU(lu, level + 1, calc_timeout) |
|
421 | 424 |
|
422 | 425 |
return result |
423 | 426 |
|
424 |
def ExecOpCode(self, op, cbs, timeout=None, priority=None):
|
|
427 |
def ExecOpCode(self, op, cbs, timeout=None): |
|
425 | 428 |
"""Execute an opcode. |
426 | 429 |
|
427 | 430 |
@type op: an OpCode instance |
... | ... | |
430 | 433 |
@param cbs: Runtime callbacks |
431 | 434 |
@type timeout: float or None |
432 | 435 |
@param timeout: Maximum time to acquire all locks, None for no timeout |
433 |
@type priority: number or None |
|
434 |
@param priority: Priority for acquiring lock(s) |
|
435 | 436 |
@raise LockAcquireTimeout: In case locks couldn't be acquired in specified |
436 | 437 |
amount of time |
437 | 438 |
|
... | ... | |
456 | 457 |
# and in a shared fashion otherwise (to prevent concurrent run with |
457 | 458 |
# an exclusive LU. |
458 | 459 |
self._AcquireLocks(locking.LEVEL_CLUSTER, locking.BGL, |
459 |
not lu_class.REQ_BGL, calc_timeout(), |
|
460 |
priority) |
|
460 |
not lu_class.REQ_BGL, calc_timeout()) |
|
461 | 461 |
elif lu_class.REQ_BGL: |
462 | 462 |
raise errors.ProgrammerError("Opcode '%s' requires BGL, but locks are" |
463 | 463 |
" disabled" % op.OP_ID) |
... | ... | |
468 | 468 |
assert lu.needed_locks is not None, "needed_locks not set by LU" |
469 | 469 |
|
470 | 470 |
try: |
471 |
result = self._LockAndExecLU(lu, locking.LEVEL_INSTANCE, calc_timeout, |
|
472 |
priority) |
|
471 |
result = self._LockAndExecLU(lu, locking.LEVEL_INSTANCE, calc_timeout) |
|
473 | 472 |
finally: |
474 | 473 |
if self._ec_id: |
475 | 474 |
self.context.cfg.DropECReservations(self._ec_id) |
Also available in: Unified diff