(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.109 and 1.116.4.2

version 1.109, 2006/06/28 00:15:41 version 1.116.4.2, 2007/12/14 20:47:55
Line 29 
Line 29 
 // //
 //============================================================================== //==============================================================================
 // //
 // Author: Mike Brasher (mbrasher@bmc.com)  
 //  
 // Modified By: Mike Day (monitor_2) mdday@us.ibm.com  
 //              Amit K Arora (Bug#1153) amita@in.ibm.com  
 //              Alagaraja Ramasubramanian (alags_raj@in.ibm.com) for Bug#1090  
 //              Sushma Fernandes (sushma@hp.com) for Bug#2057  
 //              Josephine Eskaline Joyce (jojustin@in.ibm.com) for PEP#101  
 //              Roger Kumpf, Hewlett-Packard Company (roger_kumpf@hp.com)  
 //  
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
 #include "Network.h" #include "Network.h"
Line 51 
Line 42 
 #include <Pegasus/Common/MessageQueueService.h> #include <Pegasus/Common/MessageQueueService.h>
 #include <Pegasus/Common/Exception.h> #include <Pegasus/Common/Exception.h>
 #include "ArrayIterator.h" #include "ArrayIterator.h"
   #include <errno.h>
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
  
Line 60 
Line 52 
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
 // _getError()  
 //  
 ////////////////////////////////////////////////////////////////////////////////  
   
 static inline int _getError()  
 {  
 #ifdef PEGASUS_OS_TYPE_WINDOWS  
     return WSAGetLastError();  
 #else  
     return errno;  
 #endif  
 }  
   
 ////////////////////////////////////////////////////////////////////////////////  
 //  
 // Monitor // Monitor
 // //
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
Line 107 
Line 84 
  
 Monitor::~Monitor() Monitor::~Monitor()
 { {
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");      uninitializeTickler();
       Socket::uninitializeInterface();
       PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                     "returning from monitor destructor");
   }
   void Monitor::uninitializeTickler()
   {
       PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
  
     try{      try
       {
         if(_tickle_peer_socket >= 0)         if(_tickle_peer_socket >= 0)
         {         {
             Socket::close(_tickle_peer_socket);             Socket::close(_tickle_peer_socket);
Line 125 
Line 110 
     }     }
     catch(...)     catch(...)
     {     {
         Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                   "Failed to close tickle sockets");                   "Failed to close tickle sockets");
     }     }
  
     Socket::uninitializeInterface();  
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
                   "returning from monitor destructor");  
 } }
  
 void Monitor::initializeTickler(){  void Monitor::initializeTickler()
   {
     /*     /*
        NOTE: On any errors trying to        NOTE: On any errors trying to
              setup out tickle connection,              setup out tickle connection,
Line 142 
Line 125 
     */     */
  
     /* setup the tickle server/listener */     /* setup the tickle server/listener */
       // try until the tcpip is restarted
       do
       {
     // get a socket for the server side     // get a socket for the server side
     if((_tickle_server_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){          if ((_tickle_server_socket =
         //handle error                   Socket::createSocket(PF_INET, SOCK_STREAM, 0)) ==
         MessageLoaderParms parms("Common.Monitor.TICKLE_CREATE",               PEGASUS_INVALID_SOCKET)
           {
               MessageLoaderParms parms(
                   "Common.Monitor.TICKLE_CREATE",
                                  "Received error number $0 while creating the internal socket.",                                  "Received error number $0 while creating the internal socket.",
                                  _getError());                  getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // set TCP_NODELAY  
     int opt = 1;  
     setsockopt(_tickle_server_socket, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, sizeof(opt));  
   
     // initialize the address     // initialize the address
     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
Line 173 
Line 157 
     // bind server side to socket     // bind server side to socket
     if((::bind(_tickle_server_socket,     if((::bind(_tickle_server_socket,
                reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),                reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                sizeof(_tickle_server_addr))) < 0){                   sizeof(_tickle_server_addr))) < 0)
         // handle error          {
 #ifdef PEGASUS_OS_ZOS #ifdef PEGASUS_OS_ZOS
     MessageLoaderParms parms("Common.Monitor.TICKLE_BIND_LONG",              MessageLoaderParms parms(
                                  "Received error:$0 while binding the internal socket.",strerror(errno));                  "Common.Monitor.TICKLE_BIND_LONG",
                   "Received error:$0 while binding the internal socket.",
                   strerror(errno));
 #else #else
         MessageLoaderParms parms("Common.Monitor.TICKLE_BIND",              MessageLoaderParms parms(
                   "Common.Monitor.TICKLE_BIND",
                                  "Received error number $0 while binding the internal socket.",                                  "Received error number $0 while binding the internal socket.",
                                  _getError());                  getSocketError());
 #endif #endif
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // tell the kernel we are a server     // tell the kernel we are a server
     if((::listen(_tickle_server_socket,3)) < 0){          if ((::listen(_tickle_server_socket, 3)) < 0)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_LISTEN",              MessageLoaderParms parms(
                          "Received error number $0 while listening to the internal socket.",                  "Common.Monitor.TICKLE_LISTEN",
                                  _getError());                  "Received error number $0 while listening to the internal "
                       "socket.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // make sure we have the correct socket for our server     // make sure we have the correct socket for our server
     int sock = ::getsockname(_tickle_server_socket,          int sock = ::getsockname(
               _tickle_server_socket,
                    reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),                    reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                    &_addr_size);                    &_addr_size);
     if(sock < 0){          if (sock < 0)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_SOCKNAME",              MessageLoaderParms parms(
                          "Received error number $0 while getting the internal socket name.",                  "Common.Monitor.TICKLE_SOCKNAME",
                                  _getError());                  "Received error number $0 while getting the internal socket "
                       "name.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     /* set up the tickle client/connector */     /* set up the tickle client/connector */
  
     // get a socket for our tickle client     // get a socket for our tickle client
     if((_tickle_client_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){          if ((_tickle_client_socket =
         // handle error                   Socket::createSocket(PF_INET, SOCK_STREAM, 0)) ==
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",              PEGASUS_INVALID_SOCKET)
                          "Received error number $0 while creating the internal client socket.",          {
                                  _getError());              MessageLoaderParms parms(
                   "Common.Monitor.TICKLE_CLIENT_CREATE",
                   "Received error number $0 while creating the internal client "
                       "socket.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // set TCP_NODELAY  
     setsockopt(_tickle_client_socket, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, sizeof(opt));  
   
     // setup the address of the client     // setup the address of the client
     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
Line 236 
Line 229 
     // bind socket to client side     // bind socket to client side
     if((::bind(_tickle_client_socket,     if((::bind(_tickle_client_socket,
                reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),                reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),
                sizeof(_tickle_client_addr))) < 0){                   sizeof(_tickle_client_addr))) < 0)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",              MessageLoaderParms parms(
                          "Received error number $0 while binding the internal client socket.",                  "Common.Monitor.TICKLE_CLIENT_BIND",
                                  _getError());                  "Received error number $0 while binding the internal client "
                       "socket.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // connect to server side     // connect to server side
     if((::connect(_tickle_client_socket,     if((::connect(_tickle_client_socket,
                   reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),                   reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                   sizeof(_tickle_server_addr))) < 0){                   sizeof(_tickle_server_addr))) < 0)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",              MessageLoaderParms parms(
                          "Received error number $0 while connecting the internal client socket.",                  "Common.Monitor.TICKLE_CLIENT_CONNECT",
                                  _getError());                  "Received error number $0 while connecting the internal "
                       "client socket.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     /* set up the slave connection */     /* set up the slave connection */
     memset(&_tickle_peer_addr, 0, sizeof(_tickle_peer_addr));     memset(&_tickle_peer_addr, 0, sizeof(_tickle_peer_addr));
     SocketLength peer_size = sizeof(_tickle_peer_addr);     SocketLength peer_size = sizeof(_tickle_peer_addr);
     pegasus_sleep(1);          Threads::sleep(1);
  
     // this call may fail, we will try a max of 20 times to establish this peer connection          // this call may fail, we will try a max of 20 times to establish
           // this peer connection
     if((_tickle_peer_socket = ::accept(_tickle_server_socket,     if((_tickle_peer_socket = ::accept(_tickle_server_socket,
             reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),             reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),
             &peer_size)) < 0){                   &peer_size)) < 0)
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)          {
         // Only retry on non-windows platforms.              if (_tickle_peer_socket == PEGASUS_SOCKET_ERROR &&
         if(_tickle_peer_socket == -1 && errno == EAGAIN)                  getSocketError() == PEGASUS_NETWORK_TRYAGAIN)
         {         {
           int retries = 0;           int retries = 0;
           do           do
           {           {
             pegasus_sleep(1);                      Threads::sleep(1);
             _tickle_peer_socket = ::accept(_tickle_server_socket,                      _tickle_peer_socket = ::accept(
                           _tickle_server_socket,
                 reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),                 reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),
                 &peer_size);                 &peer_size);
             retries++;             retries++;
           } while(_tickle_peer_socket == -1 && errno == EAGAIN && retries < 20);                  } while (_tickle_peer_socket == PEGASUS_SOCKET_ERROR &&
                            getSocketError() == PEGASUS_NETWORK_TRYAGAIN &&
                            retries < 20);
               }
               // TCP/IP is down, destroy sockets and retry again.
               if (_tickle_peer_socket == PEGASUS_SOCKET_ERROR &&
                   getSocketError() == PEGASUS_NETWORK_TCPIP_STOPPED)
               {
                   // destroy everything
                   uninitializeTickler();
                   // retry again.
                   continue;
         }         }
 #endif  
     }     }
     if(_tickle_peer_socket == -1){          if (_tickle_peer_socket == PEGASUS_SOCKET_ERROR)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_ACCEPT",              MessageLoaderParms parms(
                          "Received error number $0 while accepting the internal socket connection.",                  "Common.Monitor.TICKLE_ACCEPT",
                                  _getError());                  "Received error number $0 while accepting the internal "
                       "socket connection.",
                   getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
     // add the tickler to the list of entries to be monitored and set to IDLE because Monitor only          else
           {
               // socket is ok
               break;
           }
       } while (1); // try until TCP/IP is restarted
   
       Socket::disableBlocking(_tickle_peer_socket);
       Socket::disableBlocking(_tickle_client_socket);
   
       // add the tickler to the list of entries to be monitored and set to
       // IDLE because Monitor only
     // checks entries with IDLE state for events     // checks entries with IDLE state for events
     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);
     entry._status = _MonitorEntry::IDLE;     entry._status = _MonitorEntry::IDLE;
   
       // is the tickler initalized as first socket on startup ?
       if (_entries.size()==0)
       {
          // if yes, append a new entry
     _entries.append(entry);     _entries.append(entry);
 } }
       else
       {
          // if not, overwrite the tickler entry with new socket
          _entries[0]=entry;
       }
   }
  
 void Monitor::tickle(void)  void Monitor::tickle()
 { {
     static char _buffer[] =     static char _buffer[] =
     {     {
Line 302 
Line 335 
     };     };
  
     AutoMutex autoMutex(_tickle_mutex);     AutoMutex autoMutex(_tickle_mutex);
     Socket::disableBlocking(_tickle_client_socket);  
     Socket::write(_tickle_client_socket,&_buffer, 2);     Socket::write(_tickle_client_socket,&_buffer, 2);
     Socket::enableBlocking(_tickle_client_socket);  
 } }
  
 void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status ) void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )
Line 313 
Line 344 
     _entries[index]._status = status;     _entries[index]._status = status;
 } }
  
 Boolean Monitor::run(Uint32 milliseconds)  void Monitor::run(Uint32 milliseconds)
 { {
  
     Boolean handled_events = false;  
     int i = 0;     int i = 0;
  
     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};
Line 328 
Line 358 
  
     ArrayIterator<_MonitorEntry> entries(_entries);     ArrayIterator<_MonitorEntry> entries(_entries);
  
     // Check the stopConnections flag.  If set, clear the Acceptor monitor entries      // Check the stopConnections flag.  If set, clear the Acceptor monitor
       // entries
     if (_stopConnections.get() == 1)     if (_stopConnections.get() == 1)
     {     {
         for ( int indx = 0; indx < (int)entries.size(); indx++)         for ( int indx = 0; indx < (int)entries.size(); indx++)
Line 376 
Line 407 
  
                                         if (h._responsePending == true)                                         if (h._responsePending == true)
                                         {                                         {
                                                 Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "                  PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                                                                                                         "Ignoring connection delete request because "                      "Monitor::run - Ignoring connection delete request "
                                                                                                         "responses are still pending. "                          "because responses are still pending. "
                                                                                                         "connection=0x%p, socket=%d\n",                                                                                                         "connection=0x%p, socket=%d\n",
                                                                                                         (void *)&h, h.getSocket());                      (void *)&h, h.getSocket()));
                                                 continue;                                                 continue;
                                         }                                         }
                                         h._connectionClosePending = false;                                         h._connectionClosePending = false;
Line 400 
Line 431 
           // unlocked will not result in an ArrayIndexOutOfBounds           // unlocked will not result in an ArrayIndexOutOfBounds
           // exception.           // exception.
  
           autoEntryMutex.unlock();              _entry_mut.unlock();
           o.enqueue(message);           o.enqueue(message);
           autoEntryMutex.lock();              _entry_mut.lock();
           // After enqueue a message and the autoEntryMutex has been released and locked again,  
           // the array of _entries can be changed. The ArrayIterator has be reset with the original _entries.              // After enqueue a message and the autoEntryMutex has been
               // released and locked again, the array of _entries can be
               // changed. The ArrayIterator has be reset with the original
               // _entries.
           entries.reset(_entries);           entries.reset(_entries);
        }        }
     }     }
Line 436 
Line 470 
     */     */
     maxSocketCurrentPass++;     maxSocketCurrentPass++;
  
     autoEntryMutex.unlock();      _entry_mut.unlock();
  
     //     //
     // The first argument to select() is ignored on Windows and it is not     // The first argument to select() is ignored on Windows and it is not
Line 448 
Line 482 
 #else #else
     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
 #endif #endif
     autoEntryMutex.lock();      _entry_mut.lock();
     // After enqueue a message and the autoEntryMutex has been released and locked again,  
     // the array of _entries can be changed. The ArrayIterator has be reset with the original _entries      struct timeval timeNow;
       Time::gettimeofday(&timeNow);
   
       // After enqueue a message and the autoEntryMutex has been released and
       // locked again, the array of _entries can be changed. The ArrayIterator
       // has be reset with the original _entries
     entries.reset(_entries);     entries.reset(_entries);
  
     if (events == PEGASUS_SOCKET_ERROR)     if (events == PEGASUS_SOCKET_ERROR)
     {     {
        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
           "Monitor::run - errorno = %d has occurred on select.", errno);              "Monitor::run - errorno = %d has occurred on select.", errno));
        // The EBADF error indicates that one or more or the file        // The EBADF error indicates that one or more or the file
        // descriptions was not valid. This could indicate that        // descriptions was not valid. This could indicate that
        // the entries structure has been corrupted or that        // the entries structure has been corrupted or that
Line 466 
Line 505 
     }     }
     else if (events)     else if (events)
     {     {
        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
           "Monitor::run select event received events = %d, monitoring %d idle entries",              "Monitor::run select event received events = %d, monitoring %d "
            events, _idleEntries);                  "idle entries",
               events, _idleEntries));
        for( int indx = 0; indx < (int)entries.size(); indx++)        for( int indx = 0; indx < (int)entries.size(); indx++)
        {        {
           // The Monitor should only look at entries in the table that are IDLE (i.e.,              // The Monitor should only look at entries in the table that are
           // owned by the Monitor).              // IDLE (i.e., owned by the Monitor).
           if((entries[indx]._status.get() == _MonitorEntry::IDLE) &&           if((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
              (FD_ISSET(entries[indx].socket, &fdread)))              (FD_ISSET(entries[indx].socket, &fdread)))
           {           {
              MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);              MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);
              Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                  PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                   "Monitor::run indx = %d, queueId =  %d, q = %p",                   "Monitor::run indx = %d, queueId =  %d, q = %p",
                   indx, entries[indx].queueId, q);                      indx, entries[indx].queueId, q));
              PEGASUS_ASSERT(q !=0);              PEGASUS_ASSERT(q !=0);
  
              try              try
              {              {
                 if(entries[indx]._type == Monitor::CONNECTION)                 if(entries[indx]._type == Monitor::CONNECTION)
                 {                 {
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                      "entries[indx].type for indx = %d is Monitor::CONNECTION", indx);                              "entries[indx].type for indx = %d is "
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                                  "Monitor::CONNECTION",
                               indx));
                    // Do not update the entry just yet. The entry gets updated once                          HTTPConnection *dst =
                    // the request has been read.                              reinterpret_cast<HTTPConnection *>(q);
                    //entries[indx]._status = _MonitorEntry::BUSY;                          dst->_entry_index = indx;
   
                    // If allocate_and_awaken failure, retry on next iteration                          // Update idle start time because we have received some
 /* Removed for PEP 183.                          // data. Any data is good data at this point, and we'll
                    if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(                          // keep the connection alive, even if we've exceeded
                            (void *)q, _dispatch))                          // the idleConnectionTimeout, which will be checked
                    {                          // when we call closeConnectionOnTimeout() next.
                       Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,                          Time::gettimeofday(&dst->_idleStartTime);
                           "Monitor::run: Insufficient resources to process request.");  
                       entries[indx]._status = _MonitorEntry::IDLE;                          // Check for accept pending (ie. SSL handshake pending)
                       return true;                          // or idle connection timeouts for sockets from which
                    }                          // we received data (avoiding extra queue lookup below).
 */                          if (!dst->closeConnectionOnTimeout(&timeNow))
 // Added for PEP 183                          {
                    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);                          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                              "Monitor::_dispatch: entering run() for "
                          "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, q = %p",                                  "indx = %d, queueId = %d, q = %p",
                    dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);                              dst->_entry_index,
                                   dst->_monitor->
                                       _entries[dst->_entry_index].queueId,
                               dst));
   
                    try                    try
                    {                    {
                        dst->run(1);                        dst->run(1);
                    }                    }
                    catch (...)                    catch (...)
                    {                    {
                        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                              PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                        "Monitor::_dispatch: exception received");                        "Monitor::_dispatch: exception received");
                    }                    }
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                          PEG_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));
                    // It is possible the entry status may not be set to busy.                          }
                    // The following will fail in that case.  
                    // PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);  
                    // 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.  
   
                    // The following is not relevant as the worker thread or the  
                    // reader thread will update the status of the entry.  
                    //if (dst->_connectionClosePending)  
                    //{  
                    //  dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;  
                    //}  
                    //else  
                    //{  
                    //  dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;  
                    //}  
 // end Added for PEP 183  
                 }                 }
                 else if( entries[indx]._type == Monitor::INTERNAL){                      else if (entries[indx]._type == Monitor::INTERNAL)
                       {
                         // set ourself to BUSY,                         // set ourself to BUSY,
                         // read the data                         // read the data
                         // and set ourself back to IDLE                         // and set ourself back to IDLE
  
                         entries[indx]._status = _MonitorEntry::BUSY;                         entries[indx]._status = _MonitorEntry::BUSY;
                         static char buffer[2];                         static char buffer[2];
                         Socket::disableBlocking(entries[indx].socket);                          Sint32 amt =
                         Sint32 amt = Socket::read(entries[indx].socket,&buffer, 2);                              Socket::read(entries[indx].socket,&buffer, 2);
                         Socket::enableBlocking(entries[indx].socket);  
                           if (amt == PEGASUS_SOCKET_ERROR &&
                               getSocketError() == PEGASUS_NETWORK_TCPIP_STOPPED)
                           {
                               PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                                   "Monitor::run: Tickler socket got an IO error. "
                                       "Going to re-create Socket and wait for "
                                       "TCP/IP restart.");
                               uninitializeTickler();
                               initializeTickler();
                           }
                           else
                           {
                         entries[indx]._status = _MonitorEntry::IDLE;                         entries[indx]._status = _MonitorEntry::IDLE;
                 }                 }
                       }
                 else                 else
                 {                 {
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                      "Non-connection entry, indx = %d, has been received.", indx);                              "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);
                    entries[indx]._status = _MonitorEntry::BUSY;                    entries[indx]._status = _MonitorEntry::BUSY;
                    autoEntryMutex.unlock();                          _entry_mut.unlock();
                    q->enqueue(msg);                    q->enqueue(msg);
                    autoEntryMutex.lock();                          _entry_mut.lock();
            // After enqueue a message and the autoEntryMutex has been released and locked again,  
            // the array of entries can be changed. The ArrayIterator has be reset with the original _entries                          // After enqueue a message and the autoEntryMutex has
                           // been released and locked again, the array of
                           // entries can be changed. The ArrayIterator has be
                           // reset with the original _entries
            entries.reset(_entries);            entries.reset(_entries);
                    entries[indx]._status = _MonitorEntry::IDLE;                    entries[indx]._status = _MonitorEntry::IDLE;
   
                    return true;  
                 }                 }
              }              }
              catch(...)              catch(...)
              {              {
              }              }
              handled_events = true;              }
               // else check for accept pending (ie. SSL handshake pending) or
               // idle connection timeouts for sockets from which we did not
               // receive data.
               else if ((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
                   entries[indx]._type == Monitor::CONNECTION)
               {
                   MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);
                   HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);
                   dst->_entry_index = indx;
                   dst->closeConnectionOnTimeout(&timeNow);
               }
           }
       }
       // else if "events" is zero (ie. select timed out) then we still need
       // to check if there are any pending SSL handshakes that have timed out.
       else
       {
           for (int indx = 0; indx < (int)entries.size(); indx++)
           {
               if ((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
                   entries[indx]._type == Monitor::CONNECTION)
               {
                   MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);
                   HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);
                   dst->_entry_index = indx;
                   dst->closeConnectionOnTimeout(&timeNow);
           }           }
        }        }
     }     }
   
     return(handled_events);  
 } }
  
 void Monitor::stopListeningForConnections(Boolean wait) void Monitor::stopListeningForConnections(Boolean wait)
Line 616 
Line 684 
    // current connections requested    // current connections requested
    _solicitSocketCount++;  // bump the count    _solicitSocketCount++;  // bump the count
    int size = (int)_entries.size();    int size = (int)_entries.size();
    if((int)_solicitSocketCount >= (size-1)){      if ((int)_solicitSocketCount >= (size-1))
         for(int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++){      {
           for (int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++)
           {
                 _MonitorEntry entry(0, 0, 0);                 _MonitorEntry entry(0, 0, 0);
                 _entries.append(entry);                 _entries.append(entry);
         }         }
Line 642 
Line 712 
       {       {
       }       }
    }    }
    _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful      // decrease the count, if we are here we didn't do anything meaningful
       _solicitSocketCount--;
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
    return -1;    return -1;
   
 } }
  
 void Monitor::unsolicitSocketMessages(SocketHandle socket) void Monitor::unsolicitSocketMessages(SocketHandle socket)
 { {
   
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
     AutoMutex autoMut(_entry_mut);     AutoMutex autoMut(_entry_mut);
  
     /*     /*
         Start at index = 1 because _entries[0] is the tickle entry which never needs          Start at index = 1 because _entries[0] is the tickle entry which
         to be EMPTY;          never needs to be EMPTY;
     */     */
     unsigned int index;     unsigned int index;
     for(index = 1; index < _entries.size(); index++)     for(index = 1; index < _entries.size(); index++)
Line 672 
Line 741 
  
     /*     /*
         Dynamic Contraction:         Dynamic Contraction:
         To remove excess entries we will start from the end of the _entries array          To remove excess entries we will start from the end of the _entries
         and remove all entries with EMPTY status until we find the first NON EMPTY.          array and remove all entries with EMPTY status until we find the
         This prevents the positions, of the NON EMPTY entries, from being changed.          first NON EMPTY.  This prevents the positions, of the NON EMPTY
           entries, from being changed.
     */     */
     index = _entries.size() - 1;     index = _entries.size() - 1;
     while(_entries[index]._status.get() == _MonitorEntry::EMPTY){      while (_entries[index]._status.get() == _MonitorEntry::EMPTY)
       {
         if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)         if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)
                 _entries.remove(index);                 _entries.remove(index);
         index--;         index--;
Line 686 
Line 757 
 } }
  
 // Note: this is no longer called with PEP 183. // Note: this is no longer called with PEP 183.
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)  ThreadReturnType 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,      PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
         "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, q = %p",          "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, "
         dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);              "q = %p",
           dst->_entry_index,
           dst->_monitor->_entries[dst->_entry_index].queueId,
           dst));
   
    try    try
    {    {
       dst->run(1);       dst->run(1);
    }    }
    catch (...)    catch (...)
    {    {
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
           "Monitor::_dispatch: exception received");           "Monitor::_dispatch: exception received");
    }    }
    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,      PEG_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));
  
    PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);      PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() ==
           _MonitorEntry::BUSY);
  
    // Once the HTTPConnection thread has set the status value to either    // Once the HTTPConnection thread has set the status value to either
    // Monitor::DYING or Monitor::IDLE, it has returned control of the connection      // Monitor::DYING or Monitor::IDLE, it has returned control of the
    // to the Monitor.  It is no longer permissible to access the connection      // connection to the Monitor.  It is no longer permissible to access the
    // or the entry in the _entries table.      // connection or the entry in the _entries table.
    if (dst->_connectionClosePending)    if (dst->_connectionClosePending)
    {    {
       dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;          dst->_monitor->_entries[dst->_entry_index]._status =
               _MonitorEntry::DYING;
    }    }
    else    else
    {    {
       dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;          dst->_monitor->_entries[dst->_entry_index]._status =
               _MonitorEntry::IDLE;
    }    }
    return 0;    return 0;
 } }

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