(file) Return to Monitor.cpp CVS log (file) (dir) Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/Monitor.cpp between version 1.32.2.2 and 1.48

version 1.32.2.2, 2002/10/29 21:13:17 version 1.48, 2003/09/22 17:39:39
Line 27 
Line 27 
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
   #include <Pegasus/Common/Config.h>
   
   #include <cstring>
 #include "Monitor.h" #include "Monitor.h"
   #include "MessageQueue.h"
   #include "Socket.h"
   #include <Pegasus/Common/Tracer.h>
   #include <Pegasus/Common/HTTPConnection.h>
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
 # if defined(FD_SETSIZE) && FD_SETSIZE != 1024 # if defined(FD_SETSIZE) && FD_SETSIZE != 1024
Line 48 
Line 55 
 # include <netinet/in.h> # include <netinet/in.h>
 # include <netdb.h> # include <netdb.h>
 # include <arpa/inet.h> # include <arpa/inet.h>
 # include <unistd.h>  
 #endif #endif
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
Line 60 
Line 66 
  
  
 static struct timeval create_time = {0, 1}; static struct timeval create_time = {0, 1};
 static struct timeval destroy_time = {15, 0};  static struct timeval destroy_time = {300, 0};
 static struct timeval deadlock_time = {0, 0}; static struct timeval deadlock_time = {0, 0};
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
Line 86 
Line 92 
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 Monitor::Monitor() Monitor::Monitor()
    : _async(false)     : _module_handle(0), _controller(0), _async(false), _stopConnections(0)
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
     _entries.reserveCapacity(32);     _entries.reserveCapacity(32);
     int i = 0;      for( int i = 0; i < 32; i++ )
     for( ; i < 32; i++ )  
     {     {
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
Line 100 
Line 105 
 } }
  
 Monitor::Monitor(Boolean async) Monitor::Monitor(Boolean async)
    : _async(async)     : _module_handle(0), _controller(0), _async(async), _stopConnections(0)
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
 //     _entries.reserveCapacity(32);      _entries.reserveCapacity(32);
 //     int i = 0;      for( int i = 0; i < 32; i++ )
 //     for( ; i < 32; i++ )      {
 //     {         _MonitorEntry entry(0, 0, 0);
 //        _MonitorEntry entry(0, 0, 0);         _entries.append(entry);
 //        _entries.append(entry);      }
 //     }  
   
     if( _async == true )     if( _async == true )
   
     {     {
        _thread_pool = new ThreadPool(0,        _thread_pool = new ThreadPool(0,
                                      "Monitor",                                      "Monitor",
                                      1,                                       0,
                                      0,                                      0,
                                      create_time,                                      create_time,
                                      destroy_time,                                      destroy_time,
Line 132 
Line 134 
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                   "deregistering with module controller");                   "deregistering with module controller");
  
       if(_module_handle != NULL)
       {
          _controller->deregister_module(PEGASUS_MODULENAME_MONITOR);
          _controller = 0;
          delete _module_handle;
       }
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");
  
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
Line 150 
Line 157 
    gettimeofday(&now, NULL);    gettimeofday(&now, NULL);
    int dead_threads = 0;    int dead_threads = 0;
  
    if( now.tv_sec - last.tv_sec > 300 )     if( now.tv_sec - last.tv_sec > 120 )
    {    {
       PEGASUS_STD(cout) << "Monitor Thread Pool currently has " <<  
          _thread_pool->running_count() +  
          _thread_pool->pool_count() << " Threads." << PEGASUS_STD(endl);  
       gettimeofday(&last, NULL);       gettimeofday(&last, NULL);
       try       try
       {       {
Line 174 
Line 178 
  
     Boolean handled_events = false;     Boolean handled_events = false;
     int i = 0;     int i = 0;
       #if defined(PEGASUS_OS_OS400) || defined(PEGASUS_OS_HPUX)
       struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};
   #else
     struct timeval tv = {0,1};     struct timeval tv = {0,1};
   #endif
     fd_set fdread;     fd_set fdread;
     FD_ZERO(&fdread);     FD_ZERO(&fdread);
       _entry_mut.lock(pegasus_thread_self());
   
       // Check the stopConnections flag.  If set, clear the Acceptor monitor entries
       if (_stopConnections == 1)
       {
           for ( int indx = 0; indx < (int)_entries.size(); indx++)
           {
               if (_entries[indx]._type == Monitor::ACCEPTOR)
               {
                   if ( _entries[indx]._status.value() != _MonitorEntry::EMPTY)
                   {
                      if ( _entries[indx]._status.value() == _MonitorEntry::IDLE ||
                           _entries[indx]._status.value() == _MonitorEntry::DYING )
                      {
                          // remove the entry
                          _entries[indx]._status = _MonitorEntry::EMPTY;
                      }
                      else
                      {
                          // set status to DYING
                         _entries[indx]._status.value() == _MonitorEntry::DYING;
                      }
                  }
              }
           }
           _stopConnections = 0;
       }
  
     for( int indx = 0; indx < (int)_entries.size(); indx++)     for( int indx = 0; indx < (int)_entries.size(); indx++)
     {     {
Line 187 
Line 221 
        }        }
     }     }
  
   
     int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);     int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
Line 202 
Line 237 
              MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);              MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
              if(q == 0)              if(q == 0)
              {              {
                 PEGASUS_STD(cout) << "Monitor:: found an empty connection slot" << PEGASUS_STD(endl);  
                 try                 try
                 {                 {
                       _entries[indx]._status = _MonitorEntry::EMPTY;                       _entries[indx]._status = _MonitorEntry::EMPTY;
                 }                 }
                 catch(...)                 catch(...)
                 {                 {
   
                 }                 }
                 return true;                  continue;
              }              }
              try              try
              {              {
Line 223 
Line 258 
                       MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();                       MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();
                       Message* message= new CloseConnectionMessage(_entries[indx].socket);                       Message* message= new CloseConnectionMessage(_entries[indx].socket);
                       message->dest = o.getQueueId();                       message->dest = o.getQueueId();
                         _entry_mut.unlock();
                       o.enqueue(message);                       o.enqueue(message);
                       return true;                       return true;
                    }                    }
Line 235 
Line 271 
                    events |= SocketMessage::READ;                    events |= SocketMessage::READ;
                    Message *msg = new SocketMessage(_entries[indx].socket, events);                    Message *msg = new SocketMessage(_entries[indx].socket, events);
                    _entries[indx]._status = _MonitorEntry::BUSY;                    _entries[indx]._status = _MonitorEntry::BUSY;
                      _entry_mut.unlock();
   
                    q->enqueue(msg);                    q->enqueue(msg);
                    _entries[indx]._status = _MonitorEntry::IDLE;                    _entries[indx]._status = _MonitorEntry::IDLE;
                    return true;                    return true;
Line 247 
Line 285 
           }           }
        }        }
     }     }
       _entry_mut.unlock();
     return(handled_events);     return(handled_events);
 } }
  
   void Monitor::stopListeningForConnections()
   {
       PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");
   
       _stopConnections = 1;
   
       PEG_METHOD_EXIT();
   }
   
  
 int  Monitor::solicitSocketMessages( int  Monitor::solicitSocketMessages(
     Sint32 socket,     Sint32 socket,
Line 258 
Line 306 
     int type)     int type)
 { {
  
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
  
    int index = -1;    int index = -1;
      _entry_mut.lock(pegasus_thread_self());
   
    for(index = 0; index < (int)_entries.size(); index++)    for(index = 0; index < (int)_entries.size(); index++)
    {    {
       try       try
       {       {
          if(_entries[index]._status.value() == _MonitorEntry::EMPTY)          if(_entries[index]._status.value() == _MonitorEntry::EMPTY)
          {          {
   
             _entries[index].socket = socket;             _entries[index].socket = socket;
             _entries[index].queueId  = queueId;             _entries[index].queueId  = queueId;
             _entries[index]._type = type;             _entries[index]._type = type;
             _entries[index]._status = _MonitorEntry::IDLE;             _entries[index]._status = _MonitorEntry::IDLE;
               _entry_mut.unlock();
   
             return index;             return index;
          }          }
       }       }
       catch(...)       catch(...)
       {       {
       }       }
   
    }    }
         _entry_mut.unlock();
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
    return index;    return index;
 } }
Line 286 
Line 339 
 void Monitor::unsolicitSocketMessages(Sint32 socket) void Monitor::unsolicitSocketMessages(Sint32 socket)
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
       _entry_mut.lock(pegasus_thread_self());
  
     for(int index = 0; index < (int)_entries.size(); index++)     for(int index = 0; index < (int)_entries.size(); index++)
     {     {
        if(_entries[index].socket == socket)        if(_entries[index].socket == socket)
        {        {
           _entries[index]._status = _MonitorEntry::EMPTY;           _entries[index]._status = _MonitorEntry::EMPTY;
             break;
        }        }
     }     }
       _entry_mut.unlock();
 PEG_METHOD_EXIT(); PEG_METHOD_EXIT();
 if( _async  == true )  
    PEGASUS_STD(cout) << "Monitor:: running " << _thread_pool->running_count() <<  
    " idle " << _thread_pool->pool_count() << PEGASUS_STD(endl);  
   
 } }
  
   
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm) PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)
 { {
    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);
Line 315 
Line 365 
 } }
  
  
   
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   ////************************* monitor 2 *****************************////
   
   
   m2e_rep::m2e_rep(void)
     :Base(), state(IDLE)
   
   {
   }
   
   m2e_rep::m2e_rep(monitor_2_entry_type _type,
                    pegasus_socket _sock,
                    void* _accept,
                    void* _dispatch)
     : Base(), type(_type), state(IDLE), psock(_sock),
       accept_parm(_accept), dispatch_parm(_dispatch)
   {
   
   }
   
   m2e_rep::~m2e_rep(void)
   {
   }
   
   m2e_rep::m2e_rep(const m2e_rep& r)
     : Base()
   {
     if(this != &r){
       type = r.type;
       psock = r.psock;
       accept_parm = r.accept_parm;
       dispatch_parm = r.dispatch_parm;
       state = IDLE;
   
     }
   }
   
   
   m2e_rep& m2e_rep::operator =(const m2e_rep& r)
   {
     if(this != &r) {
       type = r.type;
       psock = r.psock;
       accept_parm = r.accept_parm;
       dispatch_parm = r.dispatch_parm;
       state = IDLE;
     }
     return *this;
   }
   
   Boolean m2e_rep::operator ==(const m2e_rep& r)
   {
     if(this == &r)
       return true;
     return false;
   }
   
   Boolean m2e_rep::operator ==(void* r)
   {
     if((void*)this == r)
       return true;
     return false;
   }
   
   m2e_rep::operator pegasus_socket() const
   {
     return psock;
   }
   
   
   monitor_2_entry::monitor_2_entry(void)
   {
     _rep = new m2e_rep();
   }
   
   monitor_2_entry::monitor_2_entry(pegasus_socket& _psock,
                                    monitor_2_entry_type _type,
                                    void* _accept_parm, void* _dispatch_parm)
   {
     _rep = new m2e_rep(_type, _psock, _accept_parm, _dispatch_parm);
   }
   
   monitor_2_entry::monitor_2_entry(const monitor_2_entry& e)
   {
     if(this != &e){
       Inc(this->_rep = e._rep);
     }
   }
   
   monitor_2_entry::~monitor_2_entry(void)
   {
     Dec(_rep);
   }
   
   monitor_2_entry& monitor_2_entry::operator=(const monitor_2_entry& e)
   {
     if(this != &e){
       Dec(_rep);
       Inc(this->_rep = e._rep);
     }
     return *this;
   }
   
   Boolean monitor_2_entry::operator ==(const monitor_2_entry& me) const
   {
     if(this == &me)
       return true;
     return false;
   }
   
   Boolean monitor_2_entry::operator ==(void* k) const
   {
     if((void *)this == k)
       return true;
     return false;
   }
   
   
   monitor_2_entry_type monitor_2_entry::get_type(void) const
   {
     return _rep->type;
   }
   
   void monitor_2_entry::set_type(monitor_2_entry_type t)
   {
     _rep->type = t;
   }
   
   
   monitor_2_entry_state  monitor_2_entry::get_state(void) const
   {
     return (monitor_2_entry_state) _rep->state.value();
   }
   
   void monitor_2_entry::set_state(monitor_2_entry_state t)
   {
     _rep->state = t;
   }
   
   void* monitor_2_entry::get_accept(void) const
   {
     return _rep->accept_parm;
   }
   
   void monitor_2_entry::set_accept(void* a)
   {
     _rep->accept_parm = a;
   }
   
   
   void* monitor_2_entry::get_dispatch(void) const
   {
     return _rep->dispatch_parm;
   }
   
   void monitor_2_entry::set_dispatch(void* a)
   {
     _rep->dispatch_parm = a;
   }
   
   pegasus_socket monitor_2_entry::get_sock(void) const
   {
     return _rep->psock;
   }
   
   
   void monitor_2_entry::set_sock(pegasus_socket& s)
   {
     _rep->psock = s;
   
   }
   
   
   monitor_2::monitor_2(void)
     : _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),
       _ready(true),  _die(0), _requestCount(0)
   {
     try {
   
       bsd_socket_factory _factory;
   
       // set up the listener/acceptor
       pegasus_socket temp = pegasus_socket(&_factory);
   
       temp.socket(PF_INET, SOCK_STREAM, 0);
       // initialize the address
       memset(&_tickle_addr, 0, sizeof(_tickle_addr));
   #ifdef PEGASUS_OS_ZOS
       _tickle_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");
   #else
       _tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
   #endif
       _tickle_addr.sin_family = PF_INET;
       _tickle_addr.sin_port = 0;
   
       PEGASUS_SOCKLEN_SIZE _addr_size = sizeof(_tickle_addr);
   
       temp.bind((struct sockaddr *)&_tickle_addr, sizeof(_tickle_addr));
       temp.listen(3);
       temp.getsockname((struct sockaddr*)&_tickle_addr, &_addr_size);
   
       // set up the connector
   
       pegasus_socket tickler = pegasus_socket(&_factory);
       tickler.socket(PF_INET, SOCK_STREAM, 0);
       struct sockaddr_in _addr;
       memset(&_addr, 0, sizeof(_addr));
   #ifdef PEGASUS_OS_ZOS
       _addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");
   #else
       _addr.sin_addr.s_addr = inet_addr("127.0.0.1");
   #endif
       _addr.sin_family = PF_INET;
       _addr.sin_port = 0;
       tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));
       tickler.connect((struct sockaddr*)&_tickle_addr, sizeof(_tickle_addr));
   
       _tickler.set_sock(tickler);
       _tickler.set_type(INTERNAL);
       _tickler.set_state(BUSY);
   
       struct sockaddr_in peer;
       memset(&peer, 0, sizeof(peer));
       PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);
   
       pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);
       monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);
       _tickle->set_state(BUSY);
   
       _listeners.insert_first(_tickle);
   
     }
     catch(...){  }
   }
   
   monitor_2::~monitor_2(void)
   {
     try {
       monitor_2_entry* temp = _listeners.remove_first();
       while(temp){
         delete temp;
         temp = _listeners.remove_first();
       }
     }
     catch(...){  }
   }
   
   
   void monitor_2::run(void)
   {
     monitor_2_entry* temp;
     while(_die.value() == 0) {
        struct timeval tv = {0, 0};
   
       // place all sockets in the select set
       FD_ZERO(&rd_fd_set);
       try {
         _listeners.lock(pegasus_thread_self());
         temp = _listeners.next(0);
         while(temp != 0 ){
           if(temp->get_state() == CLOSED ){
             monitor_2_entry* closed = temp;
             temp = _listeners.next(closed);
             _listeners.remove_no_lock(closed);
             delete closed;
           }
           if(temp == 0)
              break;
           Sint32 fd = (Sint32) temp->get_sock();
           if(fd >= 0 )
              FD_SET(fd , &rd_fd_set);
           temp = _listeners.next(temp);
         }
         _listeners.unlock();
       }
       catch(...){
         return;
       }
       // important -  the dispatch routine has pointers to all the
       // entries that are readable. These entries can be changed but
       // the pointer must not be tampered with.
   
       int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);
       try {
         _listeners.lock(pegasus_thread_self());
         temp = _listeners.next(0);
         while(temp != 0 ){
           Sint32 fd = (Sint32) temp->get_sock();
           if(fd >= 0 && FD_ISSET(fd, &rd_fd_set)) {
             temp->set_state(BUSY);
             FD_CLR(fd,  &rd_fd_set);
             monitor_2_entry* ready = new monitor_2_entry(*temp);
             _ready.insert_first((void*)ready);
             _requestCount++;
           }
           temp = _listeners.next(temp);
         }
         _listeners.unlock();
       }
       catch(...){
         return;
       }
       // now handle the sockets that are ready to read
       _dispatch();
     } // while alive
   }
   
   void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))
   {
     void* old = (void *)_session_dispatch;
     _session_dispatch = dp;
     return old;
   }
   
   void* monitor_2::set_accept_dispatch(void (*dp)(monitor_2_entry*))
   {
     void* old = (void*)_accept_dispatch;
     _accept_dispatch = dp;
     return old;
   
   }
   
   
   // important -  the dispatch routine has pointers to all the
   // entries that are readable. These entries can be changed but
   // the pointer must not be tampered with.
   void monitor_2::_dispatch(void)
   {
     monitor_2_entry* entry = (monitor_2_entry*) _ready.remove_first();
     while(entry != 0 ){
       switch(entry->get_type()) {
       case INTERNAL:
         static char buffer[2];
         entry->get_sock().read(&buffer, 2);
         break;
       case LISTEN:
         {
           static struct sockaddr peer;
           static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);
           pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);
           monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());
           if(temp && _accept_dispatch != 0)
             _accept_dispatch(temp);
         }
         break;
       case SESSION:
         if(_session_dispatch != 0 )
           _session_dispatch(entry);
         else {
           static char buffer[4096];
           int bytes = entry->get_sock().read(&buffer, 4096);
         }
   
         break;
       case UNTYPED:
       default:
         break;
       }
       _requestCount--;
       delete entry;
       entry = (monitor_2_entry*) _ready.remove_first();
     }
   }
   
   void monitor_2::stop(void)
   {
     _die = 1;
     tickle();
   
     // shut down the listener list, free the list nodes
     _tickler.get_sock().close();
     _listeners.shutdown_queue();
   }
   
   void monitor_2::tickle(void)
   {
     static char _buffer[] =
       {
         '0','0'
       };
   
     _tickler.get_sock().write(&_buffer, 2);
   }
   
   
   monitor_2_entry*  monitor_2::add_entry(pegasus_socket& ps,
                                          monitor_2_entry_type type,
                                          void* accept_parm,
                                          void* dispatch_parm)
   {
     monitor_2_entry* m2e = new monitor_2_entry(ps, type, accept_parm, dispatch_parm);
   
     try{
       _listeners.insert_first(m2e);
     }
     catch(...){
       delete m2e;
       return 0;
     }
     tickle();
     return m2e;
   }
   
   Boolean monitor_2::remove_entry(Sint32 s)
   {
     monitor_2_entry* temp;
     try {
       _listeners.try_lock(pegasus_thread_self());
       temp = _listeners.next(0);
       while(temp != 0){
         if(s == (Sint32)temp->_rep->psock ){
           temp = _listeners.remove_no_lock(temp);
           delete temp;
           _listeners.unlock();
           return true;
         }
         temp = _listeners.next(temp);
       }
       _listeners.unlock();
     }
     catch(...){
     }
     return false;
   }
   
   Uint32 monitor_2::getOutstandingRequestCount(void)
   {
     return _requestCount.value();
   
   }
   
   
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

Legend:
Removed from v.1.32.2.2  
changed lines
  Added in v.1.48
No CVS admin address has been configured
 Powered by
ViewCVS 0.9.2