(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.7 and 1.32.2.2

version 1.7, 2002/05/15 17:15:34 version 1.32.2.2, 2002/10/29 21:13:17
Line 1 
Line 1 
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM  // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,
   // The Open Group, Tivoli Systems
 // //
 // Permission is hereby granted, free of charge, to any person obtaining a copy // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to // of this software and associated documentation files (the "Software"), to
Line 26 
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
 #  error "FD_SETSIZE was not set to 1024 prior to the last inclusion \ #  error "FD_SETSIZE was not set to 1024 prior to the last inclusion \
 of <winsock.h>. It may have been indirectly included (e.g., by including \ of <winsock.h>. It may have been indirectly included (e.g., by including \
 <windows.h>). Find the inclusion of that header which is visible to this \  <windows.h>). Finthe inclusion of that header which is visible to this \
 compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \ compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \
 otherwise, less than 64 clients (the default) will be able to connect to the \ otherwise, less than 64 clients (the default) will be able to connect to the \
 CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM." CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."
Line 60 
Line 55 
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
   
   static AtomicInt _connections = 0;
   
   
   static struct timeval create_time = {0, 1};
   static struct timeval destroy_time = {15, 0};
   static struct timeval deadlock_time = {0, 0};
   
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
 // MonitorRep // MonitorRep
Line 83 
Line 86 
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 Monitor::Monitor() Monitor::Monitor()
    : _module_handle(0), _controller(0), _async(false)     : _async(false)
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = new MonitorRep;      _rep = 0;
     FD_ZERO(&_rep->rd_fd_set);      _entries.reserveCapacity(32);
     FD_ZERO(&_rep->wr_fd_set);      int i = 0;
     FD_ZERO(&_rep->ex_fd_set);      for( ; i < 32; i++ )
     FD_ZERO(&_rep->active_rd_fd_set);      {
     FD_ZERO(&_rep->active_wr_fd_set);         _MonitorEntry entry(0, 0, 0);
     FD_ZERO(&_rep->active_ex_fd_set);         _entries.append(entry);
       }
 } }
  
 Monitor::~Monitor()  Monitor::Monitor(Boolean async)
      : _async(async)
 { {
     delete _rep;      Socket::initializeInterface();
     Socket::uninitializeInterface();      _rep = 0;
   //     _entries.reserveCapacity(32);
   //     int i = 0;
   //     for( ; i < 32; i++ )
   //     {
   //        _MonitorEntry entry(0, 0, 0);
   //        _entries.append(entry);
   //     }
   
       if( _async == true )
   
       {
          _thread_pool = new ThreadPool(0,
                                        "Monitor",
                                        1,
                                        0,
                                        create_time,
                                        destroy_time,
                                        deadlock_time);
       }
       else
          _thread_pool = 0;
 } }
  
   Monitor::~Monitor()
   {
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                     "deregistering with module controller");
   
  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");
  
 //<<< Tue May 14 20:38:26 2002 mdd >>>      Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
 //  register with module controller      Socket::uninitializeInterface();
 //  when it is time to enqueue the message,      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
 // use an async_thread_exec call to                    "returning from monitor destructor");
 // isolate the entire if(events) { enqueue -> fd_clear } block      if(_async == true)
 //  let the thread pool grow and shrink according to load.         delete _thread_pool;
   }
  
 Boolean Monitor::run(Uint32 milliseconds)  
 {  
  
   int Monitor::kill_idle_threads()
   {
      static struct timeval now, last;
      gettimeofday(&now, NULL);
      int dead_threads = 0;
  
    // register the monitor as a module to gain access to the cimserver's thread pool     if( now.tv_sec - last.tv_sec > 300 )
    // <<< Wed May 15 09:52:16 2002 mdd >>>  
    while(_module_handle == NULL)  
    {    {
         PEGASUS_STD(cout) << "Monitor Thread Pool currently has " <<
            _thread_pool->running_count() +
            _thread_pool->pool_count() << " Threads." << PEGASUS_STD(endl);
         gettimeofday(&last, NULL);
       try       try
       {       {
            dead_threads =  _thread_pool->kill_dead_threads();
          _controller = &(ModuleController::register_module(PEGASUS_QUEUENAME_CONTROLSERVICE,  
                                                            PEGASUS_MODULENAME_MONITOR,  
                                                            (void *)this,  
                                                            0,  
                                                            0,  
                                                            0,  
                                                            &_module_handle));  
       }       }
       catch(AlreadyExists & )        catch(IPCException& )
       {       {
          break;  
       }  
    }    }
  
      }
      return dead_threads;
   }
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS  
   
     // Windows select() has a strange little bug. It returns immediately if  
     // there are no descriptors in the set even if the timeout is non-zero.  
     // To work around this, we call Sleep() for now:  
   
     if (_entries.size() == 0)  
         Sleep(milliseconds);  
  
 #endif  Boolean Monitor::run(Uint32 milliseconds)
   {
  
     // Check for events on the selected file descriptors. Only do this if      Boolean handled_events = false;
     // there were no undispatched events from last time.      int i = 0;
  
     static int count = 0;      struct timeval tv = {0,1};
       fd_set fdread;
       FD_ZERO(&fdread);
  
     if (count == 0)      for( int indx = 0; indx < (int)_entries.size(); indx++)
     {     {
         memcpy(&_rep->active_rd_fd_set, &_rep->rd_fd_set, sizeof(fd_set));         if(_entries[indx]._status.value() == _MonitorEntry::IDLE)
         memcpy(&_rep->active_wr_fd_set, &_rep->wr_fd_set, sizeof(fd_set));         {
         memcpy(&_rep->active_ex_fd_set, &_rep->ex_fd_set, sizeof(fd_set));            FD_SET(_entries[indx].socket, &fdread);
          }
         const Uint32 SECONDS = milliseconds / 1000;      }
         const Uint32 MICROSECONDS = (milliseconds % 1000) * 1000;  
         struct timeval tv = { SECONDS, MICROSECONDS };      int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);
   
         count = select(  
             FD_SETSIZE,  
             &_rep->active_rd_fd_set,  
             &_rep->active_wr_fd_set,  
             &_rep->active_ex_fd_set,  
             &tv);  
  
         if (count == 0)  
             return false;  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
         else if (count == SOCKET_ERROR)      if(events && events != SOCKET_ERROR )
 #else #else
         else if (count == -1)      if(events && events != -1 )
 #endif #endif
         {         {
             count = 0;         for( int indx = 0; indx < (int)_entries.size(); indx++)
             return false;  
         }  
     }  
   
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)  
     {     {
         Sint32 socket = _entries[i].socket;            if(FD_ISSET(_entries[indx].socket, &fdread))
         Uint32 events = 0;  
   
         if (FD_ISSET(socket, &_rep->active_rd_fd_set))  
             events |= SocketMessage::READ;  
   
         if (FD_ISSET(socket, &_rep->active_wr_fd_set))  
             events |= SocketMessage::WRITE;  
   
         if (FD_ISSET(socket, &_rep->active_ex_fd_set))  
             events |= SocketMessage::EXCEPTION;  
   
         if (events)  
         {         {
             Tracer::trace(TRC_HTTP, Tracer::LEVEL4,               MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
                "Monitor::run - Socket Event Detected events = %d", events);               if(q == 0)
   
   
             MessageQueue* queue = MessageQueue::lookup(_entries[i].queueId);  
   
             if (!_async)  
             {             {
                if( ! queue )                  PEGASUS_STD(cout) << "Monitor:: found an empty connection slot" << PEGASUS_STD(endl);
                   unsolicitSocketMessages(_entries[i].queueId);                  try
                Message* message = new SocketMessage(socket, events);  
                queue->enqueue(message);  
   
                 if (events & SocketMessage::WRITE)  
                 {                 {
                    FD_CLR(socket, &_rep->active_wr_fd_set);                        _entries[indx]._status = _MonitorEntry::EMPTY;
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                                  "Monitor::run FD_CLR WRITE");  
                 }                 }
                   catch(...)
                 if (events & SocketMessage::EXCEPTION)  
                 {                 {
                    FD_CLR(socket, &_rep->active_ex_fd_set);  
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                               "Monitor::run FD_CLR EXECEPTION");  
                 }                 }
                   return true;
                 if (events & SocketMessage::READ)               }
                try
                 {                 {
                    FD_CLR(socket, &_rep->active_rd_fd_set);                  if(_entries[indx]._type == Monitor::CONNECTION)
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                  {
                                  "Monitor::run FD_CLR READ");                     static_cast<HTTPConnection *>(q)->_entry_index = indx;
                      if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )
                      {
                         _entries[indx]._status = _MonitorEntry::DYING;
                         MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();
                         Message* message= new CloseConnectionMessage(_entries[indx].socket);
                         message->dest = o.getQueueId();
                         o.enqueue(message);
                         return true;
                 }                 }
                      _entries[indx]._status = _MonitorEntry::BUSY;
                      _thread_pool->allocate_and_awaken((void *)q, _dispatch);
             }             }
             else             else
             {             {
                      int events = 0;
                monitor_dispatch *parms =                     events |= SocketMessage::READ;
                   new monitor_dispatch(this, queue, i, socket, events);                     Message *msg = new SocketMessage(_entries[indx].socket, events);
                _controller->async_thread_exec(*_module_handle, _dispatch, (void *)parms);                     _entries[indx]._status = _MonitorEntry::BUSY;
             }                     q->enqueue(msg);
             count--;                     _entries[indx]._status = _MonitorEntry::IDLE;
             return true;             return true;
         }         }
     }     }
                catch(...)
     return false;               {
                }
                handled_events = true;
             }
          }
       }
       return(handled_events);
 } }
  
 Boolean Monitor::solicitSocketMessages(  
   int  Monitor::solicitSocketMessages(
     Sint32 socket,     Sint32 socket,
     Uint32 events,     Uint32 events,
     Uint32 queueId)      Uint32 queueId,
       int type)
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");  
  
     // See whether a handler is already registered for this one:     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");
  
     Uint32 pos = _findEntry(socket);     int index = -1;
      for(index = 0; index < (int)_entries.size(); index++)
      {
         try
         {
            if(_entries[index]._status.value() == _MonitorEntry::EMPTY)
            {
  
     if (pos != PEGASUS_NOT_FOUND)              _entries[index].socket = socket;
               _entries[index].queueId  = queueId;
               _entries[index]._type = type;
               _entries[index]._status = _MonitorEntry::IDLE;
               return index;
            }
         }
         catch(...)
     {     {
         PEG_METHOD_EXIT();  
         return false;  
     }     }
      }
     // Set the events:  
   
     if (events & SocketMessage::READ)  
         FD_SET(socket, &_rep->rd_fd_set);  
   
     if (events & SocketMessage::WRITE)  
         FD_SET(socket, &_rep->wr_fd_set);  
   
     if (events & SocketMessage::EXCEPTION)  
         FD_SET(socket, &_rep->ex_fd_set);  
   
     // Add the entry to the list:  
   
 //    _MonitorEntry entry = { socket, queueId };  
     _MonitorEntry entry(socket, queueId);  
   
     _entries.append(entry);  
   
     // Success!  
     ModuleController* controlService =  
         new ModuleController(PEGASUS_QUEUENAME_CONTROLSERVICE);  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
     return true;     return index;
 } }
  
 Boolean Monitor::unsolicitSocketMessages(Sint32 socket)  void Monitor::unsolicitSocketMessages(Sint32 socket)
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessage");      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
   
     // Look for the given entry and remove it:  
  
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)      for(int index = 0; index < (int)_entries.size(); index++)
     {     {
         if (_entries[i].socket == socket)         if(_entries[index].socket == socket)
         {         {
             Sint32 socket = _entries[i].socket;            _entries[index]._status = _MonitorEntry::EMPTY;
             FD_CLR(socket, &_rep->rd_fd_set);  
             FD_CLR(socket, &_rep->wr_fd_set);  
             FD_CLR(socket, &_rep->ex_fd_set);  
             _entries.remove(i);  
             PEG_METHOD_EXIT();  
             return true;  
         }         }
     }     }
  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
     return false;  if( _async  == true )
 }     PEGASUS_STD(cout) << "Monitor:: running " << _thread_pool->running_count() <<
      " idle " << _thread_pool->pool_count() << PEGASUS_STD(endl);
 Uint32 Monitor::_findEntry(Sint32 socket) const  
 {  
    for (Uint32 i = 0, n = _entries.size(); i < n; i++)  
     {  
         if (_entries[i].socket == socket)  
             return i;  
     }  
  
     return PEG_NOT_FOUND;  
 } }
  
  
 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);
  
    Monitor::monitor_dispatch *parms = reinterpret_cast<Monitor::monitor_dispatch *>(parm);     dst->run(1);
      if(  dst->_monitor->_entries.size() > (Uint32)dst->_entry_index )
    if (! parms->_decoder)        dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
    {  
  
       parms->_myself->unsolicitSocketMessages(parms->_myself->_entries[parms->_entry].queueId);     return 0;
       return(0);  
    }    }
  
    Message* message = new SocketMessage(parms->_socket, parms->_events);  
    parms->_decoder->enqueue(message);  
   
    if (parms->_events & SocketMessage::WRITE)  
    {  
       FD_CLR(parms->_socket, &(parms->_myself->_rep->active_wr_fd_set));  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                     "Monitor::run FD_CLR WRITE");  
    }  
   
    if (parms->_events & SocketMessage::EXCEPTION)  
    {  
       FD_CLR(parms->_socket, &(parms->_myself->_rep->active_ex_fd_set));  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                     "Monitor::run FD_CLR EXECEPTION");  
    }  
   
    if (parms->_events & SocketMessage::READ)  
    {  
       FD_CLR(parms->_socket, &(parms->_myself->_rep->active_rd_fd_set));  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                     "Monitor::run FD_CLR READ");  
    }  
   
    delete parms;  
    return(0);  
 }  
   
   
 void Monitor::set_async(Boolean async)  
 {  
    _async = async;  
 }  
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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