feature #212: Dispatch limits for the scheduler.

src/scheduler/include/PoolXML.h

@@ -91,7 +91,9 @@ public:
 
         num_objs = get_suitable_nodes(nodes);
 
-        for (unsigned int i=0 ; i < nodes.size() ; i++)
+        for (unsigned int i=0 ;
+             i < nodes.size() && ( pool_limit <= 0 || i < pool_limit ) ;
+             i++)
         {
             add_object(nodes[i]);
         }
@@ -125,9 +127,10 @@ protected:
 
     // ------------------------------------------------------------------------
 
-    PoolXML(Client* client):ObjectXML()
+    PoolXML(Client* client, unsigned int pool_limit = 0):ObjectXML()
     {
-        this->client = client;
+        this->client     = client;
+        this->pool_limit = pool_limit;
     };
 
     virtual ~PoolXML()
@@ -161,6 +164,12 @@ protected:
      */
     Client * client;
 
+    /**
+     *  Limit of pool elements to process (request individual info)
+     *  from the pool.
+     */
+    unsigned int pool_limit;
+
     /**
      * Hash map contains the suitable [id, object] pairs.
      */

src/scheduler/include/Scheduler.h

@@ -44,11 +44,13 @@ public:
 
 protected:
 
-    Scheduler(string& url, time_t _timer):
+    Scheduler(string& url, time_t _timer,
+              int _machines_limit, int _dispatch_limit):
         hpool(0),
         vmpool(0),
-        one_url(url),
         timer(_timer),
+        machines_limit(_machines_limit),
+        dispatch_limit(_dispatch_limit),
         threshold(0.9),
         client("",url)
     {
@@ -119,14 +121,18 @@ private:
     // Configuration attributes
     // ---------------------------------------------------------------
 
-    /**
-     *  the URL of the XML-RPC server
-     */
-    string  one_url;
-
-
     time_t  timer;
 
+    /**
+     *  Limit of pending virtual machines to process from the pool.
+     */
+    unsigned int machines_limit;
+
+    /**
+     *  Limit of virtual machines to ask OpenNebula core to deploy.
+     */
+    unsigned int dispatch_limit;
+
     /**
      *  Threshold value to round up freecpu
      */

src/scheduler/include/VirtualMachinePoolXML.h

@@ -27,7 +27,10 @@ class VirtualMachinePoolXML : public PoolXML
 {
 public:
 
-    VirtualMachinePoolXML(Client* client):PoolXML(client){};
+    VirtualMachinePoolXML(
+                             Client*        client,
+                             unsigned int   machines_limit
+                         ):PoolXML(client, machines_limit){};
 
     ~VirtualMachinePoolXML(){};
 

src/scheduler/src/pool/VirtualMachinePoolXML.cc

@@ -50,7 +50,7 @@ void VirtualMachinePoolXML::add_object(xmlNodePtr node)
     {
         NebulaLog::log("VM",Log::ERROR,
                        "XML Node does not represent a valid Virtual Machine");
-        // TODO: if the xml node isn't valid, do nothing?
+
        return;
     }
 

src/scheduler/src/sched/Scheduler.cc

@@ -106,7 +106,7 @@ void Scheduler::start()
     // -----------------------------------------------------------
 
     hpool  = new HostPoolXML(&client);
-    vmpool = new VirtualMachinePoolXML(&client);
+    vmpool = new VirtualMachinePoolXML(&client, machines_limit);
 
     // -----------------------------------------------------------
     // Load scheduler policies
@@ -372,8 +372,9 @@ void Scheduler::dispatch()
     VirtualMachineXML * vm;
     ostringstream       oss;
 
-    int hid;
+    int             hid;
-    int rc;
+    int             rc;
+    unsigned int    dispatched_vms;
 
     map<int, ObjectXML*>::const_iterator  vm_it;
     const map<int, ObjectXML*>            pending_vms = vmpool->get_objects();
@@ -391,7 +392,11 @@ void Scheduler::dispatch()
 
     NebulaLog::log("SCHED",Log::INFO,oss);
 
-    for (vm_it=pending_vms.begin(); vm_it != pending_vms.end(); vm_it++)
+    dispatched_vms = 0;
+    for (vm_it=pending_vms.begin();
+         vm_it != pending_vms.end() && ( dispatch_limit <= 0 ||
+                                         dispatched_vms < dispatch_limit );
+         vm_it++)
     {
         vm = static_cast<VirtualMachineXML*>(vm_it->second);
 
@@ -399,7 +404,12 @@ void Scheduler::dispatch()
 
         if (rc == 0)
         {
-            vmpool->dispatch(vm_it->first,hid);
+            rc = vmpool->dispatch(vm_it->first,hid);
+
+            if (rc == 0)
+            {
+                dispatched_vms++;
+            }
         }
     }
 }

src/scheduler/src/sched/mm_sched.cc

@@ -31,7 +31,11 @@ class RankScheduler : public Scheduler
 {
 public:
 
-    RankScheduler(string url,time_t timer=1):Scheduler(url,timer),rp(0){};
+    RankScheduler(string url,
+                  unsigned int machines_limit,
+                  unsigned int dispatch_limit,
+                  time_t timer=1
+                  ):Scheduler(url,timer,machines_limit, dispatch_limit),rp(0){};
     
     ~RankScheduler()
     {
@@ -58,11 +62,13 @@ int main(int argc, char **argv)
     RankScheduler * ss;
     int             port = 2633;
     time_t          timer= 30;
+    unsigned int    machines_limit = 400;
+    unsigned int    dispatch_limit = 300;
     char            opt;
     
     ostringstream  oss;
         
-    while((opt = getopt(argc,argv,"p:t:")) != -1)
+    while((opt = getopt(argc,argv,"p:t:m:d:")) != -1)
     {
         switch(opt)
         {
@@ -72,8 +78,15 @@ int main(int argc, char **argv)
             case 't':
                 timer = atoi(optarg);
                 break;
+            case 'm':
+                machines_limit = atoi(optarg);
+                break;
+            case 'd':
+                dispatch_limit = atoi(optarg);
+                break;
             default:
-                cerr << "usage: " << argv[0] << " [-p port] [-t timer]\n";
+                cerr << "usage: " << argv[0] << " [-p port] [-t timer] ";
+                cerr << "[-m machines limit] [-d dispatch limit]\n";
                 exit(-1);
                 break;
         }
@@ -83,7 +96,7 @@ int main(int argc, char **argv)
     
     oss << "http://localhost:" << port << "/RPC2";
         
-    ss = new RankScheduler(oss.str(),timer);
+    ss = new RankScheduler(oss.str(),timer, machines_limit, dispatch_limit);
     
     try
     {