1 sahana.prabhakar 1.1 //%LICENSE////////////////////////////////////////////////////////////////
  2                      //
  3                      // Licensed to The Open Group (TOG) under one or more contributor license
  4                      // agreements.  Refer to the OpenPegasusNOTICE.txt file distributed with
  5                      // this work for additional information regarding copyright ownership.
  6                      // Each contributor licenses this file to you under the OpenPegasus Open
  7                      // Source License; you may not use this file except in compliance with the
  8                      // License.
  9                      //
 10                      // Permission is hereby granted, free of charge, to any person obtaining a
 11                      // copy of this software and associated documentation files (the "Software"),
 12                      // to deal in the Software without restriction, including without limitation
 13                      // the rights to use, copy, modify, merge, publish, distribute, sublicense,
 14                      // and/or sell copies of the Software, and to permit persons to whom the
 15                      // Software is furnished to do so, subject to the following conditions:
 16                      //
 17                      // The above copyright notice and this permission notice shall be included
 18                      // in all copies or substantial portions of the Software.
 19                      //
 20                      // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
 21                      // OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 22 sahana.prabhakar 1.1 // MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
 23                      // IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
 24                      // CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
 25                      // TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
 26                      // SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 27                      //
 28                      //////////////////////////////////////////////////////////////////////////
 29                      //
 30                      //%///////////////////////////////////////////////////////////////////////
 31                      
 32                      #ifndef Pegasus_Async_RequestExecutor_h
 33                      #define Pegasus_Async_RequestExecutor_h
 34                      
 35                      #include <Pegasus/Common/Condition.h>
 36                      #include <Pegasus/Common/ThreadPool.h>
 37                      #include <Pegasus/Common/CIMMessage.h>
 38                      
 39                      #include <Pegasus/ProviderManager2/Linkage.h>
 40                      #include <iostream>
 41                      
 42                      PEGASUS_NAMESPACE_BEGIN
 43 sahana.prabhakar 1.1 
 44                      /**
 45                          A response callback class that can be used to wait on request handler 
 46                          threads.The different request handler threads use the shared
 47                          callback instance to signal completion which in turn aggregates the
 48                          responses and finally signals upon reaching the expected responses 
 49                          thus retuning the aggregated response
 50                      **/
 51                      class PEGASUS_COMMON_LINKAGE ResponseAggregationCallback
 52                      {
 53                      public:
 54                          ResponseAggregationCallback();
 55                          ~ResponseAggregationCallback();
 56                      
 57                          /*
 58                              This function waits till the received response count reaches the 
 59                              expected response count and return the aggregated response.
 60                          */
 61                          CIMException waitForCompletion(Uint32 expectedResponseCount);
 62                      
 63                          /* 
 64 sahana.prabhakar 1.1         This function incremet the response count and signals completion
 65                              once the count reaches the excpected count.
 66                          */
 67                          void signalCompletion(CIMException& responseException);
 68                      
 69                      private:
 70                          CIMException _responseException;
 71                          Uint32 _expectedResponseCount;
 72                          Uint32 _currResponseCount;
 73                          Mutex _mutex;
 74                          Condition _pollingCond;
 75                      };
 76                      
 77                      /**
 78                          A multithreaded executor that processes the submitted requests on multiple
 79                          threads using the thread pool by invoking the request handler per request
 80                          and returns the aggregated response.
 81                      **/
 82                      class PEGASUS_COMMON_LINKAGE AsyncRequestExecutor
 83                      {
 84                      public:
 85 sahana.prabhakar 1.1     //base class representing the request parameter
 86                          class AsyncRequestMsg
 87                          {
 88                           public:
 89                               AsyncRequestMsg() {}
 90                               virtual ~AsyncRequestMsg() {}
 91                          };
 92                      
 93                          /**
 94                          Input :
 95                          1) Function pointer to the function that needs to be called on
 96                             different threads.
 97                          2) The pointer to an object which can be used in the 
 98                             asyncRequestCallback function.
 99                          3) The max number of threads which the Thread Pool can run 
100                             while executing the requests.
101                          **/    
102                          AsyncRequestExecutor(
103                              CIMException (*asyncRequestCallback)(void *, AsyncRequestMsg* ),
104                              void *callbackPtr,
105                              Uint32 numberOfThreads=5);
106 sahana.prabhakar 1.1 
107                          ~AsyncRequestExecutor();
108                      
109                          //execute the  requests on multiple threads by invoking the
110                          //request handler and return the aggregated response.
111                          CIMException executeRequests(Array<AsyncRequestMsg*> requests);
112                      
113                      private:
114                      
115                          static ThreadReturnType PEGASUS_THREAD_CDECL
116                              _requestProcessor(void* arg);
117                      
118                      
119                          CIMException (*_asyncRequestCallback)(void *callbackPtr, 
120                              AsyncRequestMsg* request);
121                      
122                          void *_callbackPtr;
123                      
124                          ThreadPool* _threadPool;
125                      
126                          ResponseAggregationCallback _responseCallback;
127 sahana.prabhakar 1.1 
128                          /* 
129                              Structure used to pass a request to the executor consisting of a
130                              request handler and the request msg.
131                          */
132                          struct ReqThreadParam 
133                          {
134                          public:
135                              ReqThreadParam() {}
136                      
137                              ReqThreadParam( 
138                                  CIMException (*asyncRequestCallback)(void *callbackPtr,
139                                      AsyncRequestMsg *request),
140                                  void *callbackPtr,
141                                  AsyncRequestMsg* request,
142                                  ResponseAggregationCallback* respCallback):
143                                      _asyncRequestCallback(asyncRequestCallback),
144                                      _callbackPtr(callbackPtr),
145                                      _request(request),
146                                      _responseCallback(respCallback) 
147                              {}
148 sahana.prabhakar 1.1 
149                              ~ReqThreadParam() 
150                              { }
151                      
152                              CIMException (*_asyncRequestCallback)(void *,AsyncRequestMsg*);
153                              void *_callbackPtr;
154                              AsyncRequestMsg* _request;
155                              ResponseAggregationCallback*  _responseCallback;
156                          };
157                      };
158                      
159                      PEGASUS_NAMESPACE_END
160                      
161                      #endif /* Pegasus_Async_RequestExecutor_h */