00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037 #include <queue>
00038
00039 #include "scheduler.h"
00040 #include "tags.h"
00041 #include "mess.h"
00042 #include "../../core/peo_debug.h"
00043
00044 static std :: queue <SCHED_RESOURCE> resources;
00045
00046 static std :: queue <SCHED_REQUEST> requests;
00047
00048 static unsigned initNumberOfRes = 0;
00049
00050 extern void wakeUpCommunicator();
00051
00052 void initScheduler ()
00053 {
00054
00055 resources = std :: queue <SCHED_RESOURCE> ();
00056 requests = std :: queue <SCHED_REQUEST> ();
00057 initNumberOfRes = 0;
00058
00059 for (unsigned i = 0; i < the_schema.size (); i ++)
00060 {
00061
00062 const Node & node = the_schema [i];
00063
00064 if (node.rk_sched == my_node -> rk)
00065 for (unsigned j = 0; j < node.num_workers; j ++)
00066 resources.push (std :: pair <RANK_ID, WORKER_ID> (i, j + 1));
00067 }
00068 initNumberOfRes = resources.size ();
00069 }
00070
00071 bool allResourcesFree ()
00072 {
00073 return resources.size () == initNumberOfRes;
00074 }
00075
00076 unsigned numResourcesFree ()
00077 {
00078 return resources.size ();
00079 }
00080
00081 static void update ()
00082 {
00083
00084 unsigned num_alloc = std :: min (resources.size (), requests.size ());
00085
00086 for (unsigned i = 0; i < num_alloc; i ++)
00087 {
00088
00089 SCHED_REQUEST req = requests.front ();
00090 requests.pop ();
00091
00092 SCHED_RESOURCE res = resources.front ();
00093 resources.pop ();
00094
00095 printDebugMessage ("allocating a resource.");
00096 initMessage ();
00097 pack (req.second);
00098 pack (res);
00099 sendMessage (req.first, SCHED_RESULT_TAG);
00100 }
00101 }
00102
00103 void unpackResourceRequest ()
00104 {
00105
00106 printDebugMessage ("queuing a resource request.");
00107 SCHED_REQUEST req;
00108 unpack (req);
00109 requests.push (req);
00110 update ();
00111 }
00112
00113 void unpackTaskDone ()
00114 {
00115
00116 printDebugMessage ("I'm notified a worker is now idle.");
00117 SCHED_RESOURCE res;
00118 unpack (res);
00119 resources.push (res);
00120 if (resources.size () == initNumberOfRes)
00121 printDebugMessage ("all the resources are now free.");
00122 update ();
00123 wakeUpCommunicator();
00124 }