(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.107 and 1.133

version 1.107, 2006/06/26 22:56:40 version 1.133, 2008/06/10 20:09:29
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 48 
Line 39 
 #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/HTTPAcceptor.h>
 #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 "HostAddress.h"
   #include <errno.h>
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
 static AtomicInt _connections(0);  
   
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
 // _getError()  // Tickler
 // //
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 static inline int _getError()  Tickler::Tickler()
       : _listenSocket(PEGASUS_INVALID_SOCKET),
         _clientSocket(PEGASUS_INVALID_SOCKET),
         _serverSocket(PEGASUS_INVALID_SOCKET)
 { {
 #ifdef PEGASUS_OS_TYPE_WINDOWS      try
     return WSAGetLastError()      {
 #else          _initialize();
     return errno;      }
 #endif      catch (...)
       {
           _uninitialize();
           throw;
       }
 } }
  
 ////////////////////////////////////////////////////////////////////////////////  Tickler::~Tickler()
 //  
 // Monitor  
 //  
 ////////////////////////////////////////////////////////////////////////////////  
   
 #define MAX_NUMBER_OF_MONITOR_ENTRIES  32  
 Monitor::Monitor()  
    : _stopConnections(0),  
      _stopConnectionsSem(0),  
      _solicitSocketCount(0),  
      _tickle_client_socket(-1),  
      _tickle_server_socket(-1),  
      _tickle_peer_socket(-1)  
 { {
     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;      _uninitialize();
     Socket::initializeInterface();  }
     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);  
  
     // setup the tickler  void Tickler::notify()
     initializeTickler();  {
       Socket::write(_clientSocket, "\0", 1);
   }
  
     // Start the count at 1 because initilizeTickler()  void Tickler::reset()
     // has added an entry in the first position of the  {
     // _entries array      // Clear all bytes from the tickle socket
     for( int i = 1; i < numberOfMonitorEntriesToAllocate; i++ )      char buffer[32];
       while (Socket::read(_serverSocket, buffer, 32) > 0)
     {     {
        _MonitorEntry entry(0, 0, 0);  
        _entries.append(entry);  
     }     }
 } }
  
 Monitor::~Monitor()  #if defined(PEGASUS_OS_TYPE_UNIX)
   
   // Use an anonymous pipe for the tickle connection.
   
   void Tickler::_initialize()
 { {
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");      int fds[2];
  
     try{      if (pipe(fds) == -1)
         if(_tickle_peer_socket >= 0)  
         {         {
             Socket::close(_tickle_peer_socket);          MessageLoaderParms parms(
               "Common.Monitor.TICKLE_CREATE",
               "Received error number $0 while creating the internal socket.",
               getSocketError());
           throw Exception(parms);
         }         }
         if(_tickle_client_socket >= 0)  
         {      _serverSocket = fds[0];
             Socket::close(_tickle_client_socket);      _clientSocket = fds[1];
   
       Socket::disableBlocking(_serverSocket);
         }         }
         if(_tickle_server_socket >= 0)  
   #else
   
   // Use an external loopback socket connection to allow the tickle socket to
   // be included in the select() array on non-Unix platforms.
   
   void Tickler::_initialize()
         {         {
             Socket::close(_tickle_server_socket);      //
         }      // Set up the addresses for the listen, client, and server sockets
     }      // based on whether IPv6 is enabled.
     catch(...)      //
   
       Socket::initializeInterface();
   
   # ifdef PEGASUS_ENABLE_IPV6
       struct sockaddr_storage listenAddress;
       struct sockaddr_storage clientAddress;
       struct sockaddr_storage serverAddress;
   # else
       struct sockaddr_in listenAddress;
       struct sockaddr_in clientAddress;
       struct sockaddr_in serverAddress;
   # endif
   
       int addressFamily;
       SocketLength addressLength;
   
       memset(&listenAddress, 0, sizeof (listenAddress));
   
   # ifdef PEGASUS_ENABLE_IPV6
       if (System::isIPv6StackActive())
     {     {
         Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          // Use the IPv6 loopback address for the listen sockets
                   "Failed to close tickle sockets");          HostAddress::convertTextToBinary(
               HostAddress::AT_IPV6,
               "::1",
               &reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_addr);
           listenAddress.ss_family = AF_INET6;
           reinterpret_cast<struct sockaddr_in6*>(&listenAddress)->sin6_port = 0;
   
           addressFamily = AF_INET6;
           addressLength = sizeof(struct sockaddr_in6);
     }     }
       else
   # endif
       {
           // Use the IPv4 loopback address for the listen sockets
           HostAddress::convertTextToBinary(
               HostAddress::AT_IPV4,
               "127.0.0.1",
               &reinterpret_cast<struct sockaddr_in*>(
                   &listenAddress)->sin_addr.s_addr);
           reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_family =
               AF_INET;
           reinterpret_cast<struct sockaddr_in*>(&listenAddress)->sin_port = 0;
  
     Socket::uninitializeInterface();          addressFamily = AF_INET;
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,          addressLength = sizeof(struct sockaddr_in);
                   "returning from monitor destructor");  
 } }
  
 void Monitor::initializeTickler(){      // Use the same address for the client socket as the listen socket
     /*      clientAddress = listenAddress;
        NOTE: On any errors trying to  
              setup out tickle connection,  
              throw an exception/end the server  
     */  
  
     /* setup the tickle server/listener */      //
       // Set up a listen socket to allow the tickle client and server to connect
       //
  
     // get a socket for the server side      // Create the listen socket
     if((_tickle_server_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){      if ((_listenSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==
         //handle error               PEGASUS_INVALID_SOCKET)
         MessageLoaderParms parms("Common.Monitor.TICKLE_CREATE",      {
           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      // Bind the listen socket to the loopback address
     int opt = 1;      if (::bind(
     setsockopt(_tickle_server_socket, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, sizeof(opt));              _listenSocket,
               reinterpret_cast<struct sockaddr*>(&listenAddress),
     // initialize the address              addressLength) < 0)
     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));      {
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM          MessageLoaderParms parms(
 #pragma convert(37)              "Common.Monitor.TICKLE_BIND",
 #endif  
     _tickle_server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM  
 #pragma convert(0)  
 #endif  
     _tickle_server_addr.sin_family = PF_INET;  
     _tickle_server_addr.sin_port = 0;  
   
     socklen_t _addr_size = sizeof(_tickle_server_addr);  
   
     // bind server side to socket  
     if((::bind(_tickle_server_socket,  
                reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),  
                sizeof(_tickle_server_addr))) < 0){  
         // handle error  
 #ifdef PEGASUS_OS_ZOS  
     MessageLoaderParms parms("Common.Monitor.TICKLE_BIND_LONG",  
                                  "Received error:$0 while binding the internal socket.",strerror(errno));  
 #else  
         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  
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // tell the kernel we are a server      // Listen for a connection from the tickle client
     if((::listen(_tickle_server_socket,3)) < 0){      if ((::listen(_listenSocket, 3)) < 0)
         // handle error      {
         MessageLoaderParms parms("Common.Monitor.TICKLE_LISTEN",          MessageLoaderParms parms(
               "Common.Monitor.TICKLE_LISTEN",
                          "Received error number $0 while listening to the internal socket.",                          "Received error number $0 while listening to the internal socket.",
                                  _getError());              getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // make sure we have the correct socket for our server      // Verify we have the correct listen socket
     int sock = ::getsockname(_tickle_server_socket,      SocketLength tmpAddressLength = addressLength;
                    reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),      int sock = ::getsockname(
                    &_addr_size);          _listenSocket,
     if(sock < 0){          reinterpret_cast<struct sockaddr*>(&listenAddress),
         // handle error          &tmpAddressLength);
         MessageLoaderParms parms("Common.Monitor.TICKLE_SOCKNAME",      if (sock < 0)
       {
           MessageLoaderParms parms(
               "Common.Monitor.TICKLE_SOCKNAME",
                          "Received error number $0 while getting the internal socket name.",                          "Received error number $0 while getting the internal socket name.",
                                  _getError());              getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     /* set up the tickle client/connector */      //
       // Set up the client side of the tickle connection.
       //
  
     // get a socket for our tickle client      // Create the client socket
     if((_tickle_client_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){      if ((_clientSocket = Socket::createSocket(addressFamily, SOCK_STREAM, 0)) ==
         // handle error               PEGASUS_INVALID_SOCKET)
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",      {
                          "Received error number $0 while creating the internal client socket.",          MessageLoaderParms parms(
                                  _getError());              "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      // Bind the client socket to the loopback address
     setsockopt(_tickle_client_socket, IPPROTO_TCP, TCP_NODELAY, (char*)&opt, sizeof(opt));      if (::bind(
               _clientSocket,
     // setup the address of the client              reinterpret_cast<struct sockaddr*>(&clientAddress),
     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));              addressLength) < 0)
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM      {
 #pragma convert(37)          MessageLoaderParms parms(
 #endif              "Common.Monitor.TICKLE_CLIENT_BIND",
     _tickle_client_addr.sin_addr.s_addr = inet_addr("127.0.0.1");              "Received error number $0 while binding the internal client "
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM                  "socket.",
 #pragma convert(0)              getSocketError());
 #endif          throw Exception(parms);
     _tickle_client_addr.sin_family = PF_INET;      }
     _tickle_client_addr.sin_port = 0;  
  
     // bind socket to client side      // Connect the client socket to the listen socket address
     if((::bind(_tickle_client_socket,      if (::connect(
                reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),              _clientSocket,
                sizeof(_tickle_client_addr))) < 0){              reinterpret_cast<struct sockaddr*>(&listenAddress),
         // handle error              addressLength) < 0)
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",      {
                          "Received error number $0 while binding the internal client socket.",          MessageLoaderParms parms(
                                  _getError());              "Common.Monitor.TICKLE_CLIENT_CONNECT",
               "Received error number $0 while connecting the internal client "
                   "socket.",
               getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // connect to server side      //
     if((::connect(_tickle_client_socket,      // Set up the server side of the tickle connection.
                   reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),      //
                   sizeof(_tickle_server_addr))) < 0){  
         // handle error      tmpAddressLength = addressLength;
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",  
                          "Received error number $0 while connecting the internal client socket.",      // Accept the client socket connection.
                                  _getError());      _serverSocket = ::accept(
           _listenSocket,
           reinterpret_cast<struct sockaddr*>(&serverAddress),
           &tmpAddressLength);
   
       if (_serverSocket == PEGASUS_SOCKET_ERROR)
       {
           MessageLoaderParms parms(
               "Common.Monitor.TICKLE_ACCEPT",
               "Received error number $0 while accepting the internal socket "
                   "connection.",
               getSocketError());
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     /* set up the slave connection */      //
     memset(&_tickle_peer_addr, 0, sizeof(_tickle_peer_addr));      // Close the listen socket and make the other sockets non-blocking
     socklen_t peer_size = sizeof(_tickle_peer_addr);      //
     pegasus_sleep(1);  
       Socket::close(_listenSocket);
     // this call may fail, we will try a max of 20 times to establish this peer connection      Socket::disableBlocking(_serverSocket);
     if((_tickle_peer_socket = ::accept(_tickle_server_socket,      Socket::disableBlocking(_clientSocket);
             reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),  
             &peer_size)) < 0){  
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)  
         // Only retry on non-windows platforms.  
         if(_tickle_peer_socket == -1 && errno == EAGAIN)  
         {  
           int retries = 0;  
           do  
           {  
             pegasus_sleep(1);  
             _tickle_peer_socket = ::accept(_tickle_server_socket,  
                 reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),  
                 &peer_size);  
             retries++;  
           } while(_tickle_peer_socket == -1 && errno == EAGAIN && retries < 20);  
         }         }
   
 #endif #endif
   
   void Tickler::_uninitialize()
   {
       PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
   
       try
       {
           Socket::close(_serverSocket);
           Socket::close(_clientSocket);
           Socket::close(_listenSocket);
     }     }
     if(_tickle_peer_socket == -1){      catch (...)
         // handle error      {
         MessageLoaderParms parms("Common.Monitor.TICKLE_ACCEPT",          PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                          "Received error number $0 while accepting the internal socket connection.",              "Failed to close tickle sockets");
                                  _getError());  
         throw Exception(parms);  
     }     }
     // add the tickler to the list of entries to be monitored and set to IDLE because Monitor only      Socket::uninitializeInterface();
     // checks entries with IDLE state for events  
     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);  
     entry._status = _MonitorEntry::IDLE;  
     _entries.append(entry);  
 } }
  
 void Monitor::tickle(void)  
 {  ////////////////////////////////////////////////////////////////////////////////
     static char _buffer[] =  //
   // Monitor
   //
   ////////////////////////////////////////////////////////////////////////////////
   
   #define MAX_NUMBER_OF_MONITOR_ENTRIES  32
   Monitor::Monitor()
      : _stopConnections(0),
        _stopConnectionsSem(0),
        _solicitSocketCount(0)
     {     {
       '0','0'      int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;
     };      _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);
   
       // Create a MonitorEntry for the Tickler and set its state to IDLE so the
       // Monitor will watch for its events.
       _entries.append(MonitorEntry(
           _tickler.getReadHandle(),
           1,
           MonitorEntry::STATUS_IDLE,
           MonitorEntry::TYPE_TICKLER));
  
     AutoMutex autoMutex(_tickle_mutex);      // Start the count at 1 because _entries[0] is the Tickler
     Socket::disableBlocking(_tickle_client_socket);      for (int i = 1; i < numberOfMonitorEntriesToAllocate; i++)
     Socket::write(_tickle_client_socket,&_buffer, 2);      {
     Socket::enableBlocking(_tickle_client_socket);          _entries.append(MonitorEntry());
       }
 } }
  
 void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )  Monitor::~Monitor()
 { {
     // Set the state to requested state      PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
     _entries[index]._status = status;                    "returning from monitor destructor");
 } }
  
 Boolean Monitor::run(Uint32 milliseconds)  void Monitor::tickle()
 { {
       _tickler.notify();
   }
  
     Boolean handled_events = false;  void Monitor::setState(
     int i = 0;      Uint32 index,
       MonitorEntry::Status status)
   {
       AutoMutex autoEntryMutex(_entriesMutex);
       // Set the state to requested state
       _entries[index].status = status;
   }
  
   void Monitor::run(Uint32 milliseconds)
   {
     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};
  
     fd_set fdread;     fd_set fdread;
     FD_ZERO(&fdread);     FD_ZERO(&fdread);
  
     AutoMutex autoEntryMutex(_entry_mut);      AutoMutex autoEntryMutex(_entriesMutex);
  
     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 (Uint32 indx = 0; indx < entries.size(); indx++)
         {         {
             if (entries[indx]._type == Monitor::ACCEPTOR)              if (entries[indx].type == MonitorEntry::TYPE_ACCEPTOR)
             {             {
                 if ( entries[indx]._status.get() != _MonitorEntry::EMPTY)                  if (entries[indx].status != MonitorEntry::STATUS_EMPTY)
                 {                 {
                    if ( entries[indx]._status.get() == _MonitorEntry::IDLE ||                      if (entries[indx].status == MonitorEntry::STATUS_IDLE ||
                         entries[indx]._status.get() == _MonitorEntry::DYING )                          entries[indx].status == MonitorEntry::STATUS_DYING)
                    {                    {
                        // remove the entry                        // remove the entry
                        entries[indx]._status = _MonitorEntry::EMPTY;                          entries[indx].status = MonitorEntry::STATUS_EMPTY;
                    }                    }
                    else                    else
                    {                    {
                        // set status to DYING                        // set status to DYING
                       entries[indx]._status = _MonitorEntry::DYING;                          entries[indx].status = MonitorEntry::STATUS_DYING;
                    }                    }
                }                }
            }            }
Line 355 
Line 415 
         _stopConnectionsSem.signal();         _stopConnectionsSem.signal();
     }     }
  
     for( int indx = 0; indx < (int)entries.size(); indx++)      for (Uint32 indx = 0; indx < entries.size(); indx++)
     {     {
                          const _MonitorEntry &entry = entries[indx];          const MonitorEntry& entry = entries[indx];
        if ((entry._status.get() == _MonitorEntry::DYING) &&  
                                          (entry._type == Monitor::CONNECTION))          if ((entry.status == MonitorEntry::STATUS_DYING) &&
               (entry.type == MonitorEntry::TYPE_CONNECTION))
        {        {
           MessageQueue *q = MessageQueue::lookup(entry.queueId);           MessageQueue *q = MessageQueue::lookup(entry.queueId);
           PEGASUS_ASSERT(q != 0);           PEGASUS_ASSERT(q != 0);
Line 376 
Line 437 
  
                                         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;
           MessageQueue &o = h.get_owner();              HTTPAcceptor &o = h.getOwningAcceptor();
           Message* message= new CloseConnectionMessage(entry.socket);           Message* message= new CloseConnectionMessage(entry.socket);
           message->dest = o.getQueueId();           message->dest = o.getQueueId();
  
Line 400 
Line 461 
           // unlocked will not result in an ArrayIndexOutOfBounds           // unlocked will not result in an ArrayIndexOutOfBounds
           // exception.           // exception.
  
           autoEntryMutex.unlock();              _entriesMutex.unlock();
           o.enqueue(message);           o.enqueue(message);
           autoEntryMutex.lock();              _entriesMutex.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 418 
Line 482 
         because we have to traverse the entire array.         because we have to traverse the entire array.
     */     */
     SocketHandle maxSocketCurrentPass = 0;     SocketHandle maxSocketCurrentPass = 0;
     for( int indx = 0; indx < (int)entries.size(); indx++)      for (Uint32 indx = 0; indx < entries.size(); indx++)
     {     {
        if(maxSocketCurrentPass < entries[indx].socket)        if(maxSocketCurrentPass < entries[indx].socket)
             maxSocketCurrentPass = entries[indx].socket;             maxSocketCurrentPass = entries[indx].socket;
  
        if(entries[indx]._status.get() == _MonitorEntry::IDLE)          if (entries[indx].status == MonitorEntry::STATUS_IDLE)
        {        {
            _idleEntries++;            _idleEntries++;
            FD_SET(entries[indx].socket, &fdread);            FD_SET(entries[indx].socket, &fdread);
Line 436 
Line 500 
     */     */
     maxSocketCurrentPass++;     maxSocketCurrentPass++;
  
     autoEntryMutex.unlock();      _entriesMutex.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 512 
 #else #else
     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
 #endif #endif
     autoEntryMutex.lock();      int selectErrno = getSocketError();
     // 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      _entriesMutex.lock();
   
       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 - select() returned error %d.", selectErrno));
        // 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
        // we have a synchronization error.        // we have a synchronization error.
  
        PEGASUS_ASSERT(errno != EBADF);          PEGASUS_ASSERT(selectErrno != EBADF);
     }     }
     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",
        for( int indx = 0; indx < (int)entries.size(); indx++)              events, _idleEntries));
        {          for (Uint32 indx = 0; indx < entries.size(); indx++)
           // The Monitor should only look at entries in the table that are IDLE (i.e.,          {
           // owned by the Monitor).              // The Monitor should only look at entries in the table that are
           if((entries[indx]._status.get() == _MonitorEntry::IDLE) &&              // IDLE (i.e., owned by the Monitor).
               if ((entries[indx].status == MonitorEntry::STATUS_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,  
                   "Monitor::run indx = %d, queueId =  %d, q = %p",  
                   indx, entries[indx].queueId, q);  
              PEGASUS_ASSERT(q !=0);              PEGASUS_ASSERT(q !=0);
                   PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                       "Monitor::run indx = %d, queueId = %d, q = %p",
                       indx, entries[indx].queueId, q));
  
              try              try
              {              {
                 if(entries[indx]._type == Monitor::CONNECTION)                      if (entries[indx].type == MonitorEntry::TYPE_CONNECTION)
                 {                 {
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                      "entries[indx].type for indx = %d is Monitor::CONNECTION", indx);                              "entries[%d].type is TYPE_CONNECTION",
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                              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,                                  "Entering HTTPConnection::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);                                  indx, entries[indx].queueId, q));
   
                    try                    try
                    {                    {
                        dst->run(1);                        dst->run(1);
                    }                    }
                    catch (...)                    catch (...)
                    {                    {
                        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                                  PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL2,
                        "Monitor::_dispatch: exception received");                                      "Caught exception from "
                                       "HTTPConnection::run()");
                    }                    }
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                              PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
                    "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);                                  "Exited HTTPConnection::run()");
                           }
                    // It is possible the entry status may not be set to busy.                      }
                    // The following will fail in that case.                      else if (entries[indx].type == MonitorEntry::TYPE_TICKLER)
                    // PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);                      {
                    // Once the HTTPConnection thread has set the status value to either                          _tickler.reset();
                    // 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){  
                         // set ourself to BUSY,  
                         // read the data  
                         // and set ourself back to IDLE  
   
                         entries[indx]._status = _MonitorEntry::BUSY;  
                         static char buffer[2];  
                         Socket::disableBlocking(entries[indx].socket);  
                         Sint32 amt = Socket::read(entries[indx].socket,&buffer, 2);  
                         Socket::enableBlocking(entries[indx].socket);  
                         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]._status = _MonitorEntry::BUSY;                              entries[indx].socket, events);
                    autoEntryMutex.unlock();                          entries[indx].status = MonitorEntry::STATUS_BUSY;
                           _entriesMutex.unlock();
                    q->enqueue(msg);                    q->enqueue(msg);
                    autoEntryMutex.lock();                          _entriesMutex.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  
            entries.reset(_entries);  
                    entries[indx]._status = _MonitorEntry::IDLE;  
  
                    return true;                          // After enqueue a message and the autoEntryMutex has
                           // been released and locked again, the array of
                           // entries can be changed. The ArrayIterator has to be
                           // reset with the latest _entries.
                           entries.reset(_entries);
                           entries[indx].status = MonitorEntry::STATUS_IDLE;
                 }                 }
              }              }
              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 == MonitorEntry::STATUS_IDLE) &&
                   entries[indx].type == MonitorEntry::TYPE_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 (Uint32 indx = 0; indx < entries.size(); indx++)
           {
               if ((entries[indx].status == MonitorEntry::STATUS_IDLE) &&
                   entries[indx].type == MonitorEntry::TYPE_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 607 
Line 683 
     SocketHandle socket,     SocketHandle socket,
     Uint32 events,     Uint32 events,
     Uint32 queueId,     Uint32 queueId,
     int type)      Uint32 type)
 { {
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
    AutoMutex autoMut(_entry_mut);      AutoMutex autoMut(_entriesMutex);
   
    // Check to see if we need to dynamically grow the _entries array    // Check to see if we need to dynamically grow the _entries array
    // We always want the _entries array to 2 bigger than the      // We always want the _entries array to be 2 bigger than the
    // current connections requested    // current connections requested
    _solicitSocketCount++;  // bump the count    _solicitSocketCount++;  // bump the count
    int size = (int)_entries.size();  
    if((int)_solicitSocketCount >= (size-1)){      for (Uint32 i = _entries.size(); i < _solicitSocketCount + 1; i++)
         for(int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++){      {
                 _MonitorEntry entry(0, 0, 0);          _entries.append(MonitorEntry());
                 _entries.append(entry);  
         }  
    }    }
  
    int index;      for (Uint32 index = 1; index < _entries.size(); index++)
    for(index = 1; index < (int)_entries.size(); index++)  
    {    {
       try       try
       {       {
          if(_entries[index]._status.get() == _MonitorEntry::EMPTY)              if (_entries[index].status == MonitorEntry::STATUS_EMPTY)
          {          {
             _entries[index].socket = socket;             _entries[index].socket = socket;
             _entries[index].queueId  = queueId;             _entries[index].queueId  = queueId;
             _entries[index]._type = type;                  _entries[index].type = type;
             _entries[index]._status = _MonitorEntry::IDLE;                  _entries[index].status = MonitorEntry::STATUS_IDLE;
  
             return index;                  return (int)index;
          }          }
       }       }
       catch(...)       catch(...)
       {       {
       }       }
    }    }
    _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(_entriesMutex);
  
     /*     /*
         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 reset to EMPTY;
     */     */
     unsigned int index;      for (Uint32 index = 1; index < _entries.size(); index++)
     for(index = 1; index < _entries.size(); index++)  
     {     {
        if(_entries[index].socket == socket)        if(_entries[index].socket == socket)
        {        {
           _entries[index]._status = _MonitorEntry::EMPTY;              _entries[index].reset();
           _entries[index].socket = PEGASUS_INVALID_SOCKET;  
           _solicitSocketCount--;           _solicitSocketCount--;
           break;           break;
        }        }
Line 672 
Line 743 
  
     /*     /*
         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;      for (Uint32 index = _entries.size() - 1;
     while(_entries[index]._status.get() == _MonitorEntry::EMPTY){           (_entries[index].status == MonitorEntry::STATUS_EMPTY) &&
         if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)               (index >= MAX_NUMBER_OF_MONITOR_ENTRIES);
                 _entries.remove(index);           index--)
         index--;  
     }  
     PEG_METHOD_EXIT();  
 }  
   
 // Note: this is no longer called with PEP 183.  
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)  
 { {
    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);          _entries.remove(index);
    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
         "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, q = %p",  
         dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);  
    try  
    {  
       dst->run(1);  
    }  
    catch (...)  
    {  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
           "Monitor::_dispatch: exception received");  
    }    }
    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
           "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);  
   
    PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);  
  
    // Once the HTTPConnection thread has set the status value to either      PEG_METHOD_EXIT();
    // 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)  
    {  
       dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;  
    }  
    else  
    {  
       dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;  
    }  
    return 0;  
 } }
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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