parallel.h

 
 
#pragma once
 
#include <system/action.h>
#include <system/func.h>
#include <system/object.h>
#include <system/threading/tasks/task_utils.h>
 
namespace System { namespace Threading { namespace Tasks {
 
class ASPOSECPP_SHARED_CLASS ParallelLoopResult : public Object, public System::Details::BoxableObjectBase
{
    typedef ParallelLoopResult ThisType;
    typedef Object BaseType;
 
    typedef ::System::BaseTypesInfo<BaseType> ThisTypeBaseTypesInfo;
 
public:
    ParallelLoopResult() : _completed(false) {}
 
    bool get_IsCompleted() {return _completed;}
    
    Nullable<int64_t> get_LowestBreakIteration() {return _lowestBreakIteration;}
 
protected:
    bool _completed;                            
    Nullable<int64_t> _lowestBreakIteration;    
};
 
class ASPOSECPP_SHARED_CLASS ParallelOptions : public Object
{
    typedef ParallelOptions ThisType;
    typedef Object BaseType;
 
    typedef ::System::BaseTypesInfo<BaseType> ThisTypeBaseTypesInfo;
 
public:
    ASPOSECPP_SHARED_API ParallelOptions();
 
    const SharedPtr<TaskScheduler>& get_TaskScheduler() {return _scheduler;}
    
    void set_TaskScheduler(const SharedPtr<TaskScheduler>& value) {_scheduler = value;}
    
    int32_t get_MaxDegreeOfParallelism() {return _maxDegreeOfParallelism;}
    
    void set_MaxDegreeOfParallelism(int32_t value) {_maxDegreeOfParallelism = value;}
    
    const CancellationToken& get_CancellationToken() {return _cancellationToken;}
    
    void set_CancellationToken(const CancellationToken& value) {_cancellationToken = value;}
 
protected:
    friend class Parallel;
 
    ASPOSECPP_SHARED_API SharedPtr<TaskScheduler> get_EffectiveTaskScheduler();
    
    ASPOSECPP_SHARED_API int32_t get_EffectiveMaxConcurrencyLevel();
 
private:
    SharedPtr<TaskScheduler> _scheduler;    
    int32_t _maxDegreeOfParallelism;        
    CancellationToken _cancellationToken;   
};
 
class ASPOSECPP_SHARED_CLASS Parallel
{
public:
    template <typename TSource>
    static ParallelLoopResult ForEach(const SharedPtr<Collections::Generic::IEnumerable<TSource>>& source,
        const SharedPtr<ParallelOptions>& parallelOptions, const Action<TSource>& body)
    {
        parallelOptions->get_CancellationToken().ThrowIfCancellationRequested();
 
        auto enumerator = source->GetEnumerator();
        auto tasks = MakeArray<TaskPtr>(parallelOptions->get_EffectiveMaxConcurrencyLevel());
        auto lock = MakeObject<Object>();
 
        for (auto& task : tasks)
        {
            task = MakeAsync([=, &lock, &enumerator, &body](System::Details::AsyncContext& __) mutable
            {
                if (parallelOptions->get_CancellationToken().get_IsCancellationRequested())
                {
                    __.Return(); return;
                }
 
                TSource value;
                {
                    LockContext __guard((lock).GetPointer());
                    if (enumerator->MoveNext())
                    {
                        value = enumerator->get_Current();
                    }
                    else
                    {
                        __.Return(); return;
                    }
                }
                body(value);
            });
        }
 
        Tasks::WaitAll(tasks, parallelOptions->get_CancellationToken());
 
        return ParallelLoopResult();
    }
 
    template<typename TSource>
    static ParallelLoopResult ForEach(const SharedPtr<Collections::Generic::IEnumerable<TSource>>& source, const Action<TSource>& body)
    {
        return ForEach(source, s_defaultParallelOptions, body);
    }
 
protected:
    ASPOSECPP_SHARED_API static SharedPtr<ParallelOptions> s_defaultParallelOptions;
};
 
}}} // namespace System::Threading::Tasks