Home Research Papers Posters Manuals Talks Download People Help Internal

conv-ccs/ccs-builtins.C

Go to the documentation of this file.

/*****************************************************************************
 * A few useful built-in CCS handlers.
 *****************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>

#include "converse.h"
#include "ckhashtable.h"
#include "pup.h"
#include "pup_toNetwork.h"
#include "debug-conv++.h"
#include "conv-ccs.h"
#include "sockRoutines.h"
#include "queueing.h"
#include "ccs-builtins.h"

#ifdef __MINGW_H
#include "process.h"
#endif

#if CMK_CCS_AVAILABLE

void ccs_getinfo(char *msg);

/**********************************************
  "ccs_killport"-- takes one 4-byte big-endian port number
    Register a "client kill port".  When this program exits,
    it will connect to this TCP port and write "die\n\0" it.
*/

typedef struct killPortStruct{
  skt_ip_t ip;
  unsigned int port;
  struct killPortStruct *next;
} killPortStruct;
/*Only 1 kill list per node-- no Cpv needed*/
static killPortStruct *killList=NULL;

static void ccs_killport(char *msg)
{
  killPortStruct *oldList=killList;
  int port=ChMessageInt(*(ChMessageInt_t *)(msg+CmiReservedHeaderSize));
  skt_ip_t ip;
  unsigned int connPort;
  CcsCallerId(&ip,&connPort);
  killList=(killPortStruct *)malloc(sizeof(killPortStruct));
  killList->ip=ip;
  killList->port=port;
  killList->next=oldList;
  CmiFree(msg);
}
/*Send any registered clients kill messages before we exit*/
static int noMoreErrors(int c,const char *m) {return -1;}
extern "C" void CcsImpl_kill(void)
{
  skt_set_abort(noMoreErrors);
  while (killList!=NULL)
  {
    SOCKET fd=skt_connect(killList->ip,killList->port,20);
    if (fd!=INVALID_SOCKET) {
      skt_sendN(fd,"die\n",strlen("die\n")+1);
      skt_close(fd);
    }
    killList=killList->next;
  }
}

/**********************************************
  "ccs_killpe"-- kills the executing processor
    Used for fault-tolerance testing: terminate the processor.
*/

#include <signal.h>

static void ccs_killpe(char *msg) {
#if CMK_HAS_KILL
  kill(getpid(), 9);
#else
  CmiAbort("ccs_killpe() not supported!");
#endif
}

/*************************************************
List interface:
   This lets different parts of a Charm++ program register 
"list retrieval functions", which expose some aspect of a
running program.  Example lists include: all readonly globals, 
messages waiting in a queue, or all extant array elements.

  "ccs_list_len" (4-byte character count, null-terminated ASCII string)
Return the number of items currently in the list at the given path.

  "ccs_list_items.bin" (4-byte start, 4-byte end+1,4-byte extra
data count n, n-byte extra data, 4-byte character count c, c-byte
ASCII request string [without NULL terminator])
Return the given items (from start to end) of the given queue, 
formatted as raw network binary data.

  "ccs_list_items.fmt" [parameters as for .bin]
Return the given items, formatted as tagged network binary data.

  "ccs_list_items.txt" [parameters as for .bin]
Return the given items, formatted as ASCII text.
*/

static void CpdListBoundsCheck(CpdListAccessor *l,int &lo,int &hi)
{
  if (l->checkBoundary()) {
    int len=l->getLength();
    if (lo<0) lo=0;
    if (hi>len) hi=len;
  }
}

typedef CkHashtableTslow<const char *,CpdListAccessor *> CpdListTable_t;
CpvStaticDeclare(CpdListTable_t *,cpdListTable);

static CpdListAccessor *CpdListLookup(const char *path)
{
  CpdListAccessor *acc=CpvAccess(cpdListTable)->get(path);
  if (acc==NULL) {
    CmiError("CpdListAccessor> Unrecognized list path '%s'\n",path);
    return NULL;
  }
  return acc;
}

static CpdListAccessor *CpdListLookup(const ChMessageInt_t *lenAndPath)
{
  static const int CpdListMaxLen=80;
  int len=ChMessageInt(lenAndPath[0]);
  const char *path=(const char *)(lenAndPath+1);
  char pathBuf[CpdListMaxLen+1]; //Temporary null-termination buffer
  if ((len<0) || (len>CpdListMaxLen)) {
    CmiError("CpdListAccessor> Invalid list path length %d!\n",len);
    return NULL; //Character count is invalid
  }
  strncpy(pathBuf,path,len);
  pathBuf[len]=0; //Ensure string is null-terminated
  return CpdListLookup(pathBuf);
}

//CCS External access routines:

//Get the length of the given list:
static void CpdList_ccs_list_len(char *msg)
{
  const ChMessageInt_t *req=(const ChMessageInt_t *)(msg+CmiReservedHeaderSize);
  CpdListAccessor *acc=CpdListLookup(req);
  if (acc!=NULL) {
    ChMessageInt_t reply=ChMessageInt_new(acc->getLength());
    CcsSendReply(sizeof(reply),(void *)&reply);
  }
  CmiFree(msg);
}

//Read a list contents request header:
//  first item to send, 4-byte network integer
//  last item+1 to send, 4-byte network integer
//  extra data length, 4-byte network integer
//  extra data, list-defined bytes
//  list path length, 4-byte network integer (character count)
//  list path name, null-terminated ASCII
static CpdListAccessor *CpdListHeader_ccs_list_items(char *msg,
             CpdListItemsRequest &h)
{
  int msgLen=CmiSize((void *)msg)-CmiReservedHeaderSize;
  CpdListAccessor *ret=NULL;
  const ChMessageInt_t *req=(const ChMessageInt_t *)(msg+CmiReservedHeaderSize);
  h.lo=ChMessageInt(req[0]); // first item to send
  h.hi=ChMessageInt(req[1]); // last item to send+1
  h.extraLen=ChMessageInt(req[2]); // extra data length
  if (h.extraLen>=0 
  && ((int)(3*sizeof(ChMessageInt_t)+h.extraLen))<msgLen) {
    h.extra=(void *)(req+3);  // extra data
    ret=CpdListLookup((ChMessageInt_t *)(h.extraLen+(char *)h.extra));
    if (ret!=NULL) CpdListBoundsCheck(ret,h.lo,h.hi);
  }
  return ret;
}


// Pup this cpd list's items under this request.
static void pupCpd(PUP::er &p, CpdListAccessor *acc, CpdListItemsRequest &req)
{
      p.syncComment(PUP::sync_begin_array,"CpdList");
      acc->pup(p,req);
      p.syncComment(PUP::sync_end_array);
}

static void CpdList_ccs_list_items_txt(char *msg)
{
  CpdListItemsRequest req;
  CpdListAccessor *acc=CpdListHeader_ccs_list_items(msg,req);
  if(acc == NULL) CmiPrintf("ccs-builtins> Null Accessor--bad list name (txt)\n");
  if (acc!=NULL) {
    int bufLen;
    { 
      PUP::sizerText p; pupCpd(p,acc,req); bufLen=p.size(); 
    }
    char *buf=new char[bufLen];
    { 
      PUP::toText p(buf); pupCpd(p,acc,req);
      if (p.size()!=bufLen)
        CmiError("ERROR! Sizing/packing length mismatch for %s list pup function!\n",
                acc->getPath());
    }
    CcsSendReply(bufLen,(void *)buf);
    delete[] buf;
  }
  CmiFree(msg);
}

static void CpdList_ccs_list_items_set(char *msg)
{
  CpdListItemsRequest req;
  CpdListAccessor *acc=CpdListHeader_ccs_list_items(msg,req);
  if(acc == NULL) CmiPrintf("ccs-builtins> Null Accessor--bad list name (set)\n");
  else {
    PUP_toNetwork_unpack p(req.extra);
    pupCpd(p,acc,req);
    if (p.size()!=req.extraLen)
        CmiPrintf("Size mismatch during ccs_list_items.set: client sent %d bytes, but %d bytes used!\n",
                req.extraLen,p.size());
  }
  CmiFree(msg);
}

void CpdMachineArchitecture(char *msg) {
  char reply[8]; // where we store our reply
  reply[0]=CHARMDEBUG_MAJOR;
  reply[1]=CHARMDEBUG_MINOR;
  // decide if we are 32 bit (1) or 64 bit (2)
  reply[2] = 0;
  if (sizeof(char*) == 4) reply[2] = 1;
  else if (sizeof(char*) == 8) reply[2] = 2;
  // decide if we are little endian (1) or big endian (2)
  reply[3] = 0;
  int value = 1;
  char firstByte = *((char*)&value);
  if (firstByte == 1) reply[3] = 1;
  else reply[3] = 2;
  // add the third bit if we are in bigsim
#if CMK_BIGSIM_CHARM
  reply[3] |= 4;
#endif
  // get the size of an "int"
  reply[4] = sizeof(int);
  // get the size of an "long"
  reply[5] = sizeof(long);
#if CMK_LONG_LONG_DEFINED
  // get the size of an "long long"
  reply[6] = sizeof(long long);
#else
  // Per Filippo, the debugger will be fine with this. It should never
  // come up, since configure didn't detect support for `long long` on
  // the machine.
  reply[6] = 0;
#endif
  // get the size of an "bool"
  reply[7] = sizeof(bool);
  CcsSendReply(8, (void*)reply);
  CmiFree(msg);
}

static void CpdList_ccs_list_items_fmt(char *msg)
{
  CpdListItemsRequest req;
  CpdListAccessor *acc=CpdListHeader_ccs_list_items(msg,req);
  if (acc!=NULL) {
    int bufLen;
    { 
      PUP_toNetwork_sizer ps;
      PUP_fmt p(ps); 
      pupCpd(p,acc,req);
      bufLen=ps.size(); 
    }
    char *buf=new char[bufLen];
    { 
      PUP_toNetwork_pack pp(buf); 
      PUP_fmt p(pp);
      pupCpd(p,acc,req);
      if (pp.size()!=bufLen)
        CmiError("ERROR! Sizing/packing length mismatch for %s list pup function (%d sizing, %d packing)\n",
                acc->getPath(),bufLen,pp.size());
    }
    CcsSendReply(bufLen,(void *)buf);
    delete[] buf;
  }
  CmiFree(msg);
}




//Introspection object-- extract a list of CpdLists!
class CpdList_introspect : public CpdListAccessor {
  CpdListTable_t *tab;
public:
  CpdList_introspect(CpdListTable_t *tab_) :tab(tab_) { }
  virtual const char *getPath(void) const { return "converse/lists";}
  virtual size_t getLength(void) const {
    size_t len=0;
    CkHashtableIterator *it=tab->iterator();
    while (NULL!=it->next()) len++;
    delete it;
    return len;
  }
  virtual void pup(PUP::er &p,CpdListItemsRequest &req) {
    CkHashtableIterator *it=tab->iterator();
    void *objp;
    int curObj=0;
    while (NULL!=(objp=it->next())) {
      if (curObj>=req.lo && curObj<req.hi) {
        CpdListAccessor *acc=*(CpdListAccessor **)objp;
        char *pathName=(char *)acc->getPath();
        beginItem(p,curObj);
        p.comment("name");
        p(pathName,strlen(pathName));
      }
      curObj++;
    }
  }
};




#endif /*CMK_CCS_AVAILABLE*/
/*We have to include these virtual functions, even when CCS is
disabled, to avoid bizarre link-time errors.*/

// C++ and C client API
void CpdListRegister(CpdListAccessor *acc)
#if CMK_CCS_AVAILABLE
{
  CpvAccess(cpdListTable)->put(acc->getPath())=acc;
}
#else
{ }
#endif

extern "C" void CpdListRegister_c(const char *path,
            CpdListLengthFn_c len,void *lenParam,
            CpdListItemsFn_c items,void *itemsParam,int checkBoundary)
#if CMK_CCS_AVAILABLE
{
  CpdListRegister(new CpdListAccessor_c(path,
             len,lenParam,items,itemsParam,checkBoundary!=0?true:false));
}
#else
{ }
#endif

#if CMK_CCS_AVAILABLE


// Initialization       

static void CpdListInit(void) {
  CpvInitialize(CpdListTable_t *,cpdListTable);
  CpvAccess(cpdListTable)=new CpdListTable_t(31,0.5,
              CkHashFunction_string,CkHashCompare_string);
  CpdListRegister(new CpdList_introspect(CpvAccess(cpdListTable)));

  CcsRegisterHandler("ccs_list_len",(CmiHandler)CpdList_ccs_list_len);
  CcsRegisterHandler("ccs_list_items.txt",(CmiHandler)CpdList_ccs_list_items_txt);
  CcsRegisterHandler("ccs_list_items.fmt",(CmiHandler)CpdList_ccs_list_items_fmt);
  CcsRegisterHandler("ccs_list_items.set",(CmiHandler)CpdList_ccs_list_items_set);
  CcsRegisterHandler("debug/converse/arch",(CmiHandler)CpdMachineArchitecture);
}

#if CMK_WEB_MODE
/******************************************************
Web performance monitoring interface:
        Clients will register for performance data with 
processor 0 by calling the "perf_monitor" CCS request.  
Every WEB_INTERVAL (few seconds), this code
calls all registered web performance functions on all processors.  
The resulting integers are sent back to the client as a 
CCS reply.  

The current reply format is ASCII and rather nasty:
it's the string "perf" followed by space-separated list
of the performance functions for each processor.  By default,
only two performance functions are registered: the current 
processor utilization, in percent; and the current queue length.

The actual call sequence is:
CCS Client->CWebHandler->...  (processor 0)
  ...->CWeb_Collect->... (all processors)
...->CWeb_Reduce->CWeb_Deliver (processor 0 again)
*/

#if 0
#  define WEBDEBUG(x) CmiPrintf x
#else
#  define WEBDEBUG(x) /*empty*/
#endif

#define WEB_INTERVAL 1000 /*Time, in milliseconds, between performance snapshots*/
#define MAXFNS 20 /*Largest number of performance functions to expect*/

typedef struct {
        char hdr[CmiReservedHeaderSize];
        int fromPE;/*Source processor*/
        int perfData[MAXFNS];/*Performance numbers*/
} CWeb_CollectedData;

/*This needs to be made into a list of registered clients*/
static int hasApplet=0;
static CcsDelayedReply appletReply;

typedef int (*CWebFunction)(void);
static CWebFunction CWebPerformanceFunctionArray[MAXFNS];
static int CWebNoOfFns;
static int CWeb_ReduceIndex;
static int CWeb_CollectIndex;

/*Deliver the reduced web performance data to the waiting client:
*/
static int collectedCount;
static CWeb_CollectedData **collectedValues;

static void CWeb_Deliver(void)
{
  int i,j;

  if (hasApplet) {
    WEBDEBUG(("CWeb_Deliver to applet\n"));
    /*Send the performance data off to the applet*/
    char *reply=(char *)malloc(6+14*CmiNumPes()*CWebNoOfFns);
    sprintf(reply,"perf");
  
    for(i=0; i<CmiNumPes(); i++){
      for (j=0;j<CWebNoOfFns;j++)
      {
        char buf[20];
        sprintf(buf," %d",collectedValues[i]->perfData[j]);
        strcat(reply,buf);
      }
    }
    CcsSendDelayedReply(appletReply,strlen(reply) + 1, reply);
    free(reply);
    hasApplet=0;
  }
  else
    WEBDEBUG(("CWeb_Deliver (NO APPLET)\n"));
  
  /* Free saved performance data */
  for(i = 0; i < CmiNumPes(); i++){
    CmiFree(collectedValues[i]);
    collectedValues[i] = 0;
  }
  collectedCount = 0;
}

/*On PE 0, this handler accumulates all the performace data
*/
static void CWeb_Reduce(void *msg){
  CWeb_CollectedData *cur,*prev;
  int src;
  if(CmiMyPe() != 0){
    CmiAbort("CWeb performance data sent to wrong processor...\n");
  }
  WEBDEBUG(("CWeb_Reduce"));
  cur=(CWeb_CollectedData *)msg;
  src=cur->fromPE;
  prev = collectedValues[src]; /* Previous value, ideally 0 */
  collectedValues[src] = cur;
  if(prev == 0) collectedCount++;
  else CmiFree(prev); /*<- caused by out-of-order perf. data delivery*/

  if(collectedCount == CmiNumPes()){
    CWeb_Deliver();
  }
}

/*On each PE, this handler collects the performance data
and sends it to PE 0.
*/
static void CWeb_Collect(void)
{
  CWeb_CollectedData *msg;
  int i;

  WEBDEBUG(("CWeb_Collect on %d\n",CmiMyPe()));
  msg = (CWeb_CollectedData *)CmiAlloc(sizeof(CWeb_CollectedData));
  msg->fromPE = CmiMyPe();
  
  /* Evaluate each performance function*/
  for(i = 0; i < CWebNoOfFns; i++)
    msg->perfData[i] = CWebPerformanceFunctionArray[i] ();

  /* Send result off to node 0 */  
  CmiSetHandler(msg, CWeb_ReduceIndex);
  CmiSyncSendAndFree(0, sizeof(CWeb_CollectedData), msg);

  /* Re-call this function after a delay */
  CcdCallFnAfter((CcdVoidFn)CWeb_Collect, 0, WEB_INTERVAL);
}

extern "C" void CWebPerformanceRegisterFunction(CWebFunction fn)
{
  if (CmiMyRank()!=0) return; /* Should only register from rank 0 */
  if (CWebNoOfFns>=MAXFNS) CmiAbort("Registered too many CWebPerformance functions!");
  CWebPerformanceFunctionArray[CWebNoOfFns] = fn;
  CWebNoOfFns++;
}

/*This is called on PE 0 by clients that wish
to receive performance data.
*/
static void CWebHandler(void){
  if(CcsIsRemoteRequest()) {
    static int startedCollection=0;
    
    WEBDEBUG(("CWebHandler request on %d\n",CmiMyPe()));    
    hasApplet=1;
    appletReply=CcsDelayReply();
    
    if(startedCollection == 0){
      WEBDEBUG(("Starting data collection on every processor\n"));    
      int i;
      startedCollection=1;
      collectedCount=0;
      collectedValues = (CWeb_CollectedData **)malloc(sizeof(void *) * CmiNumPes());
      for(i = 0; i < CmiNumPes(); i++)
        collectedValues[i] = 0;
      
      /*Start collecting data on each processor*/
      for(i = 0; i < CmiNumPes(); i++){
        char *msg = (char *)CmiAlloc(CmiReservedHeaderSize);
        CmiSetHandler(msg, CWeb_CollectIndex);
        CmiSyncSendAndFree(i, CmiReservedHeaderSize,msg);
      }
    }
  }
}

struct CWebModeStats {
public:
        double beginTime; 
        double startTime; 
        double usedTime; 
        int PROCESSING; 
};
CpvStaticDeclare(CWebModeStats *,cwebStats);

/* Called to begin a collection interval
*/
static void usageReset(CWebModeStats *stats,double curWallTime)
{
   stats->beginTime=curWallTime;
   stats->usedTime = 0.;
}

/* Called when processor becomes busy
*/
static void usageStart(CWebModeStats *stats,double curWallTime)
{
   stats->startTime  = curWallTime;
   stats->PROCESSING = 1;
}

/* Called when processor becomes idle
*/
static void usageStop(CWebModeStats *stats,double curWallTime)
{
   stats->usedTime   += curWallTime - stats->startTime;
   stats->PROCESSING = 0;
}

/* Call this when the program is started
 -> Whenever traceModuleInit would be called
 -> -> see conv-core/convcore.c
*/
static void initUsage()
{
   CpvInitialize(CWebModeStats *, cwebStats);
   CWebModeStats *stats=new CWebModeStats;
   CpvAccess(cwebStats)=stats;
   usageReset(stats,CmiWallTimer());
   usageStart(stats,CmiWallTimer());
   CcdCallOnConditionKeep(CcdPROCESSOR_BEGIN_BUSY,(CcdVoidFn)usageStart,stats);
   CcdCallOnConditionKeep(CcdPROCESSOR_BEGIN_IDLE,(CcdVoidFn)usageStop,stats);    
}

static int getUsage(void)
{
   int usage = 0;
   double time      = CmiWallTimer();
   CWebModeStats *stats=CpvAccess(cwebStats);
   double totalTime = time - stats->beginTime;

   if(stats->PROCESSING)
   { /* Lock in current CPU usage */
      usageStop(stats,time); usageStart(stats,time);
   }
   if(totalTime > 0.)
      usage = (int)(0.5 + 100 *stats->usedTime/totalTime);
   usageReset(stats,time);

   return usage;
}

static int getSchedQlen(void)
{
  return(CqsLength((Queue)CpvAccess(CsdSchedQueue)));
}

#endif /*CMK_WEB_MODE*/

#if ! CMK_WEB_MODE
static void CWeb_Invalid(void)
{
  CmiAbort("Invalid web mode handler invoked!\n");
}
#endif

void CWebInit(void)
{
#if CMK_WEB_MODE
  CcsRegisterHandler("perf_monitor", (CmiHandler)CWebHandler);
  
  CWeb_CollectIndex=CmiRegisterHandler((CmiHandler)CWeb_Collect);
  CWeb_ReduceIndex=CmiRegisterHandler((CmiHandler)CWeb_Reduce);
  
  initUsage();
  CWebPerformanceRegisterFunction(getUsage);
  CWebPerformanceRegisterFunction(getSchedQlen);
#else
  /* always maintain the consistent CmiHandler table */
  /* which is good for heterogeneous clusters */
  CmiRegisterHandler((CmiHandler)CWeb_Invalid);
  CmiRegisterHandler((CmiHandler)CWeb_Invalid);
#endif
}


extern "C" void CcsBuiltinsInit(char **argv)
{
  CcsRegisterHandler("ccs_getinfo",(CmiHandler)ccs_getinfo);
  CcsRegisterHandler("ccs_killport",(CmiHandler)ccs_killport);
  CcsRegisterHandler("ccs_killpe",(CmiHandler)ccs_killpe);
  CWebInit();
  CpdListInit();
}


#endif /*CMK_CCS_AVAILABLE*/

void PUP_fmt::fieldHeader(typeCode_t typeCode,int nItems) {
    // Compute and write intro byte:
    lengthLen_t ll;
    if (nItems==1) ll=lengthLen_single;
    else if (nItems<256) ll=lengthLen_byte;
    else ll=lengthLen_int;
    // CmiPrintf("Intro byte: l=%d t=%d\n",(int)ll,(int)typeCode);
    byte intro=(((int)ll)<<4)+(int)typeCode;
    p(intro);
    // Compute and write length:
    switch(ll) {
    case lengthLen_single: break; // Single item
    case lengthLen_byte: {
        byte l=nItems;
        p(l);
        } break;
    case lengthLen_int: {
        p(nItems); 
        } break; 
    };
}

void PUP_fmt::comment(const char *message) {
        int nItems=strlen(message);
        fieldHeader(typeCode_comment,nItems);
        p((char *)message,nItems);
}
void PUP_fmt::synchronize(unsigned int m) {
        fieldHeader(typeCode_sync,1);
        p(m);
}
void PUP_fmt::bytes(void *ptr,int n,size_t itemSize,PUP::dataType t) {
        switch(t) {
        case PUP::Tchar:
        case PUP::Tuchar:
        case PUP::Tbyte:
                fieldHeader(typeCode_byte,n);
                p.bytes(ptr,n,itemSize,t);
                break;
        case PUP::Tshort: case PUP::Tint:
        case PUP::Tushort: case PUP::Tuint:
        case PUP::Tbool:
                fieldHeader(typeCode_int,n);
                p.bytes(ptr,n,itemSize,t);
                break;
        // treat "long" and "pointer" as 8-bytes, in conformity with pup_toNetwork.C
        case PUP::Tlong: case PUP::Tlonglong:
        case PUP::Tulong: case PUP::Tulonglong:
                fieldHeader(typeCode_long,n);
                p.bytes(ptr,n,itemSize,t);
                break;
        case PUP::Tfloat:
                fieldHeader(typeCode_float,n);
                p.bytes(ptr,n,itemSize,t);
                break;
        case PUP::Tdouble: case PUP::Tlongdouble:
                fieldHeader(typeCode_double,n);
                p.bytes(ptr,n,itemSize,t);
                break;
    case PUP::Tpointer:
        fieldHeader(typeCode_pointer,n);
        p.bytes(ptr,n,itemSize,t);
        break;
        default: CmiAbort("Unrecognized type code in PUP_fmt::bytes");
        };
}

---