(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.44 and 1.59

version 1.44, 2003/08/22 16:15:22 version 1.59, 2003/10/10 17:38:59
Line 23 
Line 23 
 // //
 // Author: Mike Brasher (mbrasher@bmc.com) // Author: Mike Brasher (mbrasher@bmc.com)
 // //
 // Modified By:  // Modified By: Mike Day (monitor_2) mdday@us.ibm.com
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
Line 91 
Line 91 
 // //
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
   #define MAX_NUMBER_OF_MONITOR_ENTRIES  32
 Monitor::Monitor() Monitor::Monitor()
    : _module_handle(0), _controller(0), _async(false)     : _module_handle(0), _controller(0), _async(false), _stopConnections(0)
 { {
       int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
     _entries.reserveCapacity(32);      _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);
     for( int i = 0; i < 32; i++ )      for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )
     {     {
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
Line 105 
Line 107 
 } }
  
 Monitor::Monitor(Boolean async) Monitor::Monitor(Boolean async)
    : _module_handle(0), _controller(0), _async(async)     : _module_handle(0), _controller(0), _async(async), _stopConnections(0)
 { {
       int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
     _entries.reserveCapacity(32);      _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);
     for( int i = 0; i < 32; i++ )      for( int i = 0; i < numberOfMonitorEntriesToAllocate; i++ )
     {     {
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
Line 187 
Line 190 
     FD_ZERO(&fdread);     FD_ZERO(&fdread);
     _entry_mut.lock(pegasus_thread_self());     _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 = _MonitorEntry::DYING;
                      }
                  }
              }
           }
           _stopConnections = 0;
       }
   
       Uint32 _idleEntries = 0;
   
     for( int indx = 0; indx < (int)_entries.size(); indx++)     for( int indx = 0; indx < (int)_entries.size(); indx++)
     {     {
        if(_entries[indx]._status.value() == _MonitorEntry::IDLE)        if(_entries[indx]._status.value() == _MonitorEntry::IDLE)
        {        {
             _idleEntries++;
           FD_SET(_entries[indx].socket, &fdread);           FD_SET(_entries[indx].socket, &fdread);
        }        }
     }     }
  
       _entry_mut.unlock();
     int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);     int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);
      _entry_mut.lock(pegasus_thread_self());
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
     if(events && events != SOCKET_ERROR )      if(events == SOCKET_ERROR)
 #else #else
     if(events && events != -1 )      if(events == -1)
 #endif #endif
     {     {
          Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
             "Monitor::run - errorno = %d has occurred on select.", errno);
          // The EBADF error indicates that one or more or the file
          // descriptions was not valid. This could indicate that
          // the _entries structure has been corrupted or that
          // we have a synchronization error.
   
          PEGASUS_ASSERT(errno != EBADF);
       }
       else if (events)
       {
          Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
             "Monitor::run select event received events = %d, monitoring %d idle entries",
              events, _idleEntries);
        for( int indx = 0; indx < (int)_entries.size(); indx++)        for( int indx = 0; indx < (int)_entries.size(); indx++)
        {        {
           if(FD_ISSET(_entries[indx].socket, &fdread))            // The Monitor should only look at entries in the table that are IDLE (i.e.,
             // owned by the Monitor).
             if((_entries[indx]._status.value() == _MonitorEntry::IDLE) &&
                (FD_ISSET(_entries[indx].socket, &fdread)))
           {           {
              MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);              MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
              if(q == 0)               Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
              {                    "Monitor::run indx = %d, queueId =  %d, q = %p",
                 try                    indx, _entries[indx].queueId, q);
                 {               PEGASUS_ASSERT(q !=0);
                    _entries[indx]._status = _MonitorEntry::EMPTY;  
                 }  
                 catch(...)  
                 {  
  
                 }  
                 continue;  
              }  
              try              try
              {              {
                 if(_entries[indx]._type == Monitor::CONNECTION)                 if(_entries[indx]._type == Monitor::CONNECTION)
                 {                 {
                      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                        "_entries[indx].type for indx = %d is Monitor::CONNECTION", indx);
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                    static_cast<HTTPConnection *>(q)->_entry_index = indx;
                    if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )                    if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )
                    {                    {
                         Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                             "Monitor::run processing dying value > 0 for indx = %d, connection being closed.",
                             indx);
                       _entries[indx]._status = _MonitorEntry::DYING;                       _entries[indx]._status = _MonitorEntry::DYING;
                       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);
Line 241 
Line 289 
                 }                 }
                 else                 else
                 {                 {
                      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                        "Non-connection entry, indx = %d, has been received.", indx);
                    int events = 0;                    int events = 0;
                    events |= SocketMessage::READ;                    events |= SocketMessage::READ;
                    Message *msg = new SocketMessage(_entries[indx].socket, events);                    Message *msg = new SocketMessage(_entries[indx].socket, events);
Line 263 
Line 313 
     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(
Line 274 
Line 332 
  
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
  
    int index = -1;  
    _entry_mut.lock(pegasus_thread_self());    _entry_mut.lock(pegasus_thread_self());
  
    for(index = 0; index < (int)_entries.size(); index++)     for(int index = 0; index < (int)_entries.size(); index++)
    {    {
       try       try
       {       {
Line 299 
Line 356 
    }    }
       _entry_mut.unlock();       _entry_mut.unlock();
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
    return index;     return -1;
 } }
  
 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());     _entry_mut.lock(pegasus_thread_self());
  
Line 312 
Line 370 
        if(_entries[index].socket == socket)        if(_entries[index].socket == socket)
        {        {
           _entries[index]._status = _MonitorEntry::EMPTY;           _entries[index]._status = _MonitorEntry::EMPTY;
             _entries[index].socket = -1;
           break;           break;
        }        }
     }     }
Line 322 
Line 381 
 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);
      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
           "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, q = %p",
           dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);
      try
      {
    dst->run(1);    dst->run(1);
    if(  dst->_monitor->_entries.size() > (Uint32)dst->_entry_index )     }
       dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;     catch (...)
      {
         Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
             "Monitor::_dispatch: exception received");
      }
      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
             "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);
  
      dst->_monitor->_entry_mut.lock(pegasus_thread_self());
      // It shouldn't be necessary to set status = _MonitorEntry::IDLE
      // if the connection is being closed.  However, the current logic
      // in Monitor::run requires this value to be set for the close
      // to be processed.
   
      PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);
      dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
      if (dst->_connectionClosePending)
      {
         dst->_dying = 1;
      }
      dst->_monitor->_entry_mut.unlock();
    return 0;    return 0;
 } }
  
Line 341 
Line 423 
 ////************************* monitor 2 *****************************//// ////************************* monitor 2 *****************************////
  
  
   
   
   
 m2e_rep::m2e_rep(void) m2e_rep::m2e_rep(void)
   :Base(), state(IDLE)   :Base(), state(IDLE)
  
Line 509 
Line 594 
  
 } }
  
   //static monitor_2* _m2_instance;
   
   AsyncDQueue<HTTPConnection2> monitor_2::_connections(true, 0);
   
   // monitor_2* monitor_2::get_monitor2(void)
   // {
   //    if(_m2_instance == 0 )
   //       _m2_instance = new monitor_2();
   //    PEGASUS_STD(cout) << "_m2_instance: " << _m2_instance << PEGASUS_STD(endl);
   
   //       return _m2_instance;
   // }
  
 monitor_2::monitor_2(void) monitor_2::monitor_2(void)
   : _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),   : _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),
     _ready(true),  _die(0), _requestCount(0)      _ready(true, 0), _die(0), _requestCount(0)
 { {
   try {   try {
  
Line 524 
Line 621 
     temp.socket(PF_INET, SOCK_STREAM, 0);     temp.socket(PF_INET, SOCK_STREAM, 0);
     // initialize the address     // initialize the address
     memset(&_tickle_addr, 0, sizeof(_tickle_addr));     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
   #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
   #pragma convert(37)
   #endif
     _tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");     _tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
   #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
   #pragma convert(0)
   #endif
   #endif
     _tickle_addr.sin_family = PF_INET;     _tickle_addr.sin_family = PF_INET;
     _tickle_addr.sin_port = 0;     _tickle_addr.sin_port = 0;
  
Line 540 
Line 647 
     tickler.socket(PF_INET, SOCK_STREAM, 0);     tickler.socket(PF_INET, SOCK_STREAM, 0);
     struct sockaddr_in _addr;     struct sockaddr_in _addr;
     memset(&_addr, 0, sizeof(_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");     _addr.sin_addr.s_addr = inet_addr("127.0.0.1");
   #endif
     _addr.sin_family = PF_INET;     _addr.sin_family = PF_INET;
     _addr.sin_port = 0;     _addr.sin_port = 0;
     tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));     tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));
Line 555 
Line 666 
     PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);     PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);
  
     pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);     pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);
   
     monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);     monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);
     _tickle->set_state(BUSY);     _tickle->set_state(BUSY);
  
Line 581 
Line 693 
 { {
   monitor_2_entry* temp;   monitor_2_entry* temp;
   while(_die.value() == 0) {   while(_die.value() == 0) {
     struct timeval tv = {0, 0};  
        struct timeval tv_idle = { 60, 0 };
  
     // place all sockets in the select set     // place all sockets in the select set
     FD_ZERO(&rd_fd_set);     FD_ZERO(&rd_fd_set);
Line 593 
Line 706 
           monitor_2_entry* closed = temp;           monitor_2_entry* closed = temp;
           temp = _listeners.next(closed);           temp = _listeners.next(closed);
           _listeners.remove_no_lock(closed);           _listeners.remove_no_lock(closed);
             HTTPConnection2 *cn = monitor_2::remove_connection((Sint32)(closed->get_sock()));
             delete cn;
           delete closed;           delete closed;
         }         }
         FD_SET((Sint32) temp->get_sock()  , &rd_fd_set);          if(temp == 0)
              break;
           Sint32 fd = (Sint32) temp->get_sock();
           if(fd >= 0 )
              FD_SET(fd , &rd_fd_set);
         temp = _listeners.next(temp);         temp = _listeners.next(temp);
       }       }
       _listeners.unlock();       _listeners.unlock();
Line 606 
Line 725 
     // important -  the dispatch routine has pointers to all the     // important -  the dispatch routine has pointers to all the
     // entries that are readable. These entries can be changed but     // entries that are readable. These entries can be changed but
     // the pointer must not be tampered with.     // the pointer must not be tampered with.
       if(_connections.count() )
     int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);     int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);
       else
          int events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, &tv_idle);
   
       if(_die.value())
       {
          break;
       }
   
     try {     try {
       _listeners.lock(pegasus_thread_self());       _listeners.lock(pegasus_thread_self());
       temp = _listeners.next(0);       temp = _listeners.next(0);
       while(temp != 0 ){       while(temp != 0 ){
         Sint32 fd = (Sint32) temp->get_sock();         Sint32 fd = (Sint32) temp->get_sock();
           if(fd >= 0 && FD_ISSET(fd, &rd_fd_set)) {
         if(FD_ISSET(fd, &rd_fd_set)) {  
           temp->set_state(BUSY);           temp->set_state(BUSY);
           FD_CLR(fd,  &rd_fd_set);           FD_CLR(fd,  &rd_fd_set);
           monitor_2_entry* ready = new monitor_2_entry(*temp);           monitor_2_entry* ready = new monitor_2_entry(*temp);
           _ready.insert_first((void*)ready);            try
             {
                _ready.insert_first(ready);
             }
             catch(...)
             {
             }
   
           _requestCount++;           _requestCount++;
         }         }
         temp = _listeners.next(temp);         temp = _listeners.next(temp);
Line 629 
Line 762 
       return;       return;
     }     }
     // now handle the sockets that are ready to read     // now handle the sockets that are ready to read
       if(_ready.count())
     _dispatch();     _dispatch();
       else
       {
          if(_connections.count() == 0 )
             _idle_dispatch(_idle_parm);
       }
   } // while alive   } // while alive
   
 } }
  
 void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*)) void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))
Line 645 
Line 785 
   void* old = (void*)_accept_dispatch;   void* old = (void*)_accept_dispatch;
   _accept_dispatch = dp;   _accept_dispatch = dp;
   return old;   return old;
   }
   
   void* monitor_2::set_idle_dispatch(void (*dp)(void*))
   {
      void* old = (void*)_idle_dispatch;
      _idle_dispatch = dp;
      return old;
   }
  
   void* monitor_2::set_idle_parm(void* parm)
   {
      void* old = _idle_parm;
      _idle_parm = parm;
      return old;
 } }
  
  
Line 654 
Line 807 
 // the pointer must not be tampered with. // the pointer must not be tampered with.
 void monitor_2::_dispatch(void) void monitor_2::_dispatch(void)
 { {
   monitor_2_entry* entry = (monitor_2_entry*) _ready.remove_first();     monitor_2_entry* entry;
   
      try
      {
   
            entry = _ready.remove_first();
      }
      catch(...)
      {
      }
   
   while(entry != 0 ){   while(entry != 0 ){
     switch(entry->get_type()) {     switch(entry->get_type()) {
     case INTERNAL:     case INTERNAL:
       static char buffer[2];       static char buffer[2];
         entry->get_sock().disableBlocking();
       entry->get_sock().read(&buffer, 2);       entry->get_sock().read(&buffer, 2);
         entry->get_sock().enableBlocking();
       break;       break;
     case LISTEN:     case LISTEN:
       {       {
         static struct sockaddr peer;         static struct sockaddr peer;
         static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);         static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);
           entry->get_sock().disableBlocking();
         pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);         pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);
           entry->get_sock().enableBlocking();
         monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());         monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());
         if(temp && _accept_dispatch != 0)         if(temp && _accept_dispatch != 0)
           _accept_dispatch(temp);           _accept_dispatch(temp);
Line 686 
Line 853 
     }     }
     _requestCount--;     _requestCount--;
     delete entry;     delete entry;
     entry = (monitor_2_entry*) _ready.remove_first();  
       if(_ready.count() == 0 )
          break;
   
       try
       {
          entry = _ready.remove_first();
       }
       catch(...)
       {
       }
   
   }   }
 } }
  
Line 694 
Line 872 
 { {
   _die = 1;   _die = 1;
   tickle();   tickle();
   
   // shut down the listener list, free the list nodes   // shut down the listener list, free the list nodes
   _tickler.get_sock().close();   _tickler.get_sock().close();
   _listeners.shutdown_queue();   _listeners.shutdown_queue();
Line 707 
Line 884 
       '0','0'       '0','0'
     };     };
  
     _tickler.get_sock().disableBlocking();
   
   _tickler.get_sock().write(&_buffer, 2);   _tickler.get_sock().write(&_buffer, 2);
     _tickler.get_sock().enableBlocking();
   
 } }
  
  
Line 758 
Line 939 
 } }
  
  
   HTTPConnection2* monitor_2::remove_connection(Sint32 sock)
   {
   
      HTTPConnection2* temp;
      try
      {
         monitor_2::_connections.lock(pegasus_thread_self());
         temp = monitor_2::_connections.next(0);
         while(temp != 0 )
         {
            if(sock == temp->getSocket())
            {
               temp = monitor_2::_connections.remove_no_lock(temp);
               monitor_2::_connections.unlock();
               return temp;
            }
            temp = monitor_2::_connections.next(temp);
         }
         monitor_2::_connections.unlock();
      }
      catch(...)
      {
      }
      return 0;
   }
   
   Boolean monitor_2::insert_connection(HTTPConnection2* connection)
   {
      try
      {
         monitor_2::_connections.insert_first(connection);
      }
      catch(...)
      {
         return false;
      }
      return true;
   }
   
   
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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