1 h.sterling 1.1 //%2005////////////////////////////////////////////////////////////////////////
2 //
3 // Copyright (c) 2000, 2001, 2002 BMC Software; Hewlett-Packard Development
4 // Company, L.P.; IBM Corp.; The Open Group; Tivoli Systems.
5 // Copyright (c) 2003 BMC Software; Hewlett-Packard Development Company, L.P.;
6 // IBM Corp.; EMC Corporation, The Open Group.
7 // Copyright (c) 2004 BMC Software; Hewlett-Packard Development Company, L.P.;
8 // IBM Corp.; EMC Corporation; VERITAS Software Corporation; The Open Group.
9 // Copyright (c) 2005 Hewlett-Packard Development Company, L.P.; IBM Corp.;
10 // EMC Corporation; VERITAS Software Corporation; The Open Group.
11 //
12 // Permission is hereby granted, free of charge, to any person obtaining a copy
13 // of this software and associated documentation files (the "Software"), to
14 // deal in the Software without restriction, including without limitation the
15 // rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
16 // sell copies of the Software, and to permit persons to whom the Software is
17 // furnished to do so, subject to the following conditions:
18 //
19 // THE ABOVE COPYRIGHT NOTICE AND THIS PERMISSION NOTICE SHALL BE INCLUDED IN
20 // ALL COPIES OR SUBSTANTIAL PORTIONS OF THE SOFTWARE. THE SOFTWARE IS PROVIDED
21 // "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT
22 h.sterling 1.1 // LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
23 // PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
24 // HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
25 // ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
26 // WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
27 //
28 //==============================================================================
29 //
30 // Author: Heather Sterling (hsterl@us.ibm.com)
31 //
32 // Modified By:
33 //
34 //%/////////////////////////////////////////////////////////////////////////////
35
36 #include <Pegasus/Common/Config.h>
37 //#include <cstdlib>
38 //#include <dlfcn.h>
39 #include <Pegasus/Common/System.h>
40 #include <Pegasus/Common/FileSystem.h>
41 #include <Pegasus/Common/Tracer.h>
42 #include <Pegasus/Common/Logger.h>
43 h.sterling 1.1 #include <Pegasus/Common/XmlReader.h>
44 #include <Pegasus/Common/XmlParser.h>
45 #include <Pegasus/Common/XmlWriter.h>
46 #include <Pegasus/Common/IPC.h>
47
48 #include "ConsumerManager.h"
49
50 PEGASUS_NAMESPACE_BEGIN
51 PEGASUS_USING_STD;
52
53 //ATTN: Can we just use a properties file instead?? If we only have one property, we may want to just parse it ourselves.
54 // We may need to add more properties, however. Potential per consumer properties: unloadOk, idleTimout, retryCount, etc
55 static struct OptionRow optionsTable[] =
56 //optionname defaultvalue rqd type domain domainsize clname hlpmsg
57 {
58 {"location", "", false, Option::STRING, 0, 0, "location", "library name for the consumer"},
59 };
60
61 const Uint32 NUM_OPTIONS = sizeof(optionsTable) / sizeof(optionsTable[0]);
62
63 //retry settings
64 h.sterling 1.1 static const Uint32 DEFAULT_MAX_RETRY_COUNT = 5;
|
65 h.sterling 1.6 static const Uint32 DEFAULT_RETRY_LAPSE = 300000; //ms = 5 minutes
|
66 h.sterling 1.1
67
68
69 ConsumerManager::ConsumerManager(const String& consumerDir, const String& consumerConfigDir, Boolean enableConsumerUnload, Uint32 idleTimeout) :
70 _consumerDir(consumerDir),
71 _consumerConfigDir(consumerConfigDir),
72 _enableConsumerUnload(enableConsumerUnload),
73 _idleTimeout(idleTimeout),
74 _forceShutdown(true)
75 {
76 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::ConsumerManager");
77
78 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Consumer library directory: " + consumerDir);
79 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Consumer configuration directory: " + consumerConfigDir);
80
81 Tracer::trace(__FILE__,__LINE__,TRC_LISTENER,Tracer::LEVEL4,
82 "Consumer unload enabled %d: idle timeout %d",
83 enableConsumerUnload,
84 idleTimeout);
85
86
|
87 h.sterling 1.6 //ATTN: Bugzilla 3765 - Uncomment when OptionManager has a reset capability
88 //_optionMgr.registerOptions(optionsTable, NUM_OPTIONS);
|
89 h.sterling 1.1
|
90 kumpf 1.3 struct timeval deallocateWait = {15, 0};
91 _thread_pool = new ThreadPool(0, "ConsumerManager", 0, 0, deallocateWait);
|
92 h.sterling 1.1
93 _init();
94
95 PEG_METHOD_EXIT();
96 }
97
98 ConsumerManager::~ConsumerManager()
99 {
100 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::~ConsumerManager");
101
102 unloadAllConsumers();
103
|
104 h.sterling 1.5 if (_thread_pool != NULL)
105 {
|
106 kumpf 1.3 delete _thread_pool;
|
107 h.sterling 1.5 }
|
108 h.sterling 1.1
109 ConsumerTable::Iterator i = _consumers.start();
110 for (; i!=0; i++)
111 {
112 DynamicConsumer* consumer = i.value();
113 delete consumer;
114 }
115
116 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Deleted all consumers");
117
118 ModuleTable::Iterator j = _modules.start();
119 for (;j!=0;j++)
120 {
121 ConsumerModule* module = j.value();
122 delete module;
123 }
124
125 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Deleted all modules");
126
127 PEG_METHOD_EXIT();
128 }
129 h.sterling 1.1
130 void ConsumerManager::_init()
131 {
132 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_init");
133
134 //check if there are any outstanding indications
135 Array<String> files;
136 Uint32 pos;
137 String consumerName;
138
139 if (FileSystem::getDirectoryContents(_consumerConfigDir, files))
140 {
141 for (Uint32 i = 0; i < files.size(); i++)
142 {
143 pos = files[i].find(".dat");
144 if (pos != PEG_NOT_FOUND)
145 {
146 consumerName = files[i].subString(0, pos);
147
148 try
149 {
150 h.sterling 1.1 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Attempting to load indication for!" + consumerName + "!");
151 getConsumer(consumerName);
152
153 } catch (...)
154 {
155 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Cannot load consumer from file " + files[i]);
156 }
157 }
158 }
159 }
160
161 PEG_METHOD_EXIT();
162 }
163
164 String ConsumerManager::getConsumerDir()
165 {
166 return _consumerDir;
167 }
168
169 String ConsumerManager::getConsumerConfigDir()
170 {
171 h.sterling 1.1 return _consumerConfigDir;
172 }
173
174 Boolean ConsumerManager::getEnableConsumerUnload()
175 {
176 return _enableConsumerUnload;
177 }
178
179 Uint32 ConsumerManager::getIdleTimeout()
180 {
181 return _idleTimeout;
182 }
183
184 /** Retrieves the library name associated with the consumer name. By default, the library name
185 * is the same as the consumer name. However, you may specify a different library name in a consumer
186 * configuration file. This file must be named "MyConsumer.txt" and contain the following:
187 * location="libraryName"
188 *
189 * The config file is optional and is generally only needed in cases where there are strict requirements
190 * on library naming.
191 *
192 h.sterling 1.1 * It is the responsibility of the caller to catch any exceptions thrown by this method.
193 */
194 String ConsumerManager::_getConsumerLibraryName(const String & consumerName)
195 {
196 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::getConsumerLibraryName");
197
198 //default library name is consumer name
199 String libraryName = consumerName;
200
201 //check whether an alternative library name was specified in an optional consumer config file
202 String configFile = FileSystem::getAbsolutePath((const char*)_consumerConfigDir.getCString(), String(consumerName + ".conf"));
203 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Looking for config file " + configFile);
204
205 if (FileSystem::exists(configFile) && FileSystem::canRead(configFile))
206 {
207 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Found config file for consumer " + consumerName);
208
209 try
210 {
|
211 h.sterling 1.6 //Bugzilla 3765 - Change this to use a member var when OptionManager has a reset option
212 OptionManager _optionMgr;
213 _optionMgr.registerOptions(optionsTable, NUM_OPTIONS); //comment this line out later
|
214 h.sterling 1.1 _optionMgr.mergeFile(configFile);
215 _optionMgr.checkRequiredOptions();
216
217 if (!_optionMgr.lookupValue("location", libraryName) || (libraryName == String::EMPTY))
218 {
219 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Warning: Using default library name since none was specified in " + configFile);
220 libraryName = consumerName;
221 }
222
223 } catch (Exception & ex)
224 {
225 throw Exception(MessageLoaderParms("DynListener.ConsumerManager.INVALID_CONFIG_FILE",
226 "Error reading $0: $1.",
227 configFile,
228 ex.getMessage()));
229 }
230 } else
231 {
232 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "No config file exists for " + consumerName);
233 }
234
235 h.sterling 1.1 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "The library name for " + consumerName + " is " + libraryName);
236
237 PEG_METHOD_EXIT();
238 return libraryName;
239 }
240
241 /** Returns the DynamicConsumer for the consumerName. If it already exists, we return the one in the cache. If it
242 * DNE, we create it and initialize it, and add it to the table.
243 * @throws Exception if we cannot successfully create and initialize the consumer
244 */
245 DynamicConsumer* ConsumerManager::getConsumer(const String& consumerName)
246 {
247 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::getConsumer");
248
249 DynamicConsumer* consumer = 0;
250 CIMIndicationConsumerProvider* consumerRef = 0;
251 Boolean cached = false;
252 Boolean entryExists = false;
253
254 AutoMutex lock(_consumerTableMutex);
255
256 h.sterling 1.1 if (_consumers.lookup(consumerName, consumer))
257 {
258 //why isn't this working??
259 entryExists = true;
260
261 if (consumer && consumer->isLoaded())
262 {
263 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Consumer exists in the cache and is already loaded: " + consumerName);
264 cached = true;
265 }
266 } else
267 {
268 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Consumer not found in cache, creating " + consumerName);
269 consumer = new DynamicConsumer(consumerName);
270 //ATTN: The above is a memory leak if _initConsumer throws an exception
271 //need to delete it in that case
272 }
273
274 if (!cached)
275 {
276 _initConsumer(consumerName, consumer);
277 h.sterling 1.1
278 if (!entryExists)
279 {
280 _consumers.insert(consumerName, consumer);
281 }
282 }
283
284 consumer->updateIdleTimer();
285
286 PEG_METHOD_EXIT();
287 return consumer;
288 }
289
290 /** Initializes a DynamicConsumer.
291 * Caller assumes responsibility for mutexing the operation as well as ensuring the consumer does not already exist.
292 * @throws Exception if the consumer cannot be initialized
293 */
294 void ConsumerManager::_initConsumer(const String& consumerName, DynamicConsumer* consumer)
295 {
296 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_initConsumer");
297
298 h.sterling 1.1 CIMIndicationConsumerProvider* base = 0;
299 ConsumerModule* module = 0;
300
301 //lookup provider module in cache (if it exists, it returns the cached module, otherwise it creates and returns a new one)
302 String libraryName = _getConsumerLibraryName(consumerName);
303 module = _lookupModule(libraryName);
304
305 //build library path
306 String libraryPath = FileSystem::getAbsolutePath((const char*)_consumerDir.getCString(), FileSystem::buildLibraryFileName(libraryName));
307 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Loading library: " + libraryPath);
308
309 //load module
310 try
311 {
312 base = module->load(consumerName, libraryPath);
313 consumer->set(module, base);
314
315 } catch (Exception& ex)
316 {
317 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error loading consumer module: " + ex.getMessage());
318
319 h.sterling 1.1 throw Exception(MessageLoaderParms("DynListener.ConsumerManager.CANNOT_LOAD_MODULE",
320 "Cannot load module ($0:$1): Unknown exception.",
321 consumerName,
322 libraryName));
323 } catch (...)
324 {
325 throw Exception(MessageLoaderParms("DynListener.ConsumerManager.CANNOT_LOAD_MODULE",
326 "Cannot load module ($0:$1): Unknown exception.",
327 consumerName,
328 libraryName));
329 }
330
331 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Successfully loaded consumer module " + libraryName);
332
333 //initialize consumer
334 try
335 {
336 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Initializing Consumer " + consumerName);
337
338 consumer->initialize();
339
340 h.sterling 1.1 //ATTN: need to change this
341 Semaphore* semaphore = new Semaphore(0); //blocking
342
343 consumer->setShutdownSemaphore(semaphore);
344
345 //start the worker thread
|
346 konrad.r 1.7 if (_thread_pool->allocate_and_awaken(consumer,
|
347 h.sterling 1.1 _worker_routine,
|
348 konrad.r 1.7 semaphore) != PEGASUS_THREAD_OK)
349 {
350 Logger::put(Logger::STANDARD_LOG, System::CIMSERVER, Logger::TRACE,
351 "Not enough threads for consumer.");
352
353 Tracer::trace(TRC_LISTENER, Tracer::LEVEL2,
354 "Could not allocate thread for consumer.");
355
356 consumer->setShutdownSemaphore(0);
357 delete semaphore;
358 throw Exception(MessageLoaderParms("DynListener.ConsumerManager.CANNOT_ALLOCATE_THREAD",
359 "Not enough threads for consumer worker routine."));
360 }
|
361 h.sterling 1.1
362 //load any outstanding requests
363 Array<CIMInstance> outstandingIndications = _deserializeOutstandingIndications(consumerName);
364 if (outstandingIndications.size())
365 {
366 //the consumer will signal itself in _loadOustandingIndications
367 consumer->_loadOutstandingIndications(outstandingIndications);
368 }
369
370 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Successfully initialized consumer " + consumerName);
371
372 } catch (...)
373 {
374 module->unloadModule();
375 consumer->reset();
376 throw Exception(MessageLoaderParms("DynListener.ConsumerManager.CANNOT_INITIALIZE_CONSUMER",
377 "Cannot initialize consumer ($0).",
378 consumerName));
379 }
380
381 PEG_METHOD_EXIT();
382 h.sterling 1.1 }
383
384
385 /** Returns the ConsumerModule with the given library name. If it already exists, we return the one in the cache. If it
386 * DNE, we create it and add it to the table.
387 * @throws Exception if we cannot successfully create and initialize the consumer
388 */
389 ConsumerModule* ConsumerManager::_lookupModule(const String & libraryName)
390 {
391 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_lookupModule");
392
393 AutoMutex lock(_moduleTableMutex);
394
395 ConsumerModule* module = 0;
396
397 //see if consumer module is cached
398 if (_modules.lookup(libraryName, module))
399 {
400 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4,
401 "Found Consumer Module" + libraryName + " in Consumer Manager Cache");
402
403 h.sterling 1.1 } else
404 {
405 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4,
406 "Creating Consumer Provider Module " + libraryName);
407
408 module = new ConsumerModule();
409 _modules.insert(libraryName, module);
410 }
411
412 PEG_METHOD_EXIT();
413 return(module);
414 }
415
416 /** Returns true if there are active consumers
417 */
418 Boolean ConsumerManager::hasActiveConsumers()
419 {
420 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::hasActiveConsumers");
421
422 AutoMutex lock(_consumerTableMutex);
423 DynamicConsumer* consumer = 0;
424 h.sterling 1.1
425 try
426 {
427 for (ConsumerTable::Iterator i = _consumers.start(); i != 0; i++)
428 {
429 consumer = i.value();
430
431 if (consumer && consumer->isLoaded() && (consumer->getPendingIndications() > 0))
432 {
433 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Found active consumer: " + consumer->_name);
434 PEG_METHOD_EXIT();
435 return true;
436 }
437 }
438 } catch (...)
439 {
440 // Unexpected exception; do not assume that no providers are loaded
441 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Unexpected Exception in hasActiveConsumers.");
442 PEG_METHOD_EXIT();
443 return true;
444 }
445 h.sterling 1.1
446 PEG_METHOD_EXIT();
447 return false;
448 }
449
450 /** Returns true if there are loaded consumers
451 */
452 Boolean ConsumerManager::hasLoadedConsumers()
453 {
454 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::hasLoadedConsumers");
455
456 AutoMutex lock(_consumerTableMutex);
457 DynamicConsumer* consumer = 0;
458
459 try
460 {
461 for (ConsumerTable::Iterator i = _consumers.start(); i != 0; i++)
462 {
463 consumer = i.value();
464
465 if (consumer && consumer->isLoaded())
466 h.sterling 1.1 {
467 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Found loaded consumer: " + consumer->_name);
468 PEG_METHOD_EXIT();
469 return true;
470 }
471 }
472 } catch (...)
473 {
474 // Unexpected exception; do not assume that no providers are loaded
475 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Unexpected Exception in hasLoadedConsumers.");
476 PEG_METHOD_EXIT();
477 return true;
478 }
479
480 PEG_METHOD_EXIT();
481 return false;
482 }
483
484
485 /** Shutting down a consumer consists of four major steps:
486 * 1) Send the shutdown signal. This causes the worker routine to break out of the loop and exit.
487 h.sterling 1.1 * 2) Wait for the worker thread to end. This may take a while if it's processing an indication. This
488 * is optional in a shutdown scenario. If the listener is shutdown with a -f force, the listener
489 * will not wait for the consumer to finish before shutting down. Note that a normal shutdown only allows
490 * the current consumer indication to finish. All other queued indications are serialized to a log and
491 * are sent when the consumer is reoaded.
492 * 3) Terminate the consumer provider interface.
493 * 4) Decrement the module refcount (the module will automatically unload when it's refcount == 0)
494 *
495 * In a scenario where more multiple consumers are loaded, the shutdown signal should be sent to all
496 * of the consumers so the threads can finish simultaneously.
497 *
498 * ATTN: Should the normal shutdown wait for everything in the queue to be processed? Just new indications
499 * to be processed? I am not inclined to this solution since it could take a LOT of time. By serializing
500 * and deserialing indications between shutdown and startup, I feel like we do not need to process ALL
501 * queued indications on shutdown.
502 */
503
504 /** Unloads all consumers.
505 */
506 void ConsumerManager::unloadAllConsumers()
507 {
508 h.sterling 1.1 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::unloadAllConsumers");
509
510 AutoMutex lock(_consumerTableMutex);
511
512 if (!_consumers.size())
513 {
514 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "There are no consumers to unload.");
515 PEG_METHOD_EXIT();
516 return;
517 }
518
519 if (!_forceShutdown)
520 {
521 //wait until all the consumers have finished processing the events in their queue
522 //ATTN: Should this have a timeout even though it's a force??
523 while (hasActiveConsumers())
524 {
525 pegasus_sleep(500);
526 }
527 }
528
529 h.sterling 1.1 Array<DynamicConsumer*> loadedConsumers;
530
531 ConsumerTable::Iterator i = _consumers.start();
532 DynamicConsumer* consumer = 0;
533
534 for (; i!=0; i++)
535 {
536 consumer = i.value();
537 if (consumer && consumer->isLoaded())
538 {
539 loadedConsumers.append(consumer);
540 }
541 }
542
543 if (loadedConsumers.size())
544 {
545 try
546 {
547 _unloadConsumers(loadedConsumers);
548
549 } catch (Exception& ex)
550 h.sterling 1.1 {
551 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Error unloading consumers.");
552 }
553 } else
554 {
555 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "There are no consumers to unload.");
556 }
557
558 PEG_METHOD_EXIT();
559 }
560
561 /** Unloads idle consumers.
562 */
563 void ConsumerManager::unloadIdleConsumers()
564 {
565 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::unloadIdleConsumers");
566
567 AutoMutex lock(_consumerTableMutex);
568
569 if (!_consumers.size())
570 {
571 h.sterling 1.1 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "There are no consumers to unload.");
572 PEG_METHOD_EXIT();
573 return;
574 }
575
576 Array<DynamicConsumer*> loadedConsumers;
577
578 ConsumerTable::Iterator i = _consumers.start();
579 DynamicConsumer* consumer = 0;
580
581 for (; i!=0; i++)
582 {
583 consumer = i.value();
584 if (consumer && consumer->isLoaded() && consumer->isIdle())
585 {
586 loadedConsumers.append(consumer);
587 }
588 }
589
590 if (loadedConsumers.size())
591 {
592 h.sterling 1.1 try
593 {
594 _unloadConsumers(loadedConsumers);
595
596 } catch (Exception& ex)
597 {
598 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Error unloading consumers.");
599 }
600 } else
601 {
602 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "There are no consumers to unload.");
603 }
604
605 PEG_METHOD_EXIT();
606 }
607
608 /** Unloads a single consumer.
609 */
610 void ConsumerManager::unloadConsumer(const String& consumerName)
611 {
612 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::unloadConsumer");
613 h.sterling 1.1
614 AutoMutex lock(_consumerTableMutex);
615
616 DynamicConsumer* consumer = 0;
617
618 //check whether the consumer exists
619 if (!_consumers.lookup(consumerName, consumer))
620 {
621 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Error: cannot unload consumer, unknown consumer " + consumerName);
622 return;
623 }
624
625 //check whether the consumer is loaded
626 if (consumer && consumer->isLoaded()) //ATTN: forceShutdown?
627 {
628 //unload the consumer
629 Array<DynamicConsumer*> loadedConsumers;
630 loadedConsumers.append(consumer);
631
632 try
633 {
634 h.sterling 1.1 _unloadConsumers(loadedConsumers);
635
636 } catch (Exception& ex)
637 {
638 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Error unloading consumers.");
639 }
640
641 } else
642 {
643 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Error: cannot unload consumer " + consumerName);
644 }
645
646 PEG_METHOD_EXIT();
647 }
648
649 /** Unloads the consumers in the given array.
650 * The consumerTable mutex MUST be locked prior to entering this method.
651 */
652 void ConsumerManager::_unloadConsumers(Array<DynamicConsumer*> consumersToUnload)
653 {
654 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_unloadConsumers");
655 h.sterling 1.1
656 //tell consumers to shutdown
657 for (Uint32 i = 0; i < consumersToUnload.size(); i++)
658 {
659 consumersToUnload[i]->sendShutdownSignal();
660 }
661
662 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Sent shutdown signal to all consumers.");
663
664 //wait for all the consumer worker threads to complete
665 //since we can only shutdown after they are all complete, it does not matter if the first, fifth, or last
666 //consumer takes the longest; the wait time is equal to the time it takes for the busiest consumer to stop
667 //processing its requests.
668 for (Uint32 i = 0; i < consumersToUnload.size(); i++)
669 {
670 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Unloading consumer " + consumersToUnload[i]->getName());
671
672 //wait for the consumer worker thread to end
673 try
674 {
675 Semaphore* _shutdownSemaphore = consumersToUnload[i]->getShutdownSemaphore();
676 h.sterling 1.1 if (_shutdownSemaphore)
677 {
678 _shutdownSemaphore->time_wait(10000);
679 }
680
681 } catch (TimeOut &)
682 {
683 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Timed out while attempting to stop consumer thread.");
684 }
685
686 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Terminating consumer.");
687
688 try
689 {
690 //terminate consumer provider interface
691 consumersToUnload[i]->terminate();
692
693 //unload consumer provider module
694 PEGASUS_ASSERT(consumersToUnload[i]->_module != 0);
695 consumersToUnload[i]->_module->unloadModule();
696
697 h.sterling 1.1 //serialize outstanding indications
698 _serializeOutstandingIndications(consumersToUnload[i]->getName(), consumersToUnload[i]->_retrieveOutstandingIndications());
699
700 //reset the consumer
701 consumersToUnload[i]->reset();
702
703 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Consumer library successfully unloaded.");
704
705 } catch (Exception& e)
706 {
707 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error unloading consumer: " + e.getMessage());
708 //ATTN: throw exception? log warning?
709 }
710 }
711
712 PEG_METHOD_EXIT();
713 }
714
715 /** Serializes oustanding indications to a <MyConsumer>.dat file
716 */
717 void ConsumerManager::_serializeOutstandingIndications(const String& consumerName, Array<CIMInstance> indications)
718 h.sterling 1.1 {
719 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_serializeOutstandingIndications");
720
721 if (!indications.size())
722 {
723 PEG_METHOD_EXIT();
724 return;
725 }
726
727 String fileName = FileSystem::getAbsolutePath((const char*)_consumerConfigDir.getCString(), String(consumerName + ".dat"));
728 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Consumer dat file: " + fileName);
729
730 Array<char> buffer;
731
732 // Open the log file and serialize remaining
733 FILE* fileHandle = 0;
734 fileHandle = fopen((const char*)fileName.getCString(), "w");
735
736 if (!fileHandle)
737 {
738 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Unable to open log file for " + consumerName);
739 h.sterling 1.1
740 } else
741 {
742 Tracer::trace(__FILE__,__LINE__,TRC_LISTENER,Tracer::LEVEL3,
743 "Serializing %d outstanding requests for %s",
744 indications.size(),
745 (const char*)consumerName.getCString());
746
747 //we have to put the array of instances under a valid root element or the parser complains
748 XmlWriter::append(buffer, "<IRETURNVALUE>\n");
749
750 for (Uint32 i = 0; i < indications.size(); i++)
751 {
752 XmlWriter::appendValueNamedInstanceElement(buffer, indications[i]);
753 }
754
755 XmlWriter::append(buffer, "</IRETURNVALUE>\0");
756
757 fputs((const char*)buffer.getData(), fileHandle);
758
759 fclose(fileHandle);
760 h.sterling 1.1 }
761
762 PEG_METHOD_EXIT();
763 }
764
765 /** Reads outstanding indications from a <MyConsumer>.dat file
766 */
767 Array<CIMInstance> ConsumerManager::_deserializeOutstandingIndications(const String& consumerName)
768 {
769 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_deserializeOutstandingIndications");
770
771 String fileName = FileSystem::getAbsolutePath((const char*)_consumerConfigDir.getCString(), String(consumerName + ".dat"));
772 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Consumer dat file: " + fileName);
773
774 Array<CIMInstance> cimInstances;
775
776 // Open the log file and serialize remaining indications
777 if (FileSystem::exists(fileName) && FileSystem::canRead(fileName))
778 {
779 Array<char> text;
780 CIMInstance cimInstance;
781 h.sterling 1.1 XmlEntry entry;
782
783 try
784 {
785 FileSystem::loadFileToMemory(text, fileName); //ATTN: Is this safe to use; what about CRLFs?
786 text.append('\0');
787
788 //parse the file
789 XmlParser parser((char*)text.getData());
790 XmlReader::expectStartTag(parser, entry, "IRETURNVALUE");
791
792 while (XmlReader::getNamedInstanceElement(parser, cimInstance))
793 {
794 cimInstances.append(cimInstance);
795 }
796
797 XmlReader::expectEndTag(parser, "IRETURNVALUE");
798
799 Tracer::trace(__FILE__,__LINE__,TRC_LISTENER,Tracer::LEVEL3,
800 "Consumer %s has %d outstanding indications",
801 (const char*)consumerName.getCString(),
802 h.sterling 1.1 cimInstances.size());
803
804 //delete the file
805 FileSystem::removeFile(fileName);
806
807 } catch (Exception& ex)
808 {
809 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL3, "Error parsing dat file: " + ex.getMessage() + " " + consumerName);
810
811 } catch (...)
812 {
813 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error parsing dat file: Unknown Exception " + consumerName);
814 }
815 }
816
817 PEG_METHOD_EXIT();
818 return cimInstances;
819 }
820
821
822
823 h.sterling 1.1 /**
824 * This is the main worker thread of the consumer. By having only one thread per consumer, we eliminate a ton
825 * of synchronization issues and make it easy to prevent the consumer from performing two mutually exclusive
826 * operations at once. This also prevents one bad consumer from taking the entire listener down. That being said,
827 * it is up to the programmer to write smart consumers, and to ensure that their actions don't deadlock the worker thread.
828 *
829 * If a consumer receives a lot of traffic, or it's consumeIndication() method takes a considerable amount of time to
830 * complete, it may make sense to make the consumer multi-threaded. The individual consumer can immediately spawn off
831 * new threads to handle indications, and return immediately to catch the next indication. In this way, a consumer
832 * can attain extremely high performance.
833 *
834 * There are three different events that can signal us:
835 * 1) A new indication (signalled by DynamicListenerIndicationDispatcher)
836 * 2) A shutdown signal (signalled from ConsumerManager, due to a listener shutdown or an idle consumer state)
837 * 3) A retry signal (signalled from this routine itself)
838 *
839 * The idea is that all new indications are put on the front of the queue and processed first. All of the retry
840 * indications are put on the back of the queue and are only processed AFTER all new indications are sent.
841 * Before processing each indication, we check to see whether or not the shutdown signal was given. If so,
842 * we immediately break out of the loop, and another compenent serializes the remaining indications to a file.
843 *
844 h.sterling 1.1 * An indication gets retried if the consumer throws a CIM_ERR_FAILED exception.
845 *
|
846 h.sterling 1.5 * This function makes sure it waits until the default retry lapse has passed to avoid issues with the following scenario:
847 * 20 new indications come in, 10 of them are successful, 10 are not.
|
848 h.sterling 1.1 * We were signalled 20 times, so we will pass the time_wait 20 times. Perceivably, the process time on each indication
849 * could be minimal. We could potentially proceed to process the retries after a very small time interval since
|
850 h.sterling 1.5 * we would never hit the wait for the retry timeout.
|
851 h.sterling 1.1 *
852 * ATTN: Outstanding issue with this strategy -- 20 new indications come in, 19 of them come in before the first one
853 * is processed. Because new indications are first in, first out, the 19 indications will be processed in reverse order.
|
854 h.sterling 1.5 * Is this a problem? Short answer - NO. They could arrive in reverse order anyways depending on the network stack.
|
855 h.sterling 1.1 *
856 */
857 PEGASUS_THREAD_RETURN PEGASUS_THREAD_CDECL ConsumerManager::_worker_routine(void *param)
858 {
859 PEG_METHOD_ENTER(TRC_LISTENER, "ConsumerManager::_worker_routine");
860
861 DynamicConsumer* myself = static_cast<DynamicConsumer*>(param);
862 String name = myself->getName();
863 DQueue<IndicationDispatchEvent> tmpEventQueue(true);
864
865 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "_worker_routine::entering loop for " + name);
866
867 PEGASUS_STD(cout) << "Worker thread started for consumer : " << name << endl;
868
869 while (true)
870 {
871 try
872 {
873 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::waiting " + name);
874
875 //wait to be signalled
876 h.sterling 1.1 myself->_check_queue->time_wait(DEFAULT_RETRY_LAPSE);
877
878 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::signalled " + name);
879
880 //check whether we received the shutdown signal
881 if (myself->_dieNow)
882 {
883 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::shutdown received " + name);
884 break;
885 }
886
887 //signal must have been due to an incoming event
888 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::indication received " + name);
889
890 //create a temporary queue to store failed indications
891 tmpEventQueue.empty_list();
892
893 //continue processing events until the queue is empty
894 //make sure to check for the shutdown signal before every iteration
895
896 while (myself->_eventqueue.size())
897 h.sterling 1.1 {
898 //check for shutdown signal
899 //this only breaks us out of the queue loop, but we will immediately get through the next wait from
900 //the shutdown signal itself, at which time we break out of the main loop
901 if (myself->_dieNow)
902 {
903 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Received signal to shutdown, jumping out of queue loop " + name);
904 break;
905 }
906
907 //pop next indication off the queue
908 IndicationDispatchEvent* event = 0;
909 event = myself->_eventqueue.remove_first(); //what exceptions/errors can this throw?
910
911 if (!event)
912 {
913 //this should never happen
914 continue;
915 }
916
|
917 h.sterling 1.5 //check retry status. do not retry until the retry time has elapsed
918 if (event->getRetries() > 0)
919 {
920 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Last attempt time from event is " + event->getLastAttemptTime().toString());
921 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "Current time is " + CIMDateTime::getCurrentDateTime().toString());
922 Sint64 differenceInMicroseconds = CIMDateTime::getDifference(event->getLastAttemptTime(),
923 CIMDateTime::getCurrentDateTime());
924
925 if (differenceInMicroseconds < (DEFAULT_RETRY_LAPSE * 1000))
926 {
927 //do not retry; just add to the retry queue
928 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::cannot retry event until retry time has elapsed");
929 tmpEventQueue.insert_last(event);
930 continue;
931 }
932 }
933
|
934 h.sterling 1.1 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::consumeIndication " + name);
935
936 try
937 {
938 myself->consumeIndication(event->getContext(),
939 event->getURL(),
940 event->getIndicationInstance());
941
942 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::processed indication successfully. " + name);
943
944 delete event;
945 continue;
946
947 } catch (CIMException & ce)
948 {
949 //check for failure
950 if (ce.getCode() == CIM_ERR_FAILED)
951 {
952 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "_worker_routine::consumeIndication() temporary failure: " + ce.getMessage() + " " + name);
953
954 //update event parameters
955 h.sterling 1.1 event->increaseRetries();
956
957 //determine if we have hit the max retry count
958 if (event->getRetries() >= DEFAULT_MAX_RETRY_COUNT)
959 {
960 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2,
961 "Error: the maximum retry count has been exceeded. Removing the event from the queue.");
962
963 Logger::put(
964 Logger::ERROR_LOG,
965 "ConsumerManager",
966 Logger::SEVERE,
967 "The following indication did not get processed successfully: $0",
968 event->getIndicationInstance().getPath().toString());
969
970 delete event;
971 continue;
972
973 } else
974 {
975 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL4, "_worker_routine::placing failed indication back in queue");
976 h.sterling 1.1 tmpEventQueue.insert_last(event);
977 }
978
979 } else
980 {
981 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error: consumeIndication() permanent failure: " + ce.getMessage());
982 delete event;
983 continue;
984 }
985
986 } catch (Exception & ex)
987 {
988 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error: consumeIndication() permanent failure: " + ex.getMessage());
989 delete event;
990 continue;
991
992 } catch (...)
993 {
994 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "Error: consumeIndication() failed: Unknown exception.");
995 delete event;
996 continue;
997 h.sterling 1.1 } //end try
998
999 } //while eventqueue
1000
1001 //copy the failed indications back to the main queue
1002 //since we are always adding the failed indications to the back, it should not interfere with the
1003 //dispatcher adding events to the front
1004
1005 //there is no = operator for DQueue so we must do it manually
1006 IndicationDispatchEvent* tmpEvent = 0;
1007 while (tmpEventQueue.size())
1008 {
1009 tmpEvent = tmpEventQueue.remove_first();
1010 myself->_eventqueue.insert_last(tmpEvent);
1011 }
1012
1013 } catch (TimeOut& te)
1014 {
1015 PEG_TRACE_STRING(TRC_LISTENER, Tracer::LEVEL2, "_worker_routine::Time to retry any outstanding indications.");
1016
1017 //signal the queue in the same way we would if we received a new indication
1018 h.sterling 1.1 //this allows the thread to fall into the queue processing code
1019 myself->_check_queue->signal();
1020
1021 } //time_wait
1022
1023
1024 } //shutdown
1025
1026 PEG_METHOD_EXIT();
1027 return 0;
1028 }
1029
1030
1031 PEGASUS_NAMESPACE_END
1032
1033
1034
|
Return to
ConsumerManager.cpp
CVS log
Up to [Pegasus] / pegasus / src / Pegasus / DynListener