modules/platforms/dotnet/Apache.Ignite.Core/Impl/Compute/ComputeImpl.cs
/// <summary>
/// Internal routine for closure-based task execution.
/// </summary>
/// <param name="task">Task.</param>
/// <param name="job">Job.</param>
/// <param name="jobs">Jobs.</param>
/// <param name="opId">Op code.</param>
/// <param name="jobsCount">Jobs count.</param>
/// <param name="writeAction">Custom write action.</param>
/// <returns>Future.</returns>
Internal routine for closure-based task execution.
Internal routine for closure - based task execution .
[SuppressMessage("Microsoft.Design", "CA1031:DoNotCatchGeneralExceptionTypes",
            Justification = "User code can throw any exception")]
        private Future<TReduceRes> ExecuteClosures0<TArg, TJobRes, TReduceRes>(
            IComputeTask<TArg, TJobRes, TReduceRes> task, IComputeJob job = null,
            IEnumerable<IComputeJob> jobs = null, int opId = OpUnicast, int jobsCount = 0,
            Action<BinaryWriter> writeAction = null)
        {
            Debug.Assert(job != null || jobs != null);
            var holder = new ComputeTaskHolder<TArg, TJobRes, TReduceRes>((Ignite) _prj.Ignite, this, task, default(TArg));
            var taskHandle = Marshaller.Ignite.HandleRegistry.Allocate(holder);
            var jobHandles = new List<long>(job != null ? 1 : jobsCount);
            try
            {
                Exception err = null;
                try
                {
                    var futTarget = DoOutOpObject(opId, writer =>
                    {
                        writer.WriteLong(taskHandle);
                        if (job != null)
                        {
                            writer.WriteInt(1);
                            jobHandles.Add(WriteJob(job, writer));
                        }
                        else
                        {
                            writer.WriteInt(jobsCount);
                            Debug.Assert(jobs != null, "jobs != null");
                            jobHandles.AddRange(jobs.Select(jobEntry => WriteJob(jobEntry, writer)));
                        }
                        holder.JobHandles(jobHandles);
                        if (writeAction != null)
                            writeAction(writer);
                    });
                    holder.Future.SetTarget(new Listenable(futTarget));
                }
                catch (Exception e)
                {
                    err = e;
                }
                if (err != null)
                {
                    foreach (var hnd in jobHandles) 
                        Marshaller.Ignite.HandleRegistry.Release(hnd);
                    holder.CompleteWithError(taskHandle, err);
                }
            }
            catch (Exception e)
            {
                holder.CompleteWithError(taskHandle, e);
            }
            return holder.Future;
        }
[ SuppressMessage ( " " , " " , Justification = " " ) ] private Future < TReduceRes > ExecuteClosures0 < TArg , TJobRes , TReduceRes > ( IComputeTask < TArg , TJobRes , TReduceRes > task , IComputeJob job = null , IEnumerable < IComputeJob > jobs = null , int opId = OpUnicast , int jobsCount = 0 , Action < BinaryWriter > writeAction = null ) { Debug . Assert ( job != null || jobs != null ) ; var holder = new ComputeTaskHolder < TArg , TJobRes , TReduceRes > ( ( Ignite ) _prj . Ignite , this , task , default ( TArg ) ) ; var taskHandle = Marshaller . Ignite . HandleRegistry . Allocate ( holder ) ; var jobHandles = new List < long > ( job != null ? 1 : jobsCount ) ; try { Exception err = null ; try { var futTarget = DoOutOpObject ( opId , writer => { writer . WriteLong ( taskHandle ) ; if ( job != null ) { writer . WriteInt ( 1 ) ; jobHandles . Add ( WriteJob ( job , writer ) ) ; } else { writer . WriteInt ( jobsCount ) ; Debug . Assert ( jobs != null , " " ) ; jobHandles . AddRange ( jobs . Select ( jobEntry => WriteJob ( jobEntry , writer ) ) ) ; } holder . JobHandles ( jobHandles ) ; if ( writeAction != null ) writeAction ( writer ) ; } ) ; holder . Future . SetTarget ( new Listenable ( futTarget ) ) ; } catch ( Exception e ) { err = e ; } if ( err != null ) { foreach ( var hnd in jobHandles ) Marshaller . Ignite . HandleRegistry . Release ( hnd ) ; holder . CompleteWithError ( taskHandle , err ) ; } } catch ( Exception e ) { holder . CompleteWithError ( taskHandle , e ) ; } return holder . Future ; }
Internal routine for closure-based task execution.
Internal routine for closure - based task execution .
// Manual job handles release because they were not assigned to the task yet. // This exception means that out-op failed.
None
task
IComputeTask<TArg, TJobRes, TReduceRes>
None
job
IComputeJob
None
jobs
IEnumerable<IComputeJob>
None
opId
int
None
jobsCount
int
None
writeAction
Action<BinaryWriter>
Custom write action.
Custom write action .