(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.1.2.1 and 1.39

version 1.1.2.1, 2001/07/26 17:19:09 version 1.39, 2003/05/06 16:36:44
Line 1 
Line 1 
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001 The Open group, BMC Software, Tivoli Systems, IBM  // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,
   // The Open Group, Tivoli Systems
 // //
 // Permission is hereby granted, free of charge, to any person obtaining a copy // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to // of this software and associated documentation files (the "Software"), to
Line 26 
Line 27 
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
 #define FD_SETSIZE 1024  #include <Pegasus/Common/Config.h>
 #include <winsock.h>  
 #include <cstring> #include <cstring>
 #include "Monitor.h" #include "Monitor.h"
 #include "MessageQueue.h" #include "MessageQueue.h"
   #include "Socket.h"
   #include <Pegasus/Common/Tracer.h>
   #include <Pegasus/Common/HTTPConnection.h>
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
   # if defined(FD_SETSIZE) && FD_SETSIZE != 1024
   #  error "FD_SETSIZE was not set to 1024 prior to the last inclusion \
   of <winsock.h>. It may have been indirectly included (e.g., by including \
   <windows.h>). Finthe inclusion of that header which is visible to this \
   compilation unit and #define FD_SETZIE to 1024 prior to that inclusion; \
   otherwise, less than 64 clients (the default) will be able to connect to the \
   CIMOM. PLEASE DO NOT SUPPRESS THIS WARNING; PLEASE FIX THE PROBLEM."
   
   # endif
   # define FD_SETSIZE 1024
   # include <windows.h>
   #else
   # include <sys/types.h>
   # include <sys/socket.h>
   # include <sys/time.h>
   # include <netinet/in.h>
   # include <netdb.h>
   # include <arpa/inet.h>
   #endif
  
 PEGASUS_USING_STD; PEGASUS_USING_STD;
  
 PEGASUS_NAMESPACE_BEGIN PEGASUS_NAMESPACE_BEGIN
  
 ////////////////////////////////////////////////////////////////////////////////  
 //  
 // Local routines:  
 //  
 ////////////////////////////////////////////////////////////////////////////////  
  
 static inline int select_wrapper(  static AtomicInt _connections = 0;
     int nfds,  
     fd_set* rd_fd_set,  
     fd_set* wr_fd_set,  
     fd_set* ex_fd_set,  
     const struct timeval* tv)  
 {  
     return select(FD_SETSIZE, rd_fd_set, wr_fd_set, ex_fd_set, tv);  
 }  
  
 ////////////////////////////////////////////////////////////////////////////////  
 //  
 // Routines for starting and stoping socket interface.  
 //  
 ////////////////////////////////////////////////////////////////////////////////  
   
 extern Uint32 _socketInterfaceInitCount;  
   
 static void _SocketInterfaceInc()  
 {  
     if (_socketInterfaceInitCount == 0)  
     {  
         WSADATA tmp;  
   
         if (WSAStartup(0x202, &tmp) == SOCKET_ERROR)  
             WSACleanup();  
     }  
   
     _socketInterfaceInitCount++;  
 }  
   
 static void _SocketInterfaceDec()  
 {  
     _socketInterfaceInitCount--;  
  
     if (_socketInterfaceInitCount == 0)  static struct timeval create_time = {0, 1};
         WSACleanup();  static struct timeval destroy_time = {300, 0};
 }  static struct timeval deadlock_time = {0, 0};
  
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
 // //
Line 104 
Line 91 
 //////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
  
 Monitor::Monitor() Monitor::Monitor()
      : _module_handle(0), _controller(0), _async(false)
 { {
     _SocketInterfaceInc();      Socket::initializeInterface();
       _rep = 0;
     _rep = new MonitorRep;      _entries.reserveCapacity(32);
     FD_ZERO(&_rep->rd_fd_set);      for( int i = 0; i < 32; i++ )
     FD_ZERO(&_rep->wr_fd_set);      {
     FD_ZERO(&_rep->ex_fd_set);         _MonitorEntry entry(0, 0, 0);
     FD_ZERO(&_rep->active_rd_fd_set);         _entries.append(entry);
     FD_ZERO(&_rep->active_wr_fd_set);      }
     FD_ZERO(&_rep->active_ex_fd_set);  
 } }
  
 MonitorHandler::~MonitorHandler()  Monitor::Monitor(Boolean async)
      : _module_handle(0), _controller(0), _async(async)
   {
       Socket::initializeInterface();
       _rep = 0;
       _entries.reserveCapacity(32);
       for( int i = 0; i < 32; i++ )
 { {
     _SocketInterfaceDec();         _MonitorEntry entry(0, 0, 0);
          _entries.append(entry);
 } }
       if( _async == true )
 Monitor::~Monitor()  
 { {
     delete _rep;         _thread_pool = new ThreadPool(0,
                                        "Monitor",
                                        0,
                                        0,
                                        create_time,
                                        destroy_time,
                                        deadlock_time);
       }
       else
          _thread_pool = 0;
 } }
  
 Boolean Monitor::select(Uint32 milliseconds)  Monitor::~Monitor()
 { {
     // Windows select() has a strange little bug. It returns immediately if      Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
     // there are no descriptors in the set even if the timeout is non-zero.                    "deregistering with module controller");
     // To work around this, we call Sleep() for now:  
  
     if (_entries.size() == 0)      if(_module_handle != NULL)
         Sleep(milliseconds);      {
          _controller->deregister_module(PEGASUS_MODULENAME_MONITOR);
          _controller = 0;
          delete _module_handle;
       }
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "deleting rep");
  
     // Check for events on the selected file descriptors. Only do this if      Tracer::trace(TRC_HTTP, Tracer::LEVEL4, "uninitializing interface");
     // there were no undispatched events from last time.      Socket::uninitializeInterface();
       Tracer::trace(TRC_HTTP, Tracer::LEVEL4,
                     "returning from monitor destructor");
       if(_async == true)
          delete _thread_pool;
   }
  
     static int count = 0;  
  
     if (count == 0)  int Monitor::kill_idle_threads()
     {     {
         memcpy(&_rep->active_rd_fd_set, &_rep->rd_fd_set, sizeof(fd_set));     static struct timeval now, last;
         memcpy(&_rep->active_wr_fd_set, &_rep->wr_fd_set, sizeof(fd_set));     gettimeofday(&now, NULL);
         memcpy(&_rep->active_ex_fd_set, &_rep->ex_fd_set, sizeof(fd_set));     int dead_threads = 0;
   
         const Uint32 SEC = milliseconds / 1000;  
         const Uint32 USEC = (milliseconds % 1000) * 1000;  
         struct timeval tv = { SEC, USEC };  
  
         count = select_wrapper(     if( now.tv_sec - last.tv_sec > 120 )
             FD_SETSIZE,  
             &_rep->active_rd_fd_set,  
             &_rep->active_wr_fd_set,  
             &_rep->active_ex_fd_set,  
             &tv);  
   
         if (count == 0)  
             return false;  
         else if (count == SOCKET_ERROR)  
         {         {
             count = 0;        gettimeofday(&last, NULL);
             return false;        try
         }        {
            dead_threads =  _thread_pool->kill_dead_threads();
     }     }
         catch(IPCException& )
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)  
     {     {
         Sint32 socket = _entries[i].socket;        }
         Uint32 events = 0;  
   
         if (FD_ISSET(socket, &_rep->active_rd_fd_set))  
             events |= READ;  
  
         if (FD_ISSET(socket, &_rep->active_wr_fd_set))     }
             events |= WRITE;     return dead_threads;
   }
  
         if (FD_ISSET(socket, &_rep->active_ex_fd_set))  
             events |= EXCEPTION;  
  
         if (events)  Boolean Monitor::run(Uint32 milliseconds)
         {         {
             MessageQueue* queue = MessageQueue::lookup(_entries[i].queueId);  
  
             if (!queue)      Boolean handled_events = false;
                 unsolicitSocketMessages(_entries[i].queueId);      int i = 0;
       #if defined(PEGASUS_OS_OS400) || defined(PEGASUS_OS_HPUX)
       struct timeval tv = {milliseconds/1000, milliseconds%1000*1000};
   #else
       struct timeval tv = {0, 1};
   #endif
       fd_set fdread;
       FD_ZERO(&fdread);
       _entry_mut.lock(pegasus_thread_self());
  
             if (events & WRITE)      for( int indx = 0; indx < (int)_entries.size(); indx++)
             {             {
                 FD_CLR(socket, &_rep->active_wr_fd_set);         if(_entries[indx]._status.value() == _MonitorEntry::IDLE)
          {
             FD_SET(_entries[indx].socket, &fdread);
          }
             }             }
  
             if (events & EXCEPTION)  
       int events = select(FD_SETSIZE, &fdread, NULL, NULL, &tv);
   
   #ifdef PEGASUS_OS_TYPE_WINDOWS
       if(events && events != SOCKET_ERROR )
   #else
       if(events && events != -1 )
   #endif
       {
          for( int indx = 0; indx < (int)_entries.size(); indx++)
          {
             if(FD_ISSET(_entries[indx].socket, &fdread))
             {             {
                 FD_CLR(socket, &_rep->active_ex_fd_set);               MessageQueue *q = MessageQueue::lookup(_entries[indx].queueId);
                if(q == 0)
                {
                   try
                   {
                      _entries[indx]._status = _MonitorEntry::EMPTY;
             }             }
                   catch(...)
                   {
  
             if (events & READ)                  }
                   continue;
                }
                try
                {
                   if(_entries[indx]._type == Monitor::CONNECTION)
             {             {
                 FD_CLR(socket, &_rep->active_rd_fd_set);                     static_cast<HTTPConnection *>(q)->_entry_index = indx;
                      if(static_cast<HTTPConnection *>(q)->_dying.value() > 0 )
                      {
                         _entries[indx]._status = _MonitorEntry::DYING;
                         MessageQueue & o = static_cast<HTTPConnection *>(q)->get_owner();
                         Message* message= new CloseConnectionMessage(_entries[indx].socket);
                         message->dest = o.getQueueId();
                         _entry_mut.unlock();
                         o.enqueue(message);
                         return true;
                      }
                      _entries[indx]._status = _MonitorEntry::BUSY;
                      _thread_pool->allocate_and_awaken((void *)q, _dispatch);
             }             }
                   else
                   {
                      int events = 0;
                      events |= SocketMessage::READ;
                      Message *msg = new SocketMessage(_entries[indx].socket, events);
                      _entries[indx]._status = _MonitorEntry::BUSY;
                      _entry_mut.unlock();
  
             count--;                     q->enqueue(msg);
                      _entries[indx]._status = _MonitorEntry::IDLE;
             return true;             return true;
         }         }
     }     }
                catch(...)
     return false;               {
 } }
                handled_events = true;
             }
          }
       }
       _entry_mut.unlock();
       return(handled_events);
   }
   
  
 Boolean Monitor::solicitSocketMessages(  
   int  Monitor::solicitSocketMessages(
     Sint32 socket,     Sint32 socket,
     Uint32 events,     Uint32 events,
     Uint32 queueId);      Uint32 queueId,
       int type)
 { {
     // See whether a handler is already registered for this one:  
   
     Uint32 pos = _findEntry(socket);  
   
     if (pos != PEGUSUS_NOT_FOUND)  
         return false;  
   
     // Set the events:  
   
     if (events & READ)  
         FD_SET(socket, &_rep->rd_fd_set);  
   
     if (events & WRITE)  
         FD_SET(socket, &_rep->wr_fd_set);  
  
     if (events & EXCEPTION)     PEG_METHOD_ENTER(TRC_HTTP, "Monitor::solicitSocketMessages");
         FD_SET(socket, &_rep->ex_fd_set);  
  
     // Add the entry to the list:     int index = -1;
      _entry_mut.lock(pegasus_thread_self());
  
     _MonitorEntry entry = { socket, queueId };     for(index = 0; index < (int)_entries.size(); index++)
     _entries.append(entry);     {
         try
         {
            if(_entries[index]._status.value() == _MonitorEntry::EMPTY)
            {
               _entries[index].socket = socket;
               _entries[index].queueId  = queueId;
               _entries[index]._type = type;
               _entries[index]._status = _MonitorEntry::IDLE;
               _entry_mut.unlock();
  
     // Success!              return index;
            }
         }
         catch(...)
         {
         }
  
     return true;     }
         _entry_mut.unlock();
      PEG_METHOD_EXIT();
      return index;
 } }
  
 Boolean Monitor::unolicitSocketMessages(Sint32 socket)  void Monitor::unsolicitSocketMessages(Sint32 socket)
 { {
     // Look for the given entry and remove it:      PEG_METHOD_ENTER(TRC_HTTP, "Monitor::unsolicitSocketMessages");
       _entry_mut.lock(pegasus_thread_self());
  
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)      for(int index = 0; index < (int)_entries.size(); index++)
     {     {
         if (_entries[i].socket == socket)         if(_entries[index].socket == socket)
         {         {
             Sint32 socket = _entries[i].socket;            _entries[index]._status = _MonitorEntry::EMPTY;
             FD_CLR(socket, &_rep->rd_fd_set);            break;
             FD_CLR(socket, &_rep->wr_fd_set);  
             FD_CLR(socket, &_rep->ex_fd_set);  
             _entries.remove(i);  
             return true;  
         }         }
     }     }
       _entry_mut.unlock();
     return false;      PEG_METHOD_EXIT();
 } }
  
 Uint32 Monitor::_findEntry(Sint32 socket) const  PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL Monitor::_dispatch(void *parm)
 { {
     for (Uint32 i = 0, n = _entries.size(); i < n; i++)     HTTPConnection *dst = reinterpret_cast<HTTPConnection *>(parm);
     {  
         if (_entries[i].socket == socket)     dst->run(1);
             return i;     if(  dst->_monitor->_entries.size() > (Uint32)dst->_entry_index )
     }        dst->_monitor->_entries[dst->_entry_index]._status = _MonitorEntry::IDLE;
  
     return PEG_NOT_FOUND;     return 0;
 } }
  
   
   
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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