(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.62 and 1.70

version 1.62, 2003/10/16 22:46:41 version 1.70, 2004/01/30 15:41:28
Line 1 
Line 1 
 //%/////////////////////////////////////////////////////////////////////////////  //%2003////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,  // Copyright (c) 2000, 2001, 2002  BMC Software, Hewlett-Packard Development
 // The Open Group, Tivoli Systems  // Company, L. P., IBM Corp., The Open Group, Tivoli Systems.
   // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L. P.;
   // IBM Corp.; EMC Corporation, The Open Group.
 // //
 // 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 35 
Line 37 
 #include "Socket.h" #include "Socket.h"
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
 #include <Pegasus/Common/HTTPConnection.h> #include <Pegasus/Common/HTTPConnection.h>
   #include <Pegasus/Common/MessageQueueService.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 118 
Line 121 
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
     }     }
     if( _async == true )  
     {  
        _thread_pool = new ThreadPool(0,  
                                      "Monitor",  
                                      0,  
                                      0,  
                                      create_time,  
                                      destroy_time,  
                                      deadlock_time);  
     }  
     else  
        _thread_pool = 0;  
 } }
  
 Monitor::~Monitor() Monitor::~Monitor()
Line 149 
Line 140 
     Socket::uninitializeInterface();     Socket::uninitializeInterface();
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                   "returning from monitor destructor");                   "returning from monitor destructor");
     if(_async == true)  
        delete _thread_pool;  
 }  
   
   
 int Monitor::kill_idle_threads()  
 {  
    static struct timeval now, last;  
    gettimeofday(&now, NULL);  
    int dead_threads = 0;  
   
    if( now.tv_sec - last.tv_sec > 120 )  
    {  
       gettimeofday(&last, NULL);  
       try  
       {  
          dead_threads =  _thread_pool->kill_dead_threads();  
       }  
       catch(IPCException& )  
       {  
       }  
   
    }  
    return dead_threads;  
 } }
  
   
 Boolean Monitor::run(Uint32 milliseconds) Boolean Monitor::run(Uint32 milliseconds)
 { {
  
Line 216 
Line 182 
         _stopConnections = 0;         _stopConnections = 0;
     }     }
  
       for( int indx = 0; indx < (int)_entries.size(); indx++)
       {
          if ((_entries[indx]._status.value() == _MonitorEntry::DYING) &&
                   (_entries[indx]._type == Monitor::CONNECTION))
          {
             MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
             PEGASUS_ASSERT(q != 0);
             MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();
             Message* message= new CloseConnectionMessage(_entries[indx].socket);
             message->dest = o.getQueueId();
   
             // HTTPAcceptor is responsible for closing the connection.
             // The lock is released to allow HTTPAcceptor to call
             // unsolicitSocketMessages to free the entry.
             // Once HTTPAcceptor completes processing of the close
             // connection, the lock is re-requested and processing of
             // the for loop continues.  This is safe with the current
             // implementation of the _entries object.  Note that the
             // loop condition accesses the _entries.size() on each
             // iteration, so that a change in size while the mutex is
             // unlocked will not result in an ArrayIndexOutOfBounds
             // exception.
   
             _entry_mut.unlock();
             o.enqueue(message);
             _entry_mut.lock(pegasus_thread_self());
          }
       }
   
     Uint32 _idleEntries = 0;     Uint32 _idleEntries = 0;
  
     for( int indx = 0; indx < (int)_entries.size(); indx++)     for( int indx = 0; indx < (int)_entries.size(); indx++)
Line 228 
Line 223 
     }     }
  
     // Fixed in monitor_2 but added because Monitor is still the default monitor.     // Fixed in monitor_2 but added because Monitor is still the default monitor.
     // When fdread.fd_count is 0 don't imediatly return, otherwize this loops out of control      // When _idleEntries is 0 don't immediately return, otherwise this loops out of control
     // kicking off kill idle thread threads.  E.g. There is nothing to select on when the cimserver     // kicking off kill idle thread threads.  E.g. There is nothing to select on when the cimserver
     // is shutting down.     // is shutting down.
     if( fdread.fd_count == 0 )      if( _idleEntries == 0 )
     {     {
         Thread::sleep( milliseconds );         Thread::sleep( milliseconds );
         _entry_mut.unlock();         _entry_mut.unlock();
Line 282 
Line 277 
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                      "_entries[indx].type for indx = %d is Monitor::CONNECTION", indx);                      "_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 )                     _entries[indx]._status = _MonitorEntry::BUSY;
                      // If allocate_and_awaken failure, retry on next iteration
                      if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(
                              (void *)q, _dispatch))
                    {                    {
                       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                        Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,
                           "Monitor::run processing dying value > 0 for indx = %d, connection being closed.",                            "Monitor::run: Insufficient resources to process request.");
                           indx);                        _entries[indx]._status = _MonitorEntry::IDLE;
                       _entries[indx]._status = _MonitorEntry::DYING;  
                       MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();  
                       Message* message= new CloseConnectionMessage(_entries[indx].socket);  
                       message->dest = o.getQueueId();  
                       _entry_mut.unlock();                       _entry_mut.unlock();
                       o.enqueue(message);  
                       return true;                       return true;
                    }                    }
                    _entries[indx]._status = _MonitorEntry::BUSY;  
                    _thread_pool->allocate_and_awaken((void *)q, _dispatch);  
                 }                 }
                 else                 else
                 {                 {
Line 407 
Line 398 
    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
           "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);           "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);    PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);
    dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;  
      // Once the HTTPConnection thread has set the status value to either
      // Monitor::DYING or Monitor::IDLE, it has returned control of the connection
      // to the Monitor.  It is no longer permissible to access the connection
      // or the entry in the _entries table.
    if (dst->_connectionClosePending)    if (dst->_connectionClosePending)
    {    {
       dst->_dying = 1;        dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;
      }
      else
      {
         dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
    }    }
    dst->_monitor->_entry_mut.unlock();  
    return 0;    return 0;
 } }
  
Line 797 
Line 789 
  
 } }
  
   int  monitor_2::solicitSocketMessages(
       Sint32 socket,
       Uint32 events,
       Uint32 queueId,
       int type)
   {
   
      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
   
      _entry_mut.lock(pegasus_thread_self());
   
      for(int index = 0; index < (int)_entries.size(); index++)
      {
         try
         {
            if(_entries[index]._status.value() == monitor_2_entry::EMPTY)
            {
               _entries[index].socket = socket;
               //_entries[index].queueId  = queueId;
               //_entries[index]._type = type;
               _entries[index]._status = _MonitorEntry::IDLE;
               _entry_mut.unlock();
   
               return index;
            }
         }
         catch(...)
         {
         }
   
      }
      _entry_mut.unlock();
      PEG_METHOD_EXIT();
      return -1;
   }
   
   
   void monitor_2::unsolicitSocketMessages(Sint32 socket)
   {
   
       PEG_METHOD_ENTER(TRC_HTTP, "monitor_2::unsolicitSocketMessages");
       _entry2_mut.lock(pegasus_thread_self());
   
       for(int index = 0; index < (int)_entries2.size(); index++)
       {
          if(_entries2[index].socket == socket)
          {
             _entries2[index]._status = monitor_2_entry::EMPTY;
             _entries2[index].socket = -1;
             break;
          }
       }
       _entry2_mut.unlock();
       PEG_METHOD_EXIT();
   }
   
 void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*)) void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))
 { {
   void* old = (void *)_session_dispatch;   void* old = (void *)_session_dispatch;
Line 876 
Line 924 
         static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);         static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);
         entry->get_sock().disableBlocking();         entry->get_sock().disableBlocking();
         pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);         pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);
   #ifdef PEGASUS_OS_TYPE_WINDOWS
       if((Sint32)connected  == SOCKET_ERROR)
   #else
           if((Sint32)connected == -1 )
   #endif
           {
              delete entry;
              break;
           }
   
         entry->get_sock().enableBlocking();         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)

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