(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 and 1.103.10.11

version 1.103, 2006/01/30 16:17:05 version 1.103.10.11, 2006/06/16 12:39:06
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);
   Mutex Monitor::_cout_mut;
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
    #define PIPE_INCREMENT 1
   #endif
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
Line 94 
Line 103 
      _tickle_server_socket(-1),      _tickle_server_socket(-1),
      _tickle_peer_socket(-1)      _tickle_peer_socket(-1)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     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 109 
Line 122 
        _MonitorEntry entry(0, 0, 0);        _MonitorEntry entry(0, 0, 0);
        _entries.append(entry);        _entries.append(entry);
     }     }
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
 Monitor::~Monitor() Monitor::~Monitor()
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");     Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
  
     try{     try{
Line 138 
Line 159 
     Socket::uninitializeInterface();     Socket::uninitializeInterface();
     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,     Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                   "returning from monitor destructor");                   "returning from monitor destructor");
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::~Monitor(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
 void Monitor::initializeTickler(){ void Monitor::initializeTickler(){
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     /*     /*
        NOTE: On any errors trying to        NOTE: On any errors trying to
              setup out tickle connection,              setup out tickle connection,
Line 323 
Line 352 
     _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);
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::initializeTickler(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
 void Monitor::tickle(void) void Monitor::tickle(void)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     static char _buffer[] =     static char _buffer[] =
     {     {
       '0','0'       '0','0'
Line 336 
Line 373 
     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);
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::tickle(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
 void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status ) void Monitor::setState( Uint32 index, _MonitorEntry::entry_status status )
Line 346 
Line 387 
  
 Boolean Monitor::run(Uint32 milliseconds) Boolean Monitor::run(Uint32 milliseconds)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
  
     Boolean handled_events = false;     Boolean handled_events = false;
     int i = 0;     int i = 0;
Line 448 
Line 493 
         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.
     */     */
       //Array<HANDLE> pipeEventArray;
     PEGASUS_SOCKET maxSocketCurrentPass = 0;     PEGASUS_SOCKET maxSocketCurrentPass = 0;
     for( int indx = 0; indx < (int)entries.size(); indx++)      int indx;
   
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
   
       //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
       //index of the fist named pipe connection in the entries array
       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++)
       {
   
   
   #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);
              /*
              if(!entries[indx].namedPipe.isConnectionPipe)
              {
                  BOOL rc = ::ReadFile(
                          entries[indx].namedPipe.getPipe(),
                          &entries[indx].namedPipe.raw,
                          MAX_BUFFER_SIZE,
                          &entries[indx].namedPipe.bytesRead,
                          &entries[indx].namedPipe.getOverlap());
   
                  cout << "just called read on index " << indx << endl;
   
                   //&entries[indx].namedPipe.bytesRead = &size;
                  if(!rc)
                  {
   
                     cout << "ReadFile failed for : "  << GetLastError() << "."<< endl;
   
                  }
   
              }
             */
   
       pipeEntryCount++;
   
   
   
             // cout << "Monitor::run pipeEntrycount is " << pipeEntryCount <<
             // " this is the type " << entries[indx]._type << " this is index " << indx << endl;
   
          }
          else
   
   #endif
          {
   
        if(maxSocketCurrentPass < entries[indx].socket)        if(maxSocketCurrentPass < entries[indx].socket)
             maxSocketCurrentPass = entries[indx].socket;             maxSocketCurrentPass = entries[indx].socket;
  
Line 459 
Line 594 
            _idleEntries++;            _idleEntries++;
            FD_SET(entries[indx].socket, &fdread);            FD_SET(entries[indx].socket, &fdread);
        }        }
   
          }
     }     }
  
     /*     /*
Line 475 
Line 612 
     // 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.
     //     //
 #ifdef PEGASUS_OS_TYPE_WINDOWS #ifdef PEGASUS_OS_TYPE_WINDOWS
     int events = select(0, &fdread, NULL, NULL, &tv);      //int events = select(0, &fdread, NULL, NULL, &tv);
        int events = 0;
           DWORD dwWait=NULL;
       int pEvents = 0;
   
    //       cout << "events after select" << events << endl;
       cout << "Calling WaitForMultipleObjects\n";
   
       //this should be in a try block
   
       dwWait = WaitForMultipleObjects(MaxPipes,
                                      hEvents,      //ABB:- array of event objects
                                      FALSE,        // ABB:-does not wait for all
                                      20000);        //ABB:- timeout value
   
       if(dwWait == WAIT_TIMEOUT)
           {
           cout << "Wait WAIT_TIMEOUT\n";
   
                      // Sleep(2000);
               //continue;
   
                //return false;  // I think we do nothing.... Mybe there is a socket connection... so
                // cant return.
           }
           else if (dwWait == WAIT_FAILED)
           {
               cout << "Wait Failed returned\n";
               cout << "failed with " << GetLastError() << "." << endl;
               pEvents = -1;
               return false;
           }
           else
           {
               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;
                   //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;
   
               }
   
           }
                   //
   
   
      // Sleep(2000);
   
       //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);     int events = select(maxSocketCurrentPass, &fdread, NULL, NULL, &tv);
 #endif #endif
Line 483 
Line 685 
     // 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,
     // the array of _entries can be changed. The ArrayIterator has be reset with the original _entries     // the array of _entries can be changed. The ArrayIterator has be reset with the original _entries
     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 498 
Line 717 
  
        PEGASUS_ASSERT(errno != EBADF);        PEGASUS_ASSERT(errno != EBADF);
     }     }
     else if (events)      else if ((events)||(pEvents))
     {     {
   
        //  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) &&            if(((entries[indx]._status.get() == _MonitorEntry::IDLE) &&
              (FD_ISSET(entries[indx].socket, &fdread)))               FD_ISSET(entries[indx].socket, &fdread)&& (events)) ||
                (entries[indx].isNamedPipeConnection() && entries[indx].pipeSet && (pEvents)))
             {
                 MessageQueue *q;
                 cout << "IN Monior::run inside - for int indx = " <<indx <<
                     "and queue ID is " << entries[indx].queueId << endl;
                 try{
   
                    q = MessageQueue::lookup(entries[indx].queueId);
                 }
                catch (Exception e)
           {           {
              MessageQueue *q = MessageQueue::lookup(entries[indx].queueId);                   cout << " this is what lookup gives - " << e.getMessage() << endl;
                    exit(1);
                }
                catch(...)
                {
                    cout << "MessageQueue::lookup gives strange exception " << endl;
                    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;
              PEGASUS_ASSERT(q !=0);              PEGASUS_ASSERT(q !=0);
  
              try              try
              {              {
                     cout <<" this is the type " << entries[indx]._type <<
                         "for index " << indx << 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)
                 {                 {
                       cout << "In Monitor::run Monitor::CONNECTION clause" << endl;
   
                    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 544 
Line 794 
                    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
                      cout << "In Monitor::run after dst->setNamedPipe string read is " <<
                          entries[indx].namedPipe.raw << endl;
   
                    try                    try
                    {                    {
                          cout << "In Monitor::run about to call 'dst->run(1)' "  << endl;
                        dst->run(1);                        dst->run(1);
                    }                    }
                    catch (...)                    catch (...)
Line 556 
Line 816 
                    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);
  
                      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 580 
Line 842 
                         // 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;
  
                         entries[indx]._status = _MonitorEntry::BUSY;                         entries[indx]._status = _MonitorEntry::BUSY;
                         static char buffer[2];                         static char buffer[2];
Line 590 
Line 853 
                 }                 }
                 else                 else
                 {                 {
               cout << "In Monitor::run else clause of CONNECTION if statments" << endl;
                    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;
              cout << " In Monitor::run Just before checking if NamedPipeConnection" << "for Index "<<indx<< endl;
   
              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 neew 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*/
       /******************************************************
       ********************************************************/
   
                   //DWORD size = 0;
   
           BOOL rc = ::ReadFile(
                   entries[indx].namedPipe.getPipe(),
                   &entries[indx].namedPipe.raw,
                   MAX_BUFFER_SIZE,
                   &entries[indx].namedPipe.bytesRead,
                   &entries[indx].namedPipe.getOverlap());
   
           cout << "just called read on index " << indx << endl;
   
            //&entries[indx].namedPipe.bytesRead = &size;
           if(!rc)
           {
   
              cout << "ReadFile failed for : "  << GetLastError() << "."<< endl;
   
           }
   
   
   
       /******************************************************
       ********************************************************/
   
   
   
   
                      continue;
   
                  }
                  cout << " In Monitor::run about to create a Pipe message" << endl;
                  events |= NamedPipeMessage::READ;
                  msg = new NamedPipeMessage(entries[indx].namedPipe, events);
              }
              else
              {
                  cout << " In Monitor::run ..its a socket message" << endl;
                    events |= SocketMessage::READ;                    events |= SocketMessage::READ;
                    Message *msg = new SocketMessage(entries[indx].socket, events);                         msg = new SocketMessage(entries[indx].socket, events);
              }
   
                    entries[indx]._status = _MonitorEntry::BUSY;                    entries[indx]._status = _MonitorEntry::BUSY;
                    autoEntryMutex.unlock();                    autoEntryMutex.unlock();
                    q->enqueue(msg);                    q->enqueue(msg);
Line 604 
Line 924 
            entries.reset(_entries);            entries.reset(_entries);
                    entries[indx]._status = _MonitorEntry::IDLE;                    entries[indx]._status = _MonitorEntry::IDLE;
  
                      {
                          AutoMutex automut(Monitor::_cout_mut);
                          PEGASUS_STD(cout) << "Exiting:  Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
                      }
                    return true;                    return true;
                 }                 }
              }              }
Line 615 
Line 939 
        }        }
     }     }
  
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::run(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     return(handled_events);     return(handled_events);
 } }
  
 void Monitor::stopListeningForConnections(Boolean wait) void Monitor::stopListeningForConnections(Boolean wait)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
     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 634 
Line 966 
     }     }
  
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::stopListeningForConnections(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
  
Line 643 
Line 979 
     Uint32 queueId,     Uint32 queueId,
     int type)     int type)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
    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 669 
Line 1009 
             _entries[index]._type = type;             _entries[index]._type = type;
             _entries[index]._status = _MonitorEntry::IDLE;             _entries[index]._status = _MonitorEntry::IDLE;
  
               {
                   AutoMutex automut(Monitor::_cout_mut);
                   PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
               }
             return index;             return index;
          }          }
       }       }
Line 678 
Line 1022 
    }    }
    _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();
      {
          AutoMutex automut(Monitor::_cout_mut);
          PEGASUS_STD(cout) << "Exiting:  Monitor::solicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
      }
    return -1;    return -1;
  
 } }
  
 void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket) void Monitor::unsolicitSocketMessages(PEGASUS_SOCKET socket)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
  
     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
     AutoMutex autoMut(_entry_mut);     AutoMutex autoMut(_entry_mut);
Line 717 
Line 1069 
         index--;         index--;
     }     }
     PEG_METHOD_EXIT();     PEG_METHOD_EXIT();
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Exiting:  Monitor::unsolicitSocketMessages(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
 } }
  
 // 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)
 { {
       {
           AutoMutex automut(Monitor::_cout_mut);
           PEGASUS_STD(cout) << "Entering: Monitor::_dispatch(): (tid:" << Uint32(pegasus_thread_self()) << ")" << PEGASUS_STD(endl);
       }
    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 755 
Line 1115 
    return 0;    return 0;
 } }
  
   
   //This method is anlogsu 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,
       int type)
   {
      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitPipeMessages");
      AutoMutex autoMut(_entry_mut);
      // Check to see if we need to dynamically grow the _entries array
      // We always want the _entries array to 2 bigger than the
      // current connections requested
      PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages at the begining" << PEGASUS_STD(endl);
   
   
   
      _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);
           }
      }
   
      int index;
      for(index = 1; index < (int)_entries.size(); index++)
      {
         try
         {
            if(_entries[index]._status.get() == _MonitorEntry::EMPTY)
            {
               _entries[index].socket = NULL;
               _entries[index].namedPipe = namedPipe;
               _entries[index].namedPipeConnection = true;
               _entries[index].queueId  = queueId;
               _entries[index]._type = type;
               _entries[index]._status = _MonitorEntry::IDLE;
   
               PEGASUS_STD(cout) << "In Monitor::solicitPipeMessages after seting up  _entries[index] index = " << index << PEGASUS_STD(endl);
   
               return index;
            }
         }
         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);
   
      PEG_METHOD_EXIT();
      return -1;
   
   }
   
   
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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