(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.30 and 1.31.2.1

version 1.30, 2002/08/20 21:47:09 version 1.31.2.1, 2002/10/25 20:49:43
Line 96 
Line 96 
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
     _entries.reserveCapacity(128);      _entries.reserveCapacity(32);
       int i = 0;
       for( ; i < 32; i++ )
       {
          _MonitorEntry entry(0, 0, 0);
          _entries.append(entry);
       }
 } }
  
 Monitor::Monitor(Boolean async) Monitor::Monitor(Boolean async)
Line 105 
Line 110 
 { {
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;     _rep = 0;
     _entries.reserveCapacity(128);      _entries.reserveCapacity(32);
       int i = 0;
       for( ; i < 32; i++ )
       {
          _MonitorEntry entry(0, 0, 0);
          _entries.append(entry);
       }
   
     if( _async == true )     if( _async == true )
     {     {
        _thread_pool = new ThreadPool(0,        _thread_pool = new ThreadPool(0,
                                      "Monitor",                                      "Monitor",
                                      0,                                       1,
                                      0,                                      0,
                                      create_time,                                      create_time,
                                      destroy_time,                                      destroy_time,
Line 148 
Line 160 
    gettimeofday(&now, NULL);    gettimeofday(&now, NULL);
    int dead_threads = 0;    int dead_threads = 0;
  
    if( now.tv_sec - last.tv_sec > 0 )     if( now.tv_sec - last.tv_sec > 300 )
    {    {
         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
       {       {
   
          dead_threads =  _thread_pool->kill_dead_threads();          dead_threads =  _thread_pool->kill_dead_threads();
       }       }
       catch(IPCException& )       catch(IPCException& )
Line 175 
Line 189 
     fd_set fdread;     fd_set fdread;
     FD_ZERO(&fdread);     FD_ZERO(&fdread);
  
     _entries_mut.lock(pegasus_thread_self());  
   
     for( int indx = 0; indx < (int)_entries.size(); indx++)     for( int indx = 0; indx < (int)_entries.size(); indx++)
     {     {
        if(_entries[indx]._status == _MonitorEntry::IDLE)         if(_entries[indx]._status.value() == _MonitorEntry::IDLE)
        {        {
           FD_SET(_entries[indx].socket, &fdread);           FD_SET(_entries[indx].socket, &fdread);
        }        }
Line 200 
Line 212 
              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
                   {
                 _entries[indx]._status = _MonitorEntry::EMPTY;                 _entries[indx]._status = _MonitorEntry::EMPTY;
                 _entries_mut.unlock();                  }
                   catch(...)
                   {
                   }
                 return true;                 return true;
              }              }
                try
                {
              if(_entries[indx]._type == Monitor::CONNECTION)              if(_entries[indx]._type == Monitor::CONNECTION)
              {              {
                 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 )
                 {                 {
                    _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);
                    message->dest = o.getQueueId();                    message->dest = o.getQueueId();
                    _entries_mut.unlock();  
                    o.enqueue(message);                    o.enqueue(message);
                    return true;                    return true;
                 }                 }
                 _entries[indx]._status = _MonitorEntry::BUSY;                 _entries[indx]._status = _MonitorEntry::BUSY;
                 _entries_mut.unlock();  
                 _thread_pool->allocate_and_awaken((void *)q, _dispatch);                 _thread_pool->allocate_and_awaken((void *)q, _dispatch);
                 _entries_mut.lock(pegasus_thread_self());  
              }              }
              else              else
              {              {
                 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);
                 _entries_mut.unlock();                     _entries[indx]._status = _MonitorEntry::BUSY;
                 q->enqueue(msg);                 q->enqueue(msg);
                      _entries[indx]._status = _MonitorEntry::IDLE;
                 return true;                 return true;
              }              }
                }
                catch(...)
                {
                }
              handled_events = true;              handled_events = true;
           }           }
        }        }
     }     }
     _entries_mut.unlock();  
     return(handled_events);     return(handled_events);
 } }
  
Line 251 
Line 270 
  
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solictSocketMessage");
  
    _MonitorEntry entry(socket, queueId, type);     int index = -1;
    entry._status = _MonitorEntry::IDLE;  
    _entries_mut.lock(pegasus_thread_self());  
   
    Boolean found = false;  
   
    int index ;  
    for(index = 0; index < (int)_entries.size(); index++)    for(index = 0; index < (int)_entries.size(); index++)
    {    {
       if(_entries[index]._status == _MonitorEntry::EMPTY)        try
         {
            if(_entries[index]._status.value() == _MonitorEntry::EMPTY)
       {       {
          _entries[index].operator =(entry);  
          found = true;              _entries[index].socket = socket;
          break;              _entries[index].queueId  = queueId;
               _entries[index]._type = type;
               _entries[index]._status = _MonitorEntry::IDLE;
               return index;
       }       }
    }    }
    if(found == false)        catch(...)
    {    {
       _entries.append(entry);  
       index = _entries.size() - 1;  
    }    }
    _connections++;     }
    _entries_mut.unlock();  
   
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
    return index;    return index;
 } }
Line 283 
Line 297 
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
  
     _entries_mut.lock(pegasus_thread_self());  
   
     Boolean found = false;  
   
     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)
        {        {
           found = true;  
           _connections--;  
           _entries[index]._status = _MonitorEntry::EMPTY;           _entries[index]._status = _MonitorEntry::EMPTY;
        }        }
     }     }
     _entries_mut.unlock();  
     PEGASUS_ASSERT(found == true);  
  
     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);
   
    dst->run(1);    dst->run(1);
      if(  dst->_monitor->_entries.size() > (Uint32)dst->_entry_index )
         dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
   
    return 0;    return 0;
 } }
  

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