Changed CldEnqueue to three parameters ( no pack function )
authorJosh Yelon <jyelon@uiuc.edu>
Thu, 2 Jul 1998 02:51:59 +0000 (02:51 +0000)
committerJosh Yelon <jyelon@uiuc.edu>
Thu, 2 Jul 1998 02:51:59 +0000 (02:51 +0000)
12 files changed:
src/QuickThreads/configure
src/arch/net/machine.c
src/ck-core/boc.c
src/ck-core/charm.h
src/ck-core/ck.c
src/ck-core/vid.c
src/conv-core/conv-conds.c
src/conv-core/converse.h
src/conv-ldb/cldb.c
src/conv-ldb/cldb.rand.c
src/conv-ldb/cldb.spray.c [new file with mode: 0644]
src/scripts/Makefile

index c063c31d3413525eee0312bdc004c8549465d0ef..451b706bb61ee29b4c8987f5b331c9a21f0c177f 100755 (executable)
@@ -1,6 +1,6 @@
 #!/bin/sh
 
-if [ ! -f mkfiles/$1 ] ; then
+if [ ! -r mkfiles/$1 ] ; then
     echo "Usage: configure <version>"
     echo ""
     echo "Available versions are:"
index 68e2699bc02191a97ff6f4d9e60370d495e572b8..c247ac014ed266e1dab900346c9e1df2f05a1cba 100644 (file)
@@ -1508,11 +1508,11 @@ static int InternalScanf(fmt, l)
   CmiLock(Cmi_scanf_mutex);
   ctrl_sendone(120, "scanf %s %d %s", Cmi_self_IP_str, ctrlport, fmt);
   while (Cmi_scanf_data == 0) jsleep(0, 250000);
-  i = sscanf(Cmi_scanf_data, fmt,
+  i = sscanf((char*)Cmi_scanf_data, fmt,
             ptr[ 0], ptr[ 1], ptr[ 2], ptr[ 3], ptr[ 4], ptr[ 5],
             ptr[ 6], ptr[ 7], ptr[ 8], ptr[ 9], ptr[10], ptr[11],
             ptr[12], ptr[13], ptr[14], ptr[15], ptr[16], ptr[17]);
-  free(Cmi_scanf_data);
+  free((char*)Cmi_scanf_data);
   Cmi_scanf_data=0;
   CmiUnlock(Cmi_scanf_mutex);
   return i;
index d1582d6d2dfa183af2e625028edebbe945e07cd8..83ff9da31a072ed0e7389bfadb00a84f8e2772f5 100644 (file)
@@ -356,7 +356,7 @@ ChareNumType bocnum;
     trace_creation(GetEnv_msgType(env), ep, env);
   
   CmiSetHandler(env, CpvAccess(HANDLE_INCOMING_MSG_Index));
-  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index));
   if((type!=QdBocMsg)&&(type!=QdBroadcastBocMsg)&&(type!=LdbMsg))
     QDCountThisCreation(1);
 }
@@ -390,7 +390,7 @@ ChareNumType bocnum;
   if(CpvAccess(traceOn))
     trace_creation(GetEnv_msgType(env), ep, env);
   CmiSetHandler(env,CpvAccess(HANDLE_INCOMING_MSG_Index));
-  CldEnqueue(CLD_BROADCAST_ALL, env, CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(CLD_BROADCAST_ALL, env, CpvAccess(CkInfo_Index));
   if((type!=QdBocMsg)&&(type!=QdBroadcastBocMsg)&&(type!=LdbMsg))
     QDCountThisCreation(CmiNumPes());
 }
@@ -451,7 +451,7 @@ char *mydata;
   if(CpvAccess(traceOn))
     trace_creation(GetEnv_msgType(env), ep, env);
   CmiSetHandler(env,CpvAccess(HANDLE_INCOMING_MSG_Index));
-  CldEnqueue(CLD_BROADCAST_ALL, env, CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(CLD_BROADCAST_ALL, env, CpvAccess(CkInfo_Index));
   
   QDCountThisCreation(CmiNumPes());
 }
index 87eb1877a35aebafd63e4f27fdeaf6feee2f1292..331b3f29da794ae5f57784f2b0457e3ed01a4c78 100644 (file)
@@ -64,8 +64,8 @@
 #define MONOTONIC      1
 #define TABLE          2
 
-#define NULL_VID       NULL
-#define NULL_PE        NULL
+#define NULL_VID        NULL
+#define NULL_PE         CK_PE_ANY
 #define NULL_PACK_ID    0  
 
 #define CK_PE_SPECIAL(x) ((x)>=0xFFF0)
index cd4669eaca1a282d242a7daa19b49689fea6172a..23c0c3d56bff0c70e3a49cdbf2d7170734f3404b 100644 (file)
@@ -7,7 +7,7 @@
 
 void CkUnpack(ENVELOPE **msg);
 void CkPack(ENVELOPE **msg);
-void CkInfo(void *msg, int *len, int *qs, int *pb, unsigned int **pp);
+void CkInfo(void *msg, CldPackFn *pfn, int *len, int *qs, int *pb, unsigned int **pp);
 
 extern void *FIFO_Create();
 extern CHARE_BLOCK *CreateChareBlock();
@@ -15,19 +15,16 @@ extern CHARE_BLOCK *CreateChareBlock();
 CpvStaticDeclare(int, num_exits);
 CpvStaticDeclare(int, num_endcharms);
 CpvDeclare(int, CkInfo_Index);
-CpvDeclare(int, CkPack_Index);
 
 void ckModuleInit()
 {
    CpvInitialize(int, num_exits);
    CpvInitialize(int, num_endcharms);
    CpvInitialize(int, CkInfo_Index);
-   CpvInitialize(int, CkPack_Index);
 
    CpvAccess(num_exits)=0;
    CpvAccess(num_endcharms)=0;
    CpvAccess(CkInfo_Index) = CldRegisterInfoFn(CkInfo);
-   CpvAccess(CkPack_Index) = CldRegisterPackFn((CldPackFn)CkPack);
 }
 
 
@@ -269,7 +266,7 @@ int destPE;
   SetEnv_msgType(env, NewChareMsg);
   if (destPE == CK_PE_ANY) destPE = CLD_ANYWHERE;
   CmiSetHandler(env, CpvAccess(HANDLE_INCOMING_MSG_Index));
-  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index));
 }
 
 
@@ -291,7 +288,7 @@ ChareIDType * pChareID;
   if(CpvAccess(traceOn))
     trace_creation(GetEnv_msgType(env), Entry, env);
   CmiSetHandler(env, CpvAccess(HANDLE_INCOMING_MSG_Index));
-  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(destPE, env, CpvAccess(CkInfo_Index));
 }
 
 /*****************************************************************/
@@ -386,10 +383,11 @@ void CkPack(ENVELOPE **henv)
 }
 
 
-void CkInfo(void *msg, int *len,
+void CkInfo(void *msg, CldPackFn *pfn, int *len,
            int *queueing, int *priobits, unsigned int **prioptr)
 {
   ENVELOPE *env = (ENVELOPE *)msg;
+  *pfn = (CldPackFn)CkPack;
   *len = GetEnv_TotalSize(env);
   *queueing = GetEnv_queueing(env);
   *priobits = GetEnv_priosize(env);
index b4d2f130223c9223a9594e5fb6b975f117ebb9cd..b71e8e1b71bea765183795b4d13effe68ee806ed 100644 (file)
  * REVISION HISTORY:
  *
  * $Log$
- * Revision 2.15  1998-06-15 22:16:44  milind
+ * Revision 2.16  1998-07-02 02:52:03  jyelon
+ * Changed CldEnqueue to three parameters ( no pack function )
+ *
+ * Revision 2.15  1998/06/15 22:16:44  milind
  * Reduced Charm++ overhead by reducing variable accesses.
  *
  * Revision 2.14  1998/03/18 15:38:14  milind
@@ -126,8 +129,7 @@ ENVELOPE *env;
   SetEnv_chare_magic_number(env, GetID_chare_magic_number(vid->x.realID));
   if(CpvAccess(traceOn))
     trace_creation(GetEnv_msgType(env), GetEnv_EP(env), env);
-  CldEnqueue(GetID_onPE(vid->x.realID), env,
-            CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+  CldEnqueue(GetID_onPE(vid->x.realID), env, CpvAccess(CkInfo_Index));
   QDCountThisCreation(1);
 }
 
@@ -171,8 +173,7 @@ void *data_area;
        SetEnv_chare_magic_number(env, chare_magic);
         if(CpvAccess(traceOn))
          trace_creation(GetEnv_msgType(env), GetEnv_EP(env), env);
-       CldEnqueue(chare_pe, env,
-                  CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+       CldEnqueue(chare_pe, env, CpvAccess(CkInfo_Index));
        QDCountThisCreation(1);
    }
    FIFO_Destroy(vidqueue);
@@ -207,6 +208,5 @@ CHARE_BLOCK *vidBlockPtr;
     if(CpvAccess(traceOn))
       trace_creation(GetEnv_msgType(env), GetEnv_EP(env), env);
     CmiSetHandler(env, CpvAccess(HANDLE_INCOMING_MSG_Index));
-    CldEnqueue(vidPE, env,
-              CpvAccess(CkInfo_Index), CpvAccess(CkPack_Index));
+    CldEnqueue(vidPE, env, CpvAccess(CkInfo_Index));
 }
index e127960a3c47f1af66ac44e6ea131c6c6226fb57..1b51e22b23fa17999afaa89c3d7105caf1653085 100644 (file)
  * REVISION HISTORY:
  *
  * $Log$
- * Revision 2.10  1998-03-04 17:17:13  milind
+ * Revision 2.11  1998-07-02 02:52:16  jyelon
+ * Changed CldEnqueue to three parameters ( no pack function )
+ *
+ * Revision 2.10  1998/03/04 17:17:13  milind
  * Fixed the size_t errors.
  *
  * Revision 2.9  1997/07/22 14:22:33  kale
@@ -89,7 +92,7 @@ CpvStaticDeclare(int, numHeapEntries);
 
 CpvDeclare(int, CcdNumChecks);
 
-extern double CmiTimer();
+extern double CmiWallTimer();
 
 void CcdModuleInit()
 {
@@ -189,7 +192,7 @@ void CcdCallFnAfter(fnp, arg, deltaT)
     unsigned int deltaT;
 {
   double tPrime, currT;
-  currT  = CmiTimer();                    /* get current time */
+  currT  = CmiWallTimer();                /* get current time */
   tPrime = currT + (double)deltaT/1000.0; /* add delta to determine what time
                                            to actually execute fn */
   InsertInHeap(tPrime, fnp, arg); /* insert into tmr hp */
@@ -207,7 +210,7 @@ void CcdCallBacks()
   
 /* This was formerly TimerChecks() */
   if ( CpvAccess(numHeapEntries) > 0 ) {
-    currTime = CmiTimer();
+    currTime = CmiWallTimer();
   
     while ((CpvAccess(numHeapEntries) > 0) && CpvAccess(timerHeap)[1].timeVal < currTime)
     {
index c01be9ecf08172c2c93ac3b379e627fdf98f774b..a35e06d258fd7583a95f6a7e17791979c0e3e70e 100644 (file)
  * REVISION HISTORY:
  *
  * $Log$
- * Revision 2.90  1998-06-15 23:49:23  milind
+ * Revision 2.91  1998-07-02 02:52:18  jyelon
+ * Changed CldEnqueue to three parameters ( no pack function )
+ *
+ * Revision 2.90  1998/06/15 23:49:23  milind
  * Fixed charm++ message macros to adhere to the new LDB structure.
  *
  * Revision 2.89  1998/06/15 23:11:15  wilmarth
@@ -488,6 +491,7 @@ double   CmiCpuTimer   CMK_PROTO(());
 #define CsdEnqueueLifo(x)     (CqsEnqueueLifo(CpvAccess(CsdSchedQueue),(x)))
 #define CsdEnqueue(x)         (CqsEnqueueFifo(CpvAccess(CsdSchedQueue),(x)))
 #define CsdEmpty()            (CqsEmpty(CpvAccess(CsdSchedQueue)))
+#define CsdLength()           (CqsLength(CpvAccess(CsdSchedQueue)))
 
 #if CMK_CMIPRINTF_IS_A_BUILTIN
 void  CmiPrintf CMK_PROTO((char *, ...));
@@ -795,18 +799,19 @@ void CfutureInit();
 #define CLD_BROADCAST (-2)
 #define CLD_BROADCAST_ALL (-3)
 
+typedef void (*CldPackFn)(void *msg);
+
 typedef void (*CldInfoFn)(void *msg, 
+                         CldPackFn *packer,
                          int *len,
                          int *queueing,
                          int *priobits, 
                          unsigned int **prioptr);
 
-typedef void (*CldPackFn)(void **msg);
-
 int CldRegisterInfoFn(CldInfoFn fn);
 int CldRegisterPackFn(CldPackFn fn);
 
-void CldEnqueue(int pe, void *msg, int infofn, int packfn);
+void CldEnqueue(int pe, void *msg, int infofn);
 
 /****** CMM: THE MESSAGE MANAGER ******/
 
index 4542fbc94f3b881f3de133fe681c32392673c4b8..231b1a3a79fb69b9d9b4ffa19d5c2b03b412f9e6 100644 (file)
@@ -1,36 +1,27 @@
 #include "converse.h"
-/* ITEMS 1-3 below are no longer true.
- * How to write a load-balancer:
- *
- * 1. Every load-balancer must contain a definition of struct CldField.
- *    This structure describes what kind of data will be piggybacked on
- *    the messages that go through the load balancer.  The structure
- *    must include certain predefined fields.  Put the word
- *    CLD_STANDARD_FIELD_STUFF at the front of the struct definition
- *    to include these predefined fields.
- *
- * 2. Every load-balancer must contain a definition of the global variable
- *    Cld_fieldsize.  You must initialize this to sizeof(struct CldField).
- *    This is not a CPV or CSV variable, it's a plain old C global.
- *
- * 3. When you send a message, you'll probably want to temporarily
- *    switch the handler.  The following function will switch the handler
- *    while saving the old one:
- *
- *       CldSwitchHandler(msg, field, newhandler);
- *
- *    Field must be a pointer to the gap in the message where the CldField
- *    is to be stored.  The switch routine will use this region to store
- *    the old handler, as well as some other stuff.  When the message
- *    gets handled, you can switch the handler back like this:
- *    
- *       CldRestoreHandler(msg, &field);
+
+/*
+ * CldSwitchHandler takes a message and a new handler number.  It
+ * changes the handler number to the new handler number and move the
+ * old to the Xhandler part of the header.  When the message gets
+ * handled, the handler should call CldRestoreHandler to put the old
+ * handler back.
  *
- *    This will not only restore the handler, it will also tell you
- *    where in the message the CldField was stored.
+ * CldPutToken puts a message in the scheduler queue in such a way
+ * that it can be retreived from the queue.  Once the message gets
+ * handled, it can no longer be retreived.  CldGetToken removes a
+ * message that was placed in the scheduler queue in this way.
+ * CldCountTokens tells you how many tokens are currently retreivable.
  *
- * 4. Don't forget that CldEnqueue must support directed transmission of
- *    messages as well as undirected, and broadcasts too.
+ * Caution: these functions are using the function "CmiReference"
+ * which I just added to the Cmi memory allocator (it increases the
+ * reference count field, making it possible to free the memory
+ * twice.)  I'm not sure how well this is going to work.  I need this
+ * because the message should not be freed until it's out of the
+ * scheduler queue AND out of the user's hands.  It needs to stay
+ * around while it's in the scheduler queue because it may contain
+ * a priority.  I should probably rewrite these subroutines so that
+ * they simply copy the priority, I would feel safer that way.
  *
  */
 
@@ -44,16 +35,6 @@ int CldRegisterPackFn(CldPackFn fn)
   return CmiRegisterHandler((CmiHandler)fn);
 }
 
-/* CldSwitchHandler takes a message and a new handler number.  It
- * changes the handler number to the new handler number and move the
- * old to the Xhandler part of the header.  When the message gets
- * handled, the handler should call CldRestoreHandler to put the old
- * handler back.
- *
- * These next subroutines are balanced on a thin wire.  They're
- * correct, but the slightest disturbance in the offsets could break them.
- * */
-
 void CldSwitchHandler(char *cmsg, int handler)
 {
   CmiSetXHandler(cmsg, CmiGetHandler(cmsg));
@@ -65,23 +46,6 @@ void CldRestoreHandler(char *cmsg)
   CmiSetHandler(cmsg, CmiGetXHandler(cmsg));
 }
 
-/* CldPutToken puts a message in the scheduler queue in such a way
- * that it can be retreived from the queue.  Once the message gets
- * handled, it can no longer be retreived.  CldGetToken removes a
- * message that was placed in the scheduler queue in this way.
- * CldCountTokens tells you how many tokens are currently retreivable.
- *
- * Caution: these functions are using the function "CmiReference"
- * which I just added to the Cmi memory allocator (it increases the
- * reference count field, making it possible to free the memory
- * twice.)  I'm not sure how well this is going to work.  I need this
- * because the message should not be freed until it's out of the
- * scheduler queue AND out of the user's hands.  It needs to stay
- * around while it's in the scheduler queue because it may contain
- * a priority.
- *
- */
-
 void Cldhandler(void *);
  
 typedef struct CldToken_s {
@@ -125,7 +89,8 @@ void CldPutToken(void *msg)
   CldInfoFn ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
   CldToken tok = (CldToken)CmiAlloc(sizeof(struct CldToken_s));
   int len, queueing, priobits; unsigned int *prioptr;
-  
+  CldPackFn pfn;
+
   tok->msg = msg;
 
   /* add token to the doubly-linked circle */
@@ -137,7 +102,7 @@ void CldPutToken(void *msg)
   
   /* add token to the scheduler */
   CmiSetHandler(tok, proc->tokenhandleridx);
-  ifn(msg, &len, &queueing, &priobits, &prioptr);
+  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
   CsdEnqueueGeneral(tok, queueing, priobits, prioptr);
 }
 
index b6057573f745a3d0758bd1ab4c4283d607e96168..e8e96a5da57df0158ddd8478a470a14cd2ad183f 100644 (file)
@@ -1,89 +1,33 @@
 #include "converse.h"
 
-/* CldEnqueue is the load balancer.  It accepts a message, and
- * enqueues the message on some processor of its own choosing.
- * Processors are chosen in such a way as to help distribute the
- * workload.
- *
- * The user of the load balancer must set aside some space in the
- * message for load balancer to store temporary data.  The number of
- * bytes that must be reserved is CLD_FIELDSIZE.  There is no
- * stipulation about where in the message this field must be.
- *
- * If the message is to be prioritized, then the message must also
- * contain its priority.  This can be omitted if the message is to
- * have the null priority.  There is no stipulation about where in
- * the message the priority must be stored.
- *
- * The user of the load balancer must write an "info" function.
- * The load-balancer will call the info function when it needs to
- * know any of the following pieces of information:
- *
- *    - the length of the message, in bytes.
- *    - the queueing strategy of the message.
- *    - where the priority is, in the message.
- *    - where the ldb field is, in the message.
- *
- * The info function must be able to determine all this information.
- * The info function must be registered, much like a converse handler,
- * using CldRegisterInfoFn.  It must have this prototype:
- *
- *    void Info(void *msg, 
- *              int *len,
- *              void *ldbfield,
- *              int *queuing,
- *              int *priobits,
- *              int **prioptr);
- *
- * The user of the load balancer must also write a "pack" function.
- * When the load balancer is about to send the message across
- * processor boundaries, it will call the pack function.  The
- * pack function may modify the message in any way.  The pack function
- * must be registered using CldRegisterPackFn, and must have this
- * prototype:
- *
- *    void Pack(void **msg);
- *
- * Normally, CldEnqueue is used for load-balancing.  It can also be
- * used as a convenience that helps you enqueue a message on a processor
- * of your choosing.  The parameter 'pe' lets you specify which processor
- * the message is to go to.  If the value CLD_ANYWHERE is given, then
- * the message is load-balanced.  If it is CLD_BROADCAST, the message
- * is broadcast to all other processors.  If it is CLD_BROADCAST_ALL,
- * the message is broadcast to all processors.  If it is a processor
- * number, the message is sent to a specific location.
- *
- */
-
 void CldHandler(char *msg)
 {
   int len, queueing, priobits;
-  unsigned int *prioptr; CldInfoFn ifn;
+  unsigned int *prioptr; CldInfoFn ifn; CldPackFn pfn;
   CmiGrabBuffer((void **)&msg);
   CldRestoreHandler(msg);
   ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
-  ifn(msg, &len, &queueing, &priobits, &prioptr);
+  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
   CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
 }
 
 CpvDeclare(int, CldHandlerIndex);
 
-void CldEnqueue(int pe, void *msg, int infofn, int packfn)
+void CldEnqueue(int pe, void *msg, int infofn)
 {
   int len, queueing, priobits; unsigned int *prioptr;
   CldInfoFn ifn = (CldInfoFn)CmiHandlerToFunction(infofn);
   CldPackFn pfn;
-/* This is just for debugging anyway: mab
-  if (CmiGetHandler(msg) >= CpvAccess(CmiHandlerMax)) *((int*)0)=0;
-*/
   if (pe == CLD_ANYWHERE) pe = (((rand()+CmiMyPe())&0x7FFFFFFF)%CmiNumPes());
   if (pe == CmiMyPe()) {
-    ifn(msg, &len, &queueing, &priobits, &prioptr);
+    ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
     CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
   } else {
-    pfn = (CldPackFn)CmiHandlerToFunction(packfn);
-    pfn(&msg);
-    ifn(msg, &len, &queueing, &priobits, &prioptr);
+    ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+    if (pfn) {
+      pfn(&msg);
+      ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+    }
     CldSwitchHandler(msg, CpvAccess(CldHandlerIndex));
     CmiSetInfo(msg,infofn);
     if (pe==CLD_BROADCAST) CmiSyncBroadcastAndFree(len, msg);
diff --git a/src/conv-ldb/cldb.spray.c b/src/conv-ldb/cldb.spray.c
new file mode 100644 (file)
index 0000000..54b78f6
--- /dev/null
@@ -0,0 +1,166 @@
+#include "converse.h"
+#include <math.h>
+
+#define CYCLE_MILLISECONDS 500
+#define DEBUGGING_OUTPUT 0
+
+typedef struct 
+{
+  int mype;
+  int EnqueueHandler;
+  int ReduceHandler;
+  int AverageHandler;
+  int HopHandler;
+  double load_reported;
+  double load_total;
+  int    load_count;
+  int    spantree_parent;
+  int    spantree_children;
+  int    spantree_root;
+  int    rebalance;
+}
+peinfo;
+
+CpvStaticDeclare(peinfo, peinf);
+
+struct loadmsg {
+  char core[CmiMsgHeaderSizeBytes];
+  double load_total;
+};
+
+struct reqmsg {
+  char core[CmiMsgHeaderSizeBytes];
+};
+
+void CldPropagateLoad(double load);
+
+void CldInitiateReduction()
+{
+  double load = CsdLength();
+  peinfo *pinf = &(CpvAccess(peinf));
+  pinf->load_reported = load;
+  CldPropagateLoad(load);
+}
+
+void CldPropagateLoad(double load)
+{
+  struct loadmsg msg;
+  peinfo *pinf = &(CpvAccess(peinf));
+  pinf->load_total += load;
+  pinf->load_count ++;
+  if (pinf->load_count == pinf->spantree_children + 1) {
+    msg.load_total   = pinf->load_total;
+    if (pinf->mype == pinf->spantree_root) {
+      if (DEBUGGING_OUTPUT) CmiPrintf("---\n");
+      CmiSetHandler(&msg, pinf->AverageHandler);
+      CmiSyncBroadcastAll(sizeof(msg), &msg);
+    } else {
+      CmiSetHandler(&msg, pinf->ReduceHandler);
+      CmiSyncSend(pinf->spantree_parent, sizeof(msg), &msg);
+    }
+    pinf->load_total = 0;
+    pinf->load_count = 0;
+  }
+}
+
+void CldReduceHandler(struct loadmsg *msg)
+{
+  CldPropagateLoad(msg->load_total);
+}
+
+void CldAverageHandler(struct loadmsg *msg)
+{
+  peinfo *pinf = &(CpvAccess(peinf));
+  double load = CsdLength();
+  double average = (msg->load_total / CmiNumPes());
+  int rebalance;
+  if (load < (average+10) * 1.2) rebalance=0;
+  else rebalance = (load - average);
+  if (DEBUGGING_OUTPUT)
+    CmiPrintf("PE %d load=%6d average=%6d rebalance=%d\n", 
+             CmiMyPe(), CsdLength(), (int)average, rebalance);
+  pinf->rebalance = rebalance;
+  CcdCallFnAfter((CcdVoidFn)CldInitiateReduction, 0, CYCLE_MILLISECONDS);
+}
+
+void CldEnqueueHandler(char *msg)
+{
+  int len, queueing, priobits; unsigned int *prioptr;
+  CldInfoFn ifn; CldPackFn pfn;
+  CmiGrabBuffer((void **)&msg);
+  ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
+  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+  CmiSetHandler(msg, CmiGetXHandler(msg));
+  CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
+}
+
+void CldHopHandler(char *msg)
+{
+  peinfo *pinf = &(CpvAccess(peinf));
+  int len, queueing, priobits; unsigned int *prioptr;
+  CldInfoFn ifn; CldPackFn pfn; int pe;
+
+  CmiGrabBuffer((void **)&msg);
+  if (pinf->rebalance) {
+    ifn = (CldInfoFn)CmiHandlerToFunction(CmiGetInfo(msg));
+    ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+    if (pfn) {
+      pfn(&msg);
+      ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+    }
+    do pe = ((lrand48()&0x7FFFFFFF)%CmiNumPes());
+    while (pe == pinf->mype);
+    CmiSyncSendAndFree(pe, len, msg);
+    pinf->rebalance--;
+  } else {
+    CmiSetHandler(msg, CmiGetXHandler(msg));
+    CmiHandleMessage(msg);
+  }
+}
+
+void CldEnqueue(int pe, void *msg, int infofn)
+{
+  int len, queueing, priobits; unsigned int *prioptr;
+  CldInfoFn ifn; CldPackFn pfn;
+  peinfo *pinf = &(CpvAccess(peinf));
+  ifn = (CldInfoFn)CmiHandlerToFunction(infofn);
+  ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+  if (pe != CLD_ANYWHERE) {
+    if (pfn && (pe != CmiMyPe())) {
+      pfn(&msg);
+      ifn(msg, &pfn, &len, &queueing, &priobits, &prioptr);
+    }
+    CmiSetInfo(msg, infofn);
+    CmiSetXHandler(msg, CmiGetHandler(msg));
+    CmiSetHandler(msg, pinf->EnqueueHandler);
+    if (pe==CLD_BROADCAST) CmiSyncBroadcastAndFree(len, msg);
+    else if (pe==CLD_BROADCAST_ALL) CmiSyncBroadcastAllAndFree(len, msg);
+    else CmiSyncSendAndFree(pe, len, msg);
+  } else {
+    CmiSetInfo(msg, infofn);
+    CmiSetXHandler(msg, CmiGetHandler(msg));
+    CmiSetHandler(msg, pinf->HopHandler);
+    CsdEnqueueGeneral(msg, queueing, priobits, prioptr);
+  }
+}
+
+void CldModuleInit()
+{
+  peinfo *pinf;
+  CpvInitialize(peinfo, peinf);
+  srand48(time(0)+CmiMyPe());
+  pinf = &CpvAccess(peinf);
+  pinf->mype = CmiMyPe();
+  pinf->EnqueueHandler = CmiRegisterHandler(CldEnqueueHandler);
+  pinf->ReduceHandler  = CmiRegisterHandler(CldReduceHandler);
+  pinf->AverageHandler = CmiRegisterHandler(CldAverageHandler);
+  pinf->HopHandler     = CmiRegisterHandler(CldHopHandler);
+  pinf->load_total = 0.0;
+  pinf->load_count = 0;
+  pinf->spantree_children = CmiNumSpanTreeChildren(CmiMyPe());
+  pinf->spantree_parent = CmiSpanTreeParent(CmiMyPe());
+  pinf->spantree_root = CmiSpanTreeRoot();
+  pinf->rebalance = 0;
+  CldModuleGeneralInit();
+  CldInitiateReduction();
+}
index 14370dc460955c9ad1ddd3846be34cd101a8eeac..8d3f66859e12aa625ba5ba0f602e87e1b7c2d7ba 100644 (file)
@@ -133,7 +133,7 @@ CVHEADERS=converse.h conv-mach.h conv-mach.csh
 
 TRACELIBS=libtrace-none.a libtrace-summary.a libtrace-projections.a
 
-CVLIBS=libconv-core.a libconv-cplus-n.a libconv-cplus-y.a libldb-rand.o libldb-graph.o $(TRACELIBS)
+CVLIBS=libconv-core.a libconv-cplus-n.a libconv-cplus-y.a libldb-rand.o libldb-spray.o $(TRACELIBS)
 
 LIBCONV_CORE=threads.o convcore.o conv-conds.o spantree.o queueing.o fifo.o msgmgr.o memory.o cpm.o cpthreads.o futures.o cldb.o
 
@@ -206,6 +206,9 @@ libldb-rand.o: cldb.rand.c $(CVHEADERS)
 libldb-graph.o: cldb.graph.c $(CVHEADERS)
        $(CHARMC) -cp ../lib/ -o $@ cldb.graph.c
 
+libldb-spray.o: cldb.spray.c $(CVHEADERS)
+       $(CHARMC) -cp ../lib/ -o $@ cldb.spray.c
+
 ###############################################################################
 #
 # Charm Libraries