paradyn/src/DMthread/DMmain.C

   1 /*
   2  * Copyright (c) 1996 Barton P. Miller
   3  *
   4  * We provide the Paradyn Parallel Performance Tools (below
   5  * described as Paradyn") on an AS IS basis, and do not warrant its
   6  * validity or performance.  We reserve the right to update, modify,
   7  * or discontinue this software at any time.  We shall have no
   8  * obligation to supply such updates or modifications or any other
   9  * form of support to you.
  10  *
  11  * This license is for research uses.  For such uses, there is no
  12  * charge. We define "research use" to mean you may freely use it
  13  * inside your organization for whatever purposes you see fit. But you
  14  * may not re-distribute Paradyn or parts of Paradyn, in any form
  15  * source or binary (including derivatives), electronic or otherwise,
  16  * to any other organization or entity without our permission.
  17  *
  18  * (for other uses, please contact us at paradyn@cs.wisc.edu)
  19  *
  20  * All warranties, including without limitation, any warranty of
  21  * merchantability or fitness for a particular purpose, are hereby
  22  * excluded.
  23  *
  24  * By your use of Paradyn, you understand and agree that we (or any
  25  * other person or entity with proprietary rights in Paradyn) are
  26  * under no obligation to provide either maintenance services,
  27  * update services, notices of latent defects, or correction of
  28  * defects for Paradyn.
  29  *
  30  * Even if advised of the possibility of such damages, under no
  31  * circumstances shall we (or any other person or entity with
  32  * proprietary rights in the software licensed hereunder) be liable
  33  * to you or any third party for direct, indirect, or consequential
  34  * damages of any character regardless of type of action, including,
  35  * without limitation, loss of profits, loss of use, loss of good
  36  * will, or computer failure or malfunction.  You agree to indemnify
  37  * us (and any other person or entity with proprietary rights in the
  38  * software licensed hereunder) for any and all liability it may
  39  * incur to third parties resulting from your use of Paradyn.
  40  */
  41
  42 #include <assert.h>
  43 extern "C" {
  44 double   quiet_nan();
  45 #include <malloc.h>
  46 #include <stdio.h>
  47 }
  48
  49 #include "thread/h/thread.h"
  50 #include "paradyn/src/TCthread/tunableConst.h"
  51 #include "dataManager.thread.SRVR.h"
  52 #include "dyninstRPC.xdr.CLNT.h"
  53 #include "DMdaemon.h"
  54 #include "DMmetric.h"
  55 #include "DMperfstream.h"
  56 #include "DMabstractions.h"
  57 #include "paradyn/src/pdMain/paradyn.h"
  58 #include "paradyn/src/UIthread/Status.h"
  59
  60 #include "util/h/Vector.h"
  61 #include "util/h/Dictionary.h"
  62 #include "util/h/String.h"
  63 #include "DMphase.h"
  64
  65 typedef vector<string> blahType;
  66
  67 // bool parse_metrics(string metric_file);
  68 bool metMain(string &userFile);
  69
  70 // this has to be declared before baseAbstr, cmfAbstr, and rootResource
  71 int dataManager::sock_fd;
  72 int dataManager::socket;
  73 dataManager *dataManager::dm = NULL;
  74
  75 dictionary_hash<string,abstraction*> abstraction::allAbstractions(string::hash);
  76 abstraction *baseAbstr = new abstraction("BASE");
  77 abstraction *cmfAbstr = new abstraction("CMF");
  78
  79 dictionary_hash<string,metric*> metric::allMetrics(string::hash);
  80 dictionary_hash<metricInstanceHandle,metricInstance *>
  81                 metricInstance::allMetricInstances(metricInstance::mhash);
  82 dictionary_hash<perfStreamHandle,performanceStream*>
  83                 performanceStream::allStreams(performanceStream::pshash);
  84 dictionary_hash<string, resource*> resource::allResources(string::hash);
  85 dictionary_hash<string,resourceList *> resourceList::allFoci(string::hash);
  86
  87 vector<resource*> resource::resources;
  88 vector<string> resource::lib_constraints;
  89 vector<metric*> metric::metrics;
  90 vector<paradynDaemon*> paradynDaemon::allDaemons;
  91 vector<daemonEntry*> paradynDaemon::allEntries;
  92 vector<executable*> paradynDaemon::programs;
  93 unsigned paradynDaemon::procRunning;
  94 vector<resourceList *> resourceList::foci;
  95 vector<phaseInfo *> phaseInfo::dm_phases;
  96 u_int metricInstance::next_id = 1;
  97 u_int performanceStream::next_id = 0;
  98 vector<DM_enableType*> paradynDaemon::outstanding_enables;
  99 u_int paradynDaemon::next_enable_id = 0;
 100 u_int paradynDaemon::count = 0;
 101
 102 resource *resource::rootResource = new resource();
 103 timeStamp metricInstance::curr_bucket_width;
 104 timeStamp metricInstance::global_bucket_width;
 105 phaseHandle metricInstance::curr_phase_id;
 106 u_int metricInstance::num_curr_hists = 0;
 107 u_int metricInstance::num_global_hists = 0;
 108
 109 double paradynDaemon::earliestFirstTime = 0;
 110 void newSampleRate(float rate);
 111
 112 extern void histDataCallBack(sampleValue*, timeStamp, int, int, void*, bool);
 113 extern void histFoldCallBack(timeStamp, void*, bool);
 114
 115 //
 116 // IO from application processes.
 117 //
 118 void dynRPCUser::applicationIO(int,int,string data)
 119 {
 120
 121     // NOTE: this fixes a purify error with the commented out code (a memory
 122     // segment error occurs occasionally with the line "cout << rest << endl")
 123     // this is problably not the best fix,  but I can't figure out why
 124     // the error is occuring (rest is always '\0' terminated when this
 125     // error occurs)---tn
 126     fprintf(stdout,data.string_of());
 127     fflush(stdout);
 128
 129 #ifdef n_def
 130     char *ptr;
 131     char *rest;
 132     // extra should really be per process.
 133     static string extra;
 134
 135     rest = P_strdup(data.string_of());
 136
 137     char *tp = rest;
 138     ptr = P_strchr(rest, '\n');
 139     while (ptr) {
 140         *ptr = '\0';
 141         if (pid) {
 142             printf("pid %d:", pid);
 143         } else {
 144             printf("paradynd: ");
 145         }
 146         if (extra.length()) {
 147             cout << extra;
 148             extra = (char*) NULL;
 149         }
 150         cout << rest << endl;
 151         rest = ptr+1;
 152         if(rest)
 153             ptr = P_strchr(rest, '\n');
 154         else
 155             ptr = 0;
 156     }
 157     extra += rest;
 158     delete tp;
 159     rest = 0;
 160 #endif
 161
 162 }
 163
 164 extern status_line *DMstatus;
 165
 166 void dynRPCUser::resourceBatchMode(bool onNow)
 167 {
 168    printf("error calling virtual func: dynRPCUser::resourceBatchMode\n");
 169 }
 170
 171 //
 172 // upcalls from remote process.
 173 //
 174 void dynRPCUser::resourceInfoCallback(int,
 175                                       vector<string> resource_name,
 176                                       string abstr, u_int type) {
 177
 178 printf("error calling virtual func: dynRPCUser::resourceInfoCallback\n");
 179
 180 }
 181
 182 void dynRPCUser::mappingInfoCallback(int,
 183                                      string abstraction,
 184                                      string type,
 185                                      string key,
 186                                      string value)
 187 {
 188   AMnewMapping(abstraction.string_of(),type.string_of(),key.string_of(),
 189                value.string_of());
 190 }
 191
 192 class uniqueName {
 193   public:
 194     uniqueName(stringHandle base) { name = base; nextId = 0; }
 195     int nextId;
 196     stringHandle name;
 197 };
 198
 199 //
 200 // handles a completed enable response: updates metricInstance state
 201 // and send the calling thread the response
 202 //
 203 void DMenableResponse(DM_enableType &enable,vector<bool> &successful){
 204
 205
 206     vector<metricInstance *> &mis = (*enable.request);
 207     assert(successful.size() == mis.size());
 208     vector<metricInstInfo> *response = new vector<metricInstInfo>(mis.size());
 209
 210     // update MI state and response vector
 211     for(u_int i=0; i < mis.size(); i++){
 212         if(mis[i] && successful[i]){  // this MI could be enabled
 213           mis[i]->setEnabled();
 214           metric *metricptr = metric::getMetric(mis[i]->getMetricHandle());
 215           if(metricptr){
 216
 217             if(enable.ph_type == CurrentPhase){
 218                 u_int old_current = mis[i]->currUsersCount();
 219                 bool  current_data = mis[i]->isCurrHistogram();
 220                 mis[i]->newCurrDataCollection(metricptr->getStyle(),
 221                                               histDataCallBack,
 222                                               histFoldCallBack);
 223                 mis[i]->newGlobalDataCollection(metricptr->getStyle(),
 224                                                 histDataCallBack,
 225                                                 histFoldCallBack);
 226                 mis[i]->addCurrentUser(enable.ps_handle);
 227                 // set sample rate to match current phase hist. bucket width
 228                 if(!metricInstance::numCurrHists()){
 229                     float rate = phaseInfo::GetLastBucketWidth();
 230                     newSampleRate(rate);
 231                 }
 232                 // new active curr. histogram added if there are no previous
 233                 // curr. subscribers and either persistent_collection is clear
 234                 // or there was no curr. histogram prior to this
 235                 if((!old_current)
 236                     && (mis[i]->currUsersCount() == 1) &&
 237                     (!(mis[i]->isCollectionPersistent()) || (!current_data))){
 238                     metricInstance::incrNumCurrHists();
 239                 }
 240                 // new global histogram if this metricInstance was just enabled
 241                 if(!((*enable.enabled)[i])){
 242                     metricInstance::incrNumGlobalHists();
 243                 }
 244             }
 245             else {  // this is a global phase enable
 246                 mis[i]->newGlobalDataCollection(metricptr->getStyle(),
 247                                                 histDataCallBack,
 248                                                 histFoldCallBack);
 249                 mis[i]->addGlobalUser(enable.ps_handle);
 250
 251                 // if this is first global histogram enabled and there are no
 252                 // curr hists, then set sample rate to global bucket width
 253                 if(!metricInstance::numCurrHists()){
 254                     if(!metricInstance::numGlobalHists()){
 255                         float rate = Histogram::getGlobalBucketWidth();
 256                         newSampleRate(rate);
 257                 }}
 258                 // new global hist added: update count
 259                 if(!((*enable.enabled)[i])){
 260                     metricInstance::incrNumGlobalHists();
 261                 }
 262             }
 263             // update response vector
 264             (*response)[i].successfully_enabled = true;
 265             (*response)[i].mi_id = mis[i]->getHandle();
 266             (*response)[i].m_id = mis[i]->getMetricHandle();
 267             (*response)[i].r_id = mis[i]->getFocusHandle();
 268             (*response)[i].metric_name = mis[i]->getMetricName();
 269             (*response)[i].focus_name = mis[i]->getFocusName();
 270             (*response)[i].metric_units = metricptr->getUnits();
 271             (*response)[i].units_type = metricptr->getUnitsType();
 272
 273             // update the persistence flags: the OR of new & previous values
 274             if(enable.persistent_data){
 275                 mis[i]->setPersistentData();
 276             }
 277             if(enable.persistent_collection){
 278                 mis[i]->setPersistentCollection();
 279             }
 280             if(enable.phase_persistent_data){
 281                 mis[i]->setPhasePersistentData();
 282             }
 283           }
 284           else {
 285               cout << "mis enabled but no metric handle: "
 286                    << mis[i]->getMetricHandle() << endl;
 287               assert(0);
 288           }
 289         }
 290         else {  // was not successfully enabled
 291             (*response)[i].successfully_enabled = false;
 292             (*response)[i].mi_id = mis[i]->getHandle();
 293             (*response)[i].m_id = mis[i]->getMetricHandle();
 294             (*response)[i].r_id = mis[i]->getFocusHandle();
 295             (*response)[i].metric_name = mis[i]->getMetricName();
 296             (*response)[i].focus_name = mis[i]->getFocusName();
 297         }
 298
 299 //        if(mis[i]) {
 300 //          (*response)[i].mi_id = mis[i]->getHandle();
 301 //          (*response)[i].m_id = mis[i]->getMetricHandle();
 302 //          (*response)[i].r_id = mis[i]->getFocusHandle();
 303 //          (*response)[i].metric_name = mis[i]->getMetricName();
 304 //          (*response)[i].focus_name = mis[i]->getFocusName();
 305 //        }
 306     }
 307
 308     // make response call
 309     dictionary_hash_iter<perfStreamHandle,performanceStream*>
 310             allS(performanceStream::allStreams);
 311     perfStreamHandle h; performanceStream *ps;
 312     while(allS.next(h,ps)){
 313         if(h == (perfStreamHandle)(enable.ps_handle)){
 314             ps->callDataEnableFunc(response,enable.client_id);
 315             return;
 316     } }
 317     response = 0;
 318 }
 319
 320 //
 321 // handle an enable response from a daemon. If all daemons have responded
 322 // then make response callback to calling thread, and check the outstanding
 323 // enables list to see if this enable response satisfies any waiting requests.
 324 // and enable for an MI is successful if its done entry is true and if its
 325 // MI* is not 0
 326 //
 327 void dynRPCUser::enableDataCallback(u_int daemon_id,
 328                                     vector<int> return_id,
 329                                     vector<u_int> mi_ids,
 330                                     u_int request_id)
 331 {
 332     // find element in outstanding_enables corr. to request_id
 333     u_int which =0;
 334     DM_enableType *request_entry = 0;
 335     for(u_int i=0; i < paradynDaemon::outstanding_enables.size(); i++){
 336         if((paradynDaemon::outstanding_enables[i])->request_id == request_id){
 337             which = i;
 338             request_entry = paradynDaemon::outstanding_enables[i];
 339             break;
 340     } }
 341
 342     if(!request_entry){
 343         // a request entry can be removed if a new phase event occurs
 344         // between the enable request and response, so ignore the response
 345         return;
 346     }
 347     assert(daemon_id < paradynDaemon::allDaemons.size());
 348     paradynDaemon *pd = paradynDaemon::allDaemons[daemon_id];
 349
 350     // for each mi in request update mi's components with new daemon if
 351     // it was successfully enabled
 352     assert(mi_ids.size() == return_id.size());
 353     for(u_int j=0; j< return_id.size(); j++){
 354         if(return_id[j] != -1){
 355             metricInstanceHandle mh =  mi_ids[j];
 356             metricInstance *mi = request_entry->findMI(mh);
 357             assert(mi);
 358             component *comp = new component(pd,return_id[j], mi);
 359             bool aflag;
 360             aflag=(mi->addComponent(comp));
 361             assert(aflag);
 362             // if at least one daemon could enable, update done and enabled
 363             request_entry->setDone(mh);
 364     } }
 365
 366     // update count of outstanding daemon responses
 367     assert(request_entry->how_many);
 368     request_entry->how_many--;
 369
 370     // all daemons have responded to enable request, send result to caller
 371     if(!request_entry->how_many) {
 372         vector<bool> successful( request_entry->request->size());
 373         for(u_int k=0; k < request_entry->request->size(); k++){
 374             // if MI is 0 or if done is false
 375             if(!((*(request_entry->done))[k])
 376                 || !((*(request_entry->request))[k])){
 377                 successful[k] = false;
 378             }
 379             else {
 380                 successful[k] = true;
 381             }
 382         }
 383         // if all daemons have responded update state for request and send
 384         // result to caller
 385         // a successful enable has both the enabled flag set and an mi*
 386
 387         // clear currentlyEnabling flag and decrement the count of
 388         // waiting enables for all MI's
 389         for(u_int i1=0; i1 < request_entry->done->size(); i1++){
 390             if((*request_entry->request)[i1]){
 391                 ((*request_entry->request)[i1])->clearCurrentlyEnabling();
 392                 if(request_entry->ph_type == CurrentPhase){
 393                     ((*request_entry->request)[i1])->decrCurrWaiting();
 394                 }
 395                 else{
 396                     ((*request_entry->request)[i1])->decrGlobalWaiting();
 397                 }
 398         } }
 399
 400         // update MI state for this entry and send response to caller
 401         DMenableResponse(*request_entry,successful);
 402
 403
 404         // remove this entry from the outstanding enables list
 405         u_int size = paradynDaemon::outstanding_enables.size();
 406         paradynDaemon::outstanding_enables[which] =
 407                          paradynDaemon::outstanding_enables[size-1];
 408                 paradynDaemon::outstanding_enables.resize(size-1);
 409
 410         // for each element on outstanding_enables, check to see if there are
 411         // any outstatnding_enables that can be satisfied by this request
 412         // if so, update state, and for any outstanding_enables that are
 413         // complete, send the result to the client thread
 414         // update not_all_done
 415         for(u_int i2=0; i2 < paradynDaemon::outstanding_enables.size(); i2++){
 416             DM_enableType *next_entry = paradynDaemon::outstanding_enables[i2];
 417             next_entry->updateAny(*(request_entry->request),successful);
 418         }
 419         delete request_entry;
 420         request_entry = 0;
 421
 422         if(paradynDaemon::outstanding_enables.size()){
 423           bool done = false;
 424           u_int i3 = 0;
 425           while(!done){
 426               if((paradynDaemon::outstanding_enables[i3])->not_all_done){
 427                   i3++;
 428               }
 429               else {  // this entry's request is complete
 430                    // update MI state for this entry and send response to caller
 431                    DM_enableType *temp = paradynDaemon::outstanding_enables[i3];
 432                    successful.resize(temp->request->size());
 433                    for(u_int k2=0; k2 < successful.size(); k2++){
 434                        if(!((*(temp->done))[k2])) successful[k2] = false;
 435                        else successful[k2] = true;
 436                    }
 437                    // decrement the number of waiting for enables for
 438                    // each MI in this response
 439                    for(u_int k3=0; k3 < temp->request->size(); k3++){
 440                        if((*temp->request)[k3]){
 441                            if(temp->ph_type == CurrentPhase){
 442                                ((*temp->request)[k3])->decrCurrWaiting();
 443                            }
 444                            else{
 445                               ((*temp->request)[k3])->decrGlobalWaiting();
 446                            }
 447                    } }
 448
 449                    DMenableResponse(*temp,successful);
 450
 451                    // remove entry from outstanding_enables list
 452                    u_int newsize=paradynDaemon::outstanding_enables.size()-1;
 453                    paradynDaemon::outstanding_enables[i3] =
 454                          paradynDaemon::outstanding_enables[newsize];
 455                    paradynDaemon::outstanding_enables.resize(newsize);
 456                    delete temp;
 457               }
 458               if(i3 >= paradynDaemon::outstanding_enables.size()) done = true;
 459           }
 460         }
 461     }
 462 }
 463
 464 //
 465 // Upcall from daemon in response to getPredictedDataCost call
 466 // id - perfStreamHandle assoc. with the call
 467 // req_id - an identifier assoc. with the request
 468 // val - the cost of enabling the metric/focus pair
 469 //
 470 void dynRPCUser::getPredictedDataCostCallback(u_int id,
 471                                               u_int req_id,
 472                                               float val,
 473                                               u_int clientID)
 474 {
 475     // find the assoc. perfStream and update it's pred data cost value
 476     dictionary_hash_iter<perfStreamHandle,performanceStream*>
 477                 allS(performanceStream::allStreams);
 478     perfStreamHandle h; performanceStream *ps;
 479     while(allS.next(h,ps)){
 480         if(h == (perfStreamHandle)id){
 481             ps->predictedDataCostCallback(req_id,val,clientID);
 482             return;
 483     } }
 484     // TODO: call correct routine
 485     assert(0);
 486 }
 487
 488 //
 489 // Display errors using showError function from the UIM class
 490 // This function allows to display error messages from paradynd
 491 // using the "upcall" or "call back" mechanism.
 492 // Parameters:  errCode = Error code
 493 //              errString = Error message
 494 //              hostName = Host name where the error occur
 495 // Call: there is a macro defined in "showerror.h". This macro must be
 496 //       used when calling this function. A typical call is:
 497 //       showErrorCallback(99, "Erro message test"). This macro will
 498 //       automatically insert the additional host info required.
 499 //
 500 void dynRPCUser::showErrorCallback(int errCode,
 501                                    string errString,
 502                                    string hostName)
 503 {
 504     string msg;
 505
 506     if (errString.length() > 0) {
 507         if (hostName.length() > 0) {
 508             msg = string("<Msg from daemon on host ") + hostName +
 509                   string("> ") + errString;
 510         }
 511         else {
 512             msg = string("<Msg from daemon on host ?> ") + errString;
 513         }
 514         uiMgr->showError(errCode, P_strdup(msg.string_of()));
 515     }
 516     else {
 517         uiMgr->showError(errCode, "");
 518     }
 519
 520     //
 521     // hostName.length() should always be > 0, otherwise
 522     // hostName is not defined (i.e. "?" will be used instead).
 523     // if errString.length()==0, (i.e. errString.string_of()==""),
 524     // then we will use the default error message in errorList.tcl
 525     // This message, however, will not include any info about the current
 526     // host name.
 527     //
 528 }
 529
 530 //
 531 // Paradynd calls this igen fn when it starts a new process (more
 532 // specifically, after it starts the new process and the new process
 533 // has completed running DYNINSTinit).
 534 //
 535 void dynRPCUser::newProgramCallbackFunc(int pid,
 536                                         vector<string> argvString,
 537                                         string machine_name,
 538                                         bool calledFromExec,
 539                                         bool runMe)
 540 {
 541     // there better be a paradynd running on this machine!
 542
 543     for (unsigned i = 0; i < paradynDaemon::allDaemons.size(); i++) {
 544         paradynDaemon *pd = paradynDaemon::allDaemons[i];
 545         if (pd->machine.length() && (pd->machine == machine_name)){
 546             if (!paradynDaemon::addRunningProgram(pid, argvString, pd, calledFromExec,
 547                                                   runMe))
 548                assert(false);
 549
 550             uiMgr->enablePauseOrRun();
 551
 552             return;
 553         }
 554     }
 555
 556     // for now, abort if there is no paradynd, this should not happen
 557     printf("process started on %s, can't find paradynd there\n",
 558            machine_name.string_of());
 559     printf("paradyn error #1 encountered\n");
 560     exit(-1);
 561 }
 562
 563 void dynRPCUser::newMetricCallback(T_dyninstRPC::metricInfo info)
 564 {
 565     addMetric(info);
 566 }
 567
 568 void dynRPCUser::firstSampleCallback (int,double) {
 569
 570   assert(0 && "Invalid virtual function");
 571 }
 572
 573 void dynRPCUser::cpDataCallbackFunc(int,double,int,double,double)
 574 {
 575     assert(0 && "Invalid virtual function");
 576 }
 577
 578 // batch the sample delivery
 579 void dynRPCUser::batchSampleDataCallbackFunc(int,
 580                     vector<T_dyninstRPC::batch_buffer_entry>)
 581 {
 582     assert(0 && "Invalid virtual function");
 583 }
 584
 585 //
 586 // When a paradynd is started remotely, ie not by paradyn, this upcall
 587 // reports the information for that paradynd to paradyn
 588 //
 589 void
 590 dynRPCUser::reportSelf (string , string , int , string)
 591 {
 592   assert(0);
 593   return;
 594 }
 595
 596 void
 597 dynRPCUser::reportStatus (string)
 598 {
 599     assert(0 && "Invalid virtual function");
 600 }
 601
 602 void
 603 dynRPCUser::processStatus(int, u_int)
 604 {
 605     assert(0 && "Invalid virtual function");
 606 }
 607
 608 void
 609 dynRPCUser::endOfDataCollection(int)
 610 {
 611   assert(0 && "Invalid virtual function");
 612 }
 613
 614
 615 //
 616 // establish socket that will be advertised to paradynd's
 617 // this socket will allow paradynd's to connect to paradyn for pvm
 618 //
 619 static void
 620 DMsetupSocket (int &sockfd)
 621 {
 622   // setup "well known" socket for pvm paradynd's to connect to
 623   bool aflag;
 624   aflag = ((dataManager::dm->socket =
 625            RPC_setup_socket (sockfd, AF_INET, SOCK_STREAM)) >= 0);
 626   assert(aflag);
 627
 628   // bind fd for this thread
 629   msg_bind (sockfd, true);
 630 }
 631
 632 static void
 633 DMnewParadynd ()
 634 {
 635   // accept the connection
 636   int new_fd = RPC_getConnect(dataManager::dm->sock_fd);
 637   if (new_fd < 0)
 638     uiMgr->showError(4, "");
 639
 640   // add new daemon to dictionary of all deamons
 641   paradynDaemon::addDaemon(new_fd);
 642 }
 643
 644 bool dataManager::DM_sequential_init(const char* met_file){
 645    string mfile = met_file;
 646    return(metMain(mfile));
 647 }
 648
 649 int dataManager::DM_post_thread_create_init(int tid) {
 650
 651
 652     thr_name("Data Manager");
 653     dataManager::dm = new dataManager(tid);
 654
 655     // supports argv passed to paradynDaemon
 656     // new paradynd's may try to connect to well known port
 657     DMsetupSocket (dataManager::dm->sock_fd);
 658
 659     bool aflag;
 660     aflag=(RPC_make_arg_list(paradynDaemon::args,
 661                              dataManager::dm->socket, 1, 1, "", false));
 662     assert(aflag);
 663
 664     // start initial phase
 665     string dm_phase0 = "phase_0";
 666     phaseInfo::startPhase(0.0,dm_phase0,false,false);
 667
 668     char DMbuff[64];
 669     unsigned int msgSize = 64;
 670     msg_send (MAINtid, MSG_TAG_DM_READY, (char *) NULL, 0);
 671     unsigned int tag = MSG_TAG_ALL_CHILDREN_READY;
 672     msg_recv (&tag, DMbuff, &msgSize);
 673     return 1;
 674 }
 675
 676 //
 677 // Main loop for the dataManager thread.
 678 //
 679 void *DMmain(void* varg)
 680 {
 681     unsigned fd_first = 0;
 682     // We declare the "printChangeCollection" tunable constant here; it will
 683     // last for the lifetime of this function, which is pretty much forever.
 684     // (used to be declared as global in DMappContext.C.  Globally declared
 685     //  tunables are now a no-no).  Note that the variable name (printCC) is
 686     // unimportant.   -AT
 687     tunableBooleanConstantDeclarator printCC("printChangeCollection",
 688               "Print the name of metric/focus when enabled or disabled",
 689               false, // initial value
 690               NULL, // callback
 691               developerConstant);
 692
 693     // Now the same for "printSampleArrival"
 694     extern bool our_print_sample_arrival;
 695     our_print_sample_arrival = false;
 696     extern void printSampleArrivalCallback(bool);
 697     tunableBooleanConstantDeclarator printSA("printSampleArrival",
 698               "Print out status lines to show the arrival of samples",
 699               our_print_sample_arrival, // init val
 700               printSampleArrivalCallback,
 701               developerConstant);
 702
 703     int tid; memcpy((void*)&tid,varg, sizeof(int));
 704     dataManager::DM_post_thread_create_init(tid);
 705
 706     int ret;
 707     unsigned int tag;
 708     paradynDaemon *pd = NULL;
 709     while (1) {
 710         for(unsigned i = 0; i < paradynDaemon::allDaemons.size(); i++){
 711             pd = paradynDaemon::allDaemons[i];
 712             // handle up to max async requests that may have been buffered
 713             // while blocking on a sync request
 714             while (pd->buffered_requests()){
 715                 if(pd->process_buffered() == T_dyninstRPC::error) {
 716                     cout << "error on paradyn daemon\n";
 717                     paradynDaemon::removeDaemon(pd, true);
 718         } } }
 719
 720         tag = MSG_TAG_ANY;
 721         // ret = msg_poll(&tag, true);
 722         ret = msg_poll_preference(&tag, true,fd_first);
 723         fd_first = !fd_first;
 724         assert(ret != THR_ERR);
 725
 726         if (tag == MSG_TAG_FILE) {
 727             // must be an upcall on something speaking the dynRPC protocol.
 728             if (ret == dataManager::dm->sock_fd){
 729                 DMnewParadynd(); // set up a new daemon
 730             }
 731             else {
 732                 for(unsigned i = 0; i < paradynDaemon::allDaemons.size(); i++){
 733                     pd = paradynDaemon::allDaemons[i];
 734                     if(pd->get_fd() == ret){
 735                         if(pd->waitLoop() == T_dyninstRPC::error) {
 736                             cout << "error on paradyn daemon\n";
 737                             paradynDaemon::removeDaemon(pd, true);
 738                     }}
 739
 740                     // handle async requests that may have been buffered
 741                     // while blocking on a sync request
 742                     while(pd->buffered_requests()){
 743                         if(pd->process_buffered() == T_dyninstRPC::error) {
 744                             cout << "error on paradyn daemon\n";
 745                             paradynDaemon::removeDaemon(pd, true);
 746                     }}
 747                 }
 748             }
 749         } else if (dataManager::dm->isValidTag
 750                   ((T_dataManager::message_tags)tag)) {
 751             if (dataManager::dm->waitLoop(true,
 752                (T_dataManager::message_tags)tag) == T_dataManager::error) {
 753                 // handle error
 754                 assert(0);
 755             }
 756         } else {
 757             cerr << "Unrecognized message in DMmain.C\n";
 758             assert(0);
 759         }
 760    }
 761 }
 762
 763
 764 void addMetric(T_dyninstRPC::metricInfo &info)
 765 {
 766     // if metric already exists return
 767     if(metric::allMetrics.defines(info.name)){
 768         return;
 769     }
 770     metric *met = new metric(info);
 771
 772     // now tell all perfStreams
 773     dictionary_hash_iter<perfStreamHandle,performanceStream*>
 774                 allS(performanceStream::allStreams);
 775     perfStreamHandle h;
 776     performanceStream *ps;
 777     while(allS.next(h,ps)){
 778         if(ps->controlFunc.mFunc){
 779             // set the correct destination thread.
 780             dataManager::dm->setTid(ps->threadId);
 781             dataManager::dm->newMetricDefined(ps->controlFunc.mFunc,
 782                                               ps->Handle(),
 783                                               met->getName(),
 784                                               met->getStyle(),
 785                                               met->getAggregate(),
 786                                               met->getUnits(),
 787                                               met->getHandle(),
 788                                               met->getUnitsType());
 789         }
 790     }
 791 }
 792
 793
 794 // I don't want to parse for '/' more than once, thus the use of a string vector
 795 resourceHandle createResource(vector<string>& resource_name, string& abstr, unsigned type) {
 796   resource *parent = NULL;
 797   unsigned r_size = resource_name.size();
 798   string p_name;
 799
 800
 801   switch (r_size) {
 802     case 0:
 803         // Should this case ever occur ?
 804         assert(0); break;
 805     case 1:
 806         parent = resource::rootResource; break;
 807     default:
 808         for (unsigned ri=0; ri<(r_size-1); ri++)
 809             p_name += string("/") + resource_name[ri];
 810         parent = resource::string_to_resource(p_name);
 811         assert(parent);
 812         break;
 813     }
 814     if (!parent) assert(0);
 815
 816
 817     /* first check to see if the resource has already been defined */
 818     resource *p = resource::resources[parent->getHandle()];
 819     string myName = p_name;
 820     myName += "/";
 821     myName += resource_name[r_size - 1];
 822     resourceHandle *child = p->findChild(myName.string_of());
 823     if (child){
 824         return(*child);
 825         delete child;
 826     }
 827
 828     // if abstr is not defined then use default abstraction
 829     if(!abstr.string_of()){
 830         abstr = string("BASE");
 831     }
 832
 833     /* then create it */
 834     resource *ret =  new resource(parent->getHandle(),resource_name,
 835                                   myName,abstr, type);
 836
 837     // check to see if the suppressMagnify option should be set...if
 838     // this resource is specifed in the mdl exclude_lib option
 839     vector<string> shared_lib_constraints;
 840     if(resource::get_lib_constraints(shared_lib_constraints) &&
 841        (string(parent->getFullName()) == "/Code")){
 842             for(u_int i=0; i < shared_lib_constraints.size(); i++){
 843                 if(shared_lib_constraints[i] == ret->getName()){
 844                     ret->setSuppressMagnify();
 845                 }
 846             }
 847     }
 848
 849     /* inform others about it if they need to know */
 850     dictionary_hash_iter<perfStreamHandle,performanceStream*>
 851                         allS(performanceStream::allStreams);
 852     perfStreamHandle h;
 853     performanceStream *ps;
 854     resourceHandle r_handle = ret->getHandle();
 855     string name = ret->getFullName();
 856     while(allS.next(h,ps)){
 857         ps->callResourceFunc(parent->getHandle(),r_handle,ret->getFullName(),
 858         ret->getAbstractionName());
 859     }
 860     return(r_handle);
 861 }
 862
 863 void newSampleRate(float rate)
 864 {
 865     paradynDaemon *pd = NULL;
 866     for(unsigned i = 0; i < paradynDaemon::allDaemons.size(); i++){
 867         pd = paradynDaemon::allDaemons[i];
 868         pd->setSampleRate(rate);
 869     }
 870 }
 871
 872 #ifdef ndef
 873 // Note - the metric parser has been moved into the dataManager
 874 bool parse_metrics(string metric_file) {
 875      bool parseResult = metMain(metric_file);
 876     return parseResult;
 877 }
 878 #endif
 879
 880