Statistics
| Branch: | Tag: | Revision:

root / lib / jqueue.py @ 66abb9ff

History | View | Annotate | Download (73.5 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Module implementing the job queue handling.
23

24
Locking: there's a single, large lock in the L{JobQueue} class. It's
25
used by all other classes in this module.
26

27
@var JOBQUEUE_THREADS: the number of worker threads we start for
28
    processing jobs
29

30
"""
31

    
32
import logging
33
import errno
34
import time
35
import weakref
36
import threading
37
import itertools
38

    
39
try:
40
  # pylint: disable=E0611
41
  from pyinotify import pyinotify
42
except ImportError:
43
  import pyinotify
44

    
45
from ganeti import asyncnotifier
46
from ganeti import constants
47
from ganeti import serializer
48
from ganeti import workerpool
49
from ganeti import locking
50
from ganeti import opcodes
51
from ganeti import errors
52
from ganeti import mcpu
53
from ganeti import utils
54
from ganeti import jstore
55
from ganeti import rpc
56
from ganeti import runtime
57
from ganeti import netutils
58
from ganeti import compat
59
from ganeti import ht
60

    
61

    
62
JOBQUEUE_THREADS = 25
63
JOBS_PER_ARCHIVE_DIRECTORY = 10000
64

    
65
# member lock names to be passed to @ssynchronized decorator
66
_LOCK = "_lock"
67
_QUEUE = "_queue"
68

    
69

    
70
class CancelJob(Exception):
71
  """Special exception to cancel a job.
72

73
  """
74

    
75

    
76
def TimeStampNow():
77
  """Returns the current timestamp.
78

79
  @rtype: tuple
80
  @return: the current time in the (seconds, microseconds) format
81

82
  """
83
  return utils.SplitTime(time.time())
84

    
85

    
86
class _QueuedOpCode(object):
87
  """Encapsulates an opcode object.
88

89
  @ivar log: holds the execution log and consists of tuples
90
  of the form C{(log_serial, timestamp, level, message)}
91
  @ivar input: the OpCode we encapsulate
92
  @ivar status: the current status
93
  @ivar result: the result of the LU execution
94
  @ivar start_timestamp: timestamp for the start of the execution
95
  @ivar exec_timestamp: timestamp for the actual LU Exec() function invocation
96
  @ivar stop_timestamp: timestamp for the end of the execution
97

98
  """
99
  __slots__ = ["input", "status", "result", "log", "priority",
100
               "start_timestamp", "exec_timestamp", "end_timestamp",
101
               "__weakref__"]
102

    
103
  def __init__(self, op):
104
    """Initializes instances of this class.
105

106
    @type op: L{opcodes.OpCode}
107
    @param op: the opcode we encapsulate
108

109
    """
110
    self.input = op
111
    self.status = constants.OP_STATUS_QUEUED
112
    self.result = None
113
    self.log = []
114
    self.start_timestamp = None
115
    self.exec_timestamp = None
116
    self.end_timestamp = None
117

    
118
    # Get initial priority (it might change during the lifetime of this opcode)
119
    self.priority = getattr(op, "priority", constants.OP_PRIO_DEFAULT)
120

    
121
  @classmethod
122
  def Restore(cls, state):
123
    """Restore the _QueuedOpCode from the serialized form.
124

125
    @type state: dict
126
    @param state: the serialized state
127
    @rtype: _QueuedOpCode
128
    @return: a new _QueuedOpCode instance
129

130
    """
131
    obj = _QueuedOpCode.__new__(cls)
132
    obj.input = opcodes.OpCode.LoadOpCode(state["input"])
133
    obj.status = state["status"]
134
    obj.result = state["result"]
135
    obj.log = state["log"]
136
    obj.start_timestamp = state.get("start_timestamp", None)
137
    obj.exec_timestamp = state.get("exec_timestamp", None)
138
    obj.end_timestamp = state.get("end_timestamp", None)
139
    obj.priority = state.get("priority", constants.OP_PRIO_DEFAULT)
140
    return obj
141

    
142
  def Serialize(self):
143
    """Serializes this _QueuedOpCode.
144

145
    @rtype: dict
146
    @return: the dictionary holding the serialized state
147

148
    """
149
    return {
150
      "input": self.input.__getstate__(),
151
      "status": self.status,
152
      "result": self.result,
153
      "log": self.log,
154
      "start_timestamp": self.start_timestamp,
155
      "exec_timestamp": self.exec_timestamp,
156
      "end_timestamp": self.end_timestamp,
157
      "priority": self.priority,
158
      }
159

    
160

    
161
class _QueuedJob(object):
162
  """In-memory job representation.
163

164
  This is what we use to track the user-submitted jobs. Locking must
165
  be taken care of by users of this class.
166

167
  @type queue: L{JobQueue}
168
  @ivar queue: the parent queue
169
  @ivar id: the job ID
170
  @type ops: list
171
  @ivar ops: the list of _QueuedOpCode that constitute the job
172
  @type log_serial: int
173
  @ivar log_serial: holds the index for the next log entry
174
  @ivar received_timestamp: the timestamp for when the job was received
175
  @ivar start_timestmap: the timestamp for start of execution
176
  @ivar end_timestamp: the timestamp for end of execution
177
  @ivar writable: Whether the job is allowed to be modified
178

179
  """
180
  # pylint: disable=W0212
181
  __slots__ = ["queue", "id", "ops", "log_serial", "ops_iter", "cur_opctx",
182
               "received_timestamp", "start_timestamp", "end_timestamp",
183
               "__weakref__", "processor_lock", "writable"]
184

    
185
  def __init__(self, queue, job_id, ops, writable):
186
    """Constructor for the _QueuedJob.
187

188
    @type queue: L{JobQueue}
189
    @param queue: our parent queue
190
    @type job_id: job_id
191
    @param job_id: our job id
192
    @type ops: list
193
    @param ops: the list of opcodes we hold, which will be encapsulated
194
        in _QueuedOpCodes
195
    @type writable: bool
196
    @param writable: Whether job can be modified
197

198
    """
199
    if not ops:
200
      raise errors.GenericError("A job needs at least one opcode")
201

    
202
    self.queue = queue
203
    self.id = job_id
204
    self.ops = [_QueuedOpCode(op) for op in ops]
205
    self.log_serial = 0
206
    self.received_timestamp = TimeStampNow()
207
    self.start_timestamp = None
208
    self.end_timestamp = None
209

    
210
    self._InitInMemory(self, writable)
211

    
212
  @staticmethod
213
  def _InitInMemory(obj, writable):
214
    """Initializes in-memory variables.
215

216
    """
217
    obj.writable = writable
218
    obj.ops_iter = None
219
    obj.cur_opctx = None
220

    
221
    # Read-only jobs are not processed and therefore don't need a lock
222
    if writable:
223
      obj.processor_lock = threading.Lock()
224
    else:
225
      obj.processor_lock = None
226

    
227
  def __repr__(self):
228
    status = ["%s.%s" % (self.__class__.__module__, self.__class__.__name__),
229
              "id=%s" % self.id,
230
              "ops=%s" % ",".join([op.input.Summary() for op in self.ops])]
231

    
232
    return "<%s at %#x>" % (" ".join(status), id(self))
233

    
234
  @classmethod
235
  def Restore(cls, queue, state, writable):
236
    """Restore a _QueuedJob from serialized state:
237

238
    @type queue: L{JobQueue}
239
    @param queue: to which queue the restored job belongs
240
    @type state: dict
241
    @param state: the serialized state
242
    @type writable: bool
243
    @param writable: Whether job can be modified
244
    @rtype: _JobQueue
245
    @return: the restored _JobQueue instance
246

247
    """
248
    obj = _QueuedJob.__new__(cls)
249
    obj.queue = queue
250
    obj.id = state["id"]
251
    obj.received_timestamp = state.get("received_timestamp", None)
252
    obj.start_timestamp = state.get("start_timestamp", None)
253
    obj.end_timestamp = state.get("end_timestamp", None)
254

    
255
    obj.ops = []
256
    obj.log_serial = 0
257
    for op_state in state["ops"]:
258
      op = _QueuedOpCode.Restore(op_state)
259
      for log_entry in op.log:
260
        obj.log_serial = max(obj.log_serial, log_entry[0])
261
      obj.ops.append(op)
262

    
263
    cls._InitInMemory(obj, writable)
264

    
265
    return obj
266

    
267
  def Serialize(self):
268
    """Serialize the _JobQueue instance.
269

270
    @rtype: dict
271
    @return: the serialized state
272

273
    """
274
    return {
275
      "id": self.id,
276
      "ops": [op.Serialize() for op in self.ops],
277
      "start_timestamp": self.start_timestamp,
278
      "end_timestamp": self.end_timestamp,
279
      "received_timestamp": self.received_timestamp,
280
      }
281

    
282
  def CalcStatus(self):
283
    """Compute the status of this job.
284

285
    This function iterates over all the _QueuedOpCodes in the job and
286
    based on their status, computes the job status.
287

288
    The algorithm is:
289
      - if we find a cancelled, or finished with error, the job
290
        status will be the same
291
      - otherwise, the last opcode with the status one of:
292
          - waitlock
293
          - canceling
294
          - running
295

296
        will determine the job status
297

298
      - otherwise, it means either all opcodes are queued, or success,
299
        and the job status will be the same
300

301
    @return: the job status
302

303
    """
304
    status = constants.JOB_STATUS_QUEUED
305

    
306
    all_success = True
307
    for op in self.ops:
308
      if op.status == constants.OP_STATUS_SUCCESS:
309
        continue
310

    
311
      all_success = False
312

    
313
      if op.status == constants.OP_STATUS_QUEUED:
314
        pass
315
      elif op.status == constants.OP_STATUS_WAITING:
316
        status = constants.JOB_STATUS_WAITING
317
      elif op.status == constants.OP_STATUS_RUNNING:
318
        status = constants.JOB_STATUS_RUNNING
319
      elif op.status == constants.OP_STATUS_CANCELING:
320
        status = constants.JOB_STATUS_CANCELING
321
        break
322
      elif op.status == constants.OP_STATUS_ERROR:
323
        status = constants.JOB_STATUS_ERROR
324
        # The whole job fails if one opcode failed
325
        break
326
      elif op.status == constants.OP_STATUS_CANCELED:
327
        status = constants.OP_STATUS_CANCELED
328
        break
329

    
330
    if all_success:
331
      status = constants.JOB_STATUS_SUCCESS
332

    
333
    return status
334

    
335
  def CalcPriority(self):
336
    """Gets the current priority for this job.
337

338
    Only unfinished opcodes are considered. When all are done, the default
339
    priority is used.
340

341
    @rtype: int
342

343
    """
344
    priorities = [op.priority for op in self.ops
345
                  if op.status not in constants.OPS_FINALIZED]
346

    
347
    if not priorities:
348
      # All opcodes are done, assume default priority
349
      return constants.OP_PRIO_DEFAULT
350

    
351
    return min(priorities)
352

    
353
  def GetLogEntries(self, newer_than):
354
    """Selectively returns the log entries.
355

356
    @type newer_than: None or int
357
    @param newer_than: if this is None, return all log entries,
358
        otherwise return only the log entries with serial higher
359
        than this value
360
    @rtype: list
361
    @return: the list of the log entries selected
362

363
    """
364
    if newer_than is None:
365
      serial = -1
366
    else:
367
      serial = newer_than
368

    
369
    entries = []
370
    for op in self.ops:
371
      entries.extend(filter(lambda entry: entry[0] > serial, op.log))
372

    
373
    return entries
374

    
375
  def GetInfo(self, fields):
376
    """Returns information about a job.
377

378
    @type fields: list
379
    @param fields: names of fields to return
380
    @rtype: list
381
    @return: list with one element for each field
382
    @raise errors.OpExecError: when an invalid field
383
        has been passed
384

385
    """
386
    row = []
387
    for fname in fields:
388
      if fname == "id":
389
        row.append(self.id)
390
      elif fname == "status":
391
        row.append(self.CalcStatus())
392
      elif fname == "priority":
393
        row.append(self.CalcPriority())
394
      elif fname == "ops":
395
        row.append([op.input.__getstate__() for op in self.ops])
396
      elif fname == "opresult":
397
        row.append([op.result for op in self.ops])
398
      elif fname == "opstatus":
399
        row.append([op.status for op in self.ops])
400
      elif fname == "oplog":
401
        row.append([op.log for op in self.ops])
402
      elif fname == "opstart":
403
        row.append([op.start_timestamp for op in self.ops])
404
      elif fname == "opexec":
405
        row.append([op.exec_timestamp for op in self.ops])
406
      elif fname == "opend":
407
        row.append([op.end_timestamp for op in self.ops])
408
      elif fname == "oppriority":
409
        row.append([op.priority for op in self.ops])
410
      elif fname == "received_ts":
411
        row.append(self.received_timestamp)
412
      elif fname == "start_ts":
413
        row.append(self.start_timestamp)
414
      elif fname == "end_ts":
415
        row.append(self.end_timestamp)
416
      elif fname == "summary":
417
        row.append([op.input.Summary() for op in self.ops])
418
      else:
419
        raise errors.OpExecError("Invalid self query field '%s'" % fname)
420
    return row
421

    
422
  def MarkUnfinishedOps(self, status, result):
423
    """Mark unfinished opcodes with a given status and result.
424

425
    This is an utility function for marking all running or waiting to
426
    be run opcodes with a given status. Opcodes which are already
427
    finalised are not changed.
428

429
    @param status: a given opcode status
430
    @param result: the opcode result
431

432
    """
433
    not_marked = True
434
    for op in self.ops:
435
      if op.status in constants.OPS_FINALIZED:
436
        assert not_marked, "Finalized opcodes found after non-finalized ones"
437
        continue
438
      op.status = status
439
      op.result = result
440
      not_marked = False
441

    
442
  def Finalize(self):
443
    """Marks the job as finalized.
444

445
    """
446
    self.end_timestamp = TimeStampNow()
447

    
448
  def Cancel(self):
449
    """Marks job as canceled/-ing if possible.
450

451
    @rtype: tuple; (bool, string)
452
    @return: Boolean describing whether job was successfully canceled or marked
453
      as canceling and a text message
454

455
    """
456
    status = self.CalcStatus()
457

    
458
    if status == constants.JOB_STATUS_QUEUED:
459
      self.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
460
                             "Job canceled by request")
461
      self.Finalize()
462
      return (True, "Job %s canceled" % self.id)
463

    
464
    elif status == constants.JOB_STATUS_WAITING:
465
      # The worker will notice the new status and cancel the job
466
      self.MarkUnfinishedOps(constants.OP_STATUS_CANCELING, None)
467
      return (True, "Job %s will be canceled" % self.id)
468

    
469
    else:
470
      logging.debug("Job %s is no longer waiting in the queue", self.id)
471
      return (False, "Job %s is no longer waiting in the queue" % self.id)
472

    
473

    
474
class _OpExecCallbacks(mcpu.OpExecCbBase):
475
  def __init__(self, queue, job, op):
476
    """Initializes this class.
477

478
    @type queue: L{JobQueue}
479
    @param queue: Job queue
480
    @type job: L{_QueuedJob}
481
    @param job: Job object
482
    @type op: L{_QueuedOpCode}
483
    @param op: OpCode
484

485
    """
486
    assert queue, "Queue is missing"
487
    assert job, "Job is missing"
488
    assert op, "Opcode is missing"
489

    
490
    self._queue = queue
491
    self._job = job
492
    self._op = op
493

    
494
  def _CheckCancel(self):
495
    """Raises an exception to cancel the job if asked to.
496

497
    """
498
    # Cancel here if we were asked to
499
    if self._op.status == constants.OP_STATUS_CANCELING:
500
      logging.debug("Canceling opcode")
501
      raise CancelJob()
502

    
503
  @locking.ssynchronized(_QUEUE, shared=1)
504
  def NotifyStart(self):
505
    """Mark the opcode as running, not lock-waiting.
506

507
    This is called from the mcpu code as a notifier function, when the LU is
508
    finally about to start the Exec() method. Of course, to have end-user
509
    visible results, the opcode must be initially (before calling into
510
    Processor.ExecOpCode) set to OP_STATUS_WAITING.
511

512
    """
513
    assert self._op in self._job.ops
514
    assert self._op.status in (constants.OP_STATUS_WAITING,
515
                               constants.OP_STATUS_CANCELING)
516

    
517
    # Cancel here if we were asked to
518
    self._CheckCancel()
519

    
520
    logging.debug("Opcode is now running")
521

    
522
    self._op.status = constants.OP_STATUS_RUNNING
523
    self._op.exec_timestamp = TimeStampNow()
524

    
525
    # And finally replicate the job status
526
    self._queue.UpdateJobUnlocked(self._job)
527

    
528
  @locking.ssynchronized(_QUEUE, shared=1)
529
  def _AppendFeedback(self, timestamp, log_type, log_msg):
530
    """Internal feedback append function, with locks
531

532
    """
533
    self._job.log_serial += 1
534
    self._op.log.append((self._job.log_serial, timestamp, log_type, log_msg))
535
    self._queue.UpdateJobUnlocked(self._job, replicate=False)
536

    
537
  def Feedback(self, *args):
538
    """Append a log entry.
539

540
    """
541
    assert len(args) < 3
542

    
543
    if len(args) == 1:
544
      log_type = constants.ELOG_MESSAGE
545
      log_msg = args[0]
546
    else:
547
      (log_type, log_msg) = args
548

    
549
    # The time is split to make serialization easier and not lose
550
    # precision.
551
    timestamp = utils.SplitTime(time.time())
552
    self._AppendFeedback(timestamp, log_type, log_msg)
553

    
554
  def CheckCancel(self):
555
    """Check whether job has been cancelled.
556

557
    """
558
    assert self._op.status in (constants.OP_STATUS_WAITING,
559
                               constants.OP_STATUS_CANCELING)
560

    
561
    # Cancel here if we were asked to
562
    self._CheckCancel()
563

    
564
  def SubmitManyJobs(self, jobs):
565
    """Submits jobs for processing.
566

567
    See L{JobQueue.SubmitManyJobs}.
568

569
    """
570
    # Locking is done in job queue
571
    return self._queue.SubmitManyJobs(jobs)
572

    
573

    
574
class _JobChangesChecker(object):
575
  def __init__(self, fields, prev_job_info, prev_log_serial):
576
    """Initializes this class.
577

578
    @type fields: list of strings
579
    @param fields: Fields requested by LUXI client
580
    @type prev_job_info: string
581
    @param prev_job_info: previous job info, as passed by the LUXI client
582
    @type prev_log_serial: string
583
    @param prev_log_serial: previous job serial, as passed by the LUXI client
584

585
    """
586
    self._fields = fields
587
    self._prev_job_info = prev_job_info
588
    self._prev_log_serial = prev_log_serial
589

    
590
  def __call__(self, job):
591
    """Checks whether job has changed.
592

593
    @type job: L{_QueuedJob}
594
    @param job: Job object
595

596
    """
597
    assert not job.writable, "Expected read-only job"
598

    
599
    status = job.CalcStatus()
600
    job_info = job.GetInfo(self._fields)
601
    log_entries = job.GetLogEntries(self._prev_log_serial)
602

    
603
    # Serializing and deserializing data can cause type changes (e.g. from
604
    # tuple to list) or precision loss. We're doing it here so that we get
605
    # the same modifications as the data received from the client. Without
606
    # this, the comparison afterwards might fail without the data being
607
    # significantly different.
608
    # TODO: we just deserialized from disk, investigate how to make sure that
609
    # the job info and log entries are compatible to avoid this further step.
610
    # TODO: Doing something like in testutils.py:UnifyValueType might be more
611
    # efficient, though floats will be tricky
612
    job_info = serializer.LoadJson(serializer.DumpJson(job_info))
613
    log_entries = serializer.LoadJson(serializer.DumpJson(log_entries))
614

    
615
    # Don't even try to wait if the job is no longer running, there will be
616
    # no changes.
617
    if (status not in (constants.JOB_STATUS_QUEUED,
618
                       constants.JOB_STATUS_RUNNING,
619
                       constants.JOB_STATUS_WAITING) or
620
        job_info != self._prev_job_info or
621
        (log_entries and self._prev_log_serial != log_entries[0][0])):
622
      logging.debug("Job %s changed", job.id)
623
      return (job_info, log_entries)
624

    
625
    return None
626

    
627

    
628
class _JobFileChangesWaiter(object):
629
  def __init__(self, filename):
630
    """Initializes this class.
631

632
    @type filename: string
633
    @param filename: Path to job file
634
    @raises errors.InotifyError: if the notifier cannot be setup
635

636
    """
637
    self._wm = pyinotify.WatchManager()
638
    self._inotify_handler = \
639
      asyncnotifier.SingleFileEventHandler(self._wm, self._OnInotify, filename)
640
    self._notifier = \
641
      pyinotify.Notifier(self._wm, default_proc_fun=self._inotify_handler)
642
    try:
643
      self._inotify_handler.enable()
644
    except Exception:
645
      # pyinotify doesn't close file descriptors automatically
646
      self._notifier.stop()
647
      raise
648

    
649
  def _OnInotify(self, notifier_enabled):
650
    """Callback for inotify.
651

652
    """
653
    if not notifier_enabled:
654
      self._inotify_handler.enable()
655

    
656
  def Wait(self, timeout):
657
    """Waits for the job file to change.
658

659
    @type timeout: float
660
    @param timeout: Timeout in seconds
661
    @return: Whether there have been events
662

663
    """
664
    assert timeout >= 0
665
    have_events = self._notifier.check_events(timeout * 1000)
666
    if have_events:
667
      self._notifier.read_events()
668
    self._notifier.process_events()
669
    return have_events
670

    
671
  def Close(self):
672
    """Closes underlying notifier and its file descriptor.
673

674
    """
675
    self._notifier.stop()
676

    
677

    
678
class _JobChangesWaiter(object):
679
  def __init__(self, filename):
680
    """Initializes this class.
681

682
    @type filename: string
683
    @param filename: Path to job file
684

685
    """
686
    self._filewaiter = None
687
    self._filename = filename
688

    
689
  def Wait(self, timeout):
690
    """Waits for a job to change.
691

692
    @type timeout: float
693
    @param timeout: Timeout in seconds
694
    @return: Whether there have been events
695

696
    """
697
    if self._filewaiter:
698
      return self._filewaiter.Wait(timeout)
699

    
700
    # Lazy setup: Avoid inotify setup cost when job file has already changed.
701
    # If this point is reached, return immediately and let caller check the job
702
    # file again in case there were changes since the last check. This avoids a
703
    # race condition.
704
    self._filewaiter = _JobFileChangesWaiter(self._filename)
705

    
706
    return True
707

    
708
  def Close(self):
709
    """Closes underlying waiter.
710

711
    """
712
    if self._filewaiter:
713
      self._filewaiter.Close()
714

    
715

    
716
class _WaitForJobChangesHelper(object):
717
  """Helper class using inotify to wait for changes in a job file.
718

719
  This class takes a previous job status and serial, and alerts the client when
720
  the current job status has changed.
721

722
  """
723
  @staticmethod
724
  def _CheckForChanges(counter, job_load_fn, check_fn):
725
    if counter.next() > 0:
726
      # If this isn't the first check the job is given some more time to change
727
      # again. This gives better performance for jobs generating many
728
      # changes/messages.
729
      time.sleep(0.1)
730

    
731
    job = job_load_fn()
732
    if not job:
733
      raise errors.JobLost()
734

    
735
    result = check_fn(job)
736
    if result is None:
737
      raise utils.RetryAgain()
738

    
739
    return result
740

    
741
  def __call__(self, filename, job_load_fn,
742
               fields, prev_job_info, prev_log_serial, timeout):
743
    """Waits for changes on a job.
744

745
    @type filename: string
746
    @param filename: File on which to wait for changes
747
    @type job_load_fn: callable
748
    @param job_load_fn: Function to load job
749
    @type fields: list of strings
750
    @param fields: Which fields to check for changes
751
    @type prev_job_info: list or None
752
    @param prev_job_info: Last job information returned
753
    @type prev_log_serial: int
754
    @param prev_log_serial: Last job message serial number
755
    @type timeout: float
756
    @param timeout: maximum time to wait in seconds
757

758
    """
759
    counter = itertools.count()
760
    try:
761
      check_fn = _JobChangesChecker(fields, prev_job_info, prev_log_serial)
762
      waiter = _JobChangesWaiter(filename)
763
      try:
764
        return utils.Retry(compat.partial(self._CheckForChanges,
765
                                          counter, job_load_fn, check_fn),
766
                           utils.RETRY_REMAINING_TIME, timeout,
767
                           wait_fn=waiter.Wait)
768
      finally:
769
        waiter.Close()
770
    except (errors.InotifyError, errors.JobLost):
771
      return None
772
    except utils.RetryTimeout:
773
      return constants.JOB_NOTCHANGED
774

    
775

    
776
def _EncodeOpError(err):
777
  """Encodes an error which occurred while processing an opcode.
778

779
  """
780
  if isinstance(err, errors.GenericError):
781
    to_encode = err
782
  else:
783
    to_encode = errors.OpExecError(str(err))
784

    
785
  return errors.EncodeException(to_encode)
786

    
787

    
788
class _TimeoutStrategyWrapper:
789
  def __init__(self, fn):
790
    """Initializes this class.
791

792
    """
793
    self._fn = fn
794
    self._next = None
795

    
796
  def _Advance(self):
797
    """Gets the next timeout if necessary.
798

799
    """
800
    if self._next is None:
801
      self._next = self._fn()
802

    
803
  def Peek(self):
804
    """Returns the next timeout.
805

806
    """
807
    self._Advance()
808
    return self._next
809

    
810
  def Next(self):
811
    """Returns the current timeout and advances the internal state.
812

813
    """
814
    self._Advance()
815
    result = self._next
816
    self._next = None
817
    return result
818

    
819

    
820
class _OpExecContext:
821
  def __init__(self, op, index, log_prefix, timeout_strategy_factory):
822
    """Initializes this class.
823

824
    """
825
    self.op = op
826
    self.index = index
827
    self.log_prefix = log_prefix
828
    self.summary = op.input.Summary()
829

    
830
    # Create local copy to modify
831
    if getattr(op.input, opcodes.DEPEND_ATTR, None):
832
      self.jobdeps = op.input.depends[:]
833
    else:
834
      self.jobdeps = None
835

    
836
    self._timeout_strategy_factory = timeout_strategy_factory
837
    self._ResetTimeoutStrategy()
838

    
839
  def _ResetTimeoutStrategy(self):
840
    """Creates a new timeout strategy.
841

842
    """
843
    self._timeout_strategy = \
844
      _TimeoutStrategyWrapper(self._timeout_strategy_factory().NextAttempt)
845

    
846
  def CheckPriorityIncrease(self):
847
    """Checks whether priority can and should be increased.
848

849
    Called when locks couldn't be acquired.
850

851
    """
852
    op = self.op
853

    
854
    # Exhausted all retries and next round should not use blocking acquire
855
    # for locks?
856
    if (self._timeout_strategy.Peek() is None and
857
        op.priority > constants.OP_PRIO_HIGHEST):
858
      logging.debug("Increasing priority")
859
      op.priority -= 1
860
      self._ResetTimeoutStrategy()
861
      return True
862

    
863
    return False
864

    
865
  def GetNextLockTimeout(self):
866
    """Returns the next lock acquire timeout.
867

868
    """
869
    return self._timeout_strategy.Next()
870

    
871

    
872
class _JobProcessor(object):
873
  (DEFER,
874
   WAITDEP,
875
   FINISHED) = range(1, 4)
876

    
877
  def __init__(self, queue, opexec_fn, job,
878
               _timeout_strategy_factory=mcpu.LockAttemptTimeoutStrategy):
879
    """Initializes this class.
880

881
    """
882
    self.queue = queue
883
    self.opexec_fn = opexec_fn
884
    self.job = job
885
    self._timeout_strategy_factory = _timeout_strategy_factory
886

    
887
  @staticmethod
888
  def _FindNextOpcode(job, timeout_strategy_factory):
889
    """Locates the next opcode to run.
890

891
    @type job: L{_QueuedJob}
892
    @param job: Job object
893
    @param timeout_strategy_factory: Callable to create new timeout strategy
894

895
    """
896
    # Create some sort of a cache to speed up locating next opcode for future
897
    # lookups
898
    # TODO: Consider splitting _QueuedJob.ops into two separate lists, one for
899
    # pending and one for processed ops.
900
    if job.ops_iter is None:
901
      job.ops_iter = enumerate(job.ops)
902

    
903
    # Find next opcode to run
904
    while True:
905
      try:
906
        (idx, op) = job.ops_iter.next()
907
      except StopIteration:
908
        raise errors.ProgrammerError("Called for a finished job")
909

    
910
      if op.status == constants.OP_STATUS_RUNNING:
911
        # Found an opcode already marked as running
912
        raise errors.ProgrammerError("Called for job marked as running")
913

    
914
      opctx = _OpExecContext(op, idx, "Op %s/%s" % (idx + 1, len(job.ops)),
915
                             timeout_strategy_factory)
916

    
917
      if op.status not in constants.OPS_FINALIZED:
918
        return opctx
919

    
920
      # This is a job that was partially completed before master daemon
921
      # shutdown, so it can be expected that some opcodes are already
922
      # completed successfully (if any did error out, then the whole job
923
      # should have been aborted and not resubmitted for processing).
924
      logging.info("%s: opcode %s already processed, skipping",
925
                   opctx.log_prefix, opctx.summary)
926

    
927
  @staticmethod
928
  def _MarkWaitlock(job, op):
929
    """Marks an opcode as waiting for locks.
930

931
    The job's start timestamp is also set if necessary.
932

933
    @type job: L{_QueuedJob}
934
    @param job: Job object
935
    @type op: L{_QueuedOpCode}
936
    @param op: Opcode object
937

938
    """
939
    assert op in job.ops
940
    assert op.status in (constants.OP_STATUS_QUEUED,
941
                         constants.OP_STATUS_WAITING)
942

    
943
    update = False
944

    
945
    op.result = None
946

    
947
    if op.status == constants.OP_STATUS_QUEUED:
948
      op.status = constants.OP_STATUS_WAITING
949
      update = True
950

    
951
    if op.start_timestamp is None:
952
      op.start_timestamp = TimeStampNow()
953
      update = True
954

    
955
    if job.start_timestamp is None:
956
      job.start_timestamp = op.start_timestamp
957
      update = True
958

    
959
    assert op.status == constants.OP_STATUS_WAITING
960

    
961
    return update
962

    
963
  @staticmethod
964
  def _CheckDependencies(queue, job, opctx):
965
    """Checks if an opcode has dependencies and if so, processes them.
966

967
    @type queue: L{JobQueue}
968
    @param queue: Queue object
969
    @type job: L{_QueuedJob}
970
    @param job: Job object
971
    @type opctx: L{_OpExecContext}
972
    @param opctx: Opcode execution context
973
    @rtype: bool
974
    @return: Whether opcode will be re-scheduled by dependency tracker
975

976
    """
977
    op = opctx.op
978

    
979
    result = False
980

    
981
    while opctx.jobdeps:
982
      (dep_job_id, dep_status) = opctx.jobdeps[0]
983

    
984
      (depresult, depmsg) = queue.depmgr.CheckAndRegister(job, dep_job_id,
985
                                                          dep_status)
986
      assert ht.TNonEmptyString(depmsg), "No dependency message"
987

    
988
      logging.info("%s: %s", opctx.log_prefix, depmsg)
989

    
990
      if depresult == _JobDependencyManager.CONTINUE:
991
        # Remove dependency and continue
992
        opctx.jobdeps.pop(0)
993

    
994
      elif depresult == _JobDependencyManager.WAIT:
995
        # Need to wait for notification, dependency tracker will re-add job
996
        # to workerpool
997
        result = True
998
        break
999

    
1000
      elif depresult == _JobDependencyManager.CANCEL:
1001
        # Job was cancelled, cancel this job as well
1002
        job.Cancel()
1003
        assert op.status == constants.OP_STATUS_CANCELING
1004
        break
1005

    
1006
      elif depresult in (_JobDependencyManager.WRONGSTATUS,
1007
                         _JobDependencyManager.ERROR):
1008
        # Job failed or there was an error, this job must fail
1009
        op.status = constants.OP_STATUS_ERROR
1010
        op.result = _EncodeOpError(errors.OpExecError(depmsg))
1011
        break
1012

    
1013
      else:
1014
        raise errors.ProgrammerError("Unknown dependency result '%s'" %
1015
                                     depresult)
1016

    
1017
    return result
1018

    
1019
  def _ExecOpCodeUnlocked(self, opctx):
1020
    """Processes one opcode and returns the result.
1021

1022
    """
1023
    op = opctx.op
1024

    
1025
    assert op.status == constants.OP_STATUS_WAITING
1026

    
1027
    timeout = opctx.GetNextLockTimeout()
1028

    
1029
    try:
1030
      # Make sure not to hold queue lock while calling ExecOpCode
1031
      result = self.opexec_fn(op.input,
1032
                              _OpExecCallbacks(self.queue, self.job, op),
1033
                              timeout=timeout, priority=op.priority)
1034
    except mcpu.LockAcquireTimeout:
1035
      assert timeout is not None, "Received timeout for blocking acquire"
1036
      logging.debug("Couldn't acquire locks in %0.6fs", timeout)
1037

    
1038
      assert op.status in (constants.OP_STATUS_WAITING,
1039
                           constants.OP_STATUS_CANCELING)
1040

    
1041
      # Was job cancelled while we were waiting for the lock?
1042
      if op.status == constants.OP_STATUS_CANCELING:
1043
        return (constants.OP_STATUS_CANCELING, None)
1044

    
1045
      # Stay in waitlock while trying to re-acquire lock
1046
      return (constants.OP_STATUS_WAITING, None)
1047
    except CancelJob:
1048
      logging.exception("%s: Canceling job", opctx.log_prefix)
1049
      assert op.status == constants.OP_STATUS_CANCELING
1050
      return (constants.OP_STATUS_CANCELING, None)
1051
    except Exception, err: # pylint: disable=W0703
1052
      logging.exception("%s: Caught exception in %s",
1053
                        opctx.log_prefix, opctx.summary)
1054
      return (constants.OP_STATUS_ERROR, _EncodeOpError(err))
1055
    else:
1056
      logging.debug("%s: %s successful",
1057
                    opctx.log_prefix, opctx.summary)
1058
      return (constants.OP_STATUS_SUCCESS, result)
1059

    
1060
  def __call__(self, _nextop_fn=None):
1061
    """Continues execution of a job.
1062

1063
    @param _nextop_fn: Callback function for tests
1064
    @return: C{FINISHED} if job is fully processed, C{DEFER} if the job should
1065
      be deferred and C{WAITDEP} if the dependency manager
1066
      (L{_JobDependencyManager}) will re-schedule the job when appropriate
1067

1068
    """
1069
    queue = self.queue
1070
    job = self.job
1071

    
1072
    logging.debug("Processing job %s", job.id)
1073

    
1074
    queue.acquire(shared=1)
1075
    try:
1076
      opcount = len(job.ops)
1077

    
1078
      assert job.writable, "Expected writable job"
1079

    
1080
      # Don't do anything for finalized jobs
1081
      if job.CalcStatus() in constants.JOBS_FINALIZED:
1082
        return self.FINISHED
1083

    
1084
      # Is a previous opcode still pending?
1085
      if job.cur_opctx:
1086
        opctx = job.cur_opctx
1087
        job.cur_opctx = None
1088
      else:
1089
        if __debug__ and _nextop_fn:
1090
          _nextop_fn()
1091
        opctx = self._FindNextOpcode(job, self._timeout_strategy_factory)
1092

    
1093
      op = opctx.op
1094

    
1095
      # Consistency check
1096
      assert compat.all(i.status in (constants.OP_STATUS_QUEUED,
1097
                                     constants.OP_STATUS_CANCELING)
1098
                        for i in job.ops[opctx.index + 1:])
1099

    
1100
      assert op.status in (constants.OP_STATUS_QUEUED,
1101
                           constants.OP_STATUS_WAITING,
1102
                           constants.OP_STATUS_CANCELING)
1103

    
1104
      assert (op.priority <= constants.OP_PRIO_LOWEST and
1105
              op.priority >= constants.OP_PRIO_HIGHEST)
1106

    
1107
      waitjob = None
1108

    
1109
      if op.status != constants.OP_STATUS_CANCELING:
1110
        assert op.status in (constants.OP_STATUS_QUEUED,
1111
                             constants.OP_STATUS_WAITING)
1112

    
1113
        # Prepare to start opcode
1114
        if self._MarkWaitlock(job, op):
1115
          # Write to disk
1116
          queue.UpdateJobUnlocked(job)
1117

    
1118
        assert op.status == constants.OP_STATUS_WAITING
1119
        assert job.CalcStatus() == constants.JOB_STATUS_WAITING
1120
        assert job.start_timestamp and op.start_timestamp
1121
        assert waitjob is None
1122

    
1123
        # Check if waiting for a job is necessary
1124
        waitjob = self._CheckDependencies(queue, job, opctx)
1125

    
1126
        assert op.status in (constants.OP_STATUS_WAITING,
1127
                             constants.OP_STATUS_CANCELING,
1128
                             constants.OP_STATUS_ERROR)
1129

    
1130
        if not (waitjob or op.status in (constants.OP_STATUS_CANCELING,
1131
                                         constants.OP_STATUS_ERROR)):
1132
          logging.info("%s: opcode %s waiting for locks",
1133
                       opctx.log_prefix, opctx.summary)
1134

    
1135
          assert not opctx.jobdeps, "Not all dependencies were removed"
1136

    
1137
          queue.release()
1138
          try:
1139
            (op_status, op_result) = self._ExecOpCodeUnlocked(opctx)
1140
          finally:
1141
            queue.acquire(shared=1)
1142

    
1143
          op.status = op_status
1144
          op.result = op_result
1145

    
1146
          assert not waitjob
1147

    
1148
        if op.status == constants.OP_STATUS_WAITING:
1149
          # Couldn't get locks in time
1150
          assert not op.end_timestamp
1151
        else:
1152
          # Finalize opcode
1153
          op.end_timestamp = TimeStampNow()
1154

    
1155
          if op.status == constants.OP_STATUS_CANCELING:
1156
            assert not compat.any(i.status != constants.OP_STATUS_CANCELING
1157
                                  for i in job.ops[opctx.index:])
1158
          else:
1159
            assert op.status in constants.OPS_FINALIZED
1160

    
1161
      if op.status == constants.OP_STATUS_WAITING or waitjob:
1162
        finalize = False
1163

    
1164
        if not waitjob and opctx.CheckPriorityIncrease():
1165
          # Priority was changed, need to update on-disk file
1166
          queue.UpdateJobUnlocked(job)
1167

    
1168
        # Keep around for another round
1169
        job.cur_opctx = opctx
1170

    
1171
        assert (op.priority <= constants.OP_PRIO_LOWEST and
1172
                op.priority >= constants.OP_PRIO_HIGHEST)
1173

    
1174
        # In no case must the status be finalized here
1175
        assert job.CalcStatus() == constants.JOB_STATUS_WAITING
1176

    
1177
      else:
1178
        # Ensure all opcodes so far have been successful
1179
        assert (opctx.index == 0 or
1180
                compat.all(i.status == constants.OP_STATUS_SUCCESS
1181
                           for i in job.ops[:opctx.index]))
1182

    
1183
        # Reset context
1184
        job.cur_opctx = None
1185

    
1186
        if op.status == constants.OP_STATUS_SUCCESS:
1187
          finalize = False
1188

    
1189
        elif op.status == constants.OP_STATUS_ERROR:
1190
          # Ensure failed opcode has an exception as its result
1191
          assert errors.GetEncodedError(job.ops[opctx.index].result)
1192

    
1193
          to_encode = errors.OpExecError("Preceding opcode failed")
1194
          job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
1195
                                _EncodeOpError(to_encode))
1196
          finalize = True
1197

    
1198
          # Consistency check
1199
          assert compat.all(i.status == constants.OP_STATUS_ERROR and
1200
                            errors.GetEncodedError(i.result)
1201
                            for i in job.ops[opctx.index:])
1202

    
1203
        elif op.status == constants.OP_STATUS_CANCELING:
1204
          job.MarkUnfinishedOps(constants.OP_STATUS_CANCELED,
1205
                                "Job canceled by request")
1206
          finalize = True
1207

    
1208
        else:
1209
          raise errors.ProgrammerError("Unknown status '%s'" % op.status)
1210

    
1211
        if opctx.index == (opcount - 1):
1212
          # Finalize on last opcode
1213
          finalize = True
1214

    
1215
        if finalize:
1216
          # All opcodes have been run, finalize job
1217
          job.Finalize()
1218

    
1219
        # Write to disk. If the job status is final, this is the final write
1220
        # allowed. Once the file has been written, it can be archived anytime.
1221
        queue.UpdateJobUnlocked(job)
1222

    
1223
        assert not waitjob
1224

    
1225
        if finalize:
1226
          logging.info("Finished job %s, status = %s", job.id, job.CalcStatus())
1227
          return self.FINISHED
1228

    
1229
      assert not waitjob or queue.depmgr.JobWaiting(job)
1230

    
1231
      if waitjob:
1232
        return self.WAITDEP
1233
      else:
1234
        return self.DEFER
1235
    finally:
1236
      assert job.writable, "Job became read-only while being processed"
1237
      queue.release()
1238

    
1239

    
1240
def _EvaluateJobProcessorResult(depmgr, job, result):
1241
  """Looks at a result from L{_JobProcessor} for a job.
1242

1243
  To be used in a L{_JobQueueWorker}.
1244

1245
  """
1246
  if result == _JobProcessor.FINISHED:
1247
    # Notify waiting jobs
1248
    depmgr.NotifyWaiters(job.id)
1249

    
1250
  elif result == _JobProcessor.DEFER:
1251
    # Schedule again
1252
    raise workerpool.DeferTask(priority=job.CalcPriority())
1253

    
1254
  elif result == _JobProcessor.WAITDEP:
1255
    # No-op, dependency manager will re-schedule
1256
    pass
1257

    
1258
  else:
1259
    raise errors.ProgrammerError("Job processor returned unknown status %s" %
1260
                                 (result, ))
1261

    
1262

    
1263
class _JobQueueWorker(workerpool.BaseWorker):
1264
  """The actual job workers.
1265

1266
  """
1267
  def RunTask(self, job): # pylint: disable=W0221
1268
    """Job executor.
1269

1270
    @type job: L{_QueuedJob}
1271
    @param job: the job to be processed
1272

1273
    """
1274
    assert job.writable, "Expected writable job"
1275

    
1276
    # Ensure only one worker is active on a single job. If a job registers for
1277
    # a dependency job, and the other job notifies before the first worker is
1278
    # done, the job can end up in the tasklist more than once.
1279
    job.processor_lock.acquire()
1280
    try:
1281
      return self._RunTaskInner(job)
1282
    finally:
1283
      job.processor_lock.release()
1284

    
1285
  def _RunTaskInner(self, job):
1286
    """Executes a job.
1287

1288
    Must be called with per-job lock acquired.
1289

1290
    """
1291
    queue = job.queue
1292
    assert queue == self.pool.queue
1293

    
1294
    setname_fn = lambda op: self.SetTaskName(self._GetWorkerName(job, op))
1295
    setname_fn(None)
1296

    
1297
    proc = mcpu.Processor(queue.context, job.id)
1298

    
1299
    # Create wrapper for setting thread name
1300
    wrap_execop_fn = compat.partial(self._WrapExecOpCode, setname_fn,
1301
                                    proc.ExecOpCode)
1302

    
1303
    _EvaluateJobProcessorResult(queue.depmgr, job,
1304
                                _JobProcessor(queue, wrap_execop_fn, job)())
1305

    
1306
  @staticmethod
1307
  def _WrapExecOpCode(setname_fn, execop_fn, op, *args, **kwargs):
1308
    """Updates the worker thread name to include a short summary of the opcode.
1309

1310
    @param setname_fn: Callable setting worker thread name
1311
    @param execop_fn: Callable for executing opcode (usually
1312
                      L{mcpu.Processor.ExecOpCode})
1313

1314
    """
1315
    setname_fn(op)
1316
    try:
1317
      return execop_fn(op, *args, **kwargs)
1318
    finally:
1319
      setname_fn(None)
1320

    
1321
  @staticmethod
1322
  def _GetWorkerName(job, op):
1323
    """Sets the worker thread name.
1324

1325
    @type job: L{_QueuedJob}
1326
    @type op: L{opcodes.OpCode}
1327

1328
    """
1329
    parts = ["Job%s" % job.id]
1330

    
1331
    if op:
1332
      parts.append(op.TinySummary())
1333

    
1334
    return "/".join(parts)
1335

    
1336

    
1337
class _JobQueueWorkerPool(workerpool.WorkerPool):
1338
  """Simple class implementing a job-processing workerpool.
1339

1340
  """
1341
  def __init__(self, queue):
1342
    super(_JobQueueWorkerPool, self).__init__("Jq",
1343
                                              JOBQUEUE_THREADS,
1344
                                              _JobQueueWorker)
1345
    self.queue = queue
1346

    
1347

    
1348
class _JobDependencyManager:
1349
  """Keeps track of job dependencies.
1350

1351
  """
1352
  (WAIT,
1353
   ERROR,
1354
   CANCEL,
1355
   CONTINUE,
1356
   WRONGSTATUS) = range(1, 6)
1357

    
1358
  def __init__(self, getstatus_fn, enqueue_fn):
1359
    """Initializes this class.
1360

1361
    """
1362
    self._getstatus_fn = getstatus_fn
1363
    self._enqueue_fn = enqueue_fn
1364

    
1365
    self._waiters = {}
1366
    self._lock = locking.SharedLock("JobDepMgr")
1367

    
1368
  @locking.ssynchronized(_LOCK, shared=1)
1369
  def GetLockInfo(self, requested): # pylint: disable=W0613
1370
    """Retrieves information about waiting jobs.
1371

1372
    @type requested: set
1373
    @param requested: Requested information, see C{query.LQ_*}
1374

1375
    """
1376
    # No need to sort here, that's being done by the lock manager and query
1377
    # library. There are no priorities for notifying jobs, hence all show up as
1378
    # one item under "pending".
1379
    return [("job/%s" % job_id, None, None,
1380
             [("job", [job.id for job in waiters])])
1381
            for job_id, waiters in self._waiters.items()
1382
            if waiters]
1383

    
1384
  @locking.ssynchronized(_LOCK, shared=1)
1385
  def JobWaiting(self, job):
1386
    """Checks if a job is waiting.
1387

1388
    """
1389
    return compat.any(job in jobs
1390
                      for jobs in self._waiters.values())
1391

    
1392
  @locking.ssynchronized(_LOCK)
1393
  def CheckAndRegister(self, job, dep_job_id, dep_status):
1394
    """Checks if a dependency job has the requested status.
1395

1396
    If the other job is not yet in a finalized status, the calling job will be
1397
    notified (re-added to the workerpool) at a later point.
1398

1399
    @type job: L{_QueuedJob}
1400
    @param job: Job object
1401
    @type dep_job_id: string
1402
    @param dep_job_id: ID of dependency job
1403
    @type dep_status: list
1404
    @param dep_status: Required status
1405

1406
    """
1407
    assert ht.TString(job.id)
1408
    assert ht.TString(dep_job_id)
1409
    assert ht.TListOf(ht.TElemOf(constants.JOBS_FINALIZED))(dep_status)
1410

    
1411
    if job.id == dep_job_id:
1412
      return (self.ERROR, "Job can't depend on itself")
1413

    
1414
    # Get status of dependency job
1415
    try:
1416
      status = self._getstatus_fn(dep_job_id)
1417
    except errors.JobLost, err:
1418
      return (self.ERROR, "Dependency error: %s" % err)
1419

    
1420
    assert status in constants.JOB_STATUS_ALL
1421

    
1422
    job_id_waiters = self._waiters.setdefault(dep_job_id, set())
1423

    
1424
    if status not in constants.JOBS_FINALIZED:
1425
      # Register for notification and wait for job to finish
1426
      job_id_waiters.add(job)
1427
      return (self.WAIT,
1428
              "Need to wait for job %s, wanted status '%s'" %
1429
              (dep_job_id, dep_status))
1430

    
1431
    # Remove from waiters list
1432
    if job in job_id_waiters:
1433
      job_id_waiters.remove(job)
1434

    
1435
    if (status == constants.JOB_STATUS_CANCELED and
1436
        constants.JOB_STATUS_CANCELED not in dep_status):
1437
      return (self.CANCEL, "Dependency job %s was cancelled" % dep_job_id)
1438

    
1439
    elif not dep_status or status in dep_status:
1440
      return (self.CONTINUE,
1441
              "Dependency job %s finished with status '%s'" %
1442
              (dep_job_id, status))
1443

    
1444
    else:
1445
      return (self.WRONGSTATUS,
1446
              "Dependency job %s finished with status '%s',"
1447
              " not one of '%s' as required" %
1448
              (dep_job_id, status, utils.CommaJoin(dep_status)))
1449

    
1450
  def _RemoveEmptyWaitersUnlocked(self):
1451
    """Remove all jobs without actual waiters.
1452

1453
    """
1454
    for job_id in [job_id for (job_id, waiters) in self._waiters.items()
1455
                   if not waiters]:
1456
      del self._waiters[job_id]
1457

    
1458
  def NotifyWaiters(self, job_id):
1459
    """Notifies all jobs waiting for a certain job ID.
1460

1461
    @attention: Do not call until L{CheckAndRegister} returned a status other
1462
      than C{WAITDEP} for C{job_id}, or behaviour is undefined
1463
    @type job_id: string
1464
    @param job_id: Job ID
1465

1466
    """
1467
    assert ht.TString(job_id)
1468

    
1469
    self._lock.acquire()
1470
    try:
1471
      self._RemoveEmptyWaitersUnlocked()
1472

    
1473
      jobs = self._waiters.pop(job_id, None)
1474
    finally:
1475
      self._lock.release()
1476

    
1477
    if jobs:
1478
      # Re-add jobs to workerpool
1479
      logging.debug("Re-adding %s jobs which were waiting for job %s",
1480
                    len(jobs), job_id)
1481
      self._enqueue_fn(jobs)
1482

    
1483

    
1484
def _RequireOpenQueue(fn):
1485
  """Decorator for "public" functions.
1486

1487
  This function should be used for all 'public' functions. That is,
1488
  functions usually called from other classes. Note that this should
1489
  be applied only to methods (not plain functions), since it expects
1490
  that the decorated function is called with a first argument that has
1491
  a '_queue_filelock' argument.
1492

1493
  @warning: Use this decorator only after locking.ssynchronized
1494

1495
  Example::
1496
    @locking.ssynchronized(_LOCK)
1497
    @_RequireOpenQueue
1498
    def Example(self):
1499
      pass
1500

1501
  """
1502
  def wrapper(self, *args, **kwargs):
1503
    # pylint: disable=W0212
1504
    assert self._queue_filelock is not None, "Queue should be open"
1505
    return fn(self, *args, **kwargs)
1506
  return wrapper
1507

    
1508

    
1509
def _RequireNonDrainedQueue(fn):
1510
  """Decorator checking for a non-drained queue.
1511

1512
  To be used with functions submitting new jobs.
1513

1514
  """
1515
  def wrapper(self, *args, **kwargs):
1516
    """Wrapper function.
1517

1518
    @raise errors.JobQueueDrainError: if the job queue is marked for draining
1519

1520
    """
1521
    # Ok when sharing the big job queue lock, as the drain file is created when
1522
    # the lock is exclusive.
1523
    # Needs access to protected member, pylint: disable=W0212
1524
    if self._drained:
1525
      raise errors.JobQueueDrainError("Job queue is drained, refusing job")
1526

    
1527
    if not self._accepting_jobs:
1528
      raise errors.JobQueueError("Job queue is shutting down, refusing job")
1529

    
1530
    return fn(self, *args, **kwargs)
1531
  return wrapper
1532

    
1533

    
1534
class JobQueue(object):
1535
  """Queue used to manage the jobs.
1536

1537
  """
1538
  def __init__(self, context):
1539
    """Constructor for JobQueue.
1540

1541
    The constructor will initialize the job queue object and then
1542
    start loading the current jobs from disk, either for starting them
1543
    (if they were queue) or for aborting them (if they were already
1544
    running).
1545

1546
    @type context: GanetiContext
1547
    @param context: the context object for access to the configuration
1548
        data and other ganeti objects
1549

1550
    """
1551
    self.context = context
1552
    self._memcache = weakref.WeakValueDictionary()
1553
    self._my_hostname = netutils.Hostname.GetSysName()
1554

    
1555
    # The Big JobQueue lock. If a code block or method acquires it in shared
1556
    # mode safe it must guarantee concurrency with all the code acquiring it in
1557
    # shared mode, including itself. In order not to acquire it at all
1558
    # concurrency must be guaranteed with all code acquiring it in shared mode
1559
    # and all code acquiring it exclusively.
1560
    self._lock = locking.SharedLock("JobQueue")
1561

    
1562
    self.acquire = self._lock.acquire
1563
    self.release = self._lock.release
1564

    
1565
    # Accept jobs by default
1566
    self._accepting_jobs = True
1567

    
1568
    # Initialize the queue, and acquire the filelock.
1569
    # This ensures no other process is working on the job queue.
1570
    self._queue_filelock = jstore.InitAndVerifyQueue(must_lock=True)
1571

    
1572
    # Read serial file
1573
    self._last_serial = jstore.ReadSerial()
1574
    assert self._last_serial is not None, ("Serial file was modified between"
1575
                                           " check in jstore and here")
1576

    
1577
    # Get initial list of nodes
1578
    self._nodes = dict((n.name, n.primary_ip)
1579
                       for n in self.context.cfg.GetAllNodesInfo().values()
1580
                       if n.master_candidate)
1581

    
1582
    # Remove master node
1583
    self._nodes.pop(self._my_hostname, None)
1584

    
1585
    # TODO: Check consistency across nodes
1586

    
1587
    self._queue_size = None
1588
    self._UpdateQueueSizeUnlocked()
1589
    assert ht.TInt(self._queue_size)
1590
    self._drained = jstore.CheckDrainFlag()
1591

    
1592
    # Job dependencies
1593
    self.depmgr = _JobDependencyManager(self._GetJobStatusForDependencies,
1594
                                        self._EnqueueJobs)
1595
    self.context.glm.AddToLockMonitor(self.depmgr)
1596

    
1597
    # Setup worker pool
1598
    self._wpool = _JobQueueWorkerPool(self)
1599
    try:
1600
      self._InspectQueue()
1601
    except:
1602
      self._wpool.TerminateWorkers()
1603
      raise
1604

    
1605
  @locking.ssynchronized(_LOCK)
1606
  @_RequireOpenQueue
1607
  def _InspectQueue(self):
1608
    """Loads the whole job queue and resumes unfinished jobs.
1609

1610
    This function needs the lock here because WorkerPool.AddTask() may start a
1611
    job while we're still doing our work.
1612

1613
    """
1614
    logging.info("Inspecting job queue")
1615

    
1616
    restartjobs = []
1617

    
1618
    all_job_ids = self._GetJobIDsUnlocked()
1619
    jobs_count = len(all_job_ids)
1620
    lastinfo = time.time()
1621
    for idx, job_id in enumerate(all_job_ids):
1622
      # Give an update every 1000 jobs or 10 seconds
1623
      if (idx % 1000 == 0 or time.time() >= (lastinfo + 10.0) or
1624
          idx == (jobs_count - 1)):
1625
        logging.info("Job queue inspection: %d/%d (%0.1f %%)",
1626
                     idx, jobs_count - 1, 100.0 * (idx + 1) / jobs_count)
1627
        lastinfo = time.time()
1628

    
1629
      job = self._LoadJobUnlocked(job_id)
1630

    
1631
      # a failure in loading the job can cause 'None' to be returned
1632
      if job is None:
1633
        continue
1634

    
1635
      status = job.CalcStatus()
1636

    
1637
      if status == constants.JOB_STATUS_QUEUED:
1638
        restartjobs.append(job)
1639

    
1640
      elif status in (constants.JOB_STATUS_RUNNING,
1641
                      constants.JOB_STATUS_WAITING,
1642
                      constants.JOB_STATUS_CANCELING):
1643
        logging.warning("Unfinished job %s found: %s", job.id, job)
1644

    
1645
        if status == constants.JOB_STATUS_WAITING:
1646
          # Restart job
1647
          job.MarkUnfinishedOps(constants.OP_STATUS_QUEUED, None)
1648
          restartjobs.append(job)
1649
        else:
1650
          job.MarkUnfinishedOps(constants.OP_STATUS_ERROR,
1651
                                "Unclean master daemon shutdown")
1652
          job.Finalize()
1653

    
1654
        self.UpdateJobUnlocked(job)
1655

    
1656
    if restartjobs:
1657
      logging.info("Restarting %s jobs", len(restartjobs))
1658
      self._EnqueueJobsUnlocked(restartjobs)
1659

    
1660
    logging.info("Job queue inspection finished")
1661

    
1662
  def _GetRpc(self, address_list):
1663
    """Gets RPC runner with context.
1664

1665
    """
1666
    return rpc.JobQueueRunner(self.context, address_list)
1667

    
1668
  @locking.ssynchronized(_LOCK)
1669
  @_RequireOpenQueue
1670
  def AddNode(self, node):
1671
    """Register a new node with the queue.
1672

1673
    @type node: L{objects.Node}
1674
    @param node: the node object to be added
1675

1676
    """
1677
    node_name = node.name
1678
    assert node_name != self._my_hostname
1679

    
1680
    # Clean queue directory on added node
1681
    result = self._GetRpc(None).call_jobqueue_purge(node_name)
1682
    msg = result.fail_msg
1683
    if msg:
1684
      logging.warning("Cannot cleanup queue directory on node %s: %s",
1685
                      node_name, msg)
1686

    
1687
    if not node.master_candidate:
1688
      # remove if existing, ignoring errors
1689
      self._nodes.pop(node_name, None)
1690
      # and skip the replication of the job ids
1691
      return
1692

    
1693
    # Upload the whole queue excluding archived jobs
1694
    files = [self._GetJobPath(job_id) for job_id in self._GetJobIDsUnlocked()]
1695

    
1696
    # Upload current serial file
1697
    files.append(constants.JOB_QUEUE_SERIAL_FILE)
1698

    
1699
    # Static address list
1700
    addrs = [node.primary_ip]
1701

    
1702
    for file_name in files:
1703
      # Read file content
1704
      content = utils.ReadFile(file_name)
1705

    
1706
      result = self._GetRpc(addrs).call_jobqueue_update([node_name], file_name,
1707
                                                        content)
1708
      msg = result[node_name].fail_msg
1709
      if msg:
1710
        logging.error("Failed to upload file %s to node %s: %s",
1711
                      file_name, node_name, msg)
1712

    
1713
    self._nodes[node_name] = node.primary_ip
1714

    
1715
  @locking.ssynchronized(_LOCK)
1716
  @_RequireOpenQueue
1717
  def RemoveNode(self, node_name):
1718
    """Callback called when removing nodes from the cluster.
1719

1720
    @type node_name: str
1721
    @param node_name: the name of the node to remove
1722

1723
    """
1724
    self._nodes.pop(node_name, None)
1725

    
1726
  @staticmethod
1727
  def _CheckRpcResult(result, nodes, failmsg):
1728
    """Verifies the status of an RPC call.
1729

1730
    Since we aim to keep consistency should this node (the current
1731
    master) fail, we will log errors if our rpc fail, and especially
1732
    log the case when more than half of the nodes fails.
1733

1734
    @param result: the data as returned from the rpc call
1735
    @type nodes: list
1736
    @param nodes: the list of nodes we made the call to
1737
    @type failmsg: str
1738
    @param failmsg: the identifier to be used for logging
1739

1740
    """
1741
    failed = []
1742
    success = []
1743

    
1744
    for node in nodes:
1745
      msg = result[node].fail_msg
1746
      if msg:
1747
        failed.append(node)
1748
        logging.error("RPC call %s (%s) failed on node %s: %s",
1749
                      result[node].call, failmsg, node, msg)
1750
      else:
1751
        success.append(node)
1752

    
1753
    # +1 for the master node
1754
    if (len(success) + 1) < len(failed):
1755
      # TODO: Handle failing nodes
1756
      logging.error("More than half of the nodes failed")
1757

    
1758
  def _GetNodeIp(self):
1759
    """Helper for returning the node name/ip list.
1760

1761
    @rtype: (list, list)
1762
    @return: a tuple of two lists, the first one with the node
1763
        names and the second one with the node addresses
1764

1765
    """
1766
    # TODO: Change to "tuple(map(list, zip(*self._nodes.items())))"?
1767
    name_list = self._nodes.keys()
1768
    addr_list = [self._nodes[name] for name in name_list]
1769
    return name_list, addr_list
1770

    
1771
  def _UpdateJobQueueFile(self, file_name, data, replicate):
1772
    """Writes a file locally and then replicates it to all nodes.
1773

1774
    This function will replace the contents of a file on the local
1775
    node and then replicate it to all the other nodes we have.
1776

1777
    @type file_name: str
1778
    @param file_name: the path of the file to be replicated
1779
    @type data: str
1780
    @param data: the new contents of the file
1781
    @type replicate: boolean
1782
    @param replicate: whether to spread the changes to the remote nodes
1783

1784
    """
1785
    getents = runtime.GetEnts()
1786
    utils.WriteFile(file_name, data=data, uid=getents.masterd_uid,
1787
                    gid=getents.masterd_gid)
1788

    
1789
    if replicate:
1790
      names, addrs = self._GetNodeIp()
1791
      result = self._GetRpc(addrs).call_jobqueue_update(names, file_name, data)
1792
      self._CheckRpcResult(result, self._nodes, "Updating %s" % file_name)
1793

    
1794
  def _RenameFilesUnlocked(self, rename):
1795
    """Renames a file locally and then replicate the change.
1796

1797
    This function will rename a file in the local queue directory
1798
    and then replicate this rename to all the other nodes we have.
1799

1800
    @type rename: list of (old, new)
1801
    @param rename: List containing tuples mapping old to new names
1802

1803
    """
1804
    # Rename them locally
1805
    for old, new in rename:
1806
      utils.RenameFile(old, new, mkdir=True)
1807

    
1808
    # ... and on all nodes
1809
    names, addrs = self._GetNodeIp()
1810
    result = self._GetRpc(addrs).call_jobqueue_rename(names, rename)
1811
    self._CheckRpcResult(result, self._nodes, "Renaming files (%r)" % rename)
1812

    
1813
  @staticmethod
1814
  def _FormatJobID(job_id):
1815
    """Convert a job ID to string format.
1816

1817
    Currently this just does C{str(job_id)} after performing some
1818
    checks, but if we want to change the job id format this will
1819
    abstract this change.
1820

1821
    @type job_id: int or long
1822
    @param job_id: the numeric job id
1823
    @rtype: str
1824
    @return: the formatted job id
1825

1826
    """
1827
    if not isinstance(job_id, (int, long)):
1828
      raise errors.ProgrammerError("Job ID '%s' not numeric" % job_id)
1829
    if job_id < 0:
1830
      raise errors.ProgrammerError("Job ID %s is negative" % job_id)
1831

    
1832
    return str(job_id)
1833

    
1834
  @classmethod
1835
  def _GetArchiveDirectory(cls, job_id):
1836
    """Returns the archive directory for a job.
1837

1838
    @type job_id: str
1839
    @param job_id: Job identifier
1840
    @rtype: str
1841
    @return: Directory name
1842

1843
    """
1844
    return str(int(job_id) / JOBS_PER_ARCHIVE_DIRECTORY)
1845

    
1846
  def _NewSerialsUnlocked(self, count):
1847
    """Generates a new job identifier.
1848

1849
    Job identifiers are unique during the lifetime of a cluster.
1850

1851
    @type count: integer
1852
    @param count: how many serials to return
1853
    @rtype: str
1854
    @return: a string representing the job identifier.
1855

1856
    """
1857
    assert ht.TPositiveInt(count)
1858

    
1859
    # New number
1860
    serial = self._last_serial + count
1861

    
1862
    # Write to file
1863
    self._UpdateJobQueueFile(constants.JOB_QUEUE_SERIAL_FILE,
1864
                             "%s\n" % serial, True)
1865

    
1866
    result = [self._FormatJobID(v)
1867
              for v in range(self._last_serial + 1, serial + 1)]
1868

    
1869
    # Keep it only if we were able to write the file
1870
    self._last_serial = serial
1871

    
1872
    assert len(result) == count
1873

    
1874
    return result
1875

    
1876
  @staticmethod
1877
  def _GetJobPath(job_id):
1878
    """Returns the job file for a given job id.
1879

1880
    @type job_id: str
1881
    @param job_id: the job identifier
1882
    @rtype: str
1883
    @return: the path to the job file
1884

1885
    """
1886
    return utils.PathJoin(constants.QUEUE_DIR, "job-%s" % job_id)
1887

    
1888
  @classmethod
1889
  def _GetArchivedJobPath(cls, job_id):
1890
    """Returns the archived job file for a give job id.
1891

1892
    @type job_id: str
1893
    @param job_id: the job identifier
1894
    @rtype: str
1895
    @return: the path to the archived job file
1896

1897
    """
1898
    return utils.PathJoin(constants.JOB_QUEUE_ARCHIVE_DIR,
1899
                          cls._GetArchiveDirectory(job_id), "job-%s" % job_id)
1900

    
1901
  @staticmethod
1902
  def _GetJobIDsUnlocked(sort=True):
1903
    """Return all known job IDs.
1904

1905
    The method only looks at disk because it's a requirement that all
1906
    jobs are present on disk (so in the _memcache we don't have any
1907
    extra IDs).
1908

1909
    @type sort: boolean
1910
    @param sort: perform sorting on the returned job ids
1911
    @rtype: list
1912
    @return: the list of job IDs
1913

1914
    """
1915
    jlist = []
1916
    for filename in utils.ListVisibleFiles(constants.QUEUE_DIR):
1917
      m = constants.JOB_FILE_RE.match(filename)
1918
      if m:
1919
        jlist.append(m.group(1))
1920
    if sort:
1921
      jlist = utils.NiceSort(jlist)
1922
    return jlist
1923

    
1924
  def _LoadJobUnlocked(self, job_id):
1925
    """Loads a job from the disk or memory.
1926

1927
    Given a job id, this will return the cached job object if
1928
    existing, or try to load the job from the disk. If loading from
1929
    disk, it will also add the job to the cache.
1930

1931
    @param job_id: the job id
1932
    @rtype: L{_QueuedJob} or None
1933
    @return: either None or the job object
1934

1935
    """
1936
    job = self._memcache.get(job_id, None)
1937
    if job:
1938
      logging.debug("Found job %s in memcache", job_id)
1939
      assert job.writable, "Found read-only job in memcache"
1940
      return job
1941

    
1942
    try:
1943
      job = self._LoadJobFromDisk(job_id, False)
1944
      if job is None:
1945
        return job
1946
    except errors.JobFileCorrupted:
1947
      old_path = self._GetJobPath(job_id)
1948
      new_path = self._GetArchivedJobPath(job_id)
1949
      if old_path == new_path:
1950
        # job already archived (future case)
1951
        logging.exception("Can't parse job %s", job_id)
1952
      else:
1953
        # non-archived case
1954
        logging.exception("Can't parse job %s, will archive.", job_id)
1955
        self._RenameFilesUnlocked([(old_path, new_path)])
1956
      return None
1957

    
1958
    assert job.writable, "Job just loaded is not writable"
1959

    
1960
    self._memcache[job_id] = job
1961
    logging.debug("Added job %s to the cache", job_id)
1962
    return job
1963

    
1964
  def _LoadJobFromDisk(self, job_id, try_archived, writable=None):
1965
    """Load the given job file from disk.
1966

1967
    Given a job file, read, load and restore it in a _QueuedJob format.
1968

1969
    @type job_id: string
1970
    @param job_id: job identifier
1971
    @type try_archived: bool
1972
    @param try_archived: Whether to try loading an archived job
1973
    @rtype: L{_QueuedJob} or None
1974
    @return: either None or the job object
1975

1976
    """
1977
    path_functions = [(self._GetJobPath, True)]
1978

    
1979
    if try_archived:
1980
      path_functions.append((self._GetArchivedJobPath, False))
1981

    
1982
    raw_data = None
1983
    writable_default = None
1984

    
1985
    for (fn, writable_default) in path_functions:
1986
      filepath = fn(job_id)
1987
      logging.debug("Loading job from %s", filepath)
1988
      try:
1989
        raw_data = utils.ReadFile(filepath)
1990
      except EnvironmentError, err:
1991
        if err.errno != errno.ENOENT:
1992
          raise
1993
      else:
1994
        break
1995

    
1996
    if not raw_data:
1997
      return None
1998

    
1999
    if writable is None:
2000
      writable = writable_default
2001

    
2002
    try:
2003
      data = serializer.LoadJson(raw_data)
2004
      job = _QueuedJob.Restore(self, data, writable)
2005
    except Exception, err: # pylint: disable=W0703
2006
      raise errors.JobFileCorrupted(err)
2007

    
2008
    return job
2009

    
2010
  def SafeLoadJobFromDisk(self, job_id, try_archived, writable=None):
2011
    """Load the given job file from disk.
2012

2013
    Given a job file, read, load and restore it in a _QueuedJob format.
2014
    In case of error reading the job, it gets returned as None, and the
2015
    exception is logged.
2016

2017
    @type job_id: string
2018
    @param job_id: job identifier
2019
    @type try_archived: bool
2020
    @param try_archived: Whether to try loading an archived job
2021
    @rtype: L{_QueuedJob} or None
2022
    @return: either None or the job object
2023

2024
    """
2025
    try:
2026
      return self._LoadJobFromDisk(job_id, try_archived, writable=writable)
2027
    except (errors.JobFileCorrupted, EnvironmentError):
2028
      logging.exception("Can't load/parse job %s", job_id)
2029
      return None
2030

    
2031
  def _UpdateQueueSizeUnlocked(self):
2032
    """Update the queue size.
2033

2034
    """
2035
    self._queue_size = len(self._GetJobIDsUnlocked(sort=False))
2036

    
2037
  @locking.ssynchronized(_LOCK)
2038
  @_RequireOpenQueue
2039
  def SetDrainFlag(self, drain_flag):
2040
    """Sets the drain flag for the queue.
2041

2042
    @type drain_flag: boolean
2043
    @param drain_flag: Whether to set or unset the drain flag
2044

2045
    """
2046
    jstore.SetDrainFlag(drain_flag)
2047

    
2048
    self._drained = drain_flag
2049

    
2050
    return True
2051

    
2052
  @_RequireOpenQueue
2053
  def _SubmitJobUnlocked(self, job_id, ops):
2054
    """Create and store a new job.
2055

2056
    This enters the job into our job queue and also puts it on the new
2057
    queue, in order for it to be picked up by the queue processors.
2058

2059
    @type job_id: job ID
2060
    @param job_id: the job ID for the new job
2061
    @type ops: list
2062
    @param ops: The list of OpCodes that will become the new job.
2063
    @rtype: L{_QueuedJob}
2064
    @return: the job object to be queued
2065
    @raise errors.JobQueueFull: if the job queue has too many jobs in it
2066
    @raise errors.GenericError: If an opcode is not valid
2067

2068
    """
2069
    if self._queue_size >= constants.JOB_QUEUE_SIZE_HARD_LIMIT:
2070
      raise errors.JobQueueFull()
2071

    
2072
    job = _QueuedJob(self, job_id, ops, True)
2073

    
2074
    # Check priority
2075
    for idx, op in enumerate(job.ops):
2076
      if op.priority not in constants.OP_PRIO_SUBMIT_VALID:
2077
        allowed = utils.CommaJoin(constants.OP_PRIO_SUBMIT_VALID)
2078
        raise errors.GenericError("Opcode %s has invalid priority %s, allowed"
2079
                                  " are %s" % (idx, op.priority, allowed))
2080

    
2081
      dependencies = getattr(op.input, opcodes.DEPEND_ATTR, None)
2082
      if not opcodes.TNoRelativeJobDependencies(dependencies):
2083
        raise errors.GenericError("Opcode %s has invalid dependencies, must"
2084
                                  " match %s: %s" %
2085
                                  (idx, opcodes.TNoRelativeJobDependencies,
2086
                                   dependencies))
2087

    
2088
    # Write to disk
2089
    self.UpdateJobUnlocked(job)
2090

    
2091
    self._queue_size += 1
2092

    
2093
    logging.debug("Adding new job %s to the cache", job_id)
2094
    self._memcache[job_id] = job
2095

    
2096
    return job
2097

    
2098
  @locking.ssynchronized(_LOCK)
2099
  @_RequireOpenQueue
2100
  @_RequireNonDrainedQueue
2101
  def SubmitJob(self, ops):
2102
    """Create and store a new job.
2103

2104
    @see: L{_SubmitJobUnlocked}
2105

2106
    """
2107
    (job_id, ) = self._NewSerialsUnlocked(1)
2108
    self._EnqueueJobsUnlocked([self._SubmitJobUnlocked(job_id, ops)])
2109
    return job_id
2110

    
2111
  @locking.ssynchronized(_LOCK)
2112
  @_RequireOpenQueue
2113
  @_RequireNonDrainedQueue
2114
  def SubmitManyJobs(self, jobs):
2115
    """Create and store multiple jobs.
2116

2117
    @see: L{_SubmitJobUnlocked}
2118

2119
    """
2120
    all_job_ids = self._NewSerialsUnlocked(len(jobs))
2121

    
2122
    (results, added_jobs) = \
2123
      self._SubmitManyJobsUnlocked(jobs, all_job_ids, [])
2124

    
2125
    self._EnqueueJobsUnlocked(added_jobs)
2126

    
2127
    return results
2128

    
2129
  @staticmethod
2130
  def _FormatSubmitError(msg, ops):
2131
    """Formats errors which occurred while submitting a job.
2132

2133
    """
2134
    return ("%s; opcodes %s" %
2135
            (msg, utils.CommaJoin(op.Summary() for op in ops)))
2136

    
2137
  @staticmethod
2138
  def _ResolveJobDependencies(resolve_fn, deps):
2139
    """Resolves relative job IDs in dependencies.
2140

2141
    @type resolve_fn: callable
2142
    @param resolve_fn: Function to resolve a relative job ID
2143
    @type deps: list
2144
    @param deps: Dependencies
2145
    @rtype: list
2146
    @return: Resolved dependencies
2147

2148
    """
2149
    result = []
2150

    
2151
    for (dep_job_id, dep_status) in deps:
2152
      if ht.TRelativeJobId(dep_job_id):
2153
        assert ht.TInt(dep_job_id) and dep_job_id < 0
2154
        try:
2155
          job_id = resolve_fn(dep_job_id)
2156
        except IndexError:
2157
          # Abort
2158
          return (False, "Unable to resolve relative job ID %s" % dep_job_id)
2159
      else:
2160
        job_id = dep_job_id
2161

    
2162
      result.append((job_id, dep_status))
2163

    
2164
    return (True, result)
2165

    
2166
  def _SubmitManyJobsUnlocked(self, jobs, job_ids, previous_job_ids):
2167
    """Create and store multiple jobs.
2168

2169
    @see: L{_SubmitJobUnlocked}
2170

2171
    """
2172
    results = []
2173
    added_jobs = []
2174

    
2175
    def resolve_fn(job_idx, reljobid):
2176
      assert reljobid < 0
2177
      return (previous_job_ids + job_ids[:job_idx])[reljobid]
2178

    
2179
    for (idx, (job_id, ops)) in enumerate(zip(job_ids, jobs)):
2180
      for op in ops:
2181
        if getattr(op, opcodes.DEPEND_ATTR, None):
2182
          (status, data) = \
2183
            self._ResolveJobDependencies(compat.partial(resolve_fn, idx),
2184
                                         op.depends)
2185
          if not status:
2186
            # Abort resolving dependencies
2187
            assert ht.TNonEmptyString(data), "No error message"
2188
            break
2189
          # Use resolved dependencies
2190
          op.depends = data
2191
      else:
2192
        try:
2193
          job = self._SubmitJobUnlocked(job_id, ops)
2194
        except errors.GenericError, err:
2195
          status = False
2196
          data = self._FormatSubmitError(str(err), ops)
2197
        else:
2198
          status = True
2199
          data = job_id
2200
          added_jobs.append(job)
2201

    
2202
      results.append((status, data))
2203

    
2204
    return (results, added_jobs)
2205

    
2206
  @locking.ssynchronized(_LOCK)
2207
  def _EnqueueJobs(self, jobs):
2208
    """Helper function to add jobs to worker pool's queue.
2209

2210
    @type jobs: list
2211
    @param jobs: List of all jobs
2212

2213
    """
2214
    return self._EnqueueJobsUnlocked(jobs)
2215

    
2216
  def _EnqueueJobsUnlocked(self, jobs):
2217
    """Helper function to add jobs to worker pool's queue.
2218

2219
    @type jobs: list
2220
    @param jobs: List of all jobs
2221

2222
    """
2223
    assert self._lock.is_owned(shared=0), "Must own lock in exclusive mode"
2224
    self._wpool.AddManyTasks([(job, ) for job in jobs],
2225
                             priority=[job.CalcPriority() for job in jobs])
2226

    
2227
  def _GetJobStatusForDependencies(self, job_id):
2228
    """Gets the status of a job for dependencies.
2229

2230
    @type job_id: string
2231
    @param job_id: Job ID
2232
    @raise errors.JobLost: If job can't be found
2233

2234
    """
2235
    if not isinstance(job_id, basestring):
2236
      job_id = self._FormatJobID(job_id)
2237

    
2238
    # Not using in-memory cache as doing so would require an exclusive lock
2239

    
2240
    # Try to load from disk
2241
    job = self.SafeLoadJobFromDisk(job_id, True, writable=False)
2242

    
2243
    assert not job.writable, "Got writable job" # pylint: disable=E1101
2244

    
2245
    if job:
2246
      return job.CalcStatus()
2247

    
2248
    raise errors.JobLost("Job %s not found" % job_id)
2249

    
2250
  @_RequireOpenQueue
2251
  def UpdateJobUnlocked(self, job, replicate=True):
2252
    """Update a job's on disk storage.
2253

2254
    After a job has been modified, this function needs to be called in
2255
    order to write the changes to disk and replicate them to the other
2256
    nodes.
2257

2258
    @type job: L{_QueuedJob}
2259
    @param job: the changed job
2260
    @type replicate: boolean
2261
    @param replicate: whether to replicate the change to remote nodes
2262

2263
    """
2264
    if __debug__:
2265
      finalized = job.CalcStatus() in constants.JOBS_FINALIZED
2266
      assert (finalized ^ (job.end_timestamp is None))
2267
      assert job.writable, "Can't update read-only job"
2268

    
2269
    filename = self._GetJobPath(job.id)
2270
    data = serializer.DumpJson(job.Serialize())
2271
    logging.debug("Writing job %s to %s", job.id, filename)
2272
    self._UpdateJobQueueFile(filename, data, replicate)
2273

    
2274
  def WaitForJobChanges(self, job_id, fields, prev_job_info, prev_log_serial,
2275
                        timeout):
2276
    """Waits for changes in a job.
2277

2278
    @type job_id: string
2279
    @param job_id: Job identifier
2280
    @type fields: list of strings
2281
    @param fields: Which fields to check for changes
2282
    @type prev_job_info: list or None
2283
    @param prev_job_info: Last job information returned
2284
    @type prev_log_serial: int
2285
    @param prev_log_serial: Last job message serial number
2286
    @type timeout: float
2287
    @param timeout: maximum time to wait in seconds
2288
    @rtype: tuple (job info, log entries)
2289
    @return: a tuple of the job information as required via
2290
        the fields parameter, and the log entries as a list
2291

2292
        if the job has not changed and the timeout has expired,
2293
        we instead return a special value,
2294
        L{constants.JOB_NOTCHANGED}, which should be interpreted
2295
        as such by the clients
2296

2297
    """
2298
    load_fn = compat.partial(self.SafeLoadJobFromDisk, job_id, False,
2299
                             writable=False)
2300

    
2301
    helper = _WaitForJobChangesHelper()
2302

    
2303
    return helper(self._GetJobPath(job_id), load_fn,
2304
                  fields, prev_job_info, prev_log_serial, timeout)
2305

    
2306
  @locking.ssynchronized(_LOCK)
2307
  @_RequireOpenQueue
2308
  def CancelJob(self, job_id):
2309
    """Cancels a job.
2310

2311
    This will only succeed if the job has not started yet.
2312

2313
    @type job_id: string
2314
    @param job_id: job ID of job to be cancelled.
2315

2316
    """
2317
    logging.info("Cancelling job %s", job_id)
2318

    
2319
    job = self._LoadJobUnlocked(job_id)
2320
    if not job:
2321
      logging.debug("Job %s not found", job_id)
2322
      return (False, "Job %s not found" % job_id)
2323

    
2324
    assert job.writable, "Can't cancel read-only job"
2325

    
2326
    (success, msg) = job.Cancel()
2327

    
2328
    if success:
2329
      # If the job was finalized (e.g. cancelled), this is the final write
2330
      # allowed. The job can be archived anytime.
2331
      self.UpdateJobUnlocked(job)
2332

    
2333
    return (success, msg)
2334

    
2335
  @_RequireOpenQueue
2336
  def _ArchiveJobsUnlocked(self, jobs):
2337
    """Archives jobs.
2338

2339
    @type jobs: list of L{_QueuedJob}
2340
    @param jobs: Job objects
2341
    @rtype: int
2342
    @return: Number of archived jobs
2343

2344
    """
2345
    archive_jobs = []
2346
    rename_files = []
2347
    for job in jobs:
2348
      assert job.writable, "Can't archive read-only job"
2349

    
2350
      if job.CalcStatus() not in constants.JOBS_FINALIZED:
2351
        logging.debug("Job %s is not yet done", job.id)
2352
        continue
2353

    
2354
      archive_jobs.append(job)
2355

    
2356
      old = self._GetJobPath(job.id)
2357
      new = self._GetArchivedJobPath(job.id)
2358
      rename_files.append((old, new))
2359

    
2360
    # TODO: What if 1..n files fail to rename?
2361
    self._RenameFilesUnlocked(rename_files)
2362

    
2363
    logging.debug("Successfully archived job(s) %s",
2364
                  utils.CommaJoin(job.id for job in archive_jobs))
2365

    
2366
    # Since we haven't quite checked, above, if we succeeded or failed renaming
2367
    # the files, we update the cached queue size from the filesystem. When we
2368
    # get around to fix the TODO: above, we can use the number of actually
2369
    # archived jobs to fix this.
2370
    self._UpdateQueueSizeUnlocked()
2371
    return len(archive_jobs)
2372

    
2373
  @locking.ssynchronized(_LOCK)
2374
  @_RequireOpenQueue
2375
  def ArchiveJob(self, job_id):
2376
    """Archives a job.
2377

2378
    This is just a wrapper over L{_ArchiveJobsUnlocked}.
2379

2380
    @type job_id: string
2381
    @param job_id: Job ID of job to be archived.
2382
    @rtype: bool
2383
    @return: Whether job was archived
2384

2385
    """
2386
    logging.info("Archiving job %s", job_id)
2387

    
2388
    job = self._LoadJobUnlocked(job_id)
2389
    if not job:
2390
      logging.debug("Job %s not found", job_id)
2391
      return False
2392

    
2393
    return self._ArchiveJobsUnlocked([job]) == 1
2394

    
2395
  @locking.ssynchronized(_LOCK)
2396
  @_RequireOpenQueue
2397
  def AutoArchiveJobs(self, age, timeout):
2398
    """Archives all jobs based on age.
2399

2400
    The method will archive all jobs which are older than the age
2401
    parameter. For jobs that don't have an end timestamp, the start
2402
    timestamp will be considered. The special '-1' age will cause
2403
    archival of all jobs (that are not running or queued).
2404

2405
    @type age: int
2406
    @param age: the minimum age in seconds
2407

2408
    """
2409
    logging.info("Archiving jobs with age more than %s seconds", age)
2410

    
2411
    now = time.time()
2412
    end_time = now + timeout
2413
    archived_count = 0
2414
    last_touched = 0
2415

    
2416
    all_job_ids = self._GetJobIDsUnlocked()
2417
    pending = []
2418
    for idx, job_id in enumerate(all_job_ids):
2419
      last_touched = idx + 1
2420

    
2421
      # Not optimal because jobs could be pending
2422
      # TODO: Measure average duration for job archival and take number of
2423
      # pending jobs into account.
2424
      if time.time() > end_time:
2425
        break
2426

    
2427
      # Returns None if the job failed to load
2428
      job = self._LoadJobUnlocked(job_id)
2429
      if job:
2430
        if job.end_timestamp is None:
2431
          if job.start_timestamp is None:
2432
            job_age = job.received_timestamp
2433
          else:
2434
            job_age = job.start_timestamp
2435
        else:
2436
          job_age = job.end_timestamp
2437

    
2438
        if age == -1 or now - job_age[0] > age:
2439
          pending.append(job)
2440

    
2441
          # Archive 10 jobs at a time
2442
          if len(pending) >= 10:
2443
            archived_count += self._ArchiveJobsUnlocked(pending)
2444
            pending = []
2445

    
2446
    if pending:
2447
      archived_count += self._ArchiveJobsUnlocked(pending)
2448

    
2449
    return (archived_count, len(all_job_ids) - last_touched)
2450

    
2451
  def QueryJobs(self, job_ids, fields):
2452
    """Returns a list of jobs in queue.
2453

2454
    @type job_ids: list
2455
    @param job_ids: sequence of job identifiers or None for all
2456
    @type fields: list
2457
    @param fields: names of fields to return
2458
    @rtype: list
2459
    @return: list one element per job, each element being list with
2460
        the requested fields
2461

2462
    """
2463
    jobs = []
2464
    list_all = False
2465
    if not job_ids:
2466
      # Since files are added to/removed from the queue atomically, there's no
2467
      # risk of getting the job ids in an inconsistent state.
2468
      job_ids = self._GetJobIDsUnlocked()
2469
      list_all = True
2470

    
2471
    for job_id in job_ids:
2472
      job = self.SafeLoadJobFromDisk(job_id, True)
2473
      if job is not None:
2474
        jobs.append(job.GetInfo(fields))
2475
      elif not list_all:
2476
        jobs.append(None)
2477

    
2478
    return jobs
2479

    
2480
  @locking.ssynchronized(_LOCK)
2481
  def PrepareShutdown(self):
2482
    """Prepare to stop the job queue.
2483

2484
    Disables execution of jobs in the workerpool and returns whether there are
2485
    any jobs currently running. If the latter is the case, the job queue is not
2486
    yet ready for shutdown. Once this function returns C{True} L{Shutdown} can
2487
    be called without interfering with any job. Queued and unfinished jobs will
2488
    be resumed next time.
2489

2490
    Once this function has been called no new job submissions will be accepted
2491
    (see L{_RequireNonDrainedQueue}).
2492

2493
    @rtype: bool
2494
    @return: Whether there are any running jobs
2495

2496
    """
2497
    if self._accepting_jobs:
2498
      self._accepting_jobs = False
2499

    
2500
      # Tell worker pool to stop processing pending tasks
2501
      self._wpool.SetActive(False)
2502

    
2503
    return self._wpool.HasRunningTasks()
2504

    
2505
  @locking.ssynchronized(_LOCK)
2506
  @_RequireOpenQueue
2507
  def Shutdown(self):
2508
    """Stops the job queue.
2509

2510
    This shutdowns all the worker threads an closes the queue.
2511

2512
    """
2513
    self._wpool.TerminateWorkers()
2514

    
2515
    self._queue_filelock.Close()
2516
    self._queue_filelock = None