(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.94.8.1 and 1.121.4.1

version 1.94.8.1, 2005/10/17 21:05:32 version 1.121.4.1, 2007/11/05 17:03:29
Line 1 
Line 1 
 //%2005////////////////////////////////////////////////////////////////////////  //%2006////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development // Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
 // Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems. // Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.
Line 8 
Line 8 
 // IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group. // IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.
 // Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.; // Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.;
 // EMC Corporation; VERITAS Software Corporation; The Open Group. // EMC Corporation; VERITAS Software Corporation; The Open Group.
   // Copyright (c) 2006 Hewlett-Packard Development Company, L.P.; IBM Corp.;
   // EMC Corporation; Symantec 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 27 
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 <Pegasus/Common/Config.h> #include <Pegasus/Common/Config.h>
   
 #include <cstring> #include <cstring>
 #include "Monitor.h" #include "Monitor.h"
 #include "MessageQueue.h" #include "MessageQueue.h"
Line 48 
Line 41 
 #include <Pegasus/Common/HTTPConnection.h> #include <Pegasus/Common/HTTPConnection.h>
 #include <Pegasus/Common/MessageQueueService.h> #include <Pegasus/Common/MessageQueueService.h>
 #include <Pegasus/Common/Exception.h> #include <Pegasus/Common/Exception.h>
   #include "ArrayIterator.h"
 #ifdef PEGASUS_OS_TYPE_WINDOWS  #include "HostAddress.h"
 # if defined(FD_SETSIZE) && FD_SETSIZE != 1024  #include <errno.h>
 #  error "FD_SETSIZE was not set to 1024 prior to the last inclusion \  
 of <winsock.h>. It may have been indirectly included (e.g., by including \  
 <windows.h>). Finthe inclusion of that header which is visible to this \  
 compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \  
 otherwise, less than 64 clients (the default) will be able to connect to the \  
 CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."  
   
 # endif  
 # define FD_SETSIZE 1024  
 # include <windows.h>  
 #else  
 # include <sys/types.h>  
 # include <sys/socket.h>  
 # include <sys/time.h>  
 # include <netinet/in.h>  
 # include <netdb.h>  
 # include <arpa/inet.h>  
 #endif  
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
 // Define a platform-neutral socket length type  
 #if defined(PEGASUS_PLATFORM_ZOS_ZSERIES_IBM) || defined(PEGASUS_OS_VMS)  
 typedef size_t PEGASUS_SOCKLEN_T;  
 #elif defined(PEGASUS_PLATFORM_AIX_RS_IBMCXX) || defined(PEGASUS_OS_LINUX) || (defined(PEGASUS_OS_SOLARIS) && !defined(SUNOS_5_6))  
 typedef socklen_t PEGASUS_SOCKLEN_T;  
 #else  
 typedef int PEGASUS_SOCKLEN_T;  
 #endif  
   
 static AtomicInt _connections(0);  
   
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
 // Monitor  // Tickler
 // //
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 #define MAX_NUMBER_OF_MONITOR_ENTRIES  32  Tickler::Tickler()
 Monitor::Monitor()      : _listenSocket(PEGASUS_INVALID_SOCKET),
    : _stopConnections(0),        _clientSocket(PEGASUS_INVALID_SOCKET),
      _stopConnectionsSem(0),        _serverSocket(PEGASUS_INVALID_SOCKET)
      _solicitSocketCount(0),  
      _tickle_client_socket(-1),  
      _tickle_server_socket(-1),  
      _tickle_peer_socket(-1)  
 { {
     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;      try
     Socket::initializeInterface();  
     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);  
   
     // setup the tickler  
     initializeTickler();  
   
     // Start the count at 1 because initilizeTickler()  
     // has added an entry in the first position of the  
     // _entries array  
     for( int i = 1; i < numberOfMonitorEntriesToAllocate; i++ )  
     {     {
        _MonitorEntry entry(0, 0, 0);          _initialize();
        _entries.append(entry);  
     }  
 } }
       catch (...)
 Monitor::~Monitor()  
 { {
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");          _uninitialize();
           throw;
       }
   }
  
     try{  Tickler::~Tickler()
         if(_tickle_peer_socket >= 0)  
         {         {
             Socket::close(_tickle_peer_socket);      _uninitialize();
         }         }
         if(_tickle_client_socket >= 0)  
   #if defined(PEGASUS_OS_TYPE_UNIX) || defined(PEGASUS_OS_VMS)
   
   // Use an anonymous pipe for the tickle connection.
   
   void Tickler::_initialize()
         {         {
             Socket::close(_tickle_client_socket);      int fds[2];
         }  
         if(_tickle_server_socket >= 0)      if (pipe(fds) == -1)
         {         {
             Socket::close(_tickle_server_socket);          MessageLoaderParms parms(
               "Common.Monitor.TICKLE_CREATE",
               "Received error number $0 while creating the internal socket.",
               getSocketError());
           throw Exception(parms);
         }         }
   
       _serverSocket = fds[0];
       _clientSocket = fds[1];
     }     }
     catch(...)  
   #else
   
   // Use an external loopback socket connection to allow the tickle socket to
   // be included in the select() array on Windows.
   
   void Tickler::_initialize()
     {     {
         Tracer::trace(TRC_HTTP, Tracer::LEVEL4,      //
                   "Failed to close tickle sockets");      // Set up the addresses for the listen, client, and server sockets
       // based on whether IPv6 is enabled.
       //
   
       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())
       {
           // Use the IPv6 loopback address for the listen 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)) < 0){      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.",
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              getSocketError());
                                  errno);  
 #else  
                                  WSAGetLastError());  
 #endif  
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // initialize the address      // Bind the listen socket to the loopback address
     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));      if (::bind(
 #ifdef PEGASUS_OS_ZOS              _listenSocket,
     _tickle_server_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");              reinterpret_cast<struct sockaddr*>(&listenAddress),
 #else              addressLength) < 0)
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM      {
 #pragma convert(37)          MessageLoaderParms parms(
 #endif              "Common.Monitor.TICKLE_BIND",
     _tickle_server_addr.sin_addr.s_addr = inet_addr("127.0.0.1");  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM  
 #pragma convert(0)  
 #endif  
 #endif  
     _tickle_server_addr.sin_family = PF_INET;  
     _tickle_server_addr.sin_port = 0;  
   
     PEGASUS_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.",
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              getSocketError());
                                  errno);  
 #else  
                                  WSAGetLastError());  
 #endif  
 #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.",
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              getSocketError());
                                  errno);  
 #else  
                                  WSAGetLastError());  
 #endif  
         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.",
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              getSocketError());
                                  errno);  
 #else  
                                  WSAGetLastError());  
 #endif  
         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)) < 0){      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(
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              "Common.Monitor.TICKLE_CLIENT_CREATE",
                                  errno);              "Received error number $0 while creating the internal client "
 #else                  "socket.",
                                  WSAGetLastError());              getSocketError());
 #endif  
         throw Exception(parms);         throw Exception(parms);
     }     }
  
     // setup the address of the client      // Bind the client socket to the loopback address
     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));      if (::bind(
 #ifdef PEGASUS_OS_ZOS              _clientSocket,
     _tickle_client_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");              reinterpret_cast<struct sockaddr*>(&clientAddress),
 #else              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);
 #endif      }
     _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(
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)              "Common.Monitor.TICKLE_CLIENT_CONNECT",
                                  errno);              "Received error number $0 while connecting the internal client "
 #else                  "socket.",
                                  WSAGetLastError());              getSocketError());
 #endif  
         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.
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)      _serverSocket = ::accept(
                                  errno);          _listenSocket,
 #else          reinterpret_cast<struct sockaddr*>(&serverAddress),
                                  WSAGetLastError());          &tmpAddressLength);
 #endif  
       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
     PEGASUS_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      _listenSocket = PEGASUS_INVALID_SOCKET;
     if((_tickle_peer_socket = ::accept(_tickle_server_socket,  
             reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),      Socket::disableBlocking(_serverSocket);
             &peer_size)) < 0){      Socket::disableBlocking(_clientSocket);
 #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
       {
           if (_serverSocket != PEGASUS_INVALID_SOCKET)
           {
               Socket::close(_serverSocket);
               _serverSocket = PEGASUS_INVALID_SOCKET;
     }     }
     if(_tickle_peer_socket == -1){          if (_clientSocket != PEGASUS_INVALID_SOCKET)
         // handle error          {
         MessageLoaderParms parms("Common.Monitor.TICKLE_ACCEPT",              Socket::close(_clientSocket);
                          "Received error number $0 while accepting the internal socket connection.",              _clientSocket = PEGASUS_INVALID_SOCKET;
 #if !defined(PEGASUS_OS_TYPE_WINDOWS)          }
                                  errno);          if (_listenSocket != PEGASUS_INVALID_SOCKET)
 #else          {
                                  WSAGetLastError());              Socket::close(_listenSocket);
 #endif              _listenSocket = PEGASUS_INVALID_SOCKET;
         throw Exception(parms);          }
       }
       catch (...)
       {
           PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
               "Failed to close tickle sockets");
       }
       Socket::uninitializeInterface();
     }     }
     // 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  
     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);  ////////////////////////////////////////////////////////////////////////////////
   //
   // Monitor
   //
   ////////////////////////////////////////////////////////////////////////////////
   
   #define MAX_NUMBER_OF_MONITOR_ENTRIES  32
   Monitor::Monitor()
      : _stopConnections(0),
        _stopConnectionsSem(0),
        _solicitSocketCount(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.
       _MonitorEntry entry(_tickler.getServerSocket(), 1, INTERNAL);
     entry._status = _MonitorEntry::IDLE;     entry._status = _MonitorEntry::IDLE;
     _entries.append(entry);     _entries.append(entry);
 }  
  
 void Monitor::tickle(void)      // Start the count at 1 because _entries[0] is the Tickler
       for (int i = 1; i < numberOfMonitorEntriesToAllocate; i++)
 { {
     static char _buffer[] =         _MonitorEntry entry(0, 0, 0);
          _entries.append(entry);
       }
   }
   
   Monitor::~Monitor()
     {     {
       '0','0'      PEG_TRACE_CSTRING(TRC_HTTP, Tracer::LEVEL4,
     };                    "returning from monitor destructor");
   }
  
     AutoMutex autoMutex(_tickle_mutex);  void Monitor::tickle()
     Socket::disableBlocking(_tickle_client_socket);  {
     Socket::write(_tickle_client_socket,&_buffer, 2);      AutoMutex autoMutex(_tickleMutex);
     Socket::enableBlocking(_tickle_client_socket);      Socket::write(_tickler.getClientSocket(), "\0\0", 2);
 } }
  
 void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )  void Monitor::setState(
       Uint32 index,
       _MonitorEntry::entry_status status)
 { {
       AutoMutex autoEntryMutex(_entry_mut);
     // Set the state to requested state     // Set the state to requested state
     _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;  
   
     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};
  
     fd_set fdread;     fd_set fdread;
Line 371 
Line 382 
  
     AutoMutex autoEntryMutex(_entry_mut);     AutoMutex autoEntryMutex(_entry_mut);
  
     // Check the stopConnections flag.  If set, clear the Acceptor monitor entries      ArrayIterator<_MonitorEntry> entries(_entries);
     if (_stopConnections.value() == 1)  
       // Check the stopConnections flag.  If set, clear the Acceptor monitor
       // entries
       if (_stopConnections.get() == 1)
     {     {
         for ( int indx = 0; indx < (int)_entries.size(); indx++)          for ( int indx = 0; indx < (int)entries.size(); indx++)
         {         {
             if (_entries[indx]._type == Monitor::ACCEPTOR)              if (entries[indx]._type == Monitor::ACCEPTOR)
             {             {
                 if ( _entries[indx]._status.value() != _MonitorEntry::EMPTY)                  if ( entries[indx]._status.get() != _MonitorEntry::EMPTY)
                 {                 {
                    if ( _entries[indx]._status.value() == _MonitorEntry::IDLE ||                     if ( entries[indx]._status.get() == _MonitorEntry::IDLE ||
                         _entries[indx]._status.value() == _MonitorEntry::DYING )                          entries[indx]._status.get() == _MonitorEntry::DYING )
                    {                    {
                        // remove the entry                        // remove the entry
                        _entries[indx]._status = _MonitorEntry::EMPTY;                         entries[indx]._status = _MonitorEntry::EMPTY;
                    }                    }
                    else                    else
                    {                    {
                        // set status to DYING                        // set status to DYING
                       _entries[indx]._status = _MonitorEntry::DYING;                        entries[indx]._status = _MonitorEntry::DYING;
                    }                    }
                }                }
            }            }
Line 398 
Line 412 
         _stopConnectionsSem.signal();         _stopConnectionsSem.signal();
     }     }
  
     for( int indx = 0; indx < (int)_entries.size(); indx++)      for (int indx = 0; indx < (int)entries.size(); indx++)
     {     {
                          const _MonitorEntry &entry = _entries[indx];          const _MonitorEntry &entry = entries[indx];
        if ((entry._status.value() == _MonitorEntry::DYING) &&          if ((entry._status.get() == _MonitorEntry::DYING) &&
                                          (entry._type == Monitor::CONNECTION))                                          (entry._type == Monitor::CONNECTION))
        {        {
           MessageQueue *q = MessageQueue::lookup(entry.queueId);           MessageQueue *q = MessageQueue::lookup(entry.queueId);
Line 419 
Line 433 
  
                                         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 437 
Line 451 
           // Once HTTPAcceptor completes processing of the close           // Once HTTPAcceptor completes processing of the close
           // connection, the lock is re-requested and processing of           // connection, the lock is re-requested and processing of
           // the for loop continues.  This is safe with the current           // the for loop continues.  This is safe with the current
           // implementation of the _entries object.  Note that the              // implementation of the entries object.  Note that the
           // loop condition accesses the _entries.size() on each              // loop condition accesses the entries.size() on each
           // iteration, so that a change in size while the mutex is           // iteration, so that a change in size while the mutex is
           // 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.
               entries.reset(_entries);
        }        }
     }     }
  
Line 457 
Line 477 
         place to calculate the max file descriptor (maximum socket number)         place to calculate the max file descriptor (maximum socket number)
         because we have to traverse the entire array.         because we have to traverse the entire array.
     */     */
     int maxSocketCurrentPass = 0;      SocketHandle maxSocketCurrentPass = 0;
     for( int indx = 0; indx < (int)_entries.size(); indx++)      for (int indx = 0; indx < (int)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.value() == _MonitorEntry::IDLE)         if (entries[indx]._status.get() == _MonitorEntry::IDLE)
        {        {
           _idleEntries++;           _idleEntries++;
           FD_SET(_entries[indx].socket, &fdread);             FD_SET(entries[indx].socket, &fdread);
        }        }
     }     }
  
Line 476 
Line 496 
     */     */
     maxSocketCurrentPass++;     maxSocketCurrentPass++;
  
     autoEntryMutex.unlock();      _entry_mut.unlock();
     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);  
     autoEntryMutex.lock();  
  
       //
       // The first argument to select() is ignored on Windows and it is not
       // a socket value.  The original code assumed that the number of sockets
       // and a socket value have the same type.  On Windows they do not.
       //
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
     if(events == SOCKET_ERROR)      int events = select(0, &fdread, NULL, NULL, &tv);
 #else #else
     if(events == -1)      int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
 #endif #endif
       _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
       entries.reset(_entries);
   
       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
        // we have a synchronization error.        // we have a synchronization error.
  
        PEGASUS_ASSERT(errno != EBADF);        PEGASUS_ASSERT(errno != 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 (int indx = 0; indx < (int)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.value() == _MonitorEntry::IDLE) &&              // IDLE (i.e., owned by the Monitor).
              (FD_ISSET(_entries[indx].socket, &fdread)))              if ((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
                   (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 "
                                   "Monitor::CONNECTION",
                               indx));
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                    static_cast<HTTPConnection *>(q)->_entry_index = indx;
  
                    // Do not update the entry just yet. The entry gets updated once                          // Do not update the entry just yet. The entry gets
                    // the request has been read.                          // updated once the request has been read.
                    //_entries[indx]._status = _MonitorEntry::BUSY;                          //entries[indx]._status = _MonitorEntry::BUSY;
  
                    // If allocate_and_awaken failure, retry on next iteration                          // If allocate_and_awaken failure, retry on next
                           // iteration
 /* Removed for PEP 183. /* Removed for PEP 183.
                    if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(                          if (!MessageQueueService::get_thread_pool()->
                            (void *)q, _dispatch))                                  allocate_and_awaken((void *)q, _dispatch))
                    {                    {
                       Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,                              PEG_TRACE_CSTRING(TRC_DISCARDED_DATA,
                           "Monitor::run: Insufficient resources to process request.");                                  Tracer::LEVEL2,
                       _entries[indx]._status = _MonitorEntry::IDLE;                                  "Monitor::run: Insufficient resources to "
                                       "process request.");
                               entries[indx]._status = _MonitorEntry::IDLE;
                       return true;                       return true;
                    }                    }
 */ */
 // Added for PEP 183 // Added for PEP 183
                    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);                          HTTPConnection *dst =
                          Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                              reinterpret_cast<HTTPConnection *>(q);
                          "Monitor::_dispatch: entering run() for indx  = %d, queueId = %d, q = %p",                          PEG_TRACE((TRC_HTTP, Tracer::LEVEL4,
                    dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);                              "Monitor::_dispatch: entering run() for "
                                   "indx = %d, queueId = %d, 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));
                    // It is possible the entry status may not be set to busy.  
                    // The following will fail in that case.                          // It is possible the entry status may not be set to
                    // PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);                          // busy.  The following will fail in that case.
                    // Once the HTTPConnection thread has set the status value to either                          // PEGASUS_ASSERT(dst->_monitor->_entries[
                    // Monitor::DYING or Monitor::IDLE, it has returned control of the connection                          //     dst->_entry_index]._status.get() ==
                    // to the Monitor.  It is no longer permissible to access the connection                          //    _MonitorEntry::BUSY);
                    // or the entry in the _entries table.                          // Once the HTTPConnection thread has set the status
                           // value to either Monitor::DYING or Monitor::IDLE,
                    // The following is not relevant as the worker thread or the                          // it has returned control of the connection to the
                    // reader thread will update the status of the entry.                          // 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)                    //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;
                    //}                    //}
 // end Added for PEP 183 // 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.value() == _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);  
                         _entries[indx]._status.value() == _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]._status = _MonitorEntry::BUSY;                              entries[indx].socket, events);
                    autoEntryMutex.unlock();                          entries[indx]._status = _MonitorEntry::BUSY;
                           _entry_mut.unlock();
                    q->enqueue(msg);                    q->enqueue(msg);
                    autoEntryMutex.lock();                          _entry_mut.lock();
                    _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 be
                           // reset with the original _entries
                           entries.reset(_entries);
                           entries[indx]._status = _MonitorEntry::IDLE;
                 }                 }
              }              }
              catch(...)              catch(...)
              {              {
              }              }
              handled_events = true;  
           }           }
        }        }
     }     }
   
     return(handled_events);  
 } }
  
 void Monitor::stopListeningForConnections(Boolean wait) void Monitor::stopListeningForConnections(Boolean wait)
Line 624 
Line 678 
       // Wait for the monitor to notice _stopConnections.  Otherwise the       // Wait for the monitor to notice _stopConnections.  Otherwise the
       // caller of this function may unbind the ports while the monitor       // caller of this function may unbind the ports while the monitor
       // is still accepting connections on them.       // is still accepting connections on them.
       try        _stopConnectionsSem.wait();
         {  
           _stopConnectionsSem.time_wait(10000);  
         }  
       catch (TimeOut &)  
         {  
           // The monitor is probably busy processng a very long request, and is  
           // not accepting connections.  Let the caller unbind the ports.  
         }  
     }     }
  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
Line 640 
Line 686 
  
  
 int  Monitor::solicitSocketMessages( int  Monitor::solicitSocketMessages(
     Sint32 socket,      SocketHandle socket,
     Uint32 events,     Uint32 events,
     Uint32 queueId,     Uint32 queueId,
     int type)     int type)
Line 652 
Line 698 
    // 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 664 
Line 712 
    {    {
       try       try
       {       {
          if(_entries[index]._status.value() == _MonitorEntry::EMPTY)              if (_entries[index]._status.get() == _MonitorEntry::EMPTY)
          {          {
             _entries[index].socket = socket;             _entries[index].socket = socket;
             _entries[index].queueId  = queueId;             _entries[index].queueId  = queueId;
Line 678 
Line 726 
       {       {
       }       }
    }    }
    _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(Sint32 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 700 
Line 747 
        if(_entries[index].socket == socket)        if(_entries[index].socket == socket)
        {        {
           _entries[index]._status = _MonitorEntry::EMPTY;           _entries[index]._status = _MonitorEntry::EMPTY;
           _entries[index].socket = -1;              _entries[index].socket = PEGASUS_INVALID_SOCKET;
           _solicitSocketCount--;           _solicitSocketCount--;
           break;           break;
        }        }
Line 708 
Line 755 
  
     /*     /*
         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.value() == _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 722 
Line 771 
 } }
  
 // 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.value() == _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.94.8.1  
changed lines
  Added in v.1.121.4.1
No CVS admin address has been configured
 Powered by
ViewCVS 0.9.2