commit 2971c9132b8b798178921a389b18d893edec06fb
Author: Iustin Pop <iustin@google.com>
Date:   Thu May 21 18:02:42 2009 +0200

    Add a luxi call for multi-job submit
    
    As a workaround for the job submit timeouts that we have, this patch
    adds a new luxi call for multi-job submit; the advantage is that all the
    jobs are added in the queue and only after the workers can start
    processing them.
    
    This is definitely faster than per-job submit, where the submission of
    new jobs competes with the workers processing jobs.
    
    On a pure no-op OpDelay opcode (not on master, not on nodes), we have:
      - 100 jobs:
        - individual: submit time ~21s, processing time ~21s
        - multiple:   submit time 7-9s, processing time ~22s
      - 250 jobs:
        - individual: submit time ~56s, processing time ~57s
                      run 2:      ~54s                  ~55s
        - multiple:   submit time ~20s, processing time ~51s
                      run 2:      ~17s                  ~52s
    
    which shows that we indeed gain on the client side, and maybe even on
    the total processing time for a high number of jobs. For just 10 or so I
    expect the difference to be just noise.
    
    This will probably require increasing the timeout a little when
    submitting too many jobs - 250 jobs at ~20 seconds is close to the
    current rw timeout of 60s.
    
    Signed-off-by: Iustin Pop <iustin@google.com>
    Reviewed-by: Guido Trotter <ultrotter@google.com>

diff --git a/daemons/ganeti-masterd b/daemons/ganeti-masterd
index 6a73e4e..2305fc6 100755
--- a/daemons/ganeti-masterd
+++ b/daemons/ganeti-masterd
@@ -213,6 +213,13 @@ class ClientOps:
       ops = [opcodes.OpCode.LoadOpCode(state) for state in args]
       return queue.SubmitJob(ops)
 
+    if method == luxi.REQ_SUBMIT_MANY_JOBS:
+      logging.info("Received multiple jobs")
+      jobs = []
+      for ops in args:
+        jobs.append([opcodes.OpCode.LoadOpCode(state) for state in ops])
+      return queue.SubmitManyJobs(jobs)
+
     elif method == luxi.REQ_CANCEL_JOB:
       job_id = args
       logging.info("Received job cancel request for %s", job_id)
diff --git a/lib/jqueue.py b/lib/jqueue.py
index 3364a93..90757a5 100644
--- a/lib/jqueue.py
+++ b/lib/jqueue.py
@@ -943,9 +943,8 @@ class JobQueue(object):
       utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
     return True
 
-  @utils.LockedMethod
   @_RequireOpenQueue
-  def SubmitJob(self, ops):
+  def _SubmitJobUnlocked(self, ops):
     """Create and store a new job.
 
     This enters the job into our job queue and also puts it on the new
@@ -959,7 +958,7 @@ class JobQueue(object):
 
     """
     if self._IsQueueMarkedDrain():
-      raise errors.JobQueueDrainError()
+      raise errors.JobQueueDrainError("Job queue is drained, refusing job")
 
     # Check job queue size
     size = len(self._ListJobFiles())
@@ -987,6 +986,37 @@ class JobQueue(object):
 
     return job.id
 
+  @utils.LockedMethod
+  @_RequireOpenQueue
+  def SubmitJob(self, ops):
+    """Create and store a new job.
+
+    @see: L{_SubmitJobUnlocked}
+
+    """
+    return self._SubmitJobUnlocked(ops)
+
+  @utils.LockedMethod
+  @_RequireOpenQueue
+  def SubmitManyJobs(self, jobs):
+    """Create and store multiple jobs.
+
+    @see: L{_SubmitJobUnlocked}
+
+    """
+    results = []
+    for ops in jobs:
+      try:
+        data = self._SubmitJobUnlocked(ops)
+        status = True
+      except errors.GenericError, err:
+        data = str(err)
+        status = False
+      results.append((status, data))
+
+    return results
+
+
   @_RequireOpenQueue
   def UpdateJobUnlocked(self, job):
     """Update a job's on disk storage.
diff --git a/lib/luxi.py b/lib/luxi.py
index 308de9f..1c3ca6d 100644
--- a/lib/luxi.py
+++ b/lib/luxi.py
@@ -45,6 +45,7 @@ KEY_SUCCESS = "success"
 KEY_RESULT = "result"
 
 REQ_SUBMIT_JOB = "SubmitJob"
+REQ_SUBMIT_MANY_JOBS = "SubmitManyJobs"
 REQ_WAIT_FOR_JOB_CHANGE = "WaitForJobChange"
 REQ_CANCEL_JOB = "CancelJob"
 REQ_ARCHIVE_JOB = "ArchiveJob"
@@ -335,6 +336,12 @@ class Client(object):
     ops_state = map(lambda op: op.__getstate__(), ops)
     return self.CallMethod(REQ_SUBMIT_JOB, ops_state)
 
+  def SubmitManyJobs(self, jobs):
+    jobs_state = []
+    for ops in jobs:
+      jobs_state.append([op.__getstate__() for op in ops])
+    return self.CallMethod(REQ_SUBMIT_MANY_JOBS, jobs_state)
+
   def CancelJob(self, job_id):
     return self.CallMethod(REQ_CANCEL_JOB, job_id)
 
