(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.77 and 1.103.10.27

version 1.77, 2004/10/01 18:25:00 version 1.103.10.27, 2006/10/18 04:24:42
Line 1 
Line 1 
 //%2003////////////////////////////////////////////////////////////////////////  //%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.
 // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L. P.; // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L. P.;
 // IBM Corp.; EMC Corporation, The Open Group. // IBM Corp.; EMC Corporation, The Open Group.
   // Copyright (c) 2004 BMC Software; Hewlett-Packard Development Company, L.P.;
   // IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.
   // Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.;
   // 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 23 
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  
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
 #include <Pegasus/Common/Config.h> #include <Pegasus/Common/Config.h>
  
 #include <cstring> #include <cstring>
 #include "Monitor.h"  #include <Pegasus/Common/Monitor.h>
 #include "MessageQueue.h"  #include <Pegasus/Common/MessageQueue.h>
 #include "Socket.h"  #include <Pegasus/Common/Socket.h>
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
 #include <Pegasus/Common/HTTPConnection.h> #include <Pegasus/Common/HTTPConnection.h>
 #include <Pegasus/Common/MessageQueueService.h> #include <Pegasus/Common/MessageQueueService.h>
 #include <Pegasus/Common/Exception.h> #include <Pegasus/Common/Exception.h>
   #include <Pegasus/Common/ArrayIterator.h>
   
   //const static DWORD MAX_BUFFER_SIZE = 4096;  // 4 kilobytes
   
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
   // Maximum iterations of Pipe processing in Monitor::run
   const Uint32 maxIterations = 3;
   
   #endif
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
 # if defined(FD_SETSIZE) && FD_SETSIZE != 1024 # if defined(FD_SETSIZE) && FD_SETSIZE != 1024
 #  error "FD_SETSIZE was not set to 1024 prior to the last inclusion \ #  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 \ 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 \  <windows.h>). Find inclusion of that header which is visible to this \
 compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \ 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 \ 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." CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."
Line 67 
Line 77 
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
   static AtomicInt _connections(0);
  
 static AtomicInt _connections = 0;  Mutex Monitor::_cout_mut;
   
 static struct timeval create_time = {0, 1};  
 static struct timeval destroy_time = {300, 0};  
 static struct timeval deadlock_time = {0, 0};  
  
 ////////////////////////////////////////////////////////////////////////////////  
 //  
 // MonitorRep  
 //  
 ////////////////////////////////////////////////////////////////////////////////  
  
 struct MonitorRep  // Added for NamedPipe implementation for windows
 {  #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
     fd_set rd_fd_set;   #define PIPE_INCREMENT 1
     fd_set wr_fd_set;  #endif
     fd_set ex_fd_set;  
     fd_set active_rd_fd_set;  
     fd_set active_wr_fd_set;  
     fd_set active_ex_fd_set;  
 };  
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
Line 97 
Line 94 
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 #define MAX_NUMBER_OF_MONITOR_ENTRIES  32 #define MAX_NUMBER_OF_MONITOR_ENTRIES  32
   
 Monitor::Monitor() Monitor::Monitor()
    : _module_handle(0),     : _stopConnections(0),
      _controller(0),  
      _async(false),  
      _stopConnections(0),  
      _stopConnectionsSem(0),      _stopConnectionsSem(0),
      _solicitSocketCount(0),      _solicitSocketCount(0),
      _tickle_client_socket(-1),      _tickle_client_socket(-1),
      _tickle_server_socket(-1),      _tickle_server_socket(-1),
      _tickle_peer_socket(-1)      _tickle_peer_socket(-1)
 { {
     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
     Socket::initializeInterface();  
     _rep = 0;  
     _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);          AutoMutex automut(Monitor::_cout_mut);
        _entries.append(entry);          PEGASUS_STD(cout) << "Entering: Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
     }  
 } }
   #endif
 Monitor::Monitor(Boolean async)  
    : _module_handle(0),  
      _controller(0),  
      _async(async),  
      _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;     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;
     Socket::initializeInterface();     Socket::initializeInterface();
     _rep = 0;  
     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);
  
     // setup the tickler     // setup the tickler
Line 153 
Line 124 
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
     }     }
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
 } }
  
 Monitor::~Monitor() Monitor::~Monitor()
 { {
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                   "deregistering with module controller");  
   
     if(_module_handle != NULL)  
     {     {
        _controller->deregister_module(PEGASUS_MODULENAME_MONITOR);          AutoMutex automut(Monitor::_cout_mut);
        _controller = 0;          PEGASUS_STD(cout) << "Entering: Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
        delete _module_handle;  
     }     }
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");  #endif
   
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
  
     try{     try{
Line 193 
Line 165 
     Socket::uninitializeInterface();     Socket::uninitializeInterface();
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                   "returning from monitor destructor");                   "returning from monitor destructor");
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
 } }
  
 void Monitor::initializeTickler(){ void Monitor::initializeTickler(){
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
     /*     /*
        NOTE: On any errors trying to        NOTE: On any errors trying to
              setup out tickle connection,              setup out tickle connection,
Line 205 
Line 189 
     /* setup the tickle server/listener */     /* setup the tickle server/listener */
  
     // get a socket for the server side     // get a socket for the server side
     if((_tickle_server_socket = ::socket(PF_INET, SOCK_STREAM, 0)) < 0){      if((_tickle_server_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){
         //handle error         //handle error
         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.",
Line 219 
Line 203 
  
     // initialize the address     // initialize the address
     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));     memset(&_tickle_server_addr, 0, sizeof(_tickle_server_addr));
 #ifdef PEGASUS_OS_ZOS  
     _tickle_server_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");  
 #else  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
 #pragma convert(37) #pragma convert(37)
 #endif #endif
Line 229 
Line 210 
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
 #pragma convert(0) #pragma convert(0)
 #endif #endif
 #endif  
     _tickle_server_addr.sin_family = PF_INET;     _tickle_server_addr.sin_family = PF_INET;
     _tickle_server_addr.sin_port = 0;     _tickle_server_addr.sin_port = 0;
  
     PEGASUS_SOCKLEN_SIZE _addr_size = sizeof(_tickle_server_addr);      PEGASUS_SOCKLEN_T _addr_size = sizeof(_tickle_server_addr);
  
     // bind server side to socket     // bind server side to socket
     if((::bind(_tickle_server_socket,     if((::bind(_tickle_server_socket,
                (struct sockaddr *)&_tickle_server_addr,                 reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                sizeof(_tickle_server_addr))) < 0){                sizeof(_tickle_server_addr))) < 0){
         // handle error         // 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",         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) #if !defined(PEGASUS_OS_TYPE_WINDOWS)
Line 247 
Line 231 
 #else #else
                                  WSAGetLastError());                                  WSAGetLastError());
 #endif #endif
   #endif
         throw Exception(parms);         throw Exception(parms);
     }     }
  
Line 265 
Line 250 
  
     // make sure we have the correct socket for our server     // make sure we have the correct socket for our server
     int sock = ::getsockname(_tickle_server_socket,     int sock = ::getsockname(_tickle_server_socket,
                              (struct sockaddr*)&_tickle_server_addr,                     reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                              &_addr_size);                              &_addr_size);
     if(sock < 0){     if(sock < 0){
         // handle error         // handle error
Line 282 
Line 267 
     /* set up the tickle client/connector */     /* set up the tickle client/connector */
  
     // get a socket for our tickle client     // get a socket for our tickle client
     if((_tickle_client_socket = ::socket(PF_INET, SOCK_STREAM, 0)) < 0){      if((_tickle_client_socket = ::socket(PF_INET, SOCK_STREAM, 0)) == PEGASUS_INVALID_SOCKET){
         // handle error         // handle error
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CREATE",
                          "Received error number $0 while creating the internal client socket.",                          "Received error number $0 while creating the internal client socket.",
Line 296 
Line 281 
  
     // setup the address of the client     // setup the address of the client
     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));     memset(&_tickle_client_addr, 0, sizeof(_tickle_client_addr));
 #ifdef PEGASUS_OS_ZOS  
     _tickle_client_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");  
 #else  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
 #pragma convert(37) #pragma convert(37)
 #endif #endif
Line 306 
Line 288 
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM
 #pragma convert(0) #pragma convert(0)
 #endif #endif
 #endif  
     _tickle_client_addr.sin_family = PF_INET;     _tickle_client_addr.sin_family = PF_INET;
     _tickle_client_addr.sin_port = 0;     _tickle_client_addr.sin_port = 0;
  
     // bind socket to client side     // bind socket to client side
     if((::bind(_tickle_client_socket,     if((::bind(_tickle_client_socket,
                (struct sockaddr*)&_tickle_client_addr,                 reinterpret_cast<struct sockaddr*>(&_tickle_client_addr),
                sizeof(_tickle_client_addr))) < 0){                sizeof(_tickle_client_addr))) < 0){
         // handle error         // handle error
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_BIND",
Line 327 
Line 308 
  
     // connect to server side     // connect to server side
     if((::connect(_tickle_client_socket,     if((::connect(_tickle_client_socket,
                   (struct sockaddr*)&_tickle_server_addr,                    reinterpret_cast<struct sockaddr*>(&_tickle_server_addr),
                   sizeof(_tickle_server_addr))) < 0){                   sizeof(_tickle_server_addr))) < 0){
         // handle error         // handle error
         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",         MessageLoaderParms parms("Common.Monitor.TICKLE_CLIENT_CONNECT",
Line 342 
Line 323 
  
     /* set up the slave connection */     /* set up the slave connection */
     memset(&_tickle_peer_addr, 0, sizeof(_tickle_peer_addr));     memset(&_tickle_peer_addr, 0, sizeof(_tickle_peer_addr));
     PEGASUS_SOCKLEN_SIZE peer_size = sizeof(_tickle_peer_addr);      PEGASUS_SOCKLEN_T peer_size = sizeof(_tickle_peer_addr);
     pegasus_sleep(1);     pegasus_sleep(1);
  
     // this call may fail, we will try a max of 20 times to establish this peer connection     // this call may fail, we will try a max of 20 times to establish this peer connection
     if((_tickle_peer_socket = ::accept(_tickle_server_socket,     if((_tickle_peer_socket = ::accept(_tickle_server_socket,
                                        (struct sockaddr*)&_tickle_peer_addr,              reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),
                                        &peer_size)) < 0){                                        &peer_size)) < 0){
 #if !defined(PEGASUS_OS_TYPE_WINDOWS) #if !defined(PEGASUS_OS_TYPE_WINDOWS)
         // Only retry on non-windows platforms.         // Only retry on non-windows platforms.
Line 358 
Line 339 
           {           {
             pegasus_sleep(1);             pegasus_sleep(1);
             _tickle_peer_socket = ::accept(_tickle_server_socket,             _tickle_peer_socket = ::accept(_tickle_server_socket,
                                            (struct sockaddr*)&_tickle_peer_addr,                  reinterpret_cast<struct sockaddr*>(&_tickle_peer_addr),
                                            &peer_size);                                            &peer_size);
             retries++;             retries++;
           } while(_tickle_peer_socket == -1 && errno == EAGAIN && retries < 20);           } while(_tickle_peer_socket == -1 && errno == EAGAIN && retries < 20);
Line 379 
Line 360 
     // add the tickler to the list of entries to be monitored and set to IDLE because Monitor only     // add the tickler to the list of entries to be monitored and set to IDLE because Monitor only
     // checks entries with IDLE state for events     // checks entries with IDLE state for events
     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);
           Tracer::trace(TRC_HTTP,Tracer::LEVEL2,"!!!!!!!! TICKLE SOCKET-ID = %u",_tickle_peer_socket);
     entry._status = _MonitorEntry::IDLE;     entry._status = _MonitorEntry::IDLE;
     _entries.append(entry);     _entries.append(entry);
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
 } }
  
 void Monitor::tickle(void) void Monitor::tickle(void)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
   static char _buffer[] =   static char _buffer[] =
     {     {
       '0','0'       '0','0'
     };     };
                  Tracer::trace (TRC_HTTP, Tracer::LEVEL2,
                                      "Now Monitor::Tickle ");
       AutoMutex autoMutex(_tickle_mutex);
   Socket::disableBlocking(_tickle_client_socket);   Socket::disableBlocking(_tickle_client_socket);
                          Tracer::trace (TRC_HTTP, Tracer::LEVEL2,
                                              "Now Monitor::Tickle::Write() ");
   
   Socket::write(_tickle_client_socket,&_buffer, 2);   Socket::write(_tickle_client_socket,&_buffer, 2);
   Socket::enableBlocking(_tickle_client_socket);   Socket::enableBlocking(_tickle_client_socket);
                          Tracer::trace (TRC_HTTP, Tracer::LEVEL2,
                                      "Now Monitor::Tickled ");
   
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
   }
   
   void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )
   {
       // Set the state to requested state
       _entries[index]._status = status;
 } }
  
 Boolean Monitor::run(Uint32 milliseconds) Boolean Monitor::run(Uint32 milliseconds)
Line 401 
Line 415 
     Boolean handled_events = false;     Boolean handled_events = false;
     int i = 0;     int i = 0;
  
     struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};  
  
     fd_set fdread;     fd_set fdread;
     FD_ZERO(&fdread);     FD_ZERO(&fdread);
  
     _entry_mut.lock(pegasus_thread_self());      AutoMutex autoEntryMutex(_entry_mut);
   
       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 == 1)      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 435 
Line 450 
         _stopConnectionsSem.signal();         _stopConnectionsSem.signal();
     }     }
  
     for( int indx = 0; indx < (int)_entries.size(); indx++)      for( int indx = 0; indx < (int)entries.size(); indx++)
     {     {
        if ((_entries[indx]._status.value() == _MonitorEntry::DYING) &&          const _MonitorEntry &entry = entries[indx];
                 (_entries[indx]._type == Monitor::CONNECTION))          if ((entry._status.get() == _MonitorEntry::DYING) &&
              (entry._type == Monitor::CONNECTION))
        {        {
           MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);  
               MessageQueue *q = MessageQueue::lookup(entry.queueId);
           PEGASUS_ASSERT(q != 0);           PEGASUS_ASSERT(q != 0);
           MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();              HTTPConnection &h = *static_cast<HTTPConnection *>(q);
           Message* message= new CloseConnectionMessage(_entries[indx].socket);  
               if (h._connectionClosePending == false)
                           {
                               continue;
                           }
   
   
               // NOTE: do not attempt to delete while there are pending responses
               // coming thru. The last response to come thru after a
               // _connectionClosePending will reset _responsePending to false
               // and then cause the monitor to rerun this code and clean up.
               // (see HTTPConnection.cpp)
   
               if (h._responsePending == true)
               {
   // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
                   if  (!entry.namedPipeConnection)
                   {
   #endif
                       Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "
                           "Ignoring connection delete request because "
                           "responses are still pending. "
                           "connection=0x%p, socket=%d\n",
                           (void *)&h, h.getSocket());
   
   // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
                   }
                   else
                   {
                       Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "
                           "Ignoring connection delete request because "
                           "responses are still pending. "
                           "connection=0x%p, NamedPipe=%d\n",
                           (void *)&h, h.getNamedPipe().getPipe());
                   }
   #endif
                   continue;
               }
               h._connectionClosePending = false;
               MessageQueue &o = h.get_owner();
                       Message* message = 0;
   
   // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
               if (!entry.namedPipeConnection)
               {
   #endif
                   message= new CloseConnectionMessage(entry.socket);
   
   // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
               }
               else
               {
   
                               message= new CloseConnectionMessage(entry.namedPipe);
   
               }
   #endif
           message->dest = o.getQueueId();           message->dest = o.getQueueId();
  
           // HTTPAcceptor is responsible for closing the connection.           // HTTPAcceptor is responsible for closing the connection.
Line 452 
Line 529 
           // 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.
  
           _entry_mut.unlock();              autoEntryMutex.unlock();
           o.enqueue(message);           o.enqueue(message);
           _entry_mut.lock(pegasus_thread_self());              autoEntryMutex.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 472 
Line 553 
         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;  
     for( int indx = 0; indx < (int)_entries.size(); indx++)      PEGASUS_SOCKET maxSocketCurrentPass = 0;
       int indx;
   
           // Record the indexes at which Sockets are available
           Array <Uint32> socketCountAssociator;
       int socketEntryCount=0;
   
        // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
       //This array associates named pipe connections to their place in [indx]
       //in the entries array. The value in portion zero of the array is the
       //index of the fist named pipe connection in the entries array
   
           // Record the indexes at which Pipes are available
           Array <Uint32> indexPipeCountAssociator;
       int pipeEntryCount=0;
       int MaxPipes = PIPE_INCREMENT;
       // List of Pipe Handlers
       HANDLE * hPipeList = new HANDLE[PIPE_INCREMENT];
   #endif
   
       // This loop takes care of setting the namedpipe which has to be used from the list....
       for ( indx = 0,socketEntryCount=0 ;
                                indx < (int)entries.size(); indx++)
     {     {
        if(maxSocketCurrentPass < _entries[indx].socket)  
           maxSocketCurrentPass = _entries[indx].socket;  
  
        if(_entries[indx]._status.value() == _MonitorEntry::IDLE)  // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
                   if (!entries[indx].namedPipeConnection)
           {
   #endif
               if (maxSocketCurrentPass < entries[indx].socket)
                           {
                                   maxSocketCurrentPass = entries[indx].socket;
                           }
               if(entries[indx]._status.get() == _MonitorEntry::IDLE)
        {        {
           _idleEntries++;           _idleEntries++;
           FD_SET(_entries[indx].socket, &fdread);                  FD_SET(entries[indx].socket, &fdread);
                   socketCountAssociator.append(indx);
                                   socketEntryCount++;
               }
   
   // Added for NamedPipe implementation for windows
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
           }
                   else
                   {
                       entries[indx].pipeSet = false;
                           if (pipeEntryCount >= MaxPipes)
                           {
                               MaxPipes += PIPE_INCREMENT;
                                   HANDLE* temp_pList = new HANDLE[MaxPipes];
                                   for (Uint32 i =0;i<pipeEntryCount;i++)
                                   {
                                       temp_pList[i] = hPipeList[i];
                                   }
                                   delete [] hPipeList;
                                   hPipeList = temp_pList;
        }        }
                           hPipeList[pipeEntryCount] = entries[indx].namedPipe.getPipe();
                           indexPipeCountAssociator.append(indx);
                           pipeEntryCount++;
                   }
   
   #endif
     }     }
  
     /*     /*
Line 491 
Line 628 
     */     */
     maxSocketCurrentPass++;     maxSocketCurrentPass++;
  
     _entry_mut.unlock();      autoEntryMutex.unlock();
     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);  
    _entry_mut.lock(pegasus_thread_self());      int events = -1;
           // Since the pipes have been introduced, the ratio of procesing
           // time Socket:Pipe :: 3/4:1/4 respectively
   
           Uint32 newMilliseconds = milliseconds;
           #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
   
           newMilliseconds = (milliseconds * 3)/4 ;
   
       #endif
   
           struct timeval tv = {newMilliseconds/1000, newMilliseconds%1000*1000};
   
   
           #ifdef PEGASUS_OS_TYPE_WINDOWS
                   events = select(0, &fdread, NULL, NULL, &tv);
           #else
                   events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
           #endif
   
       autoEntryMutex.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);
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
     if(events == SOCKET_ERROR)     if(events == SOCKET_ERROR)
Line 505 
Line 665 
           "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);
Line 515 
Line 675 
        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,        Tracer::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 idle entries",
            events, _idleEntries);            events, _idleEntries);
        for( int indx = 0; indx < (int)_entries.size(); indx++)           for ( int sindx = 0; sindx < socketEntryCount; sindx++)
        {        {
           // The Monitor should only look at entries in the table that are IDLE (i.e.,           // The Monitor should only look at entries in the table that are IDLE (i.e.,
           // owned by the Monitor).           // owned by the Monitor).
           if((_entries[indx]._status.value() == _MonitorEntry::IDLE) &&                       indx = socketCountAssociator[sindx];
              (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,              Tracer::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,  
                      "_entries[indx].type for indx = %d is Monitor::CONNECTION", indx);  
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                    static_cast<HTTPConnection *>(q)->_entry_index = indx;
                    _entries[indx]._status = _MonitorEntry::BUSY;  
                                                    // Do not update the entry just yet. The entry gets updated once
                                                    // the request has been read.
                                                    //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.  
                    if (!MessageQueueService::get_thread_pool()->allocate_and_awaken(  
                            (void *)q, _dispatch))  
                    {  
                       Tracer::trace(TRC_DISCARDED_DATA, Tracer::LEVEL2,  
                           "Monitor::run: Insufficient resources to process request.");  
                       _entries[indx]._status = _MonitorEntry::IDLE;  
                       _entry_mut.unlock();  
                       return true;  
                    }  
 */  
 // Added for PEP 183  
                    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);                    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(q);
                          Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                          Tracer::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, q = %p",
Line 563 
Line 716 
                         "Monitor::_dispatch: exception received");                         "Monitor::_dispatch: exception received");
                    }                    }
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                    "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);                              "Monitor::_dispatch: exited run() for index %d",
                                                       dst->_entry_index);
  
                    PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.value() == _MonitorEntry::BUSY);  
                    // Once the HTTPConnection thread has set the status value to either  
                    // Monitor::DYING or Monitor::IDLE, it has returned control of the connection  
                    // to the Monitor.  It is no longer permissible to access the connection  
                    // or the entry in the _entries table.  
                    if (dst->_connectionClosePending)  
                    {  
                      dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;  
                    }                    }
                    else                                          else if (entries[indx]._type == Monitor::INTERNAL)
                    {                    {
                      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,                         // set ourself to BUSY,
                         // read the data                         // read the data
                         // and set ourself back to IDLE                         // and set ourself back to IDLE
   
                         _entries[indx]._status == _MonitorEntry::BUSY;  
                         static char buffer[2];                         static char buffer[2];
                         Socket::disableBlocking(_entries[indx].socket);                                          Socket::disableBlocking(entries[indx].socket);
                         Sint32 amt = Socket::read(_entries[indx].socket,&buffer, 2);  
                         Socket::enableBlocking(_entries[indx].socket);                                          Sint32 amt = Socket::read(entries[indx].socket,&buffer, 2);
                         _entries[indx]._status == _MonitorEntry::IDLE;                                                  Socket::enableBlocking(entries[indx].socket);
                                                   entries[indx]._status = _MonitorEntry::IDLE;
                 }                 }
                 else                 else
                 {                 {
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                      "Non-connection entry, indx = %d, has been received.", indx);                      "Non-connection entry, indx = %d, has been received.", indx);
   
                    int events = 0;                    int events = 0;
                    events |= SocketMessage::READ;                    events |= SocketMessage::READ;
                    Message *msg = new SocketMessage(_entries[indx].socket, events);                                                  Message *msg = new SocketMessage(entries[indx].socket, events);
                    _entries[indx]._status = _MonitorEntry::BUSY;                                                  entries[indx]._status = _MonitorEntry::BUSY;
                    _entry_mut.unlock();                                                  autoEntryMutex.unlock();
   
                    q->enqueue(msg);                    q->enqueue(msg);
                    _entries[indx]._status = _MonitorEntry::IDLE;                                                  autoEntryMutex.lock();
                    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;
                                                   handled_events = true;
                                                   delete [] hPipeList;
                                                   return handled_events;
   
                 }                 }
              }              }
              catch(...)              catch(...)
Line 613 
Line 760 
              handled_events = true;              handled_events = true;
           }           }
        }        }
     }                  delete [] hPipeList;
     _entry_mut.unlock();                  return handled_events;
     return(handled_events);  
 } }
  
 void Monitor::stopListeningForConnections(Boolean wait)  
   #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
   
           //if no pipes are registered return immediately
   
           int pEvents = -1;
           int pCount = -1;
           BOOL bPeekPipe = 0;
           DWORD dwBytesAvail=0;
           // The pipe is sniffed and check if there are any data. If available, the
           // message is picked from the Queue and appropriate methods are invoked.
   
   
           // pipeProcessCount records the number of requests that are processed.
           // At the end of loop this is verified against the count of request
           // on local connection . If there are any pipes which needs to be
           // processed we would apply delay and then proceed to iterate.
   
       Uint32 pipeProcessCount =0;
   
       for (int counter = 1; counter < maxIterations ; counter ++)
 { {
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");  
     // set boolean then tickle the server to recognize _stopConnections  
     _stopConnections = 1;  
     tickle();  
  
     if (wait)  
                   // pipeIndex is used to index into indexPipeCountAssociator to fetch
                   // index of the _MonitorEntry of Monitor
           for (int pipeIndex = 0; pipeIndex < pipeEntryCount; pipeIndex++)
     {     {
       // Wait for the monitor to notice _stopConnections.  Otherwise the              dwBytesAvail = 0;
       // caller of this function may unbind the ports while the monitor                      bPeekPipe = ::PeekNamedPipe(hPipeList[pipeIndex],
       // is still accepting connections on them.                                                      NULL,
                                                                       NULL,
                                                                           NULL,
                                           &dwBytesAvail,
                                                                           NULL
                                                                          );
   
                           // If peek on NamedPipe was successfull and data is available
               if (bPeekPipe && dwBytesAvail)
                   {
   
                               Tracer::trace(TRC_HTTP,Tracer::LEVEL4," PIPE_PEEKING FOUND = %u BYTES", dwBytesAvail);
   
                               pEvents = 1;
                       entries[indexPipeCountAssociator[pipeIndex]].pipeSet = true;
                               Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                       "Monitor::run select event received events = %d, \
                                           monitoring %d idle entries",
                       pEvents,
                                           _idleEntries);
   
                                   int pIndx = indexPipeCountAssociator[pipeIndex];
   
                                   if ((entries[pIndx]._status.get() == _MonitorEntry::IDLE) &&
                                            entries[pIndx].namedPipe.isConnected() &&
                                            (pEvents))
                           {
   
                                   MessageQueue *q = 0;
   
       try       try
         {         {
           _stopConnectionsSem.time_wait(10000);  
                                           q = MessageQueue::lookup (entries[pIndx].queueId);
         }         }
       catch (TimeOut &)                      catch (Exception e)
         {         {
           // The monitor is probably busy processng a very long request, and is                                          e.getMessage();
           // not accepting connections.  Let the caller unbind the ports.  
         }         }
                       catch(...)
                       {
     }     }
  
     PEG_METHOD_EXIT();                                          Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
 }                                    "Monitor::run indx = %d, queueId =  %d,\
                                                                     q = %p",pIndx, entries[pIndx].queueId, q);
                       try
                       {
                                           if (entries[pIndx]._type == Monitor::CONNECTION)
                           {
  
                                                       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                                                                         "entries[indx].type for indx = \
                                                                                 %d is Monitor::CONNECTION",
                                                                                     pIndx);
                                                       static_cast<HTTPConnection *>(q)->_entry_index = pIndx;
                                                   HTTPConnection *dst = reinterpret_cast \
                                                                                              <HTTPConnection *>(q);
                                                   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);
  
 int  Monitor::solicitSocketMessages(                                                  try
     Sint32 socket,  
     Uint32 events,  
     Uint32 queueId,  
     int type)  
 { {
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");  
    AutoMutex autoMut(_entry_mut);                                                          dst->run(1);
    // Check to see if we need to dynamically grow the _entries array  
    // We always want the _entries array to 2 bigger than the                                                          // Record that the requested data is read/Written
    // current connections requested                                                          pipeProcessCount++;
    _solicitSocketCount++;  // bump the count  
    int size = (int)_entries.size();  
    if(_solicitSocketCount >= (size-1)){  
         for(int i = 0; i < (_solicitSocketCount - (size-1)); i++){  
                 _MonitorEntry entry(0, 0, 0);  
                 _entries.append(entry);  
         }         }
                                                   catch (...)
                                                   {
                                                           Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                                                                                 "Monitor::_dispatch: \
                                                                                              exception received");
    }    }
  
    int index;                                                  Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
    for(index = 1; index < (int)_entries.size(); index++)                                                                "Monitor::_dispatch: exited \
    {                                                                         \run() index %d",
       try                                                                                     dst->_entry_index);
       {  
          if(_entries[index]._status.value() == _MonitorEntry::EMPTY)  
                                           }
                                           else
          {          {
             _entries[index].socket = socket;                                                  /* The condition
             _entries[index].queueId  = queueId;                                                             entries[indx]._type == Monitor::INTERNAL can be
             _entries[index]._type = type;                                                             ignored for pipes as the tickler is of
             _entries[index]._status = _MonitorEntry::IDLE;                                                             Monitor::INTERNAL type. The tickler is
                                                              a socket.
                                                   */
   
                                                   Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                                                                                     "Non-connection entry, indx = %d,\
                                                                                             has been received.", pIndx);
                                                   int events = 0;
                                                   Message *msg = 0;
   
                                                       pEvents |= NamedPipeMessage::READ;
                                                       msg = new NamedPipeMessage(entries[pIndx].namedPipe, pEvents);
                                       entries[pIndx]._status = _MonitorEntry::BUSY;
                                       autoEntryMutex.unlock();
                                               q->enqueue(msg);
                                                   autoEntryMutex.lock();
                                       entries.reset(_entries);
                                       entries[pIndx]._status = _MonitorEntry::IDLE;
                                                           delete [] hPipeList;
                                   return(handled_events);
  
             return index;  
          }          }
   
   
       }       }
       catch(...)       catch(...)
       {       {
   
       }       }
    }    }
    _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful  
    PEG_METHOD_EXIT();  
    return -1;  
  
 } }
           }
  
 void Monitor::unsolicitSocketMessages(Sint32 socket)                  //Check if all the pipes had recieved the data, If no then try again
           if (pipeEntryCount == pipeProcessCount)
 { {
                       break;
                   }
  
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");  
     AutoMutex autoMut(_entry_mut);  
   
     /*  
         Start at index = 1 because _entries[0] is the tickle entry which never needs  
         to be EMPTY;  
     */  
     int index;  
     for(index = 1; index < _entries.size(); index++)  
     {  
        if(_entries[index].socket == socket)  
        {  
           _entries[index]._status = _MonitorEntry::EMPTY;  
           _entries[index].socket = -1;  
           _solicitSocketCount--;  
           break;  
        }  
     }  
  
     /*  
         Dynamic Contraction:  
         To remove excess entries we will start from the end of the _entries array  
         and remove all entries with EMPTY status until we find the first NON EMPTY.  
         This prevents the positions, of the NON EMPTY entries, from being changed.  
     */  
     index = _entries.size() - 1;  
     while(_entries[index]._status == _MonitorEntry::EMPTY){  
         if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)  
                 _entries.remove(index);  
         index--;  
     }  
   
     PEG_METHOD_EXIT();  
 } }
  
 // Note: this is no longer called with PEP 183.          delete [] hPipeList;
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)  
 {  
    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);  
    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.value() == _MonitorEntry::BUSY);  #endif
  
    // Once the HTTPConnection thread has set the status value to either      return(handled_events);
    // 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;  
 } }
  
   void Monitor::stopListeningForConnections(Boolean wait)
   
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
 ////************************* monitor 2 *****************************////  
   
   
   
   
   
 m2e_rep::m2e_rep(void)  
   :Base(), state(IDLE)  
   
 { {
 }  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
   
 m2e_rep::m2e_rep(monitor_2_entry_type _type,  
                  pegasus_socket _sock,  
                  void* _accept,  
                  void* _dispatch)  
   : Base(), type(_type), state(IDLE), psock(_sock),  
     accept_parm(_accept), dispatch_parm(_dispatch)  
 { {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   #endif
       PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");
       // set boolean then tickle the server to recognize _stopConnections
       _stopConnections = 1;
       tickle();
  
 m2e_rep::~m2e_rep(void)      if (wait)
 { {
         // Wait for the monitor to notice _stopConnections.  Otherwise the
         // caller of this function may unbind the ports while the monitor
         // is still accepting connections on them.
         _stopConnectionsSem.wait();
 } }
  
 m2e_rep::m2e_rep(const m2e_rep& r)      PEG_METHOD_EXIT();
   : Base()  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
 { {
   if(this != &r){          AutoMutex automut(Monitor::_cout_mut);
     type = r.type;          PEGASUS_STD(cout) << "Exiting:  Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
     psock = r.psock;  
     accept_parm = r.accept_parm;  
     dispatch_parm = r.dispatch_parm;  
     state = IDLE;  
   
   }   }
   #endif
 } }
  
  
 m2e_rep& m2e_rep::operator =(const m2e_rep& r)  int  Monitor::solicitSocketMessages(
       PEGASUS_SOCKET socket,
       Uint32 events,
       Uint32 queueId,
       int type)
 { {
   if(this != &r) {  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
     type = r.type;  
     psock = r.psock;  
     accept_parm = r.accept_parm;  
     dispatch_parm = r.dispatch_parm;  
     state = IDLE;  
   }  
   return *this;  
 }  
   
 Boolean m2e_rep::operator ==(const m2e_rep& r)  
 { {
   if(this == &r)          AutoMutex automut(Monitor::_cout_mut);
     return true;          PEGASUS_STD(cout) << "Entering: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
   return false;  
 } }
   #endif
 Boolean m2e_rep::operator ==(void* r)     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
 {     AutoMutex autoMut(_entry_mut);
   if((void*)this == r)     // Check to see if we need to dynamically grow the _entries array
     return true;     // We always want the _entries array to 2 bigger than the
   return false;     // current connections requested
      _solicitSocketCount++;  // bump the count
      int size = (int)_entries.size();
      if((int)_solicitSocketCount >= (size-1)){
           for(int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++){
                   _MonitorEntry entry(0, 0, 0);
                   _entries.append(entry);
 } }
   
 m2e_rep::operator pegasus_socket() const  
 {  
   return psock;  
 } }
  
      int index;
 monitor_2_entry::monitor_2_entry(void)     for(index = 1; index < (int)_entries.size(); index++)
 { {
   _rep = new m2e_rep();        try
 }  
   
 monitor_2_entry::monitor_2_entry(pegasus_socket& _psock,  
                                  monitor_2_entry_type _type,  
                                  void* _accept_parm, void* _dispatch_parm)  
 { {
   _rep = new m2e_rep(_type, _psock, _accept_parm, _dispatch_parm);           if(_entries[index]._status.get() == _MonitorEntry::EMPTY)
 }  
   
 monitor_2_entry::monitor_2_entry(const monitor_2_entry& e)  
 { {
   if(this != &e){              _entries[index].socket = socket;
     Inc(this->_rep = e._rep);              _entries[index].queueId  = queueId;
   }              _entries[index]._type = type;
 }              _entries[index]._status = _MonitorEntry::IDLE;
  
 monitor_2_entry::~monitor_2_entry(void)  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
 { {
                   AutoMutex automut(Monitor::_cout_mut);
   Dec(_rep);                  PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   #endif
 monitor_2_entry& monitor_2_entry::operator=(const monitor_2_entry& e)              return index;
 {  
   if(this != &e){  
     Dec(_rep);  
     Inc(this->_rep = e._rep);  
   }   }
   return *this;  
 } }
         catch(...)
 Boolean monitor_2_entry::operator ==(const monitor_2_entry& me) const  
 { {
   if(this == &me)  
     return true;  
   return false;  
 } }
   
 Boolean monitor_2_entry::operator ==(void* k) const  
 {  
   if((void *)this == k)  
     return true;  
   return false;  
 } }
      _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful
      PEG_METHOD_EXIT();
 monitor_2_entry_type monitor_2_entry::get_type(void) const  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
 { {
   return _rep->type;         AutoMutex automut(Monitor::_cout_mut);
          PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   #endif
      return -1;
  
 void monitor_2_entry::set_type(monitor_2_entry_type t)  
 {  
   _rep->type = t;  
 } }
  
   void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket)
 monitor_2_entry_state  monitor_2_entry::get_state(void) const  
 { {
   return (monitor_2_entry_state) _rep->state.value();  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
 }  
   
 void monitor_2_entry::set_state(monitor_2_entry_state t)  
 { {
   _rep->state = t;          AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   #endif
  
 void* monitor_2_entry::get_accept(void) const      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
 {      AutoMutex autoMut(_entry_mut);
   return _rep->accept_parm;  
 }  
  
 void monitor_2_entry::set_accept(void* a)      /*
           Start at index = 1 because _entries[0] is the tickle entry which never needs
           to be EMPTY;
       */
       unsigned int index;
       for(index = 1; index < _entries.size(); index++)
 { {
   _rep->accept_parm = a;         if(_entries[index].socket == socket)
 }  
   
   
 void* monitor_2_entry::get_dispatch(void) const  
 { {
   return _rep->dispatch_parm;            _entries[index]._status = _MonitorEntry::EMPTY;
             _entries[index].socket = PEGASUS_INVALID_SOCKET;
             _solicitSocketCount--;
             break;
 } }
   
 void monitor_2_entry::set_dispatch(void* a)  
 {  
   _rep->dispatch_parm = a;  
 } }
  
 pegasus_socket monitor_2_entry::get_sock(void) const      /*
 {          Dynamic Contraction:
   return _rep->psock;          To remove excess entries we will start from the end of the _entries array
           and remove all entries with EMPTY status until we find the first NON EMPTY.
           This prevents the positions, of the NON EMPTY entries, from being changed.
       */
       index = _entries.size() - 1;
       while(_entries[index]._status.get() == _MonitorEntry::EMPTY){
           if(_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES)
                   _entries.remove(index);
           index--;
 } }
       PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
 void monitor_2_entry::set_sock(pegasus_socket& s)  
 { {
   _rep->psock = s;          AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   
 //static monitor_2* _m2_instance;  
   
 AsyncDQueue<HTTPConnection2> monitor_2::_connections(true, 0);  
   
 monitor_2::monitor_2(void)  
   : _session_dispatch(0), _accept_dispatch(0), _listeners(true, 0),  
     _ready(true, 0), _die(0), _requestCount(0)  
 {  
   try {  
   
     bsd_socket_factory _factory;  
   
     // set up the listener/acceptor  
     pegasus_socket temp = pegasus_socket(&_factory);  
   
     temp.socket(PF_INET, SOCK_STREAM, 0);  
     // initialize the address  
     memset(&_tickle_addr, 0, sizeof(_tickle_addr));  
 #ifdef PEGASUS_OS_ZOS  
     _tickle_addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");  
 #else  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM  
 #pragma convert(37)  
 #endif  
     _tickle_addr.sin_addr.s_addr = inet_addr("127.0.0.1");  
 #ifdef PEGASUS_PLATFORM_OS400_ISERIES_IBM  
 #pragma convert(0)  
 #endif  
 #endif  
     _tickle_addr.sin_family = PF_INET;  
     _tickle_addr.sin_port = 0;  
   
     PEGASUS_SOCKLEN_SIZE _addr_size = sizeof(_tickle_addr);  
   
     temp.bind((struct sockaddr *)&_tickle_addr, sizeof(_tickle_addr));  
     temp.listen(3);  
     temp.getsockname((struct sockaddr*)&_tickle_addr, &_addr_size);  
   
     // set up the connector  
   
     pegasus_socket tickler = pegasus_socket(&_factory);  
     tickler.socket(PF_INET, SOCK_STREAM, 0);  
     struct sockaddr_in _addr;  
     memset(&_addr, 0, sizeof(_addr));  
 #ifdef PEGASUS_OS_ZOS  
     _addr.sin_addr.s_addr = inet_addr_ebcdic("127.0.0.1");  
 #else  
     _addr.sin_addr.s_addr = inet_addr("127.0.0.1");  
 #endif #endif
     _addr.sin_family = PF_INET;  
     _addr.sin_port = 0;  
     tickler.bind((struct sockaddr*)&_addr, sizeof(_addr));  
     tickler.connect((struct sockaddr*)&_tickle_addr, sizeof(_tickle_addr));  
   
     _tickler.set_sock(tickler);  
     _tickler.set_type(INTERNAL);  
     _tickler.set_state(BUSY);  
   
     struct sockaddr_in peer;  
     memset(&peer, 0, sizeof(peer));  
     PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);  
   
     pegasus_socket accepted = temp.accept((struct sockaddr*)&peer, &peer_size);  
   
     monitor_2_entry* _tickle = new monitor_2_entry(accepted, INTERNAL, 0, 0);  
   
 // No need to set _tickle's state as BUSY, since monitor_2::run() now  
 // does a select only on sockets which are in IDLE (default) state.  
 //  _tickle->set_state(BUSY);  
   
     _listeners.insert_first(_tickle);  
   
   }  
   catch(...){  }  
 } }
  
 monitor_2::~monitor_2(void)  // Note: this is no longer called with PEP 183.
   PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
    stop();      {
           AutoMutex automut(Monitor::_cout_mut);
   try {          PEGASUS_STD(cout) << "Entering: Monitor::_dispatch(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
     monitor_2_entry* temp = _listeners.remove_first();  
     while(temp){  
       delete temp;  
       temp = _listeners.remove_first();  
     }  
   }   }
   #endif
   catch(...){  }     HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);
      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   try
   {   {
      HTTPConnection2* temp = _connections.remove_first();        dst->run(1);
      while(temp)  
      {  
         delete temp;  
         temp = _connections.remove_first();  
      }  
   }   }
   catch(...)   catch(...)
   {   {
   }  
   
   
 }  
   
   
 void monitor_2::run(void)  
 {  
   monitor_2_entry* temp;  
   int _nonIdle=0, _idleCount=0, events;  
   
   while(_die.value() == 0) {  
     _nonIdle=_idleCount=0;  
   
      struct timeval tv_idle = { 60, 0 };  
   
     // place all sockets in the select set  
     FD_ZERO(&rd_fd_set);  
     try {  
       _listeners.lock(pegasus_thread_self());  
       temp = _listeners.next(0);  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
        "monitor_2::run:Creating New FD list for SELECT.");  
       while(temp != 0 ){  
         if(temp->get_state() == CLOSED ) {  
           monitor_2_entry* closed = temp;  
       temp = _listeners.next(closed);  
           _listeners.remove_no_lock(closed);  
   
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
        "monitor_2::run:Deleteing CLOSED socket fd=%d.",(Sint32)closed->get_sock());  
   
           HTTPConnection2 *cn = monitor_2::remove_connection((Sint32)(closed->get_sock()));  
           delete cn;  
           delete closed;  
         }  
         if(temp == 0)  
            break;  
   
   
         //Count the number if IDLE sockets  
         if(temp->get_state() != IDLE ) _nonIdle++;  
          else _idleCount++;  
   
         Sint32 fd = (Sint32) temp->get_sock();  
   
         //Select should be called ONLY on the FDs which are in IDLE state  
         if((fd >= 0) && (temp->get_state() == IDLE))  
         {  
           Tracer::trace(TRC_HTTP, Tracer::LEVEL4,           Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
            "monitor_2::run:Adding FD %d to the list for SELECT.",fd);            "Monitor::_dispatch: exception received");
           FD_SET(fd , &rd_fd_set);  
         }  
            temp = _listeners.next(temp);  
       }  
       _listeners.unlock();  
     }  
     catch(...){  
       return;  
     }  
   
     // important -  the dispatch routine has pointers to all the  
     // entries that are readable. These entries can be changed but  
     // the pointer must not be tampered with.  
     if(_connections.count() )  
        events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, NULL);  
     else  
        events = select(FD_SETSIZE, &rd_fd_set, NULL, NULL, &tv_idle);  
   
     if(_die.value())  
     {  
        break;  
     }  
   
 #ifdef PEGASUS_OS_TYPE_WINDOWS  
     if(events == SOCKET_ERROR)  
 #else  
     if(events == -1)  
 #endif  
     {  
        Tracer::trace(TRC_HTTP, Tracer::LEVEL2,  
           "monitor_2:run:INVALID FD. errorno = %d on select.", errno);  
        // The EBADF error indicates that one or more or the file  
        // descriptions was not valid. This could indicate that  
        // the _entries structure has been corrupted or that  
        // we have a synchronization error.  
   
      // Keeping the line below commented for time being.  
      //  PEGASUS_ASSERT(errno != EBADF);  
     }     }
     else if (events)  
     {  
        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
           "monitor_2::run select event received events = %d, monitoring %d idle entries", events, _idleCount);            "Monitor::_dispatch: exited run() for index %d", dst->_entry_index);
  
      PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);
  
     try {     // Once the HTTPConnection thread has set the status value to either
       _listeners.lock(pegasus_thread_self());     // Monitor::DYING or Monitor::IDLE, it has returned control of the connection
       temp = _listeners.next(0);     // to the Monitor.  It is no longer permissible to access the connection
       while(temp != 0 ){     // or the entry in the _entries table.
           Sint32 fd = (Sint32) temp->get_sock();     if (dst->_connectionClosePending)
           if(fd >= 0 && FD_ISSET(fd, &rd_fd_set)) {  
           if(temp->get_type() != CLIENTSESSION) temp->set_state(BUSY);  
           FD_CLR(fd,  &rd_fd_set);  
           monitor_2_entry* ready = new monitor_2_entry(*temp);  
           try  
           {  
              _ready.insert_first(ready);  
           }  
           catch(...)  
           {           {
         dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::DYING;
           }           }
   
           _requestCount++;  
         }  
         temp = _listeners.next(temp);  
       }  
       _listeners.unlock();  
     }  
     catch(...){  
       return;  
     }  
     // now handle the sockets that are ready to read  
     if(_ready.count())  
        _dispatch();  
     else     else
     {     {
        if(_connections.count() == 0 )        dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
           _idle_dispatch(_idle_parm);  
     }     }
    }  // if events     return 0;
   } // while alive  
   _die=0;  
   
 } }
  
 int  monitor_2::solicitSocketMessages(  // Added for NamedPipe implementation for windows
     Sint32 socket,  #if defined PEGASUS_OS_TYPE_WINDOWS && !defined(PEGASUS_DISABLE_LOCAL_DOMAIN_SOCKET)
     Uint32 events,  //This method is anlogus to solicitSocketMessages. It does the same thing for named Pipes
   int  Monitor::solicitPipeMessages(
       NamedPipe namedPipe,
       Uint32 events,  //not sure what has to change for this enum
     Uint32 queueId,     Uint32 queueId,
     int type)     int type)
 { {
      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitPipeMessages");
    PEG_METHOD_ENTER(TRC_HTTP, "monitor_2::solicitSocketMessages");  
  
    AutoMutex autoMut(_entry_mut);    AutoMutex autoMut(_entry_mut);
      // Check to see if we need to dynamically grow the _entries array
    for(int index = 0; index < (int)_entries.size(); index++)     // We always want the _entries array to 2 bigger than the
    {     // current connections requested
       try  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {  
          if(_entries[index]._status.value() == monitor_2_entry::EMPTY)  
          {  
             _entries[index].socket = socket;  
             //_entries[index].queueId  = queueId;  
             //_entries[index]._type = type;  
             _entries[index]._status = IDLE;  
   
             return index;  
          }  
       }  
       catch(...)  
       {  
       }  
   
    }  
    PEG_METHOD_EXIT();  
    return -1;  
 }  
   
   
 void monitor_2::unsolicitSocketMessages(Sint32 socket)  
 {  
   
     PEG_METHOD_ENTER(TRC_HTTP, "monitor_2::unsolicitSocketMessages");  
     AutoMutex autoMut(_entry2_mut);  
   
     for(int index = 0; index < (int)_entries2.size(); index++)  
     {  
        if(_entries2[index].socket == socket)  
        {  
           _entries2[index]._status = monitor_2_entry::EMPTY;  
           _entries2[index].socket = -1;  
           break;  
        }  
     }  
     PEG_METHOD_EXIT();  
 }  
   
 void* monitor_2::set_session_dispatch(void (*dp)(monitor_2_entry*))  
 {  
   void* old = (void *)_session_dispatch;  
   _session_dispatch = dp;  
   return old;  
 }  
   
 void* monitor_2::set_accept_dispatch(void (*dp)(monitor_2_entry*))  
 { {
   void* old = (void*)_accept_dispatch;     AutoMutex automut(Monitor::_cout_mut);
   _accept_dispatch = dp;     PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages at the begining" << PEGASUS_STD(endl);
   return old;  
 } }
   #endif
  
 void* monitor_2::set_idle_dispatch(void (*dp)(void*))     _solicitSocketCount++;  // bump the count
 {     int size = (int)_entries.size();
    void* old = (void*)_idle_dispatch;     if((int)_solicitSocketCount >= (size-1)){
    _idle_dispatch = dp;          for(int i = 0; i < ((int)_solicitSocketCount - (size-1)); i++){
    return old;                  _MonitorEntry entry(0, 0, 0);
                   _entries.append(entry);
 } }
   
 void* monitor_2::set_idle_parm(void* parm)  
 {  
    void* old = _idle_parm;  
    _idle_parm = parm;  
    return old;  
 } }
  
      int index;
      for(index = 1; index < (int)_entries.size(); index++)
 //-----------------------------------------------------------------  
 // Note on deleting the monitor_2_entry nodes:  
 //  Each case: in the switch statement needs to handle the deletion  
 //  of the monitor_2_entry * node differently. A SESSION dispatch  
 //  routine MUST DELETE the entry during its dispatch handling.  
 //  All other dispatch routines MUST NOT delete the entry during the  
 //  dispatch handling, but must allow monitor_2::_dispatch to delete  
 //   the entry.  
 //  
 //  The reason is pretty obscure and it is debatable whether or not  
 //  to even bother, but during cimserver shutdown the single monitor_2_entry*  
 //  will leak unless the _session_dispatch routine takes care of deleting it.  
 //  
 //  The reason is that a shutdown messages completely stops everything and  
 //  the _session_dispatch routine never returns. So monitor_2::_dispatch is  
 //  never able to do its own cleanup.  
 //  
 // << Mon Oct 13 09:33:33 2003 mdd >>  
 //-----------------------------------------------------------------  
   
 void monitor_2::_dispatch(void)  
 { {
    monitor_2_entry* entry;  
   
    try    try
    {    {
            if(_entries[index]._status.get() == _MonitorEntry::EMPTY)
          entry = _ready.remove_first();  
    }  
    catch(...)  
    {  
    }  
   
   while(entry != 0 ) {  
     switch(entry->get_type()) {  
     case INTERNAL:  
       static char buffer[2];  
       entry->get_sock().disableBlocking();  
       entry->get_sock().read(&buffer, 2);  
       entry->get_sock().enableBlocking();  
       entry->set_state(IDLE);   // Set state of the socket to IDLE so that  
                                 // monitor_2::run can add to the list of FDs  
                                 // on which select would be called.  
   
   
   
       delete entry;  
   
       break;  
     case LISTEN:  
       {  
         static struct sockaddr peer;  
         static PEGASUS_SOCKLEN_SIZE peer_size = sizeof(peer);  
         entry->get_sock().disableBlocking();  
         pegasus_socket connected = entry->get_sock().accept(&peer, &peer_size);  
         entry->set_state(IDLE);  // Set state of the LISTEN socket to IDLE  
 #ifdef PEGASUS_OS_TYPE_WINDOWS  
     if((Sint32)connected  == SOCKET_ERROR)  
 #else  
         if((Sint32)connected == -1 )  
 #endif  
         {         {
            delete entry;              _entries[index].socket = NULL;
            break;              _entries[index].namedPipe = namedPipe;
         }              _entries[index].namedPipeConnection = true;
               _entries[index].queueId  = queueId;
         entry->get_sock().enableBlocking();              _entries[index]._type = type;
         monitor_2_entry *temp = add_entry(connected, SESSION, entry->get_accept(), entry->get_dispatch());              _entries[index]._status = _MonitorEntry::IDLE;
         if(temp && _accept_dispatch != 0)  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
            _accept_dispatch(temp);  
         delete entry;  
   
       }  
       break;  
     case SESSION:  
     case CLIENTSESSION:  
        if(_session_dispatch != 0 )  
        {  
           // NOTE: _session_dispatch will delete entry - do not do it here  
              unsigned client=0;  
          if(entry->get_type() == CLIENTSESSION) client = 1;  
          Sint32 sock=(Sint32)(entry->get_sock());  
   
              _session_dispatch(entry);  
   
          if(client)  
          {          {
            HTTPConnection2 *cn = monitor_2::remove_connection(sock);              AutoMutex automut(Monitor::_cout_mut);
                if(cn) delete cn;              PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages after seting up  _entries[index] index = " << index << PEGASUS_STD(endl);
            // stop();  
            _die=1;  
          }  
        }        }
   #endif
  
       else {              return index;
         static char buffer[4096];  
         int bytes = entry->get_sock().read(&buffer, 4096);  
         delete entry;  
       }       }
   
       break;  
     case UNTYPED:  
     default:  
            delete entry;  
       break;  
     }  
     _requestCount--;  
   
     if(_ready.count() == 0 )  
        break;  
   
     try  
     {  
        entry = _ready.remove_first();  
     }     }
     catch(...)     catch(...)
     {     {
     }     }
  
   }   }
 }     _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful
      PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages nothing happed - it didn't work" << PEGASUS_STD(endl);
 void monitor_2::stop(void)  
 {  
   _die = 1;  
   tickle();  
   // shut down the listener list, free the list nodes  
   _tickler.get_sock().close();  
   _listeners.shutdown_queue();  
 }  
   
 void monitor_2::tickle(void)  
 {  
   static char _buffer[] =  
     {  
       '0','0'  
     };  
   
   _tickler.get_sock().disableBlocking();  
   
   _tickler.get_sock().write(&_buffer, 2);  
   _tickler.get_sock().enableBlocking();  
   
 }  
   
   
 monitor_2_entry*  monitor_2::add_entry(pegasus_socket& ps,  
                                        monitor_2_entry_type type,  
                                        void* accept_parm,  
                                        void* dispatch_parm)  
 {  
   Sint32 fd1,fd2;  
   
   fd2=(Sint32) ps;  
   
   monitor_2_entry* m2e = new monitor_2_entry(ps, type, accept_parm, dispatch_parm);  
   
 // The purpose of the following piece of code is to avoid duplicate entries in  
 // the _listeners list. Would it be too much of an overhead ?  
 try {  
   
      monitor_2_entry* temp;  
   
       _listeners.lock(pegasus_thread_self());  
       temp = _listeners.next(0);  
       while(temp != 0 )  
       {  
         fd1=(Sint32) temp->get_sock();  
   
         if(fd1 == fd2)  
         {  
   
            Tracer::trace(TRC_HTTP, Tracer::LEVEL3,  
           "monitor_2::add_entry:Request for duplicate entry in _listeners for %d FD.", fd1);  
             if(temp->get_state() == CLOSED)  
             {  
               temp->set_state(IDLE);  
               Tracer::trace(TRC_HTTP, Tracer::LEVEL3,  
               "monitor_2::add_entry:CLOSED state changed to IDLE for %d.", fd1);  
              }  
              _listeners.unlock();  
             delete m2e;  
             return 0;  
         }  
        temp = _listeners.next(temp);  
       }  
    }  
    catch(...)  
    {  
       delete m2e;  
       return 0;  
    }  
   
   
   _listeners.unlock();  
  
      PEG_METHOD_EXIT();
      return -1;
  
   try{  
     _listeners.insert_first(m2e);  
   }  
   catch(...){  
     delete m2e;  
     return 0;  
   }  
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,  
       "monitor_2::add_entry:SUCCESSFULLY added to _listeners list. FD = %d.", fd2);  
   tickle();  
   return m2e;  
 } }
  
 Boolean monitor_2::remove_entry(Sint32 s)  //////////////////////////////////////////////////////////////////////////////
   // Method Name      : unsolicitPipeMessages
   // Input Parameter  : namedPipe  - type NamedPipe
   // Return Type      : void
   //============================================================================
   // This method is invoked from HTTPAcceptor::handleEnqueue for server
   // when the CLOSE_CONNECTION_MESSAGE is recieved. This method is also invoked
   // from HTTPAcceptor::destroyConnections method when the CIMServer is shutdown.
   // For the CIMClient, this is invoked from HTTPConnector::handleEnqueue when the
   // CLOSE_CONNECTION_MESSAGE is recieved. This method is also invoked from
   // HTTPConnector::disconnect when CIMClient requests a disconnect request.
   // The list of _MonitorEntry is searched for the matching pipe.
   // The Handle of the identified is closed and _MonitorEntry for the
   // requested pipe is removed.
   ///////////////////////////////////////////////////////////////////////////////
   
   void Monitor::unsolicitPipeMessages(NamedPipe namedPipe)
 { {
   monitor_2_entry* temp;  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
   try {  
     _listeners.try_lock(pegasus_thread_self());  
     temp = _listeners.next(0);  
     while(temp != 0){  
       if(s == (Sint32)temp->_rep->psock ){  
         temp = _listeners.remove_no_lock(temp);  
         delete temp;  
         _listeners.unlock();  
         return true;  
       }  
       temp = _listeners.next(temp);  
     }  
     _listeners.unlock();  
   }  
   catch(...){  
   }  
   return false;  
 }  
   
 Uint32 monitor_2::getOutstandingRequestCount(void)  
 { {
   return _requestCount.value();          AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::unsolicitPipeMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
 } }
   #endif
  
       PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitPipeMessages");
       AutoMutex autoMut(_entry_mut);
  
 HTTPConnection2* monitor_2::remove_connection(Sint32 sock)      /*
 {          Start at index = 1 because _entries[0] is the tickle entry which never needs
           to be EMPTY;
    HTTPConnection2* temp;      */
    try      unsigned int index;
    {      for (index = 1; index < _entries.size(); index++)
       monitor_2::_connections.lock(pegasus_thread_self());  
       temp = monitor_2::_connections.next(0);  
       while(temp != 0 )  
       {       {
          if(sock == temp->getSocket())          if (_entries[index].namedPipe.getPipe() == namedPipe.getPipe())
          {  
             temp = monitor_2::_connections.remove_no_lock(temp);  
             monitor_2::_connections.unlock();  
             return temp;  
          }  
          temp = monitor_2::_connections.next(temp);  
       }  
       monitor_2::_connections.unlock();  
    }  
    catch(...)  
    {    {
               _entries[index]._status = _MonitorEntry::EMPTY;
               // Ensure that the client has read the data
                       ::FlushFileBuffers (namedPipe.getPipe());
                       //Disconnect to release the pipe. This doesn't release Pipe Handle
                       ::DisconnectNamedPipe (_entries[index].namedPipe.getPipe());
               // Must use CloseHandle to Close Pipe
                           ::CloseHandle(_entries[index].namedPipe.getPipe());
                       _entries[index].namedPipe.disconnect();
               _solicitSocketCount--;
               break;
    }    }
    return 0;  
 } }
  
 Boolean monitor_2::insert_connection(HTTPConnection2* connection)      /*
           Dynamic Contraction:
           To remove excess entries we will start from the end of the _entries array
           and remove all entries with EMPTY status until we find the first NON EMPTY.
           This prevents the positions, of the NON EMPTY entries, from being changed.
       */
       index = _entries.size() - 1;
       while (_entries[index]._status.get() == _MonitorEntry::EMPTY)
 { {
    try          if ((_entries[index].namedPipe.getPipe() == namedPipe.getPipe()) ||
               (_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES))
    {    {
       monitor_2::_connections.insert_first(connection);                      _entries.remove(index);
    }    }
    catch(...)          index--;
       }
       PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
    {    {
       return false;          AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::unsolicitPipeMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
    }    }
    return true;  #endif
 } }
  
   #endif
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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