(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.103.10.1 and 1.103.10.22

version 1.103.10.1, 2006/06/01 01:35:10 version 1.103.10.22, 2006/07/21 18:17:51
Line 52 
Line 52 
 #include <Pegasus/Common/Exception.h> #include <Pegasus/Common/Exception.h>
 #include "ArrayIterator.h" #include "ArrayIterator.h"
  
   
   
   //const static DWORD MAX_BUFFER_SIZE = 4096;  // 4 kilobytes
   
 #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 \
Line 78 
Line 82 
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
 static AtomicInt _connections(0); static AtomicInt _connections(0);
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
   Mutex Monitor::_cout_mut;
   #endif
  
   #ifdef PEGASUS_OS_TYPE_WINDOWS
 #define PIPE_INCREMENT 1 #define PIPE_INCREMENT 1
   #endif
 static  int MaxPipes = PIPE_INCREMENT;  
   
 static HANDLE* hEvents = new HANDLE[PIPE_INCREMENT]; //this number should be fixed  
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
Line 101 
Line 105 
      _tickle_server_socket(-1),      _tickle_server_socket(-1),
      _tickle_peer_socket(-1)      _tickle_peer_socket(-1)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;     int numberOfMonitorEntriesToAllocate = MAX_NUMBER_OF_MONITOR_ENTRIES;
     Socket::initializeInterface();     Socket::initializeInterface();
     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);     _entries.reserveCapacity(numberOfMonitorEntriesToAllocate);
Line 116 
Line 126 
        _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()
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
  
     try{     try{
Line 145 
Line 167 
     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 330 
Line 364 
     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);     _MonitorEntry entry(_tickle_peer_socket, 1, INTERNAL);
     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'
Line 343 
Line 389 
     Socket::disableBlocking(_tickle_client_socket);     Socket::disableBlocking(_tickle_client_socket);
     Socket::write(_tickle_client_socket,&_buffer, 2);     Socket::write(_tickle_client_socket,&_buffer, 2);
     Socket::enableBlocking(_tickle_client_socket);     Socket::enableBlocking(_tickle_client_socket);
   #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 ) void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )
Line 353 
Line 405 
  
 Boolean Monitor::run(Uint32 milliseconds) Boolean Monitor::run(Uint32 milliseconds)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
  
     Boolean handled_events = false;     Boolean handled_events = false;
     int i = 0;     int i = 0;
Line 414 
Line 472 
  
                                         if (h._responsePending == true)                                         if (h._responsePending == true)
                                         {                                         {
                           if (!entry.namedPipeConnection)
                           {
                                                 Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "                                                 Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "Monitor::run - "
                                                                                                         "Ignoring connection delete request because "                                                                                                         "Ignoring connection delete request because "
                                                                                                         "responses are still pending. "                                                                                                         "responses are still pending. "
                                                                                                         "connection=0x%p, socket=%d\n",                                                                                                         "connection=0x%p, socket=%d\n",
                                                                                                         (void *)&h, h.getSocket());                                                                                                         (void *)&h, h.getSocket());
                           }
                           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());
                           }
                                                 continue;                                                 continue;
                                         }                                         }
                                         h._connectionClosePending = false;                                         h._connectionClosePending = false;
           MessageQueue &o = h.get_owner();           MessageQueue &o = h.get_owner();
           Message* message= new CloseConnectionMessage(entry.socket);            Message* message;
             if (!entry.namedPipeConnection)
             {
                 message= new CloseConnectionMessage(entry.socket);
             }
             else
             {
                 message= new CloseConnectionMessage(entry.namedPipe);
   
             }
           message->dest = o.getQueueId();           message->dest = o.getQueueId();
  
           // HTTPAcceptor is responsible for closing the connection.           // HTTPAcceptor is responsible for closing the connection.
Line 456 
Line 534 
         because we have to traverse the entire array.         because we have to traverse the entire array.
     */     */
     //Array<HANDLE> pipeEventArray;     //Array<HANDLE> pipeEventArray;
     int pipeEntrycount=0;  
     PEGASUS_SOCKET maxSocketCurrentPass = 0;     PEGASUS_SOCKET maxSocketCurrentPass = 0;
     int indx;     int indx;
  
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
   
     //This array associates named pipe connections to their place in [indx]     //This array associates named pipe connections to their place in [indx]
     //in the entries array. The value in poition zero of the array is the     //in the entries array. The value in poition zero of the array is the
     //index of the fist named pipe connection in the entries array     //index of the fist named pipe connection in the entries array
     Array <Uint32> indexPipeCountAssociator;     Array <Uint32> indexPipeCountAssociator;
       int pipeEntryCount=0;
       int MaxPipes = PIPE_INCREMENT;
       HANDLE* hEvents = new HANDLE[PIPE_INCREMENT];
   
   #endif
  
     for( indx = 0; indx < (int)entries.size(); indx++)     for( indx = 0; indx < (int)entries.size(); indx++)
     {     {
        if(!entries[indx].isNamedPipeConnection())  
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
          if(entries[indx].isNamedPipeConnection())
          {
   
              //entering this clause mean that a Named Pipe connection is at entries[indx]
              //cout << "In Monitor::run in clause to to create array of for WaitformultipuleObjects" << endl;
   
              //cout << "In Monitor::run - pipe being added to array is " << entries[indx].namedPipe.getName() << endl;
   
               entries[indx].pipeSet = false;
   
              // We can Keep a counter in the Monitor class for the number of named pipes ...
              //  Which can be used here to create the array size for hEvents..( obviously before this for loop.:-) )
               if (pipeEntryCount >= MaxPipes)
               {
                  // cout << "Monitor::run 'if (pipeEntryCount >= MaxPipes)' begining - pipeEntryCount=" <<
                      // pipeEntryCount << " MaxPipes=" << MaxPipes << endl;
                    MaxPipes += PIPE_INCREMENT;
                    HANDLE* temp_hEvents = new HANDLE[MaxPipes];
   
                    for (Uint32 i =0;i<pipeEntryCount;i++)
                    {
                        temp_hEvents[i] = hEvents[i];
                    }
   
                    delete [] hEvents;
   
                    hEvents = temp_hEvents;
                   // cout << "Monitor::run 'if (pipeEntryCount >= MaxPipes)' ending"<< endl;
   
               }
   
              //pipeEventArray.append((entries[indx].namedPipe.getOverlap()).hEvent);
              hEvents[pipeEntryCount] = entries[indx].namedPipe.getOverlap()->hEvent;
   
              indexPipeCountAssociator.append(indx);
   
          pipeEntryCount++;
   
   
   
          }
          else
   
   #endif
        {        {
   
            if(maxSocketCurrentPass < entries[indx].socket)            if(maxSocketCurrentPass < entries[indx].socket)
             maxSocketCurrentPass = entries[indx].socket;             maxSocketCurrentPass = entries[indx].socket;
  
Line 477 
Line 609 
                _idleEntries++;                _idleEntries++;
                FD_SET(entries[indx].socket, &fdread);                FD_SET(entries[indx].socket, &fdread);
            }            }
        }  
        else  
        {   //entering this clause mean that a Named Pipe connection is at entries[indx]  
            //cout << "In Monitor::run in clause to to create array of for WaitformultipuleObjects" << endl;  
            cout << "In Monitor::run - pipe being added to array is " <<  
                    entries[indx].namedPipe.getName() << endl;  
            cout << " overlap event is "<<  
                 (int) (entries[indx].namedPipe.getOverlap()).hEvent << endl;  
            //pipeEventArray.append((entries[indx].namedPipe.getOverlap()).hEvent);  
            hEvents[pipeEntrycount] = entries[indx].namedPipe.getOverlap().hEvent;  
            pipeEntrycount++;  
   
            indexPipeCountAssociator.append(indx);  
            cout << "pipeEntrycount is " << pipeEntrycount << endl;  
            cout <<" this is the type " << entries[indx]._type <<  
                " this is index " << indx << endl;  
  
        }        }
     }     }
Line 510 
Line 626 
     // a socket value.  The original code assumed that the number of sockets     // 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.     // and a socket value have the same type.  On Windows they do not.
     //     //
   
       int events;
       int pEvents;
   
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
     //int events = select(0, &fdread, NULL, NULL, &tv);  
      // events = select(0, &fdread, NULL, NULL, &tv);
   
       //if (events == NULL)
       //{  // This connection uses namedPipes
   
           events = 0;
     DWORD dwWait=NULL;     DWORD dwWait=NULL;
     int events;          pEvents = 0;
  
     cout << "Calling WaitForMultipleObjects\n";  
  
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
          {
           AutoMutex automut(Monitor::_cout_mut);
           cout << "Monitor::run - Calling WaitForMultipleObjects\n";
           }
   #endif
      // }
     //this should be in a try block     //this should be in a try block
  
     dwWait = WaitForMultipleObjects(MaxPipes,      dwWait = WaitForMultipleObjects(
                    MaxPipes,
                                    hEvents,      //ABB:- array of event objects                                    hEvents,      //ABB:- array of event objects
                                    FALSE,        // ABB:-does not wait for all                                    FALSE,        // ABB:-does not wait for all
                                    20000);        //ABB:- timeout value                   milliseconds);        //ABB:- timeout value   //WW this may need be shorter
  
     if(dwWait == WAIT_TIMEOUT)     if(dwWait == WAIT_TIMEOUT)
         {         {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
           {
               AutoMutex automut(Monitor::_cout_mut);
         cout << "Wait WAIT_TIMEOUT\n";         cout << "Wait WAIT_TIMEOUT\n";
           cout << "Monitor::run before the select in TIMEOUT clause events = " << events << endl;
           }
   #endif
                   events = select(0, &fdread, NULL, NULL, &tv);
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
               AutoMutex automut(Monitor::_cout_mut);
              cout << "Monitor::run after the select in TIMEOUT clause events = " << events << endl;
   #endif
   
   
            // Sleep(2000);            // Sleep(2000);
             //continue;             //continue;
              return false;  
                //return false;  // I think we do nothing.... Mybe there is a socket connection... so
                // cant return.
         }         }
         else if (dwWait == WAIT_FAILED)         else if (dwWait == WAIT_FAILED)
         {         {
               if (GetLastError() == 6) //WW this may be too specific
               {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                   AutoMutex automut(Monitor::_cout_mut);
                   cout << "Monitor::run about to call 'select since waitForMultipleObjects failed\n";
   #endif
                   /********* NOTE
                   this time (tv) combined with the waitForMulitpleObjects timeout is
                   too long it will cause the client side to time out
                   ******************/
                   events = select(0, &fdread, NULL, NULL, &tv);
   
               }
               else
               {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                   AutoMutex automut(Monitor::_cout_mut);
             cout << "Wait Failed returned\n";             cout << "Wait Failed returned\n";
             cout << "failed with " << GetLastError() << "." << endl;             cout << "failed with " << GetLastError() << "." << endl;
   #endif
                   pEvents = -1;
             return false;             return false;
         }         }
           }
         else         else
         {         {
             int pCount = dwWait - WAIT_OBJECT_0;  // determines which pipe             int pCount = dwWait - WAIT_OBJECT_0;  // determines which pipe
             cout << " WaitForMultiPleObject returned activity on server pipe: "<<  
                 pCount<< endl;  
   
             events = 1;  
   
             //this statment gets the pipe entry that was trigered  
             entries[indexPipeCountAssociator[pCount]].pipeSet = true;  
   
             if (pCount > 0) //this means activity on pipe is CIMOperation reques  
             {             {
                 cout << "In Monitor::run got Operation request" << endl;  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                 //entries[indx]._type = Monitor::CONNECTION;  
             }  
             else //this clause my not be needed in production but is used for testing  
             {             {
               cout << "In Monitor::run got Connection request" << endl;                       AutoMutex automut(Monitor::_cout_mut);
                        // cout << endl << "****************************" <<
                        //  "Monitor::run WaitForMultiPleObject returned activity on server pipe: "<<
                        //  pCount<< endl <<  endl;
                        cout << "Monitor::run WaitForMultiPleObject returned activity pipeEntrycount is " <<
                        pipeEntryCount <<
                        " this is the type " << entries[indexPipeCountAssociator[pCount]]._type << " this is index " << indexPipeCountAssociator[pCount] << endl;
             }             }
   #endif
   
                  /* There is a timeing problem here sometimes the wite in HTTPConnection i s
                not all the way done (has not _monitor->setState (_entry_index, _MonitorEntry::IDLE) )
                there for that should be done here if it is not done alread*/
   
                  if (entries[indexPipeCountAssociator[pCount]]._status.get() != _MonitorEntry::IDLE)
                  {
                      this->setState(indexPipeCountAssociator[pCount], _MonitorEntry::IDLE);
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
               AutoMutex automut(Monitor::_cout_mut);
  
                      cout << "setting state of index " << indexPipeCountAssociator[pCount]  << " to IDLE" << endl;
   #endif
         }         }
                 //  
  
  
               }
  
    // Sleep(2000);              pEvents = 1;
  
     //int events = 1;              //this statment gets the pipe entry that was trigered
     /*if (dwWait)              entries[indexPipeCountAssociator[pCount]].pipeSet = true;
     {  
         cout << "in Monitor::run about to call handlePipeConnectionEvent" << endl;  
         _handlePipeConnectionEvent(dwWait);  
     }*/  
  
           }
 #else #else
     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);      events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
 #endif #endif
     autoEntryMutex.lock();     autoEntryMutex.lock();
     // After enqueue a message and the autoEntryMutex has been released and locked again,     // After enqueue a message and the autoEntryMutex has been released and locked again,
Line 582 
Line 753 
     entries.reset(_entries);     entries.reset(_entries);
  
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
       if(pEvents == -1)
       {
           Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
             "Monitor::run - errorno = %d has occurred on select.",GetLastError() );
          // 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.
   
           // We need to generate an assert  here...
          PEGASUS_ASSERT(GetLastError()!= EBADF);
   
   
       }
   
     if(events == SOCKET_ERROR)     if(events == SOCKET_ERROR)
 #else #else
     if(events == -1)     if(events == -1)
 #endif #endif
     {     {
   
        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,        Tracer::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
Line 596 
Line 783 
  
        PEGASUS_ASSERT(errno != EBADF);        PEGASUS_ASSERT(errno != EBADF);
     }     }
     else if (events)      else if ((events)||(pEvents))
       {
   
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
     {     {
                    AutoMutex automut(Monitor::_cout_mut);
        cout << "IN Monior::run events= " << events << " pEvents= " << pEvents<< endl;
           }
   #endif
  
        cout << "IN Monior::run 'else if (events)' clause - array size is " <<  
             (int)entries.size() << endl;  
        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 indx = 0; indx < (int)entries.size(); indx++)
        {        {
              //cout << "Monitor::run at start of 'for( int indx = 0; indx ' - index = " << indx << endl;
           // 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.get() == _MonitorEntry::IDLE) &&          // cout << endl << " status of entry " << indx << " is " << entries[indx]._status.get() << endl;
              FD_ISSET(entries[indx].socket, &fdread)) ||            if((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
              (entries[indx].isNamedPipeConnection() && entries[indx].pipeSet))               ((FD_ISSET(entries[indx].socket, &fdread)&& (events)) ||
                (entries[indx].isNamedPipeConnection() && entries[indx].pipeSet && (pEvents))))
             {
   
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                 {
                    AutoMutex automut(Monitor::_cout_mut);
                    cout <<"Monitor::run - index  " << indx << " just got into 'if' statement" << endl;
                 }
   #endif
                 MessageQueue *q;
              try{
   
                    q = MessageQueue::lookup(entries[indx].queueId);
                 }
                catch (Exception e)
                {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                    AutoMutex automut(Monitor::_cout_mut);
                    cout << " this is what lookup gives - " << e.getMessage() << endl;
   #endif
                    exit(1);
                }
                catch(...)
           {           {
               cout << "IN Monior::run inside - for( int indx = 0-" << endl;  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
              MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);                   AutoMutex automut(Monitor::_cout_mut);
                    cout << "MessageQueue::lookup gives strange exception " << endl;
   #endif
                    exit(1);
                }
   
   
   
   
              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);
              //  printf("Monitor::run indx = %d, queueId =  %d, q = %p",
                //     indx, entries[indx].queueId, q);
                //cout << "Monitor::run before PEGASUS_ASSerT(q !=0) " << endl;
              PEGASUS_ASSERT(q !=0);              PEGASUS_ASSERT(q !=0);
  
   
              try              try
              {              {
                   /* {
                    AutoMutex automut(Monitor::_cout_mut);
                   cout <<" this is the type " << entries[indx]._type <<                   cout <<" this is the type " << entries[indx]._type <<
                       "for index " << indx << endl;                       "for index " << indx << endl;
                cout << "IN Monior::run right before entries[indx]._type == Monitor::CONNECTION" << endl;                cout << "IN Monior::run right before entries[indx]._type == Monitor::CONNECTION" << endl;
                    }*/
                 if(entries[indx]._type == Monitor::CONNECTION)                 if(entries[indx]._type == Monitor::CONNECTION)
                 {                 {
   
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                       {
                    cout << "In Monitor::run Monitor::CONNECTION clause" << endl;                    cout << "In Monitor::run Monitor::CONNECTION clause" << endl;
                       AutoMutex automut(Monitor::_cout_mut);
                       }
   #endif
   
                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                    Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                      "entries[indx].type for indx = %d is Monitor::CONNECTION", indx);                      "entries[indx].type for indx = %d is Monitor::CONNECTION", indx);
                    static_cast<HTTPConnection *>(q)->_entry_index = indx;                    static_cast<HTTPConnection *>(q)->_entry_index = indx;
Line 651 
Line 889 
                    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",
                    dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);                    dst->_entry_index, dst->_monitor->_entries[dst->_entry_index].queueId, dst);
   
                      /*In the case of named Pipes, the request has already been read from the pipe
                      therefor this section passed the request data to the HTTPConnection
                      NOTE: not sure if this would be better suited in a sparate private method
                      */
   
                      dst->setNamedPipe(entries[indx].namedPipe); //this step shouldn't be needd
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                      {
                          AutoMutex automut(Monitor::_cout_mut);
                      cout << "In Monitor::run after dst->setNamedPipe string read is " <<  entries[indx].namedPipe.raw << endl;
                      }
   #endif
                    try                    try
                    {                    {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                          {
                          AutoMutex automut(Monitor::_cout_mut);
                          cout << "In Monitor::run about to call 'dst->run(1)' "  << endl;
                          }
   #endif
                        dst->run(1);                        dst->run(1);
                    }                    }
                    catch (...)                    catch (...)
                    {                    {
   
                        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,                        Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                        "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);
  
                      if (entries[indx].isNamedPipeConnection())
                      {
                          entries[indx]._type = Monitor::ACCEPTOR;
                      }
   
                    // It is possible the entry status may not be set to busy.                    // It is possible the entry status may not be set to busy.
                    // The following will fail in that case.                    // The following will fail in that case.
                    // PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);                    // PEGASUS_ASSERT(dst->_monitor->_entries[dst->_entry_index]._status.get() == _MonitorEntry::BUSY);
Line 687 
Line 950 
                         // 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
             cout << " in - entries[indx]._type == Monitor::INTERNAL- " << endl;  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
               AutoMutex automut(Monitor::_cout_mut);
  
               cout << endl << " in - entries[indx]._type == Monitor::INTERNAL- " << endl << endl;
   #endif
               if (!entries[indx].isNamedPipeConnection())
               {
                         entries[indx]._status = _MonitorEntry::BUSY;                         entries[indx]._status = _MonitorEntry::BUSY;
                         static char buffer[2];                         static char buffer[2];
                         Socket::disableBlocking(entries[indx].socket);                         Socket::disableBlocking(entries[indx].socket);
Line 696 
Line 964 
                         Socket::enableBlocking(entries[indx].socket);                         Socket::enableBlocking(entries[indx].socket);
                         entries[indx]._status = _MonitorEntry::IDLE;                         entries[indx]._status = _MonitorEntry::IDLE;
                 }                 }
                   }
                 else                 else
                 {                 {
   
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
         {
               AutoMutex automut(Monitor::_cout_mut);
             cout << "In Monitor::run else clause of CONNECTION if statments" << endl;             cout << "In Monitor::run else clause of CONNECTION if statments" << endl;
         }
   #endif
                    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;
            Message *msg;            Message *msg;
  
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
             {
              AutoMutex automut(Monitor::_cout_mut);
              cout << " In Monitor::run Just before checking if NamedPipeConnection" << "for Index "<<indx<< endl;
              }
   #endif
            if (entries[indx].isNamedPipeConnection())            if (entries[indx].isNamedPipeConnection())
            {            {
                  if(!entries[indx].namedPipe.isConnectionPipe)
                  { /*if we enter this clasue it means that the named pipe that we are
                      looking at has recived a connection but is not the pipe we get connection requests over.
                      therefore we need to change the _type to CONNECTION and wait for a CIM Operations request*/
                      entries[indx]._type = Monitor::CONNECTION;
   
   
        /* This is a test  - this shows that the read file needs to be done
        before we call wiatForMultipleObjects*/
       /******************************************************
       ********************************************************/
   
   
   
           memset(entries[indx].namedPipe.raw,'\0',NAMEDPIPE_MAX_BUFFER_SIZE);
           BOOL rc = ::ReadFile(
                   entries[indx].namedPipe.getPipe(),
                   &entries[indx].namedPipe.raw,
                   NAMEDPIPE_MAX_BUFFER_SIZE,
                   &entries[indx].namedPipe.bytesRead,
                   entries[indx].namedPipe.getOverlap());
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
           {
            AutoMutex automut(Monitor::_cout_mut);
            cout << "Monitor::run just called read on index " << indx << endl;
           }
   #endif
   
            //&entries[indx].namedPipe.bytesRead = &size;
           if(!rc)
           {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
              AutoMutex automut(Monitor::_cout_mut);
              cout << "ReadFile failed for : "  << GetLastError() << "."<< endl;
   #endif
           }
   
   
   
       /******************************************************
       ********************************************************/
   
   
   
   
                    continue;
   
   
                  }
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                  {
                      AutoMutex automut(Monitor::_cout_mut);
                cout << " In Monitor::run about to create a Pipe message" << endl;                cout << " In Monitor::run about to create a Pipe message" << endl;
   
                  }
   #endif
                events |= NamedPipeMessage::READ;                events |= NamedPipeMessage::READ;
                msg = new NamedPipeMessage(entries[indx].namedPipe, events);                msg = new NamedPipeMessage(entries[indx].namedPipe, events);
            }            }
            else            else
            {            {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                  {
                  AutoMutex automut(Monitor::_cout_mut);
                  cout << " In Monitor::run ..its a socket message" << endl;
                  }
   #endif
                events |= SocketMessage::READ;                events |= SocketMessage::READ;
                        msg = new SocketMessage(entries[indx].socket, events);                        msg = new SocketMessage(entries[indx].socket, events);
            }            }
Line 725 
Line 1067 
            entries.reset(_entries);            entries.reset(_entries);
                    entries[indx]._status = _MonitorEntry::IDLE;                    entries[indx]._status = _MonitorEntry::IDLE;
  
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                      {
                          AutoMutex automut(Monitor::_cout_mut);
                          PEGASUS_STD(cout) << "Exiting:  Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
                      }
   #endif
                    return true;                    return true;
                 }                 }
              }              }
Line 736 
Line 1084 
        }        }
     }     }
  
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
     return(handled_events);     return(handled_events);
 } }
  
 void Monitor::stopListeningForConnections(Boolean wait) void Monitor::stopListeningForConnections(Boolean wait)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           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()");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::stopListeningForConnections()");
     // set boolean then tickle the server to recognize _stopConnections     // set boolean then tickle the server to recognize _stopConnections
     _stopConnections = 1;     _stopConnections = 1;
Line 755 
Line 1115 
     }     }
  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
 } }
  
  
Line 764 
Line 1130 
     Uint32 queueId,     Uint32 queueId,
     int type)     int type)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
    AutoMutex autoMut(_entry_mut);    AutoMutex autoMut(_entry_mut);
    // Check to see if we need to dynamically grow the _entries array    // Check to see if we need to dynamically grow the _entries array
Line 790 
Line 1162 
             _entries[index]._type = type;             _entries[index]._type = type;
             _entries[index]._status = _MonitorEntry::IDLE;             _entries[index]._status = _MonitorEntry::IDLE;
  
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
               {
                   AutoMutex automut(Monitor::_cout_mut);
                   PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
               }
   #endif
             return index;             return index;
          }          }
       }       }
Line 799 
Line 1177 
    }    }
    _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful    _solicitSocketCount--;  // decrease the count, if we are here we didnt do anything meaningful
    PEG_METHOD_EXIT();    PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
      {
          AutoMutex automut(Monitor::_cout_mut);
          PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
      }
   #endif
    return -1;    return -1;
  
 } }
  
 void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket) void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
  
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
     AutoMutex autoMut(_entry_mut);     AutoMutex autoMut(_entry_mut);
Line 838 
Line 1228 
         index--;         index--;
     }     }
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
 } }
  
 // 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) PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)
 { {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::_dispatch(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);    HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);
    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 885 
Line 1287 
     int type)     int type)
 { {
    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitPipeMessages");    PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitPipeMessages");
   
    AutoMutex autoMut(_entry_mut);    AutoMutex autoMut(_entry_mut);
    // Check to see if we need to dynamically grow the _entries array    // Check to see if we need to dynamically grow the _entries array
    // We always want the _entries array to 2 bigger than the    // We always want the _entries array to 2 bigger than the
    // current connections requested    // current connections requested
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
      AutoMutex automut(Monitor::_cout_mut);
    PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages at the begining" << PEGASUS_STD(endl);    PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages at the begining" << PEGASUS_STD(endl);
   #endif
  
  
    _solicitSocketCount++;  // bump the count    _solicitSocketCount++;  // bump the count
Line 915 
Line 1320 
             _entries[index].queueId  = queueId;             _entries[index].queueId  = queueId;
             _entries[index]._type = type;             _entries[index]._type = type;
             _entries[index]._status = _MonitorEntry::IDLE;             _entries[index]._status = _MonitorEntry::IDLE;
     #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
               AutoMutex automut(Monitor::_cout_mut);
             PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages after seting up  _entries[index] index = " << index << PEGASUS_STD(endl);             PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages after seting up  _entries[index] index = " << index << PEGASUS_STD(endl);
     #endif
             return index;             return index;
          }          }
       }       }
Line 934 
Line 1340 
  
 } }
  
   void Monitor::unsolicitPipeMessages(NamedPipe namedPipe)
   {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           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);
   
       /*
           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++)
       {
          if(_entries[index].namedPipe.getPipe() == namedPipe.getPipe())
          {
             _entries[index]._status = _MonitorEntry::EMPTY;
             //_entries[index].namedPipe = PEGASUS_INVALID_SOCKET;
             _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.get() == _MonitorEntry::EMPTY){
           if((_entries[index].namedPipe.getPipe() == namedPipe.getPipe()) ||
               (_entries.size() > MAX_NUMBER_OF_MONITOR_ENTRIES))
           {
               _entries.remove(index);
           }
           index--;
       }
       PEG_METHOD_EXIT();
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::unsolicitPipeMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
   #endif
   }
   
   
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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