(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.72 and 1.82.6.1

version 1.72, 2003/08/08 15:28:19 version 1.82.6.1, 2003/10/29 22:09:18
Line 1 
Line 1 
 //%////-*-c++-*-////////////////////////////////////////////////////////////////  //%2003////////////////////////////////////////////////////////////////////////
 // //
 // Copyright (c) 2000, 2001, 2002 BMC Software, Hewlett-Packard Company, IBM,  // Copyright (c) 2000, 2001, 2002  BMC Software, Hewlett-Packard Development
 // The Open Group, Tivoli Systems  // Company, L. P., IBM Corp., The Open Group, Tivoli Systems.
   // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L. P.;
   // IBM Corp.; EMC Corporation, The Open Group.
 // //
 // 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 80 
Line 82 
 } }
  
  
 void MessageQueueService::force_shutdown(void)  void MessageQueueService::force_shutdown(Boolean destroy_flag)
 { {
      return;
   
   #ifdef MESSAGEQUEUESERVICE_DEBUG
         //l10n         //l10n
    //PEGASUS_STD(cout) << "Forcing shutdown of CIMOM Message Router" << PEGASUS_STD(endl);  
    MessageLoaderParms parms("Common.MessageQueueService.FORCING_SHUTDOWN",    MessageLoaderParms parms("Common.MessageQueueService.FORCING_SHUTDOWN",
                                                         "Forcing shutdown of $0",                              "Forcing shutdown of CIMOM Message Router");
                                                         "CIMOM Message Router");  
    PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);    PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);
    MessageQueueService::_stop_polling = 1;  #endif
    MessageQueueService *svc;  
  
      MessageQueueService *svc;
      int counter = 0;
    _polling_list.lock();    _polling_list.lock();
    svc = _polling_list.next(0);    svc = _polling_list.next(0);
  
    while(svc != 0)    while(svc != 0)
    {    {
   #ifdef MESSAGEQUEUESERVICE_DEBUG
                 //l10n - reuse same MessageLoaderParms to avoid multiple creates                 //l10n - reuse same MessageLoaderParms to avoid multiple creates
         //PEGASUS_STD(cout) << "Stopping " << svc->getQueueName() << PEGASUS_STD(endl);  
         parms.msg_id = "Common.MessageQueueService.STOPPING_SERVICE";         parms.msg_id = "Common.MessageQueueService.STOPPING_SERVICE";
                 parms.default_msg = "Stopping $0";                 parms.default_msg = "Stopping $0";
                 parms.arg0 = svc->getQueueName();                 parms.arg0 = svc->getQueueName();
                 PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);                 PEGASUS_STD(cout) << MessageLoader::getMessage(parms) << PEGASUS_STD(endl);
   #endif
  
       _polling_sem.signal();       _polling_sem.signal();
       svc->_shutdown_incoming_queue();       svc->_shutdown_incoming_queue();
         counter++;
       _polling_sem.signal();       _polling_sem.signal();
       svc = _polling_list.next(svc);       svc = _polling_list.next(svc);
    }    }
    _polling_list.unlock();    _polling_list.unlock();
   
      _polling_sem.signal();
   
      MessageQueueService::_stop_polling = 1;
   
      if(destroy_flag == true)
      {
   
         svc = _polling_list.remove_last();
         while(svc)
         {
            delete svc;
            svc = _polling_list.remove_last();
         }
   
      }
 } }
  
  
Line 155 
Line 178 
 MessageQueueService::MessageQueueService(const char *name, MessageQueueService::MessageQueueService(const char *name,
                                          Uint32 queueID,                                          Uint32 queueID,
                                          Uint32 capabilities,                                          Uint32 capabilities,
                                          Uint32 mask)                                           Uint32 mask,
                                            int threads)
    : Base(name, true,  queueID),    : Base(name, true,  queueID),
  
      _mask(mask),      _mask(mask),
Line 185 
Line 209 
          _meta_dispatcher_mutex.unlock();          _meta_dispatcher_mutex.unlock();
          throw NullPointer();          throw NullPointer();
       }       }
       _thread_pool = new ThreadPool(0, "MessageQueueService", 0, 0,        _thread_pool = new ThreadPool(0, "MessageQueueService", 0, threads,
                                     create_time, destroy_time, deadlock_time);                                     create_time, destroy_time, deadlock_time);
  
       _polling_thread = new Thread(polling_routine,       _polling_thread = new Thread(polling_routine,
Line 201 
Line 225 
       //l10n       //l10n
       //throw BindFailedException("MessageQueueService Base Unable to register with  Meta Dispatcher");       //throw BindFailedException("MessageQueueService Base Unable to register with  Meta Dispatcher");
       MessageLoaderParms parms("Common.MessageQueueService.UNABLE_TO_REGISTER",       MessageLoaderParms parms("Common.MessageQueueService.UNABLE_TO_REGISTER",
                                                    "$0 Unable to register with  $1",                                 "MessageQueueService Base Unable to register with  Meta Dispatcher");
                                                    "MessageQueueService Base",  
                                                    "Meta Dispatcher");  
  
       throw BindFailedException(parms);       throw BindFailedException(parms);
    }    }
Line 220 
Line 242 
 MessageQueueService::~MessageQueueService(void) MessageQueueService::~MessageQueueService(void)
 { {
    _die = 1;    _die = 1;
    // IBM-KR: This causes a new message (IO_CLOSE) to be spawned, which  
    // doesn't get picked up anyone. The idea was that the message would be     if (_incoming_queue_shutdown.value() == 0 )
    // picked up handle_AsyncIoctl which closes the queue and does cleaning.     {
    // That described behavior has never surfaced itself. If it does appear,        _shutdown_incoming_queue();
    // uncomment the if ( ..) { } block below.     }
   
    // Note: The handle_AsyncIcotl does get called when force_shutdown(void) gets  
    // called during Pegasus shutdown procedure (in case you ever wondered).  
   
    //if (_incoming_queue_shutdown.value() == 0 )  
    //{  
    //   _shutdown_incoming_queue();  
    //}  
    _callback_ready.signal();    _callback_ready.signal();
 //   _callback_thread.join();  
  
    _meta_dispatcher_mutex.lock(pegasus_thread_self());    _meta_dispatcher_mutex.lock(pegasus_thread_self());
    _service_count--;    _service_count--;
    if (_service_count.value() == 0 )    if (_service_count.value() == 0 )
    {    {
   
       _stop_polling++;       _stop_polling++;
       _polling_sem.signal();       _polling_sem.signal();
       _polling_thread->join();       _polling_thread->join();
Line 248 
Line 262 
       _meta_dispatcher->_shutdown_routed_queue();       _meta_dispatcher->_shutdown_routed_queue();
       delete _meta_dispatcher;       delete _meta_dispatcher;
       _meta_dispatcher = 0;       _meta_dispatcher = 0;
   
       delete _thread_pool;       delete _thread_pool;
       _thread_pool = 0;       _thread_pool = 0;
    }    }
Line 263 
Line 278 
 void MessageQueueService::_shutdown_incoming_queue(void) void MessageQueueService::_shutdown_incoming_queue(void)
 { {
  
   
    if (_incoming_queue_shutdown.value() > 0 )    if (_incoming_queue_shutdown.value() > 0 )
       return ;       return ;
    AsyncIoctl *msg = new AsyncIoctl(get_next_xid(),    AsyncIoctl *msg = new AsyncIoctl(get_next_xid(),
Line 285 
Line 301 
  
    _incoming.insert_last_wait(msg->op);    _incoming.insert_last_wait(msg->op);
  
 //   _req_thread.join();  
   
 } }
  
  
Line 487 
Line 501 
       else       else
       {       {
          PEGASUS_ASSERT(rq != 0 );          PEGASUS_ASSERT(rq != 0 );
          // ATTN: optimization  
          // << Wed Mar  6 15:00:39 2002 mdd >>  
          // put thread and queue into the asyncopnode structure.  
          //  (static_cast<AsyncMessage *>(rq))->_myself = operation->_thread_ptr;  
          //   (static_cast<AsyncMessage *>(rq))->_service = operation->_service_ptr;  
          // done << Tue Mar 12 14:49:07 2002 mdd >>  
          operation->unlock();          operation->unlock();
          _handle_async_request(static_cast<AsyncRequest *>(rq));          _handle_async_request(static_cast<AsyncRequest *>(rq));
       }       }
Line 632 
Line 640 
       _polling_sem.signal();       _polling_sem.signal();
       return true;       return true;
    }    }
 //    else  
 //    {  
 //       if(  (rq != 0 && (true == MessageQueueService::messageOK(rq))) ||  
 //         (rp != 0 && ( true == MessageQueueService::messageOK(rp) )) &&  
 //         _die.value() == 0)  
 //       {  
 //       MessageQueueService::_incoming.insert_last_wait(op);  
 //       return true;  
 //       }  
 //    }  
   
    return false;    return false;
 } }
  
Line 653 
Line 650 
    return true;    return true;
 } }
  
   
 // made pure virtual  
 // << Wed Mar  6 15:11:31 2002 mdd >>  
 // void MessageQueueService::handleEnqueue(Message *msg)  
 // {  
 //    if ( msg )  
 //       delete msg;  
 // }  
   
 // made pure virtual  
 // << Wed Mar  6 15:11:56 2002 mdd >>  
 // void MessageQueueService::handleEnqueue(void)  
 // {  
 //     Message *msg = dequeue();  
 //     handleEnqueue(msg);  
 // }  
   
 void MessageQueueService::handle_heartbeat_request(AsyncRequest *req) void MessageQueueService::handle_heartbeat_request(AsyncRequest *req)
 { {
    // default action is to echo a heartbeat response    // default action is to echo a heartbeat response
Line 699 
Line 679 
    {    {
       case AsyncIoctl::IO_CLOSE:       case AsyncIoctl::IO_CLOSE:
       {       {
          // save my bearings  
 //       Thread *myself = req->op->_thread_ptr;  
          MessageQueueService *service = static_cast<MessageQueueService *>(req->op->_service_ptr);          MessageQueueService *service = static_cast<MessageQueueService *>(req->op->_service_ptr);
  
          // respond to this message.  #ifdef MESSAGEQUEUESERVICE_DEBUG
            PEGASUS_STD(cout) << service->getQueueName() << " Received AsyncIoctl::IO_CLOSE " << PEGASUS_STD(endl);
   #endif
   
            // respond to this message. this is fire and forget, so we don't need to delete anything.
            // this takes care of two problems that were being found
            // << Thu Oct  9 10:52:48 2003 mdd >>
          _make_response(req, async_results::OK);          _make_response(req, async_results::OK);
          // ensure we do not accept any further messages          // ensure we do not accept any further messages
  
Line 727 
Line 712 
             }             }
             if( operation )             if( operation )
             {             {
 //             operation->_thread_ptr = myself;  
                operation->_service_ptr = service;                operation->_service_ptr = service;
                service->_handle_incoming_operation(operation);                service->_handle_incoming_operation(operation);
             }             }
Line 737 
Line 721 
  
          // shutdown the AsyncDQueue          // shutdown the AsyncDQueue
          service->_incoming.shutdown_queue();          service->_incoming.shutdown_queue();
          AsyncOpNode *op = req->op;  
          op->_request.remove_first();  
          op->release();  
          return_op(op);  
          delete req;  
          // exit the thread !  
 //       myself->exit_self( (PEGASUS_THREAD_RETURN) 1 );  
          return;          return;
       }       }
  
Line 754 
Line 731 
  
 void MessageQueueService::handle_CimServiceStart(CimServiceStart *req) void MessageQueueService::handle_CimServiceStart(CimServiceStart *req)
 { {
   
   #ifdef MESSAGEQUEUESERVICE_DEBUG
      PEGASUS_STD(cout) << getQueueName() << "received START" << PEGASUS_STD(endl);
   #endif
   
    // clear the stoped bit and update    // clear the stoped bit and update
    _capabilities &= (~(module_capabilities::stopped));    _capabilities &= (~(module_capabilities::stopped));
    _make_response(req, async_results::OK);    _make_response(req, async_results::OK);
Line 763 
Line 745 
 } }
 void MessageQueueService::handle_CimServiceStop(CimServiceStop *req) void MessageQueueService::handle_CimServiceStop(CimServiceStop *req)
 { {
   #ifdef MESSAGEQUEUESERVICE_DEBUG
      PEGASUS_STD(cout) << getQueueName() << "received STOP" << PEGASUS_STD(endl);
   #endif
    // set the stopeed bit and update    // set the stopeed bit and update
    _capabilities |= module_capabilities::stopped;    _capabilities |= module_capabilities::stopped;
    _make_response(req, async_results::CIM_STOPPED);    _make_response(req, async_results::CIM_STOPPED);
Line 988 
Line 973 
  
    Boolean destroy_op = false;    Boolean destroy_op = false;
  
    if (request->op == false)     if (request->op == 0)
    {    {
       request->op = get_op();       request->op = get_op();
       request->op->_request.insert_first(request);       request->op->_request.insert_first(request);
Line 1185 
Line 1170 
  
 Uint32 MessageQueueService::get_next_xid(void) Uint32 MessageQueueService::get_next_xid(void)
 { {
      static Mutex _monitor;
      Uint32 value;
      _monitor.lock(pegasus_thread_self());
   
    _xid++;    _xid++;
    return _xid.value();     value =  _xid.value();
      _monitor.unlock();
      return value;
   
 } }
  
 PEGASUS_NAMESPACE_END PEGASUS_NAMESPACE_END

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