1 krisbash 1.1 /*
2 **==============================================================================
3 **
4 ** Open Management Infrastructure (OMI)
5 **
6 ** Copyright (c) Microsoft Corporation
7 **
8 ** Licensed under the Apache License, Version 2.0 (the "License"); you may not
9 ** use this file except in compliance with the License. You may obtain a copy
10 ** of the License at
11 **
12 ** http://www.apache.org/licenses/LICENSE-2.0
13 **
14 ** THIS CODE IS PROVIDED *AS IS* BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 ** KIND, EITHER EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION ANY IMPLIED
16 ** WARRANTIES OR CONDITIONS OF TITLE, FITNESS FOR A PARTICULAR PURPOSE,
17 ** MERCHANTABLITY OR NON-INFRINGEMENT.
18 **
19 ** See the Apache 2 License for the specific language governing permissions
20 ** and limitations under the License.
21 **
22 krisbash 1.1 **==============================================================================
23 */
24
25 #include "SubMgr.h"
26 #include <indication/common/indilog.h>
27
28
29 unsigned int UintThreadID()
30 {
31 ThreadID id = Thread_ID();
32 return *((unsigned int*)(void*)&id);
33 }
34
35 /*
36 *
37 * SubscriptionManager Functions
38 *
39 */
40
41 SubscriptionManager* SubMgr_New()
42 {
43 krisbash 1.1 SubscriptionManager* subMgr = (SubscriptionManager*)PAL_Calloc(1, sizeof(SubscriptionManager));
44 if (!subMgr)
45 LOGD_ALLOC_OOM;
46 return subMgr;
47 }
48
49 _Use_decl_annotations_
50 void SubMgr_Delete(SubscriptionManager* mgr )
51 {
52 SubMgr_Finalize( mgr );
53 PAL_Free(mgr);
54 }
55
56 /*
57 * Perform generic initialization of a SubscriptionManager object.
58 */
59 _Use_decl_annotations_
60 void SubMgr_Init(
61 SubscriptionManager* mgr,
62 Provider* provider )
63 {
64 krisbash 1.1 DEBUG_ASSERT (mgr);
65 RecursiveLock_Init( &mgr->enablelock );
66 mgr->terminating = MI_FALSE;
67 mgr->enabled = MI_FALSE;
68 mgr->provider = provider;
69 SubscriptionList_Init( &mgr->subscrList );
70 }
71
72 _Use_decl_annotations_
73 void SubMgr_Finalize(
74 SubscriptionManager* self)
75 {
76 DEBUG_ASSERT( self );
77 if (self->aggrCtx)
78 {
79 AggrContext_Delete(self->aggrCtx);
80 self->aggrCtx = NULL;
81 }
82 if (self->lifecycleCtx)
83 {
84 LifeContext_Delete(self->lifecycleCtx);
85 krisbash 1.1 self->lifecycleCtx = NULL;
86 }
87 SubscriptionList_Finalize( &self->subscrList );
88 }
89
90 _Use_decl_annotations_
91 MI_Result SubMgr_CreateSubscription(
92 SubscriptionManager* mgr,
93 Provider* provider,
94 InteractionOpenParams* interactionParams,
95 SubscriptionContext** subscrContext )
96 {
97 SubMgrSubscription* subscription = NULL;
98 MI_Result result = MI_RESULT_OK;
99 Context_Type ctxType;
100 SubscribeReq* msg = (SubscribeReq*)interactionParams->msg;
101 SubscriptionContext* subContext;
102
103 DEBUG_ASSERT ( mgr && provider && interactionParams && interactionParams->msg && subscrContext );
104 DEBUG_ASSERT( SubscribeReqTag == interactionParams->msg->tag );
105
106 krisbash 1.1 /*
107 * Create the SubscriptionContext and prepare it for use in case the
108 * provider attempts to PostIndication immediately after
109 * EnableIndications without waiting for Subscribe.
110 */
111 subContext = (SubscriptionContext*)Batch_GetClear(msg->base.base.batch, sizeof(SubscriptionContext));
112 if (!subContext)
113 {
114 trace_SubscrContextFailed();
115 return MI_RESULT_SERVER_LIMITS_EXCEEDED;
116 }
117
118 /* This refCount is released in CONTEXT_STRANDAUX_INVOKESUBSCRIBE. */
119 subscription = SubMgrSubscription_New(msg);
120 if (!subscription)
121 {
122 trace_SubMgrSubscription_AllocFailed();
123 return MI_RESULT_FAILED;
124 }
125
126 if (SubMgrSubscription_IsQueryValid(subscription, provider->classDecl) == MI_FALSE)
127 krisbash 1.1 {
128 SubMgrSubscription_Release(subscription);
129 return MI_RESULT_INVALID_QUERY;
130 }
131
132 *subscrContext = subContext;
133
134 subscription->subscribeCtx = *subscrContext; // TODO: should this action be wrapped?
135
136 ctxType = SubscriptionTargetType_IsLifecycle((SubscriptionTargetType)msg->targetType ) ? CTX_TYPE_IND_LIFECYCLE : CTX_TYPE_IND_SUBSCRIPTION; // TODO: Move to Context.h?
137 result = SubscrContext_Init( *subscrContext, provider, ctxType, interactionParams, subscription);
138 if ( MI_RESULT_OK != result)
139 {
140 trace_SubscriptionContext_InitFailed();
141 subscription->subscribeCtx = NULL;
142 SubMgrSubscription_Release(subscription);
143 return result;
144 }
145 SubMgrSubscription_SetState( subscription, SubscriptionState_Initialized );
146
147 return MI_RESULT_OK;
148 krisbash 1.1 }
149
150 _Use_decl_annotations_
151 MI_Result SubMgr_DeleteSubscription(
152 SubscriptionManager* mgr,
153 SubMgrSubscription* subscription )
154 {
155 MI_Result r;
156 DEBUG_ASSERT ( mgr && subscription );
157
158 trace_SubMgr_DeleteSubscription_Start(UintThreadID(), mgr, subscription, (MI_Uint32)mgr->subscrList.count, mgr->subscrList.head, mgr->subscrList.tail);
159
160 r = SubscriptionList_DeleteSubscription( &mgr->subscrList, subscription, mgr );
161
162 return r;
163 }
164
165 _Use_decl_annotations_
166 MI_Result SubMgr_CancelAllSubscriptions(
167 SubscriptionManager* mgr,
168 MI_Result result,
169 krisbash 1.1 const ZChar* errorMessage,
170 const MI_Instance* cimError )
171 {
172 DEBUG_ASSERT (mgr);
173 trace_SubMgr_CancelAllSubscriptions( UintThreadID(), mgr);
174
175 return SubscriptionList_CancelAllSubscription( &mgr->subscrList );
176 }
177
178 static SubMgrSubscription* _SubscriptionList_GetSubscription(
179 _In_ const SubscriptionList* self,
180 _In_ MI_Uint64 subscriptionID)
181 {
182 SubMgrSubscription* subscription;
183 SubscriptionList* list = (SubscriptionList*)self;
184 ReadWriteLock_AcquireRead(&list->lock);
185 for (subscription = (SubMgrSubscription*)self->head; subscription; subscription = subscription->next)
186 {
187 if (subscription->subscriptionID == subscriptionID)
188 {
189 break;
190 krisbash 1.1 }
191 }
192 ReadWriteLock_ReleaseRead(&list->lock);
193
194 return subscription;
195 }
196
197 static SubMgrSubscription* _SubscriptionList_GetSubscriptionByContext(
198 _In_ const SubscriptionList* self,
199 _In_ SubscriptionContext* subCtx )
200 {
201 SubMgrSubscription* subscription;
202 SubscriptionList* list = (SubscriptionList*)self;
203
204 /* thread safely get subscriptions and add ref count */
205 ReadWriteLock_AcquireRead( &list->lock );
206 subscription = (SubMgrSubscription*) self->head;
207 while( subscription && (subscription->subscribeCtx != subCtx) )
208 {
209 subscription = subscription->next;
210 }
211 krisbash 1.1 if (subscription)
212 SubMgrSubscription_Addref(subscription);
213 ReadWriteLock_ReleaseRead( &list->lock );
214
215 return subscription;
216 }
217
218 /*
219 * Examines the SubscriptionManager to determine if it has any matching
220 * subscriptions.
221 */
222 _Use_decl_annotations_
223 SubMgrSubscription* SubMgr_GetSubscription(
224 const SubscriptionManager* mgr,
225 MI_Uint64 subscriptionID )
226 {
227 DEBUG_ASSERT( mgr );
228 return _SubscriptionList_GetSubscription( &mgr->subscrList, subscriptionID);
229 }
230
231 /* search subscription based on subctx, add refcount to subscription if found */
232 krisbash 1.1 _Use_decl_annotations_
233 SubMgrSubscription* SubMgr_GetSubscriptionByContext(
234 const SubscriptionManager* mgr,
235 SubscriptionContext* subCtx)
236 {
237 SubMgrSubscription* subscription;
238
239 DEBUG_ASSERT( mgr && subCtx );
240
241 subscription = _SubscriptionList_GetSubscriptionByContext( &mgr->subscrList, subCtx );
242
243 if ( NULL == subscription )
244 trace_SubMgr_GetSubscriptionByContext_NotFound(UintThreadID(), (void*)mgr, subCtx);
245
246 return subscription;
247 }
248
249 /* read subscription list snapshot into an array, no refcount added to subscription(s) */
250 _Use_decl_annotations_
251 MI_Result SubMgr_GetSubscriptionList(
252 const SubscriptionManager* mgr,
253 krisbash 1.1 SubMgrSubscriptionPtr** subs,
254 size_t* count)
255 {
256 return SubscriptionList_GetList( & mgr->subscrList, MI_TRUE, subs, count );
257 }
258
259
260 /*
261 * Perform generic initialization of an SubscriptionManager object.
262 */
263 _Use_decl_annotations_
264 SubMgrSubscription* _SubMgrSubscription_New(
265 SubscribeReq* msg,
266 CallSite cs)
267 {
268 SubMgrSubscription* subscription = (SubMgrSubscription*)PAL_Calloc(1, sizeof(SubMgrSubscription));
269 const char* file = "";
270 size_t line = 0;
271
272 #if defined(CONFIG_ENABLE_DEBUG)
273 file = cs.file;
274 krisbash 1.1 line = cs.line;
275 #endif /* defined(CONFIG_ENABLE_DEBUG) */
276
277 if (!subscription)
278 {
279 LOGD_ALLOC_OOM;
280 return NULL;
281 }
282
283 /*
284 * Set refcount to 1, which will be released in context.c:CONTEXT_STRANDAUX_INVOKESUBSCRIBE.
285 */
286 subscription->refcount = 1;
287 subscription->finalmsgbit = 0;
288 subscription->unsubscribebit = 0;
289
290 SubMgrSubscription_SetState( subscription, SubscriptionState_Initialized );
291
292 subscription->msg = msg;
293 subscription->subscriptionID = msg->subscriptionID;
294
295 krisbash 1.1 subscription->filter = InstanceFilter_New( &(msg->base.base) );
296 if (!subscription->filter)
297 {
298 trace_InstanceFilter_AllocFailed();
299 SubMgrSubscription_Release( subscription );
300 return NULL;
301 }
302
303 RecursiveLock_Init(&subscription->postlock);
304
305 trace_SubMgrSubscription_New(file, (MI_Uint32)line, (void*)subscription, tcs(msg->className), (unsigned int)subscription->refcount);
306
307 /* filter must be initialized separately based on the request message. */
308 return subscription;
309 }
310
311 _Use_decl_annotations_
312 void _SubMgrSubscription_Addref(
313 SubMgrSubscription* subscription,
314 CallSite cs)
315 {
316 krisbash 1.1 ptrdiff_t count = Atomic_Inc(&subscription->refcount);
317 const char* file = "";
318 size_t line = 0;
319
320 #if defined(CONFIG_ENABLE_DEBUG)
321 file = cs.file;
322 line = cs.line;
323 #endif /* defined(CONFIG_ENABLE_DEBUG) */
324
325 trace_SubMgrSubscription_Addref(file, (MI_Uint32)line, (void*)subscription, (unsigned int)count);
326 }
327
328
329 _Use_decl_annotations_
330 void _SubMgrSubscription_Release(
331 SubMgrSubscription* subscription,
332 CallSite cs)
333 {
334 void *sub = subscription;
335 const char* file = "";
336 size_t line = 0;
337 krisbash 1.1
338 #if defined(CONFIG_ENABLE_DEBUG)
339 file = cs.file;
340 line = cs.line;
341 #endif /* defined(CONFIG_ENABLE_DEBUG) */
342
343 if (subscription)
344 {
345 /* Finalize the subscription if ref count decreased to 0 */
346 ptrdiff_t count = Atomic_Dec(&subscription->refcount);
347 if (count == 0)
348 {
349 trace_SubMgrSubscription_Release_Finalized( UintThreadID(), subscription );
350
351 if (subscription->filter)
352 {
353 InstanceFilter_Destroy( subscription->filter );
354 }
355
356 if (subscription->subscribeCtx)
357 {
358 krisbash 1.1 SubscrContext_Close(subscription->subscribeCtx);
359 subscription->subscribeCtx = NULL;
360 }
361 PAL_Free(subscription);
362 }
363 trace_SubMgrSubscription_Release(file, (MI_Uint32)line, sub, (unsigned int)count);
364 }
365
366 /* Done to satisfy OACR since it doesn't detect a change to subscription when
367 * it is free'd. This just modified the local variable subscription, not the
368 * actual ptr in the caller. */
369 subscription = NULL;
370 }
371
372 _Use_decl_annotations_
373 MI_Result SubMgrSubscription_SetState(
374 SubMgrSubscription* subscription,
375 SubscriptionState state )
376 {
377 subscription->state = state;
378 return MI_RESULT_OK;
379 krisbash 1.1 }
380
381 _Use_decl_annotations_
382 SubscriptionTargetType SubMgrSubscription_GetSupportedTypes(
383 SubMgrSubscription* subscription )
384 {
385 if (subscription &&
386 subscription->msg)
387 {
388 return (SubscriptionTargetType)subscription->msg->targetType;
389 }
390 else
391 {
392 return SUBSCRIP_TARGET_UNSUPPORTED;
393 }
394 }
395
396 _Use_decl_annotations_
397 void SubMgrSubscription_AcuquirePostLock(SubMgrSubscription* self)
398 {
399 RecursiveLock_Acquire(&self->postlock);
400 krisbash 1.1
401 trace_SubMgrSubscription_AcquirePostLock(UintThreadID(), (void*)self, self->subscribeCtx);
402 }
403
404 _Use_decl_annotations_
405 void SubMgrSubscription_ReleasePostLock(SubMgrSubscription* self)
406 {
407 trace_SubMgrSubscription_ReleasePostLock(UintThreadID(), (void*)self, self->subscribeCtx);
408
409 RecursiveLock_Release(&self->postlock);
410 }
411
412 _Use_decl_annotations_
413 MI_Boolean SubMgrSubscription_ShouldCallUnsubscribe(
414 SubMgrSubscription* self)
415 {
416 MI_Boolean shouldunsubscribe = MI_FALSE;
417
418 if ( Atomic_CompareAndSwap(&self->unsubscribebit, 0, 1) == 0 )
419 shouldunsubscribe = MI_TRUE;
420 else
421 krisbash 1.1 trace_SubMgrSubscription_ShouldCallUnsubscribe_AlreadyUnsubscribed( UintThreadID(), self );
422
423 trace_SubMgrSubscription_ShouldCallUnsubscribe(UintThreadID(), self, shouldunsubscribe);
424
425 return shouldunsubscribe;
426 }
427
428 _Use_decl_annotations_
429 MI_Boolean SubMgrSubscription_ShouldSendFinalMsg(
430 SubMgrSubscription* self)
431 {
432 MI_Boolean shouldsendfinalmsg = MI_FALSE;
433 if ( Atomic_CompareAndSwap(&self->finalmsgbit, 0, 1) == 0 )
434 shouldsendfinalmsg = MI_TRUE;
435 else
436 trace_SubMgrSubscription_ShouldSendFinalMsg_AlreadySent( UintThreadID(), self );
437
438 trace_SubMgrSubscription_ShouldSendFinalMsg(UintThreadID(), self, shouldsendfinalmsg);
439
440 return shouldsendfinalmsg;
441 }
442 krisbash 1.1
443 _Use_decl_annotations_
444 MI_Boolean SubMgrSubscription_CancelStarted(
445 SubMgrSubscription* self)
446 {
447 MI_Boolean cancelstarted = MI_FALSE;
448 if ( ReadWithFence(&self->unsubscribebit) == 1 )
449 cancelstarted = MI_TRUE;
450 else if ( ReadWithFence(&self->finalmsgbit) == 1 )
451 cancelstarted = MI_TRUE;
452 return cancelstarted;
453 }
454
455 /* Validate the filter using the provided class decl. */
456 _Use_decl_annotations_
457 MI_Boolean SubMgrSubscription_IsQueryValid(
458 SubMgrSubscription* self,
459 const MI_ClassDecl* cd )
460 {
461 WQL* wql = InstanceFilter_GetWQL(self->filter);
462 if (wql)
463 krisbash 1.1 {
464 if (WQL_Validate(wql, cd) == 0)
465 {
466 return MI_TRUE;
467 }
468 trace_QueryValidationFailed(tcs(wql->text));
469 }
470 return MI_FALSE;
471 }
472
473 /*
474 * There are 3 places needs to take enablelock
475 * 1- Subscribe request
476 * 2- post result on aggregation context
477 * 3- last subscription was removed
478 *
479 * For 2 & 3, enablebit to make sure only take the lock once
480 *
481 * parameters:
482 * terminate -- MI_TRUE means being called by Provider_TerminateIndication
483 *
484 krisbash 1.1 * return value:
485 * MI_TRUE -- acquired the lock, otherwise no lock acquired
486 *
487 */
488 const TChar* _OpNames[] = {
489 PAL_T("SubscribeProvider"),
490 PAL_T("DisableProvider"),
491 PAL_T("Terminate")
492 };
493
494 _Use_decl_annotations_
495 MI_Boolean SubMgr_AcquireEnableLock(
496 SubscriptionManager* self,
497 AcquireEnableLockOperationType optype)
498 {
499 trace_SubscriptionManager_AcquireEnableLock_Start(UintThreadID(), (void*)self, _OpNames[(int)optype]);
500
501 RecursiveLock_Acquire( &self->enablelock );
502
503 switch( optype )
504 {
505 krisbash 1.1 case AcquireFromSubscribe:
506 if ( MI_TRUE == SubMgr_IsTerminating( self ) )
507 {
508 trace_SubscriptionManager_AcquireEnableLock_AlreadyTerminated( UintThreadID(), self );
509 RecursiveLock_Release( &self->enablelock );
510 return MI_FALSE;
511 }
512 break;
513 case AcquireFromDisable:
514 if ( MI_FALSE == SubMgr_IsSubListEmpty( self ) )
515 {
516 if ( MI_FALSE == SubMgr_IsTerminating( self ) )
517 {
518 trace_SubscriptionManager_AcquireEnableLock_IgnoreDisableCall( UintThreadID(), self );
519 RecursiveLock_Release( &self->enablelock );
520 return MI_FALSE;
521 }
522 else
523 DEBUG_ASSERT (0);
524 }
525 break;
526 krisbash 1.1 case AcquireFromTerminate:
527 if ( MI_FALSE == SubMgr_IsTerminating( self ) )
528 {
529 SubMgr_SetTerminating( self, MI_TRUE );
530 if ( MI_FALSE == SubMgr_IsSubListEmpty( self ) )
531 {
532 trace_SubscriptionManager_AcquireEnableLock_CancelAll( UintThreadID(), self );
533 SubMgr_CancelAllSubscriptions( self, MI_RESULT_FAILED, NULL, NULL );
534 }
535 /* no subscription added yet, can happen during first subscribe */
536 else
537 {
538 Provider_InvokeDisable( self->provider );
539 }
540 }
541 if ( MI_TRUE == SubMgr_IsEnabled(self) )
542 SubMgr_SetEnabled( self, MI_FALSE );
543 break;
544 }
545
546 trace_SubscriptionManager_AcquireEnableLock_Complete(UintThreadID(), (void*)self, _OpNames[(int)optype]);
547 krisbash 1.1
548 return MI_TRUE;
549 }
550
551 _Use_decl_annotations_
552 void SubMgr_ReleaseEnableLock(
553 SubscriptionManager* self)
554 {
555 trace_SubscriptionManager_ReleaseEnableLock(UintThreadID(), (void*)self);
556 RecursiveLock_Release(&self->enablelock);
557 }
558
559 _Use_decl_annotations_
560 AggregationContext* SubMgr_CreateAggrContext(
561 SubscriptionManager* self)
562 {
563 MI_Result result;
564 Provider* provider = self->provider;
565 AggregationContext* aggrContext = AggrContext_New();
566 DEBUG_ASSERT( !self->aggrCtx );
567
568 krisbash 1.1 if (!aggrContext)
569 {
570 trace_OutOfMemory();
571 return NULL;
572 }
573
574 result = AggrContext_Init(aggrContext, provider, self);
575 if (MI_RESULT_OK != result)
576 {
577 trace_AggregationContext_InitFailed();
578 AggrContext_Delete(aggrContext);
579 return NULL;
580 }
581
582 self->aggrCtx = aggrContext;
583 return aggrContext;
584 }
585
586 _Use_decl_annotations_
587 AggregationContext* SubMgr_RemoveAggrContext(
588 SubscriptionManager* self)
589 krisbash 1.1 {
590 AggregationContext* aggrctx = self->aggrCtx;
591 if ( aggrctx )
592 self->aggrCtx = NULL;
593 return aggrctx;
594 }
595
596 _Use_decl_annotations_
597 MI_Boolean SubMgr_IsSubListEmpty(
598 SubscriptionManager* mgr )
599 {
600 return (mgr->subscrList.count == 0);
601 }
602
603 _Use_decl_annotations_
604 MI_Boolean SubMgr_IsEnabled(
605 const SubscriptionManager* mgr )
606 {
607 return mgr->enabled;
608 }
609
610 krisbash 1.1 _Use_decl_annotations_
611 void SubMgr_SetEnabled(
612 SubscriptionManager* mgr,
613 MI_Boolean enabled)
614 {
615 mgr->enabled = enabled;
616 }
617
618 _Use_decl_annotations_
619 MI_Boolean SubMgr_IsTerminating(
620 const SubscriptionManager* mgr )
621 {
622 return mgr->terminating;
623 }
624
625 _Use_decl_annotations_
626 void SubMgr_SetTerminating(
627 SubscriptionManager* mgr,
628 MI_Boolean terminating)
629 {
630 mgr->terminating = terminating;
631 krisbash 1.1 }
632
633 _Use_decl_annotations_
634 void SubMgr_SetAllCancelledSafe(
635 SubscriptionManager* mgr,
636 MI_Boolean allcancelled)
637 {
638 DEBUG_ASSERT ( mgr );
639 SubscriptionList_SetAllCancelledSafe( &mgr->subscrList, allcancelled );
640 }
641
642 _Use_decl_annotations_
643 void SubMgr_SetEnableThread(
644 SubscriptionManager* mgr)
645 {
646 mgr->enableThreadID = Thread_ID();
647 }
648
649 _Use_decl_annotations_
650 MI_Boolean SubMgr_IsEnableThread(
651 const SubscriptionManager* mgr)
652 krisbash 1.1 {
653 ThreadID threadId = Thread_ID();
654 return Thread_Equal( &threadId, (ThreadID*)&mgr->enableThreadID ) ? MI_TRUE : MI_FALSE;
655 }
656
657 _Use_decl_annotations_
658 MI_Boolean SubMgr_CanPostIndication(
659 const SubscriptionManager* mgr)
660 {
661 MI_Boolean isEnableThread = SubMgr_IsEnableThread(mgr);
662 MI_Boolean canPostIndication = isEnableThread ? MI_FALSE : MI_TRUE;
663 if ( MI_FALSE == canPostIndication )
664 trace_SubMgr_CanPostIndication_Fail( UintThreadID() );
665 return canPostIndication;
666 }
667
668 static MI_Result _SubscriptionList_EnsureArray(
669 _In_ SubscriptionList* self)
670 {
671 size_t capacity = self->capacity;
672 SubMgrSubscriptionPtr* subarray;
673 krisbash 1.1 if ( self->count <= capacity )
674 return MI_RESULT_OK;
675
676 if ( capacity == 0 )
677 capacity = 32;
678 else
679 capacity *= 2;
680
681 if ( capacity < self->count )
682 {
683 /* integer overflow */
684 trace_SubscriptionList_EnsureArray_Overflow( UintThreadID() );
685 return MI_RESULT_FAILED;
686 }
687
688 if ( self->subarray )
689 {
690 PAL_Free(self->subarray);
691 self->subarray = NULL;
692 self->capacity = 0;
693 }
694 krisbash 1.1
695 subarray = (SubMgrSubscriptionPtr*)PAL_Malloc( sizeof (SubMgrSubscriptionPtr) * capacity );
696 if (subarray == NULL)
697 {
698 trace_OutOfMemory();
699 return MI_RESULT_SERVER_LIMITS_EXCEEDED;
700 }
701
702 self->subarray = subarray;
703 self->capacity = capacity;
704 return MI_RESULT_OK;
705 }
706
707 _Use_decl_annotations_
708 void SubscriptionList_Init(
709 SubscriptionList* self)
710 {
711 memset( self, 0, sizeof( SubscriptionList ) );
712 ReadWriteLock_Init( &self->lock );
713 }
714
715 krisbash 1.1 _Use_decl_annotations_
716 void SubscriptionList_Finalize(
717 SubscriptionList* self)
718 {
719 if ( self->subarray )
720 {
721 PAL_Free(self->subarray);
722 self->subarray = NULL;
723 self->capacity = 0;
724 }
725 }
726
727 _Use_decl_annotations_
728 void SubscriptionList_AddSubscription(
729 SubscriptionList* self,
730 SubMgrSubscription* subscription)
731 {
732 DEBUG_ASSERT ( self );
733 /* thread safely add subscription to list */
734 ReadWriteLock_AcquireWrite(&self->lock);
735 DEBUG_ASSERT ( MI_FALSE == self->allcancelled );
736 krisbash 1.1 List_Append( &self->head, &self->tail, (ListElem*)subscription );
737 SubMgrSubscription_Addref(subscription);
738 self->count++;
739 ReadWriteLock_ReleaseWrite(&self->lock);
740 }
741
742 _Use_decl_annotations_
743 MI_Result SubscriptionList_DeleteSubscription(
744 SubscriptionList* self,
745 SubMgrSubscription* subscription,
746 SubscriptionManager* mgr )
747 {
748 SubMgrSubscription* sub;
749 /* thread safely remove the subscription from list */
750 ReadWriteLock_AcquireWrite( &self->lock );
751 sub = (SubMgrSubscription*) self->head;
752 /* find the subscription first */
753 while( sub )
754 {
755 if ( sub == subscription )
756 break;
757 krisbash 1.1 sub = sub->next;
758 }
759
760 /* Remove from the list if found */
761 if ( sub )
762 {
763 List_Remove( &self->head, &self->tail, (ListElem*)sub );
764 self->count--;
765
766 /* Done after remove so that it can operate on the updated list */
767 if (mgr->lifecycleCtx)
768 {
769 LifeContext_UpdateSupportedTypes(mgr->lifecycleCtx);
770 }
771 }
772
773 ReadWriteLock_ReleaseWrite( &self->lock );
774
775 if (sub)
776 {
777 SubMgrSubscription_Release(sub);
778 krisbash 1.1 return MI_RESULT_OK;
779 }
780 return MI_RESULT_NOT_FOUND;
781 }
782
783 _Use_decl_annotations_
784 MI_Result SubscriptionList_CancelAllSubscription(
785 SubscriptionList* self )
786 {
787 SubMgrSubscriptionPtr* sublist;
788 size_t i, count;
789 MI_Result r;
790
791 DEBUG_ASSERT ( MI_FALSE == self->allcancelled );
792
793 r = SubscriptionList_GetList( self, MI_TRUE, &sublist, &count );
794 if ( r != MI_RESULT_OK )
795 return r;
796
797 SubscriptionList_SetAllCancelledSafe( self, MI_TRUE );
798
799 krisbash 1.1 for ( i = 0; i < count; i++ )
800 {
801 /*TODO: what if a subscription was being cancelled. confirm */
802 Strand_ScheduleCancel( &sublist[i]->subscribeCtx->baseCtx.strand );
803 SubMgrSubscription_Release( sublist[i] );
804 }
805
806 return MI_RESULT_OK;
807 }
808
809 _Use_decl_annotations_
810 MI_Result SubscriptionList_GetList(
811 const SubscriptionList* self,
812 MI_Boolean addref,
813 SubMgrSubscriptionPtr** subs,
814 size_t* count)
815 {
816 MI_Result r;
817 SubscriptionList* list = (SubscriptionList*)self;
818
819 *subs = NULL;
820 krisbash 1.1 *count = 0;
821
822 //
823 // thread safely read subscriptions
824 //
825 ReadWriteLock_AcquireRead( &list->lock );
826
827 r = _SubscriptionList_EnsureArray( list );
828
829 if ( r == MI_RESULT_OK )
830 {
831 size_t i = 0;
832 SubMgrSubscription* subscription = (SubMgrSubscription*) list->head;
833 while(subscription)
834 {
835 if ( MI_TRUE == addref )
836 SubMgrSubscription_Addref( subscription );
837 list->subarray[i++] = subscription;
838 subscription = subscription->next;
839 }
840 DEBUG_ASSERT( i == list->count );
841 krisbash 1.1 *count = list->count;
842 *subs = list->subarray;
843 }
844
845 ReadWriteLock_ReleaseRead( &list->lock );
846
847 return r;
848 }
849
850 _Use_decl_annotations_
851 void SubscriptionList_SetAllCancelled(
852 SubscriptionList* self,
853 MI_Boolean allcancelled)
854 {
855 self->allcancelled = allcancelled;
856 }
857
858 void SubscriptionList_SetAllCancelledSafe(
859 _Inout_ SubscriptionList* self,
860 _In_ MI_Boolean allcancelled)
861 {
862 krisbash 1.1 ReadWriteLock_AcquireWrite( &self->lock );
863 SubscriptionList_SetAllCancelled(self, allcancelled);
864 ReadWriteLock_ReleaseWrite( &self->lock );
865 }
866
|
Return to
SubMgr.c
CVS log
Up to [OMI] / omi / provmgr