TAO/utils/logWalker/HostProcess.cpp

   1 #include "HostProcess.h"
   2 #include "Invocation.h"
   3 #include "PeerProcess.h"
   4 #include "Thread.h"
   5 #include "Session.h"
   6
   7 #include "ace/OS_NS_stdio.h"
   8
   9 PeerNode::PeerNode (long h, PeerProcess *p)
  10   :handle_ (h),
  11    peer_ (p)
  12 {
  13 }
  14
  15 HostProcess::HostProcess (const ACE_CString &src, long pid)
  16   : pid_(pid),
  17     logfile_name_(src),
  18     start_time_ (ACE_Time_Value::zero)
  19 {
  20 }
  21
  22 HostProcess::~HostProcess (void)
  23 {
  24   for (AddrList::ITERATOR i(this->listen_endpoints_);
  25        !i.done();
  26        i++)
  27     {
  28       delete reinterpret_cast<Endpoint *>(i.next()->item_);
  29     }
  30   for (ThreadList::ITERATOR i(this->threads_);
  31        !i.done();
  32        i++)
  33     {
  34       delete reinterpret_cast<Thread *>(i.next()->item_);
  35     }
  36 #if 0
  37   for (PeerProcs::ITERATOR i = by_addr_.begin(); i != servers_.end(); i++)
  38     {
  39       PeerProcs::ENTRY *entry;
  40       if (i.next(entry) == 0)
  41         break;
  42       delete entry->item();
  43     }
  44 #endif
  45 }
  46
  47 void
  48 HostProcess::proc_name (const ACE_CString &name)
  49 {
  50   this->proc_name_ = name;
  51 }
  52
  53 const ACE_CString&
  54 HostProcess::proc_name (void) const
  55 {
  56   return this->proc_name_;
  57 }
  58
  59 Thread *
  60 HostProcess::find_thread (long tid, size_t offset)
  61 {
  62   Thread *thr = 0;
  63   for (ACE_DLList_Iterator<Thread> i(threads_);
  64        !i.done();
  65        i.advance())
  66     {
  67       i.next(thr);
  68       if (thr->id() == tid)
  69         return thr;
  70     }
  71   char alias[20];
  72   ACE_OS::snprintf (alias, 20, "Thread[" ACE_SIZE_T_FORMAT_SPECIFIER_ASCII "]",
  73                    this->threads_.size() + 1);
  74   thr = new Thread (tid, alias, offset);
  75   threads_.insert_tail (thr);
  76   return thr;
  77 }
  78
  79 Thread *
  80 HostProcess::find_thread_for_peer (const ACE_CString &addr)
  81 {
  82   Thread *thr = 0;
  83   Endpoint ep (addr.c_str());
  84   for (ACE_DLList_Iterator<Thread> i(threads_);
  85        !i.done();
  86        i.advance())
  87     {
  88       i.next(thr);
  89       PeerProcess *pp = thr->peek_new_connection();
  90       if (pp == 0)
  91         continue;
  92
  93       if (pp->match_server_addr(ep))
  94         return thr;
  95     }
  96   return 0;
  97 }
  98
  99 Thread *
 100 HostProcess::find_thread_for_handle (long h)
 101 {
 102   Thread *thr = 0;
 103   for (ACE_DLList_Iterator<Thread> i(threads_);
 104        !i.done();
 105        i.advance())
 106     {
 107       i.next(thr);
 108       if (thr->active_handle() == h && thr->giop_target() != 0)
 109         return thr;
 110     }
 111   return thr;
 112 }
 113
 114 PeerProcess *
 115 HostProcess::find_peer (const ACE_CString &addr)
 116 {
 117
 118   PeerProcess *pp = 0;
 119   Endpoint ep (addr.c_str());
 120   (void)this->by_addr_.find(ep,pp);
 121   return pp;
 122 }
 123
 124 PeerProcess *
 125 HostProcess::find_peer (long h)
 126 {
 127   if (this->by_handle_.size() == 0)
 128     return 0;
 129   for (PeerArray::ITERATOR i(this->by_handle_);
 130        !i.done();
 131        i++)
 132     {
 133       PeerNode *node = reinterpret_cast<PeerNode *>(i.next()->item_);
 134       if (node->handle_ == h)
 135         return node->peer_;
 136     }
 137   return 0;
 138 }
 139
 140
 141 long
 142 HostProcess::pid (void) const
 143 {
 144   return this->pid_;
 145 }
 146
 147
 148 const ACE_Time_Value &
 149 HostProcess::start_time (void) const
 150 {
 151   return this->start_time_;
 152 }
 153
 154 void
 155 HostProcess::start_time (const ACE_Time_Value &start)
 156 {
 157   this->start_time_ = start;
 158 }
 159
 160 bool
 161 HostProcess::has_endpoint (const Endpoint& addr, bool listen)
 162 {
 163   AddrList &list = listen ? this->listen_endpoints_ : this->client_endpoints_;
 164   for (ACE_DLList_Iterator<Endpoint> i(list);
 165        !i.done();
 166        ++i)
 167     {
 168       Endpoint *elem = 0;
 169       i.next (elem);
 170       if (*elem == addr)
 171         return true;
 172     }
 173   return false;
 174 }
 175
 176 void
 177 HostProcess::add_client_endpoint(const Endpoint &addr)
 178 {
 179   this->client_endpoints_.insert_tail(new Endpoint (addr));
 180 }
 181
 182 void
 183 HostProcess::add_listen_endpoint(const Endpoint &addr)
 184 {
 185   this->listen_endpoints_.insert_tail(new Endpoint (addr));
 186 }
 187
 188 void
 189 HostProcess::add_peer(long handle, PeerProcess *peer)
 190 {
 191   peer->set_owner (this);
 192   PeerProcess *existing = this->find_peer(handle);
 193   if (existing != 0)
 194     {
 195       ACE_DEBUG ((LM_DEBUG,
 196                   "add_peer, found existing for %d\n",
 197                   handle));
 198     }
 199   const Endpoint &addr = peer->is_server() ?
 200     peer->server_addr() : peer->last_client_addr();
 201   int result = this->by_addr_.bind (addr,peer);
 202   if (result < 0)
 203     ACE_ERROR ((LM_ERROR,"add_peer, cannot bind to addr %s result = %d,  %p\n",
 204                 addr.addr_.c_str(), result, "by_addr_.bind"));
 205
 206   PeerNode *node = new PeerNode (handle,peer);
 207   this->by_handle_.insert_tail(node);
 208 }
 209
 210 void
 211 HostProcess::remove_peer(long h)
 212 {
 213   if (this->by_handle_.size() == 0)
 214     return;
 215   for (PeerArray::ITERATOR i(this->by_handle_);
 216        !i.done();
 217        i++)
 218     {
 219       PeerNode *node = reinterpret_cast<PeerNode *>(i.next()->item_);
 220       if (node->handle_ == h)
 221         {
 222           this->by_handle_.remove(i.next());
 223           return;
 224         }
 225     }
 226 }
 227
 228 void
 229 HostProcess::dump_summary (ostream &strm)
 230 {
 231   strm << "Host process " << this->proc_name_ << " pid(" << this->pid_ << ") from logfile " << this->logfile_name_ <<  endl;
 232   size_t num = this->listen_endpoints_.size();
 233   if (num > 0)
 234     {
 235       strm << "  listening on ";
 236       size_t count = 0;
 237       for (ACE_DLList_Iterator <Endpoint> t_iter (this->listen_endpoints_);
 238            !t_iter.done();
 239            t_iter.advance())
 240         {
 241           Endpoint *ep = 0;
 242           t_iter.next(ep);
 243           strm << ep->addr_.c_str();
 244           if (++count < num)
 245             strm << ", ";
 246         }
 247       strm << endl;
 248     }
 249
 250   strm << "  " << threads_.size() << " threads";
 251 #if 0
 252   if (clients_.current_size() > 0)
 253     strm << ", client to " << clients_.current_size();
 254   if (servers_.current_size() > 0)
 255     strm << ", server to " << servers_.current_size();
 256 #endif
 257   strm << endl;
 258 }
 259
 260 void
 261 HostProcess::dump_ident (ostream &strm, const char *message)
 262 {
 263   strm <<"\nHost process ";
 264   if (this->proc_name_.length())
 265     strm << this->proc_name_;
 266   strm << " pid (" << this->pid_ << ") " << message << endl;
 267 }
 268
 269 void
 270 HostProcess::dump_thread_summary (ostream &strm)
 271 {
 272   this->dump_ident (strm, "thread summary:");
 273   long total_sent = 0;
 274   long total_recv = 0;
 275   size_t total_bytes_sent = 0;
 276   size_t total_bytes_recv = 0;
 277   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 278        !t_iter.done();
 279        t_iter.advance())
 280     {
 281       Thread *thr = 0;
 282       t_iter.next(thr);
 283       thr->dump_summary (strm);
 284       thr->get_summary (total_recv, total_sent, total_bytes_recv, total_bytes_sent);
 285     }
 286   strm << "Total requests sent: " << total_sent << " received: " << total_recv << endl;
 287   strm << "Total requests bytes sent: " << total_bytes_sent << " received: " << total_bytes_recv << endl;
 288 }
 289
 290 void
 291 HostProcess::split_thread_invocations (Session *session, const ACE_Time_Value& start)
 292 {
 293   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 294        !t_iter.done();
 295        t_iter.advance())
 296     {
 297       Thread *thr = 0;
 298       t_iter.next(thr);
 299       char fname[100];
 300       thr->split_filename(fname,100);
 301       ostream *strm = session->stream_for (0, this, "threads", fname);
 302       thr->dump_invocations (*strm);
 303       *strm << endl;
 304       thr->dump_incidents (*strm, start);
 305       *strm << endl;
 306       delete strm;
 307     }
 308 }
 309 void
 310 HostProcess::dump_thread_invocations (ostream &strm, const ACE_Time_Value& start)
 311 {
 312   this->dump_ident (strm, "invocations by thread:");
 313   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 314        !t_iter.done();
 315        t_iter.advance())
 316     {
 317       Thread *thr = 0;
 318       t_iter.next(thr);
 319       thr->dump_invocations (strm);
 320       strm << endl;
 321       thr->dump_incidents (strm, start);
 322       strm << endl;
 323     }
 324 }
 325
 326 void
 327 HostProcess::iterate_peers (int group,
 328                             int operation,
 329                             ostream *strm,
 330                             Session *session)
 331 {
 332   bool first = true;
 333   for (PeerProcs::ITERATOR i = this->by_addr_.begin(); i != by_addr_.end(); i++)
 334     {
 335       PeerProcs::ENTRY *entry;
 336       if (i.next(entry) == 0)
 337         continue;
 338       PeerProcess *pp = entry->item();
 339       if (pp == 0)
 340         continue;
 341       if ((group != 3) &&
 342           ((pp->is_server() && group == 1)
 343            || (!pp->is_server() && group == 2)))
 344         continue;
 345
 346       switch (operation)
 347         {
 348         case 0:
 349           entry->item()->dump_summary (*strm);
 350           break;
 351         case 1:
 352           entry->item()->dump_object_detail (*strm);
 353           *strm << endl;
 354           break;
 355         case 2:
 356           if (strm == 0)
 357             {
 358               char fname[100];
 359               entry->item()->split_filename(fname,100);
 360               strm = session->stream_for (0, this, "peers",fname);
 361               this->dump_ident (*strm, "Invocation Detail");
 362               entry->item()->dump_invocation_detail (*strm);
 363               *strm << endl;
 364               delete (strm);
 365               strm = 0;
 366             }
 367           else
 368             {
 369               if (!first)
 370                 this->dump_ident (*strm,"Invocations continued");
 371               entry->item()->dump_invocation_detail (*strm);
 372               *strm << endl;
 373             }
 374           break;
 375         case 3:
 376           entry->item()->match_hosts (session);
 377           break;
 378         }
 379       first = false;
 380     }
 381 }
 382
 383 void
 384 HostProcess::dump_peer_summary (ostream &strm)
 385 {
 386   this->dump_ident (strm, "peer processes:");
 387   size_t num_servers = 0;
 388   size_t num_clients = 0;
 389   strm << " total peers: " << this->by_addr_.current_size() << endl;
 390   for (PeerProcs::ITERATOR i = this->by_addr_.begin(); i != by_addr_.end(); i++)
 391     {
 392       PeerProcs::ENTRY *entry;
 393       if (i.next(entry) == 0)
 394         break;
 395       PeerProcess *pp = entry->item();
 396       if (pp == 0)
 397         break;
 398       if (pp->is_server())
 399         ++num_servers;
 400       else
 401         ++num_clients;
 402     }
 403
 404   strm << " from " << num_clients << " clients" << endl;
 405   this->iterate_peers (1, 0, &strm);
 406   strm << " to " << num_servers << " servers" << endl;
 407   this->iterate_peers (2, 0, &strm);
 408 }
 409
 410 void
 411 HostProcess::dump_object_detail (ostream &strm)
 412 {
 413   this->dump_ident (strm, "peer objects: ");
 414   this->iterate_peers (3, 1, &strm);
 415 }
 416
 417 void
 418 HostProcess::split_peer_invocations (Session *session)
 419 {
 420   this->iterate_peers (3, 2, 0, session);
 421 }
 422
 423 void
 424 HostProcess::dump_invocation_detail(ostream &strm)
 425 {
 426   this->dump_ident (strm, "invocations: ");
 427   this->iterate_peers (3, 2, &strm);
 428   this->dump_ident (strm, "end invocation report");
 429 }
 430
 431 void
 432 HostProcess::reconcile_peers (Session *session)
 433 {
 434   this->iterate_peers (3, 3, 0, session);
 435 }