Active Object

1. The proxy provides an interface for the accessible member functions on the Active Object. The proxy triggers the construction of a request into the activation list. The proxy runs in the client thread.
2. The method request class defines the interface for the method executing on an Active Object. This interface also contains guard methods, indicating if the job is ready to run. The request includes all context information to be executed later.
3. The activation list maintains the pending requests. The activation list decouples the client’s thread from the Active Object thread. The proxy inserts the request object, and the scheduler removes them. Consequently, the access to the activation list must be serialized.
4. The scheduler runs in the thread of the Active Object and decides which request from the activation list is executed next. The scheduler evaluates the guards of the request to determine if the request can run.
5. The servant implements the Active Object and runs in the active object’s thread. The servant implements the interface of the method request, and the scheduler invokes its member functions.
6. The proxy creates the future and s only necessary if the request returns a result. Therefore, the client receives the future and can obtain the method invocation result on the Active Object. The client can wait for the outcome or poll for it.

Dynamic Behavior

1. Request construction and scheduling: The client invokes a method on the proxy. The proxy creates a request and passes it to the scheduler. The scheduler enqueues the request on the activation list. Additionally, the proxy returns a future to the client if the request returns a result.
2. Member function execution: The scheduler determines which request becomes runnable by evaluating the guard method of the request. It removes the request from the activation list and dispatches it to the servant. 
3. Completion: When the request returns something, it is stored in the future. The client can ask for the result. When the client has the result, the request and the future can be deleted.

Pros

// activeObject.cpp

#include <algorithm>
#include <deque>
#include <functional>
#include <future>
#include <iostream>
#include <memory>
#include <mutex>
#include <numeric>
#include <random>
#include <thread>
#include <utility>
#include <vector>

using std::async;                                               // (1)
using std::boolalpha;
using std::cout;
using std::deque;
using std::distance;
using std::for_each;
using std::find_if;
using std::future;
using std::lock_guard;
using std::make_move_iterator;
using std::make_pair;
using std::move;
using std::mt19937;
using std::mutex;
using std::packaged_task;
using std::pair;
using std::random_device;
using std::sort;
using std::jthread;
using std::uniform_int_distribution;
using std::vector;

class IsPrime {                                                 // (8)
 public:
    IsPrime(int num): numb{num} {} 
    pair<bool, int> operator()() {
        for (int j = 2; j * j <= numb; ++j){
            if (numb % j == 0) return make_pair(false, numb);
        }
        return make_pair(true, numb);
    }
 private:
    int numb;       
};

class ActiveObject {
 public:
    
    future<pair<bool, int>> enqueueTask(int i) {
        IsPrime isPrime(i);
        packaged_task<pair<bool, int>()> newJob(isPrime);
        auto isPrimeFuture = newJob.get_future();
        {
            lock_guard<mutex> lockGuard(activationListMutex);
            activationList.push_back(move(newJob));            // (6)
        }
        return isPrimeFuture;
    }

    void run() {
        std::jthread j([this] {                                 // (12)
            while ( !runNextTask() );                           // (13)
        });
    }

 private:

    bool runNextTask() {                                        // (14)
        lock_guard<mutex> lockGuard(activationListMutex);
        auto empty = activationList.empty();
        if (!empty) {                                           // (15)
            auto myTask= std::move(activationList.front());
            activationList.pop_front();
            myTask();
        }
        return empty;
    }

    deque<packaged_task<pair<bool, int>()>> activationList;      //(7)
    mutex activationListMutex;
};

vector<int> getRandNumbers(int number) {
    random_device seed;
    mt19937 engine(seed());
    uniform_int_distribution<> dist(1'000'000, 1'000'000'000);  // (4)
    vector<int> numbers;
    for (long long i = 0 ; i < number; ++i) numbers.push_back(dist(engine)); 
    return numbers;
}

future<vector<future<pair<bool, int>>>> getFutures(ActiveObject& activeObject, 
                                                   int numberPrimes) {
    return async([&activeObject, numberPrimes] {
        vector<future<pair<bool, int>>> futures;
        auto randNumbers = getRandNumbers(numberPrimes);        // (3)
        for (auto numb: randNumbers){
            futures.push_back(activeObject.enqueueTask(numb));  // (5)
        }
        return futures;
    });
}
    

int main() {
    
    cout << boolalpha << '\n';
    
     ActiveObject activeObject;
        
    // a few clients enqueue work concurrently                  // (2)
    auto client1 = getFutures(activeObject, 1998);
    auto client2 = getFutures(activeObject, 2003);
    auto client3 = getFutures(activeObject, 2011);
    auto client4 = getFutures(activeObject, 2014);
    auto client5 = getFutures(activeObject, 2017);
    
    // give me the futures                                      // (9)
    auto futures = client1.get();
    auto futures2 = client2.get();
    auto futures3 = client3.get();
    auto futures4 = client4.get();
    auto futures5 = client5.get();
    
    // put all futures together                                 // (10)
    futures.insert(futures.end(),make_move_iterator(futures2.begin()), 
                                 make_move_iterator(futures2.end()));
    
    futures.insert(futures.end(),make_move_iterator(futures3.begin()), 
                                 make_move_iterator(futures3.end()));
    
    futures.insert(futures.end(),make_move_iterator(futures4.begin()), 
                                 make_move_iterator(futures4.end()));
    
    futures.insert(futures.end(),make_move_iterator(futures5.begin()), 
                                 make_move_iterator(futures5.end()));
        
    // run the promises                                         // (11)
    activeObject.run();
    
    // get the results from the futures
    vector<pair<bool, int>> futResults;
    futResults.reserve(futures.size());
    for (auto& fut: futures) futResults.push_back(fut.get());   // (16)
    
    sort(futResults.begin(), futResults.end());                 // (17)
    
    // separate the primes from the non-primes
    auto prIt = find_if(futResults.begin(), futResults.end(), 
                        [](pair<bool, int> pa){ return pa.first == true; });
 
    cout << "Number primes: " << distance(prIt, futResults.end()) << '\n';       // (19)
    cout << "Primes:" << '\n';
    for_each(prIt, futResults.end(), [](auto p){ cout << p.second << " ";} );    // (20)
    
    cout << "\n\n";
    
    cout << "Number no primes: " << distance(futResults.begin(), prIt) << '\n';  // (18)
    cout << "No primes:" << '\n';
    for_each(futResults.begin(), prIt, [](auto p){ cout << p.second << " ";} );
    
    cout << '\n';
    
}

What’s Next?

The Active Object and Monitor Object synchronize and schedule member function invocation. The main difference is that the Active Object executes its member function in a different thread, but the Monitor Object is in the same thread as the client. In my next post, I will present the Monitor Object.