(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.24 and 1.25

version 1.24, 2002/06/11 21:28:08 version 1.25, 2002/06/19 22:06:33
Line 39 
Line 39 
 # 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 62 
Line 62 
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
  
 static struct timeval create_time = {0, 10};  static AtomicInt _connections = 0;
   
   
   static struct timeval create_time = {0, 1};
 static struct timeval destroy_time = {5, 0}; static struct timeval destroy_time = {5, 0};
 static struct timeval deadlock_time = {1000, 0}; static struct timeval deadlock_time = {1000, 0};
  
Line 92 
Line 95 
    : _module_handle(0), _controller(0), _async(false)    : _module_handle(0), _controller(0), _async(false)
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = new MonitorRep;      _rep = 0;
     FD_ZERO(&_rep->rd_fd_set);      _entries.reserve(128);
     FD_ZERO(&_rep->wr_fd_set);  
     FD_ZERO(&_rep->ex_fd_set);  
     FD_ZERO(&_rep->active_rd_fd_set);  
     FD_ZERO(&_rep->active_wr_fd_set);  
     FD_ZERO(&_rep->active_ex_fd_set);  
 } }
  
 Monitor::Monitor(Boolean async) Monitor::Monitor(Boolean async)
    : _module_handle(0), _controller(0), _async(async)    : _module_handle(0), _controller(0), _async(async)
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = new MonitorRep;      _rep = 0;
     FD_ZERO(&_rep->rd_fd_set);      _entries.reserve(128);
     FD_ZERO(&_rep->wr_fd_set);  
     FD_ZERO(&_rep->ex_fd_set);  
     FD_ZERO(&_rep->active_rd_fd_set);  
     FD_ZERO(&_rep->active_wr_fd_set);  
     FD_ZERO(&_rep->active_ex_fd_set);  
     if( _async == true )     if( _async == true )
     {     {
        _thread_pool = new ThreadPool(0,        _thread_pool = new ThreadPool(0,
                                      "Monitor",                                      "Monitor",
                                      0,                                      0,
                                      20,                                       0,
                                      create_time,                                      create_time,
                                      destroy_time,                                      destroy_time,
                                      deadlock_time);                                      deadlock_time);
Line 139 
Line 133 
     }     }
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");
  
     delete _rep;  
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
     Socket::uninitializeInterface();     Socket::uninitializeInterface();
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
Line 172 
Line 165 
 } }
  
  
 //<<< Tue May 14 20:38:26 2002 mdd >>>  
 //  register with module controller  
 //  when it is time to enqueue the message,  
 // use an async_thread_exec call to  
 // isolate the entire if(events) { enqueue -> fd_clear } block  
 //  let the thread pool grow and shrink according to load.  
   
 Boolean Monitor::run(Uint32 milliseconds) Boolean Monitor::run(Uint32 milliseconds)
 { {
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS      Boolean handled_events = false;
       int i = 0;
     // 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  
  
     // Check for events on the selected file descriptors. Only do this if      struct timeval tv = {0,1};
     // there were no undispatched events from last time.      fd_set fdread;
       FD_ZERO(&fdread);
  
     int count = 0;      _entries_mut.lock(pegasus_thread_self());
  
     memcpy(&_rep->active_rd_fd_set, &_rep->rd_fd_set, sizeof(fd_set));      for( int indx = 0; indx < (int)_entries.size(); indx++)
 //    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));  
   
     const Uint32 SECONDS = milliseconds / 1000;  
     const Uint32 MICROSECONDS = (milliseconds % 1000) * 1000;  
     struct timeval tv = { SECONDS, MICROSECONDS };  
   
     count = select(  
        FD_SETSIZE,  
        &_rep->active_rd_fd_set,  
 //       &_rep->active_wr_fd_set,  
        NULL,  
        &_rep->active_ex_fd_set,  
        &tv);  
     if(count == 0)  
     {     {
        return false;         if(_entries[indx]._status == _MonitorEntry::IDLE)
     }  
 #ifdef PEGASUS_OS_TYPE_WINDOWS  
     else if (count == SOCKET_ERROR)  
 #else  
     else if (count == -1)  
 #endif  
     {     {
        return false;            FD_SET(_entries[indx].socket, &fdread);
     }     }
   
   
     Boolean handled_events = false;  
     try { _connection_mutex.try_lock(pegasus_thread_self()); }  
     catch(AlreadyLocked){  
       pegasus_sleep(1);  
       return false;  
     }     }
       _entries_mut.unlock();
  
     for (Uint32 i = 0, n = _entries.size(); i < _entries.size(); i++)      int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);
     {  
         Sint32 socket = _entries[i].socket;  
         Uint32 events = 0;  
  
         if(_entries[i].dying.value() > 0 )  #ifdef PEGASUS_OS_TYPE_WINDOWS
       if(events && events != SOCKET_ERROR )
   #else
       if(events && events != -1 )
   #endif
         {         {
            if(_entries[i]._type == Monitor::CONNECTION)         for( int indx = 0; indx < (int)_entries.size(); indx++)
            {            {
             if(FD_ISSET(_entries[indx].socket, &fdread))
               MessageQueue *q = MessageQueue::lookup(_entries[i].queueId);  
               if(q && static_cast<HTTPConnection *>(q)->is_dying() &&  
                  (0 == static_cast<HTTPConnection *>(q)->refcount.value()))  
               {               {
                  static_cast<HTTPConnection *>(q)->lock_connection();               MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
                  static_cast<HTTPConnection *>(q)->unlock_connection();               if(q == 0)
                {
                  MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();                  unsolicitSocketMessages(indx);
                  Message* message= new CloseConnectionMessage(static_cast<HTTPConnection *>(q)->getSocket());  
                  message->dest = o.getQueueId();  
                  _connection_mutex.unlock();  
   
                  o.enqueue(message);  
                  return true;                  return true;
                  i--;  
                  n = _entries.size();  
               }  
            }            }
         }  
   
         if (FD_ISSET(socket, &_rep->active_rd_fd_set))  
             events |= SocketMessage::READ;  
  
         if (FD_ISSET(socket, &_rep->active_ex_fd_set))               if(_entries[indx]._type == Monitor::CONNECTION)
             events |= SocketMessage::EXCEPTION;  
   
         if (events)  
         {         {
             Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                  static_cast<HTTPConnection *>(q)->_entry_index = indx;
                           "Monitor::run - Socket Event Detected events = %d", events);                  if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )
             if (events & SocketMessage::READ)  
             {             {
                FD_CLR(socket, &_rep->active_rd_fd_set);                     _entries[indx]._status = _MonitorEntry::DYING;
                Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                     MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();
                              "Monitor::run FD_CLR READ");                     Message* message= new CloseConnectionMessage(_entries[indx].socket);
             }                     message->dest = o.getQueueId();
             else if (events & SocketMessage::EXCEPTION)                     o.enqueue(message);
             {  
                FD_CLR(socket, &_rep->active_ex_fd_set);  
                Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                              "Monitor::run FD_CLR EXECEPTION");  
             }  
             MessageQueue* queue = MessageQueue::lookup(_entries[i].queueId);  
             if( ! queue )  
             {  
                Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                              "Monitor::run lookup for connection entry failed, unsoliciting");  
                _connection_mutex.unlock();  
                unsolicitSocketMessages(socket);  
                return true;                return true;
             }             }
  
             if(_async == true && _entries[i]._type == Monitor::CONNECTION)                  _entries[indx]._status = _MonitorEntry::BUSY;
             {                  _thread_pool->allocate_and_awaken((void *)q, _dispatch);
   
                if( static_cast<HTTPConnection *>(queue)->refcount.value() == 0 )  
                {  
                   Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                                 "Monitor::run dispatching thread to idle connection");  
                   static_cast<HTTPConnection *>(queue)->refcount++;  
                   if( false == static_cast<HTTPConnection *>(queue)->is_dying())  
                      _thread_pool->allocate_and_awaken((void *)queue, _dispatch);  
                   else  
                      static_cast<HTTPConnection *>(queue)->refcount--;  
                }  
                else  
                   pegasus_sleep(1);  
             }             }
             else             else
             {             {
               _connection_mutex.unlock();                  int events = 0;
  
                Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                  events |= SocketMessage::READ;
                              "Monitor::run enqueueing to non-connection HTTP class");                  Message *msg = new SocketMessage(_entries[indx].socket, events);
                Message* message = new SocketMessage(socket, events);                  _entries[indx]._status = _MonitorEntry::BUSY;
                queue->enqueue(message);                  q->enqueue(msg);
                   _entries[indx]._status = _MonitorEntry::IDLE;
                return true;                return true;
   
             }  
             count--;  
             pegasus_yield();  
         }         }
         handled_events = true;         handled_events = true;
     }     }
     _connection_mutex.unlock();         }
       }
     return(handled_events);     return(handled_events);
 } }
  
 Boolean Monitor::solicitSocketMessages(  
   int  Monitor::solicitSocketMessages(
     Sint32 socket,     Sint32 socket,
     Uint32 events,     Uint32 events,
     Uint32 queueId,     Uint32 queueId,
     int type)     int type)
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");  
   
     // See whether a handler is already registered for this one:  
     Uint32 pos = _findEntry(socket);  
   
     if (pos != PEGASUS_NOT_FOUND)  
     {  
         PEG_METHOD_EXIT();  
         return false;  
     }  
  
     // Set the events:     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");
   
     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, type);     _MonitorEntry entry(socket, queueId, type);
     _entries.append(entry);     entry._status = _MonitorEntry::IDLE;
      _entries_mut.lock(pegasus_thread_self());
  
     // Success!     Boolean found = false;
  
     PEG_METHOD_EXIT();     int index ;
     return true;     for(index = 0; index < (int)_entries.size(); index++)
 }  
   
 Boolean Monitor::unsolicitSocketMessages(Sint32 socket)  
 {  
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessage");  
   
     // Look for the given entry and remove it:  
   
     _connection_mutex.lock(pegasus_thread_self());  
   
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)  
     {     {
         if (_entries[i].socket == socket)        if(_entries[index]._status == _MonitorEntry::EMPTY)
         {         {
             Sint32 socket = _entries[i].socket;           _entries[index] = entry;
             FD_CLR(socket, &_rep->rd_fd_set);           found = true;
             FD_CLR(socket, &_rep->wr_fd_set);           break;
             FD_CLR(socket, &_rep->ex_fd_set);  
             _entries.remove(i);  
             // ATTN-RK-P3-20020521: Need "Socket::close(socket);" here?  
             Socket::close(socket);  
             PEG_METHOD_EXIT();  
             _connection_mutex.unlock();  
             return true;  
         }         }
     }     }
      if(found == false)
      {
         _entries.append(entry);
         index = _entries.size() - 1;
      }
      _entries_mut.unlock();
      _connections++;
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
     _connection_mutex.unlock();     return index;
   
     return false;  
 } }
  
 Uint32 Monitor::_findEntry(Sint32 socket)  void Monitor::unsolicitSocketMessages(Sint32 socket)
 { {
   _connection_mutex.lock(pegasus_thread_self());      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
   
       _entries_mut.lock(pegasus_thread_self());
  
    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)
           {           {
             _connection_mutex.unlock();            _entries[index]._status = _MonitorEntry::EMPTY;
             return i;  
           }           }
     }     }
    _connection_mutex.unlock();      _entries_mut.unlock();
     return PEG_NOT_FOUND;  
       PEG_METHOD_EXIT();
 } }
  
  
   
 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);
    if( true == dst->is_dying())  
    {  
       dst->refcount--;  
       return 0;  
    }  
    if( false == dst->is_dying())  
    {  
       if(false == dst->run(1))  
          pegasus_sleep(1);  
  
    }     dst->run(1);
    dst->refcount--;     if( dst->_monitor->_entries.size() > dst->_entry_index )
         dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
   
    return 0;    return 0;
 } }
  

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