(file) Return to MessageQueueService.cpp CVS log (file) (dir) Up to [Pegasus] / pegasus / src / Pegasus / Common

Diff for /pegasus/src/Pegasus/Common/MessageQueueService.cpp between version 1.109 and 1.117

version 1.109, 2005/05/31 14:20:04 version 1.117, 2005/10/14 17:40:18
Line 32 
Line 32 
 // Modified By: // Modified By:
 //              Amit K Arora, IBM (amita@in.ibm.com) for Bug#1090,#2657 //              Amit K Arora, IBM (amita@in.ibm.com) for Bug#1090,#2657
 //              Josephine Eskaline Joyce, IBM (jojustin@in.ibm.com) for Bug#3259 //              Josephine Eskaline Joyce, IBM (jojustin@in.ibm.com) for Bug#3259
   //              Jim Wunderlich (Jim_Wunderlich@prodigy.net)
 // //
 //%///////////////////////////////////////////////////////////////////////////// //%/////////////////////////////////////////////////////////////////////////////
  
   // #include <iostream.h>
 #include "MessageQueueService.h" #include "MessageQueueService.h"
 #include <Pegasus/Common/Tracer.h> #include <Pegasus/Common/Tracer.h>
 #include <Pegasus/Common/MessageLoader.h> //l10n #include <Pegasus/Common/MessageLoader.h> //l10n
Line 59 
Line 61 
    return _thread_pool;    return _thread_pool;
 } }
  
   //
   // MAX_THREADS_PER_SVC_QUEUE
   //
   // JR Wunderlich Jun 6, 2005
   //
   
   #define MAX_THREADS_PER_SVC_QUEUE_LIMIT 5000
   #define MAX_THREADS_PER_SVC_QUEUE_DEFAULT 5
   
   #ifndef MAX_THREADS_PER_SVC_QUEUE
   # define MAX_THREADS_PER_SVC_QUEUE MAX_THREADS_PER_SVC_QUEUE_DEFAULT
   #endif
   
   Uint32 max_threads_per_svc_queue;
   
 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL
 MessageQueueService::kill_idle_threads(void *parm) MessageQueueService::kill_idle_threads(void *parm)
 { {
Line 102 
Line 119 
          break;          break;
       }       }
  
         // The polling_routine thread must hold the lock on the
         // _polling_thread list while processing incoming messages.
         // This lock is used to give this thread ownership of
         // services on the _polling_routine list.
   
         // This is necessary to avoid confict with other threads
         // processing the _polling_list
         // (e.g., MessageQueueServer::~MessageQueueService).
   
       list->lock();       list->lock();
       MessageQueueService *service = list->next(0);       MessageQueueService *service = list->next(0);
         ThreadStatus rtn = PEGASUS_THREAD_OK;
       while(service != NULL)       while(service != NULL)
       {       {
          if (service->_incoming.count() > 0 && service->_die.value() == 0)            if ((service->_incoming.count() > 0) &&
                 (service->_die.value() == 0) &&
                 (service->_threads < max_threads_per_svc_queue))
             {
                // The _threads count is used to track the
                // number of active threads that have been allocated
                // to process messages for this service.
   
                // The _threads count MUST be incremented while
                // the polling_routine owns the _polling_thread
                // lock and has ownership of the service object.
   
                service->_threads++;
                try
          {          {
             _thread_pool->allocate_and_awaken(service, _req_proc);                   rtn = _thread_pool->allocate_and_awaken(
                         service, _req_proc, &_polling_sem);
          }          }
                catch (...)
                {
                    service->_threads--;
   
                    // allocate_and_awaken should never generate an exception.
                    PEGASUS_ASSERT(0);
                }
                // if no more threads available, break from processing loop
                if (rtn != PEGASUS_THREAD_OK )
                {
                    service->_threads--;
                    Logger::put(Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,
                       "Not enough threads to process this request. Skipping.");
   
                    Tracer::trace(TRC_MESSAGEQUEUESERVICE, Tracer::LEVEL2,
                       "Could not allocate thread for %s. " \
                       "Queue has %d messages waiting and %d threads servicing." \
                       "Skipping the service for right now. ",
                       service->getQueueName(),
                       service->_incoming.count(),
                       service->_threads.value());
   
                    pegasus_yield();
                    service = NULL;
                 }
             }
             if (service != NULL)
             {
          service = list->next(service);          service = list->next(service);
       }       }
         }
       list->unlock();       list->unlock();
   
       if (_check_idle_flag.value() != 0)       if (_check_idle_flag.value() != 0)
       {       {
          _check_idle_flag = 0;          _check_idle_flag = 0;
            // try to do idle thread clean up processing when system is not busy
            // if system is busy there may not be a thread available to allocate
            // so nothing will be done and that is OK.
   
            if ( _thread_pool->allocate_and_awaken(service, kill_idle_threads, &_polling_sem) != PEGASUS_THREAD_OK)
            {
                   Logger::put(Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,
                           "Not enough threads to kill idle threads. What an irony.");
   
                   Tracer::trace(TRC_MESSAGEQUEUESERVICE, Tracer::LEVEL2,
                           "Could not allocate thread to kill idle threads." \
                           "Skipping. ");
            }
   
  
          // If there are insufficent resources to run  
          // kill_idle_threads, then just return.  
          _thread_pool->allocate_and_awaken(service, kill_idle_threads);  
       }       }
    }    }
    myself->exit_self( (PEGASUS_THREAD_RETURN) 1 );    myself->exit_self( (PEGASUS_THREAD_RETURN) 1 );
Line 144 
Line 226 
      _incoming(true, 0),      _incoming(true, 0),
      _incoming_queue_shutdown(0)      _incoming_queue_shutdown(0)
 { {
   
    _capabilities = (capabilities | module_capabilities::async);    _capabilities = (capabilities | module_capabilities::async);
  
    _default_op_timeout.tv_sec = 30;    _default_op_timeout.tv_sec = 30;
    _default_op_timeout.tv_usec = 100;    _default_op_timeout.tv_usec = 100;
  
      max_threads_per_svc_queue = MAX_THREADS_PER_SVC_QUEUE;
   
      // if requested thread max is out of range, then set to
      // MAX_THREADS_PER_SVC_QUEUE_LIMIT
   
      if ((max_threads_per_svc_queue < 1) ||
          (max_threads_per_svc_queue > MAX_THREADS_PER_SVC_QUEUE_LIMIT))
      {
          max_threads_per_svc_queue = MAX_THREADS_PER_SVC_QUEUE_LIMIT;
      }
   
      Tracer::trace(TRC_MESSAGEQUEUESERVICE, Tracer::LEVEL2,
         "max_threads_per_svc_queue set to %u.", max_threads_per_svc_queue);
   
    AutoMutex autoMut(_meta_dispatcher_mutex);    AutoMutex autoMut(_meta_dispatcher_mutex);
  
    if (_meta_dispatcher == 0)    if (_meta_dispatcher == 0)
Line 160 
Line 257 
       {       {
          throw NullPointer();          throw NullPointer();
       }       }
         //  _thread_pool = new ThreadPool(initial_cnt, "MessageQueueService",
         //   minimum_cnt, maximum_cnt, deallocateWait);
         //
       _thread_pool =       _thread_pool =
           new ThreadPool(0, "MessageQueueService", 0, 0, deallocateWait);           new ThreadPool(0, "MessageQueueService", 0, 0, deallocateWait);
    }    }
Line 184 
Line 284 
 { {
    _die = 1;    _die = 1;
  
      // The polling_routine locks the _polling_list while
      // processing the incoming messages for services on the
      // list.  Deleting the service from the _polling_list
      // prior to processing, avoids synchronization issues
      // with the _polling_routine.
   
      _polling_list.remove(this);
   
      // ATTN: The code for closing the _incoming queue
      // is not working correctly. In OpenPegasus 2.5,
      // execution of the following code is very timing
      // dependent. This needs to be fix.
      // See Bug 4079 for details.
    if (_incoming_queue_shutdown.value() == 0)    if (_incoming_queue_shutdown.value() == 0)
    {    {
       _shutdown_incoming_queue();       _shutdown_incoming_queue();
   
    }    }
  
      // Wait until all threads processing the messages
      // for this service have completed.
   
  while (_threads.value() > 0)  while (_threads.value() > 0)
      {      {
           pegasus_yield();           pegasus_yield();
      }      }
    _polling_list.remove(this);  
    {    {
      AutoMutex autoMut(_meta_dispatcher_mutex);      AutoMutex autoMut(_meta_dispatcher_mutex);
      _service_count--;      _service_count--;
Line 262 
Line 377 
     // processed.     // processed.
      delete msg;      delete msg;
    }    }
      catch (const Permission &)
      {
        delete msg;
      }
 } }
  
  
Line 287 
Line 406 
  
         if (service->_die.value() != 0)         if (service->_die.value() != 0)
         {         {
               service->_threads--;
             return (0);             return (0);
         }         }
             service->_threads++;  
         // pull messages off the incoming queue and dispatch them. then         // pull messages off the incoming queue and dispatch them. then
         // check pending messages that are non-blocking         // check pending messages that are non-blocking
         AsyncOpNode *operation = 0;         AsyncOpNode *operation = 0;
Line 582 
Line 701 
           polling_routine,           polling_routine,
           reinterpret_cast<void *>(&_polling_list),           reinterpret_cast<void *>(&_polling_list),
           false);           false);
       while (!_polling_thread->run())        ThreadStatus tr = PEGASUS_THREAD_OK;
         while ( (tr =_polling_thread->run()) != PEGASUS_THREAD_OK)
       {       {
           if (tr == PEGASUS_THREAD_INSUFFICIENT_RESOURCES)
          pegasus_yield();          pegasus_yield();
           else
              throw Exception(MessageLoaderParms("Common.MessageQueueService.NOT_ENOUGH_THREAD",
                           "Could not allocate thread for the polling thread."));
       }       }
    }    }
 // ATTN optimization remove the message checking altogether in the base // ATTN optimization remove the message checking altogether in the base

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