4 # Copyright (C) 2006, 2007 Google Inc.
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.
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.
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
22 """Module implementing the job queue handling."""
27 from ganeti import opcodes
28 from ganeti import errors
31 """In-memory job representation.
33 This is what we use to track the user-submitted jobs (which are of
37 def __init__(self, jid, jdesc):
39 jdesc.status = opcodes.Job.STATUS_PENDING
41 self.lock = threading.Lock()
43 def SetStatus(self, status, result=None):
45 self.data.status = status
46 if result is not None:
47 self.data.result = result
52 #FIXME(iustin): make a deep copy of result
59 """Example queue implementation.
65 self.lock = threading.Lock()
66 self.new_queue = Queue.Queue()
69 """Add a new job to the queue.
71 This enters the job into our job queue and also puts it on the new
72 queue, in order for it to be picked up by the queue processors.
79 job = JobObject(rid, item)
80 self.job_queue[rid] = job
83 self.new_queue.put(job)
87 """Query a given job ID.
91 result = self.job_queue.get(rid, None)
95 def query_jobs(self, fields, names):
98 The fields and names parameters are similar to the ones passed to
105 values = [self.job_queue[j_id] for j_id in names]
107 values = self.job_queue.itervalues()
114 row.append(jdata.job_id)
115 elif fname == "status":
116 row.append(jdata.status)
117 elif fname == "opcodes":
118 row.append(",".join([op.OP_ID for op in jdata.op_list]))
120 raise errors.OpExecError("Invalid job query field '%s'" %