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

version 1.103.10.8, 2006/06/13 14:32:39 version 1.103.10.22, 2006/07/21 18:17:51
Line 54 
Line 54 
  
  
  
 const static DWORD MAX_BUFFER_SIZE = 4096;  // 4 kilobytes  //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
Line 82 
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 #ifdef PEGASUS_OS_TYPE_WINDOWS
  #define PIPE_INCREMENT 1  #define PIPE_INCREMENT 1
Line 103 
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 118 
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 147 
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 332 
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 345 
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 355 
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 416 
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 511 
Line 587 
             }             }
  
            //pipeEventArray.append((entries[indx].namedPipe.getOverlap()).hEvent);            //pipeEventArray.append((entries[indx].namedPipe.getOverlap()).hEvent);
            hEvents[pipeEntryCount] = entries[indx].namedPipe.getOverlap().hEvent;             hEvents[pipeEntryCount] = entries[indx].namedPipe.getOverlap()->hEvent;
  
            indexPipeCountAssociator.append(indx);            indexPipeCountAssociator.append(indx);
  
Line 519 
Line 595 
  
  
  
           // cout << "Monitor::run pipeEntrycount is " << pipeEntryCount <<  
           // " this is the type " << entries[indx]._type << " this is index " << indx << endl;  
   
        }        }
        else        else
  
Line 553 
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);  
      int events = 0;     // events = select(0, &fdread, NULL, NULL, &tv);
   
       //if (events == NULL)
       //{  // This connection uses namedPipes
   
           events = 0;
         DWORD dwWait=NULL;         DWORD dwWait=NULL;
     int pEvents = 0;          pEvents = 0;
  
  //       cout << "events after select" << events << endl;  
     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;
Line 581 
Line 681 
         }         }
         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;             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;  
   
             pEvents = 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;                   {
                        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;
             }             }
             else //this clause my not be needed in production but is used for testing  #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)
             {             {
          //     cout << "In Monitor::run got Connection request" << endl;                     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
             }             }
  
   
         }         }
                 //  
  
               pEvents = 1;
  
    // Sleep(2000);              //this statment gets the pipe entry that was trigered
               entries[indexPipeCountAssociator[pCount]].pipeSet = true;
  
     //int events = 1;          }
     /*if (dwWait)  
     {  
         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 662 
Line 786 
     else if ((events)||(pEvents))     else if ((events)||(pEvents))
     {     {
  
      //  cout << "IN Monior::run 'else if (events)' clause - array size is " <<  #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
      //       (int)entries.size() << endl;          {
                    AutoMutex automut(Monitor::_cout_mut);
        cout << "IN Monior::run events= " << events << " pEvents= " << pEvents<< endl;
           }
   #endif
   
        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);
Line 672 
Line 801 
            //cout << "Monitor::run at start of 'for( int indx = 0; indx ' - index = " << indx << endl;            //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)&& (events)) ||            if((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
              (entries[indx].isNamedPipeConnection() && entries[indx].pipeSet && (pEvents)))               ((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;               MessageQueue *q;
               cout << "IN Monior::run inside - for int indx = " <<indx <<  
                   "and queue ID is " << entries[indx].queueId << endl;  
               try{               try{
  
                  q = MessageQueue::lookup(entries[indx].queueId);                  q = MessageQueue::lookup(entries[indx].queueId);
               }               }
              catch (Exception e)              catch (Exception e)
              {              {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                    AutoMutex automut(Monitor::_cout_mut);
                  cout << " this is what lookup gives - " << e.getMessage() << endl;                  cout << " this is what lookup gives - " << e.getMessage() << endl;
   #endif
                  exit(1);                  exit(1);
              }              }
              catch(...)              catch(...)
              {              {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
                    AutoMutex automut(Monitor::_cout_mut);
                  cout << "MessageQueue::lookup gives strange exception " << endl;                  cout << "MessageQueue::lookup gives strange exception " << endl;
   #endif
                  exit(1);                  exit(1);
              }              }
  
  
  
              cout << "Monitor::run after MessageQueue::lookup(entries[indx].queueId)" << endl;  
               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);
              cout << "Monitor::run before PEGASUS_ASSerT(q !=0) " << endl;             //  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);
Line 741 
Line 894 
                    therefor this section passed the request data to the HTTPConnection                    therefor this section passed the request data to the HTTPConnection
                    NOTE: not sure if this would be better suited in a sparate private method                    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  
                    cout << "In Monitor::run after dst->setNamedPipe string read is " <<  
                        entries[indx].namedPipe.raw << endl;  
  
                      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;                        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 782 
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 791 
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;
            cout << " In Monitor::run Just before checking if NamedPipeConnection" << "for Index "<<indx<< endl;  
  
   #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(!entries[indx].namedPipe.isConnectionPipe)
                { /*if we enter this clasue it means that the named pipe that we are                { /*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.                    looking at has recived a connection but is not the pipe we get connection requests over.
                    therefore we neew to change the _type to CONNECTION and wait for a CIM Operations request*/                     therefore we need to change the _type to CONNECTION and wait for a CIM Operations request*/
                    entries[indx]._type = Monitor::CONNECTION;                    entries[indx]._type = Monitor::CONNECTION;
  
  
Line 814 
Line 999 
     /******************************************************     /******************************************************
     ********************************************************/     ********************************************************/
  
                 //DWORD size = 0;  
  
   
           memset(entries[indx].namedPipe.raw,'\0',NAMEDPIPE_MAX_BUFFER_SIZE);
         BOOL rc = ::ReadFile(         BOOL rc = ::ReadFile(
                 entries[indx].namedPipe.getPipe(),                 entries[indx].namedPipe.getPipe(),
                 &entries[indx].namedPipe.raw,                 &entries[indx].namedPipe.raw,
                 MAX_BUFFER_SIZE,                  NAMEDPIPE_MAX_BUFFER_SIZE,
                 &entries[indx].namedPipe.bytesRead,                 &entries[indx].namedPipe.bytesRead,
                 &entries[indx].namedPipe.getOverlap());                  entries[indx].namedPipe.getOverlap());
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
         cout << "just called read on index " << indx << endl;          {
            AutoMutex automut(Monitor::_cout_mut);
            cout << "Monitor::run just called read on index " << indx << endl;
           }
   #endif
  
          //&entries[indx].namedPipe.bytesRead = &size;          //&entries[indx].namedPipe.bytesRead = &size;
         if(!rc)         if(!rc)
         {         {
   #ifdef PEGASUS_LOCALDOMAINSOCKET_DEBUG
              AutoMutex automut(Monitor::_cout_mut);
            cout << "ReadFile failed for : "  << GetLastError() << "."<< endl;            cout << "ReadFile failed for : "  << GetLastError() << "."<< endl;
   #endif
         }         }
  
  
Line 843 
Line 1034 
  
                    continue;                    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;                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 864 
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 875 
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 894 
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 903 
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 929 
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 938 
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 977 
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 1024 
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 1054 
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 1073 
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.8  
changed lines
  Added in v.1.103.10.22
No CVS admin address has been configured
 Powered by
ViewCVS 0.9.2