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 ()
  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 () 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   PeerProcess *pp = 0;
 118   Endpoint ep (addr.c_str());
 119   (void)this->by_addr_.find(ep,pp);
 120   return pp;
 121 }
 122
 123 PeerProcess *
 124 HostProcess::find_peer (long h)
 125 {
 126   if (this->by_handle_.size() == 0)
 127     return 0;
 128   for (PeerArray::ITERATOR i(this->by_handle_);
 129        !i.done();
 130        i++)
 131     {
 132       PeerNode *node = reinterpret_cast<PeerNode *>(i.next()->item_);
 133       if (node->handle_ == h)
 134         return node->peer_;
 135     }
 136   return 0;
 137 }
 138
 139
 140 long
 141 HostProcess::pid () const
 142 {
 143   return this->pid_;
 144 }
 145
 146
 147 const ACE_Time_Value &
 148 HostProcess::start_time () const
 149 {
 150   return this->start_time_;
 151 }
 152
 153 void
 154 HostProcess::start_time (const ACE_Time_Value &start)
 155 {
 156   this->start_time_ = start;
 157 }
 158
 159 bool
 160 HostProcess::has_endpoint (const Endpoint& addr, bool listen)
 161 {
 162   AddrList &list = listen ? this->listen_endpoints_ : this->client_endpoints_;
 163   for (ACE_DLList_Iterator<Endpoint> i(list);
 164        !i.done();
 165        ++i)
 166     {
 167       Endpoint *elem = 0;
 168       i.next (elem);
 169       if (*elem == addr)
 170         return true;
 171     }
 172   return false;
 173 }
 174
 175 void
 176 HostProcess::add_client_endpoint(const Endpoint &addr)
 177 {
 178   this->client_endpoints_.insert_tail(new Endpoint (addr));
 179 }
 180
 181 void
 182 HostProcess::add_listen_endpoint(const Endpoint &addr)
 183 {
 184   this->listen_endpoints_.insert_tail(new Endpoint (addr));
 185 }
 186
 187 void
 188 HostProcess::add_peer(long handle, PeerProcess *peer)
 189 {
 190   peer->set_owner (this);
 191   PeerProcess *existing = this->find_peer(handle);
 192   if (existing != 0)
 193     {
 194       ACE_DEBUG ((LM_DEBUG,
 195                   "add_peer, found existing for %d\n",
 196                   handle));
 197     }
 198   const Endpoint &addr = peer->is_server() ?
 199     peer->server_addr() : peer->last_client_addr();
 200   int result = this->by_addr_.bind (addr,peer);
 201   if (result < 0)
 202     ACE_ERROR ((LM_ERROR,"add_peer, cannot bind to addr %s result = %d,  %p\n",
 203                 addr.addr_.c_str(), result, "by_addr_.bind"));
 204
 205   PeerNode *node = new PeerNode (handle,peer);
 206   this->by_handle_.insert_tail(node);
 207 }
 208
 209 void
 210 HostProcess::remove_peer(long h)
 211 {
 212   if (this->by_handle_.size() == 0)
 213     return;
 214   for (PeerArray::ITERATOR i(this->by_handle_);
 215        !i.done();
 216        i++)
 217     {
 218       PeerNode *node = reinterpret_cast<PeerNode *>(i.next()->item_);
 219       if (node->handle_ == h)
 220         {
 221           this->by_handle_.remove(i.next());
 222           return;
 223         }
 224     }
 225 }
 226
 227 void
 228 HostProcess::dump_summary (ostream &strm)
 229 {
 230   strm << "Host process " << this->proc_name_ << " pid(" << this->pid_ << ") from logfile " << this->logfile_name_ <<  endl;
 231   size_t num = this->listen_endpoints_.size();
 232   if (num > 0)
 233     {
 234       strm << "  listening on ";
 235       size_t count = 0;
 236       for (ACE_DLList_Iterator <Endpoint> t_iter (this->listen_endpoints_);
 237            !t_iter.done();
 238            t_iter.advance())
 239         {
 240           Endpoint *ep = 0;
 241           t_iter.next(ep);
 242           strm << ep->addr_.c_str();
 243           if (++count < num)
 244             strm << ", ";
 245         }
 246       strm << endl;
 247     }
 248
 249   strm << "  " << threads_.size() << " threads";
 250 #if 0
 251   if (clients_.current_size() > 0)
 252     strm << ", client to " << clients_.current_size();
 253   if (servers_.current_size() > 0)
 254     strm << ", server to " << servers_.current_size();
 255 #endif
 256   strm << endl;
 257 }
 258
 259 void
 260 HostProcess::dump_ident (ostream &strm, const char *message)
 261 {
 262   strm <<"\nHost process ";
 263   if (this->proc_name_.length())
 264     strm << this->proc_name_;
 265   strm << " pid (" << this->pid_ << ") " << message << endl;
 266 }
 267
 268 void
 269 HostProcess::dump_thread_summary (ostream &strm)
 270 {
 271   this->dump_ident (strm, "thread summary:");
 272   long total_sent = 0;
 273   long total_recv = 0;
 274   size_t total_bytes_sent = 0;
 275   size_t total_bytes_recv = 0;
 276   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 277        !t_iter.done();
 278        t_iter.advance())
 279     {
 280       Thread *thr = 0;
 281       t_iter.next(thr);
 282       thr->dump_summary (strm);
 283       thr->get_summary (total_recv, total_sent, total_bytes_recv, total_bytes_sent);
 284     }
 285   strm << "Total requests sent: " << total_sent << " received: " << total_recv << endl;
 286   strm << "Total requests bytes sent: " << total_bytes_sent << " received: " << total_bytes_recv << endl;
 287 }
 288
 289 void
 290 HostProcess::split_thread_invocations (Session *session, const ACE_Time_Value& start)
 291 {
 292   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 293        !t_iter.done();
 294        t_iter.advance())
 295     {
 296       Thread *thr = 0;
 297       t_iter.next(thr);
 298       char fname[100];
 299       thr->split_filename(fname,100);
 300       ostream *strm = session->stream_for (0, this, "threads", fname);
 301       thr->dump_invocations (*strm);
 302       *strm << endl;
 303       thr->dump_incidents (*strm, start);
 304       *strm << endl;
 305       delete strm;
 306     }
 307 }
 308 void
 309 HostProcess::dump_thread_invocations (ostream &strm, const ACE_Time_Value& start)
 310 {
 311   this->dump_ident (strm, "invocations by thread:");
 312   for (ACE_DLList_Iterator <Thread> t_iter (this->threads_);
 313        !t_iter.done();
 314        t_iter.advance())
 315     {
 316       Thread *thr = 0;
 317       t_iter.next(thr);
 318       thr->dump_invocations (strm);
 319       strm << endl;
 320       thr->dump_incidents (strm, start);
 321       strm << endl;
 322     }
 323 }
 324
 325 void
 326 HostProcess::iterate_peers (int group,
 327                             int operation,
 328                             ostream *strm,
 329                             Session *session)
 330 {
 331   bool first = true;
 332   for (PeerProcs::ITERATOR i = this->by_addr_.begin(); i != by_addr_.end(); i++)
 333     {
 334       PeerProcs::ENTRY *entry;
 335       if (i.next(entry) == 0)
 336         continue;
 337       PeerProcess *pp = entry->item();
 338       if (pp == 0)
 339         continue;
 340       if ((group != 3) &&
 341           ((pp->is_server() && group == 1)
 342            || (!pp->is_server() && group == 2)))
 343         continue;
 344
 345       switch (operation)
 346         {
 347         case 0:
 348           entry->item()->dump_summary (*strm);
 349           break;
 350         case 1:
 351           entry->item()->dump_object_detail (*strm);
 352           *strm << endl;
 353           break;
 354         case 2:
 355           if (strm == 0)
 356             {
 357               char fname[100];
 358               entry->item()->split_filename(fname,100);
 359               strm = session->stream_for (0, this, "peers",fname);
 360               this->dump_ident (*strm, "Invocation Detail");
 361               entry->item()->dump_invocation_detail (*strm);
 362               *strm << endl;
 363               delete (strm);
 364               strm = 0;
 365             }
 366           else
 367             {
 368               if (!first)
 369                 this->dump_ident (*strm,"Invocations continued");
 370               entry->item()->dump_invocation_detail (*strm);
 371               *strm << endl;
 372             }
 373           break;
 374         case 3:
 375           entry->item()->match_hosts (session);
 376           break;
 377         }
 378       first = false;
 379     }
 380 }
 381
 382 void
 383 HostProcess::dump_peer_summary (ostream &strm)
 384 {
 385   this->dump_ident (strm, "peer processes:");
 386   size_t num_servers = 0;
 387   size_t num_clients = 0;
 388   strm << " total peers: " << this->by_addr_.current_size() << endl;
 389   for (PeerProcs::ITERATOR i = this->by_addr_.begin(); i != by_addr_.end(); i++)
 390     {
 391       PeerProcs::ENTRY *entry;
 392       if (i.next(entry) == 0)
 393         break;
 394       PeerProcess *pp = entry->item();
 395       if (pp == 0)
 396         break;
 397       if (pp->is_server())
 398         ++num_servers;
 399       else
 400         ++num_clients;
 401     }
 402
 403   strm << " from " << num_clients << " clients" << endl;
 404   this->iterate_peers (1, 0, &strm);
 405   strm << " to " << num_servers << " servers" << endl;
 406   this->iterate_peers (2, 0, &strm);
 407 }
 408
 409 void
 410 HostProcess::dump_object_detail (ostream &strm)
 411 {
 412   this->dump_ident (strm, "peer objects: ");
 413   this->iterate_peers (3, 1, &strm);
 414 }
 415
 416 void
 417 HostProcess::split_peer_invocations (Session *session)
 418 {
 419   this->iterate_peers (3, 2, 0, session);
 420 }
 421
 422 void
 423 HostProcess::dump_invocation_detail(ostream &strm)
 424 {
 425   this->dump_ident (strm, "invocations: ");
 426   this->iterate_peers (3, 2, &strm);
 427   this->dump_ident (strm, "end invocation report");
 428 }
 429
 430 void
 431 HostProcess::reconcile_peers (Session *session)
 432 {
 433   this->iterate_peers (3, 3, 0, session);
 434 }