Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

Hi all

[brief overview]
I'm evaluating Apache Ignite framework as a replacement for Hazelcast. One
of usages where it's planned to be compared is task/job processing. We have
implemented tasks management by ourselves based on Hazelcast but not using
their MarReduce framework (such as it was very slow in version 3.5). We
developed a proprietary framework, which is similar to Ignite's
ComputeTaskSplitAdapter and ComputeJobAdapter (failover, task status,
distributed execution, etc) at the end and we would like to simplify it by
switching to Apache Ignite.
To simplify discussion - I've created a trivial project (attached here),
which creates one Task with *N* Jobs in it. Jobs are kind of "dummy" - they
do nothing and return nothing. The goal is to check an overhead of task
management (distribution, failover, etc) overall.

And here are some results:
1. 1 Node  - 1 Task - 5000 Jobs -> ~0.8 sec
2. 2 Nodes - 1 Task - 5000 Jobs -> ~10-11 sec
3. 3 Nodes - 1 Task - 5000 Jobs -> ~15 sec
4. 1 Node  - 1 Task - 10000 Jobs -> ~3 sec
5. 2 Nodes - 1 Task - 10000 Jobs -> ~40 sec

[question]
Are these measurements correct from Ignite point of view? Meaning - I didn't
miss nothing in term of configuration and this is the price which should be
paid for that amount of Jobs to be executed in distributed way (10 sec. for
5000 empty jobs for example). Or it can be tuned/speed-up in some way?

Thank you in advance!

P.S. And I realized that for a run with 1 Task and 5000 "dummy" jobs Ignite
cluster uses almost 3.5Gb heap space.

task-management.zip
<http://apache-ignite-users.70518.x6.nabble.com/file/t1316/task-management.zip>  



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hi @yakov


yakov wrote
> Yes, however, you can still return results from each job and use it.
> Please
> see javadoc for org.apache.ignite.compute.ComputeJobResult#getData

yes, it's good to have such opportunity at least on "result" step.
But still I'm very curious, why the overhead is so big when results are
being collected on reduce step.

I've done a quick profile action (without deep analysis) and it seems to me
that with "cached results" for MapReduce process there is a big overhead in
org.jsr166.* package. Meaning there are quite a lot of invocation of
"get/poll" and "put" methods from concurrent impl. of Map
(ConcurentHashMap8, ConcurentLinkedHashMap, ConcurentLinkedDeque8) - 
<http://apache-ignite-users.70518.x6.nabble.com/file/t1316/Ignite-JProfile.jpg> 
 
here is ConcurrentLinkedDeque8.poll exceptional method call
<http://apache-ignite-users.70518.x6.nabble.com/file/t1316/Ignite-JProfile-ConcurrentLinkedDeque8-poll.jpg> 

Will try to dig into it more at my spare time...



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Yakov Zhdanov <yz...@apache.org>]

Yes, however, you can still return results from each job and use it. Please
see javadoc for org.apache.ignite.compute.ComputeJobResult#getData

--Yakov

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

yakov wrote
> What are your timings now?

on two local nodes, after jvm is warmed up (~100 executions), it's running
in average 30ms instead of 6 sec when result is returned in return/reduce
phase. This is a huge improvement! 
I can take it now as a basis and start adding some additional behavior on
top of it (task status persistence, custom job exception handling etc.).

Just for my understanding - after reading the javadoc, I've understood that
this annotation is exactly for use-cases where results of distributed jobs
are not required to be returned to a caller/reducet although these jobs
still can write/read from distributed cache in collocate manner (when
https://issues.apache.org/jira/browse/IGNITE-5037 is implemented), right? 




--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Yakov Zhdanov <yz...@apache.org>]

You are welcome!

What are your timings now?

--Yakov

2017-09-07 15:01 GMT+03:00 ihorps <ih...@gmail.com>:

> hi @yakov
>
>
> yakov wrote
> > Try attaching @ComputeTaskNoResultCache to your task.
>
> Thank you for the hint. It speeds up task management processing
> drastically!
>
>
>
> --
> Sent from: http://apache-ignite-users.70518.x6.nabble.com/
>

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hi @yakov


yakov wrote
> Try attaching @ComputeTaskNoResultCache to your task.

Thank you for the hint. It speeds up task management processing drastically!



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Yakov Zhdanov <yz...@apache.org>]

Try attaching @ComputeTaskNoResultCache to your task.

Also filed a ticket - https://issues.apache.org/jira/browse/IGNITE-6284

As far as 2 - I meant empty runnables submitted to an JDK thread pool
executor - submission will require to acquire a lock and notify pool
thread. So overhead is very significant compared to an execution of a no-op
runnable. Ignite processes job execution requests coming from remote node
in public pool
(org.apache.ignite.configuration.IgniteConfiguration#getPublicThreadPoolSize)
and submits jobs to this pool mapped from local node.

--Yakov

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hi @yakov

Thank you for your feedback.

1. yes, warming up a jvm - this is what I missed at the begging (no doubts
here at all). I can confirm that it gets better in average after few dozens
of run.
2. did you mean than IgniteRunnable/IgniteCallable here (efficiency for
no-op task/job)? I'd like to use Map Reduce framework (especially when
IGNITE-5037 is implemented) such as it gives me almost out-of-the-box a
feature of Task management in terms of collocation execution, failover,
distribution. If I would go with IgniteRunnable/IgniteCallable I'd have to
take over our proprietary code, which we implemented in hazelcast and this
is what I want to avoid.
3. Can confirm that is runs ~ 1sec. faster after JVM is warmed up (4000
jobs, 1 task and to local nodes).

For now I'm satisfied with the "first-try-touch" experience. Looking forward
for IGNITE-5037.

The next topic is to compare memory consumption, such I wrote in previous
comments, running 4000 no-op jobs with help of Map Reduce API I could follow
that at least 2Gb additional memory was used for such run. Still have to
investigate why and how it works in more details.

Thanks!  



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Yakov Zhdanov <yz...@apache.org>]

Guys,

I see the following issues with the benchmark:

1. There is only one iteration. I would put it in a loop and measure at
least hundred of iterations.
2. no-op jobs are not real world example at all =) job requests are
processed in thread pool executor which is not very much effective for such
usecase.
3. Job is an inner class inside NoOpTask. Evgeniy, can you please change it
to a static class and check if it helps here?
4. Also, multinode tests should be run on multiple machines (preferably, 1
node per host).

--Yakov

Re: Task management - MapReduce & ForkJoin performance penalty

[Evgenii Zhuravlev <e....@gmail.com>]

But of course, it could be changed. The community didn't decide yet if wiki
doesn't have information about it.

2017-09-05 17:46 GMT+03:00 Evgenii Zhuravlev <e....@gmail.com>:

> I think it was planned at the end of October.
>
> Evgenii
>
> 2017-09-05 17:41 GMT+03:00 ihorps <ih...@gmail.com>:
>
>> hi, @ezhuravlev
>>
>> This is what I'm looking for, many thanks!
>>
>> Some hints when v2.3 is planned to be release (I can't find it on wiki)?
>>
>> I'd rather wait for this API in Ignite then implementing it by myself an
>> throw it later such as I'm in evaluation/prototype phase now.
>>
>> Best regards,
>> ihorps
>>
>>
>>
>> --
>> Sent from: http://apache-ignite-users.70518.x6.nabble.com/
>>
>
>

Re: Task management - MapReduce & ForkJoin performance penalty

[Evgenii Zhuravlev <e....@gmail.com>]

I think it was planned at the end of October.

Evgenii

2017-09-05 17:41 GMT+03:00 ihorps <ih...@gmail.com>:

> hi, @ezhuravlev
>
> This is what I'm looking for, many thanks!
>
> Some hints when v2.3 is planned to be release (I can't find it on wiki)?
>
> I'd rather wait for this API in Ignite then implementing it by myself an
> throw it later such as I'm in evaluation/prototype phase now.
>
> Best regards,
> ihorps
>
>
>
> --
> Sent from: http://apache-ignite-users.70518.x6.nabble.com/
>

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hi, @ezhuravlev

This is what I'm looking for, many thanks!

Some hints when v2.3 is planned to be release (I can't find it on wiki)? 

I'd rather wait for this API in Ignite then implementing it by myself an
throw it later such as I'm in evaluation/prototype phase now.

Best regards,
ihorps



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Evgenii Zhuravlev <e....@gmail.com>]

Hi,

Here is a ticket for exactly what you want, it's in progress right now:
https://issues.apache.org/jira/browse/IGNITE-5037

If you don't want to wait till it will be implemented, you can use
affinityCall(...) or affinityRun(...) and somehow reduce result after it
will be returned.

Evgenii

2017-09-04 21:52 GMT+03:00 ihorps <ih...@gmail.com>:

> hi @ezhuravlev
>
> Thank you for your reply, very appreciated!
>
> I can confirm that by adding real business logic to Jobs it's actually
> scales horizontally quite well and by adding more nodes the whole task
> finishes just faster.
>
> One more think, which I'm looking now on is running tasks with help of
> MapReduce API in collocated fashion. As far as I understood from
> documentation ( Collocate Computing and Data
> <https://apacheignite.readme.io/docs/collocate-compute-and-data>  ) this
> is
> possible only by calling affinityCall(...) or affinityRun(...), which take
> IgniteCallable or IgniteRunnable.
> I'd like to create a ComputeTask (ComputeTaskAdapter or
> ComputeTaskSplitAdapter), which would spawn ComputeJob with affinity key
> (let's say in constructor) and execute them on node with co-located data.
>
> So is this possible to do such somehow? I couldn't find for now how it can
> be done in elegant way...
>
> Thank you in advance.
>
>
>
> --
> Sent from: http://apache-ignite-users.70518.x6.nabble.com/
>

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hi @ezhuravlev

Thank you for your reply, very appreciated!

I can confirm that by adding real business logic to Jobs it's actually
scales horizontally quite well and by adding more nodes the whole task
finishes just faster.

One more think, which I'm looking now on is running tasks with help of
MapReduce API in collocated fashion. As far as I understood from
documentation ( Collocate Computing and Data
<https://apacheignite.readme.io/docs/collocate-compute-and-data>  ) this is
possible only by calling affinityCall(...) or affinityRun(...), which take
IgniteCallable or IgniteRunnable.
I'd like to create a ComputeTask (ComputeTaskAdapter or
ComputeTaskSplitAdapter), which would spawn ComputeJob with affinity key
(let's say in constructor) and execute them on node with co-located data.

So is this possible to do such somehow? I couldn't find for now how it can
be done in elegant way...

Thank you in advance.



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[ezhuravlev <e....@gmail.com>]

Hi,

I've added Thread.sleep(200) to Jobs to simulate a small load.
Here is what I've got: 
1node: 1 Task 2000 Jobs ~25 sec
2nodes(on the same machine): 1 Task 2000 Jobs ~13 sec

What I want to say here - this overhead will be not noticeable on real Jobs.

What about some configuration changes - you could change
igniteConfiguration.setPublicThreadPoolSize(). By default, it set to CPU
count(with hyperthreading), but for small tasks you can make it bigger, for
example, CPU count*2.

Also, Ignite contains module benchmarks - you can check it and find
benchmarks for ComputeGrid.

All the best,
Evgenii



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

hello

So here are results for NoOpTaks + NoOpJob on two different hosts (hardware
spec. is the same as mentioned above)
1. 1 Task - 100 Jobs -> ~0.1 sec 
2. 1 Task - 1000 Jobs -> ~4 sec 
3. 1 Task - 2000 Jobs -> ~15 sec 
4. 1 Task - 3000 Jobs -> ~36 sec 
5. 1 Task - 5000 Jobs -> ~96 sec 




--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

ezhuravlev wrote
> Also, maybe it's better to compare your current solution with Ignite on
> some real tasks? Or at least more approximate to the real use case
> 
> Evgenii

Hi @ezhuravlev
Thank you for your replay!
I'm preparing more "fair" comparison with our custom made solution but it
can't be done in a simple way due to different technical solutions at the
end (I'll try to explain it below). So it requires to migrate some basic
functionality to have relatively objective comparison (hope I'll do this
soon).


ezhuravlev wrote
> I don't really understand, what you've tried to measure here? 
> .....
> Maybe you could describe your case in detail, so we could suggest you a
> better solution?

I'll try to explain what is the goal of such simple measurement. But it will
require additional introduction to the current situation - why and how we
have come there. It won't be short but I'll try to be concise.

[Goal of simple measurement]
Try to evaluate an overhead of job management in a distributed manner.
Literally - what time is spent for job management in a cluster of ignite
computegrid framework. Idea was:
1. having 16 core (hyper-thead, 8 physical cores) machine I could make an
assumption that it can execute in a parallel at least 8x2 jobs (not sure -
just an assmaption)
2. create 1 task with empty jobs and count on that a crucial majority of
time will be spent in task/jobs management itself

As I understood from your feedback it's better to try it on different
physical hosts - I'll check it out tomorrow.

[why we need it]
Short introduction why I do this evaluation (probably not relevant at the
end but let's see).
Skipping all not relevant (yet) info I'll start describing: few years ago
there was a decision made - we need in-memory compute engine. Distributed
one. So we could speed up... yes, SQL calculation. So we got this:
1. Create an abstraction for distributed compute grid
2. Hazelcast was chosen as a basis
3. Such as computation was, by its nature, kind of a map-reduce (calculation
of reduced values on different levels, starting from store --> warehouse -->
country) we chose Hazelcast's MapReduce API
4. We had only few Tasks with few Jobs 
5. Jobs were running few minutes due to slow data load

Everything was ok as long as we got new type of tasks. Jobs in these tasks
where loading little from DB and were more CPU aggressive and amount of such
jobs was increased up to 1000. We revealed that overhead for management one
job in a cluster was > 2-3 sec., which was unacceptable (Hazelcast admitted
that MapReduce framework was buggy and not performant at that point of time)

As a rescue we decided to write our custom solution:
1. Introduce Task abstraction
2. Introduce Job abstraction as a type for sub-tasks parallelism.
3. Keep Task management in a distributed map (transaction, state [fail,
done, executing]). Distributed map is backed by persistence storage in
relational DB
4. Keep Job management in a distributed map (transnational, state [fail,
done, executing], collocated run, job steeling and more). Jobs are kept in
memory only.
5. Job has no return type (like Runnable)
6. More may come here... left out due to further simplification 

After some time I realized that our custom solution becomes really heavy for
maintenance/support and seems to be that we are reinventing the wheel. I've
figured out that Apache Ignite does quite similar stuff and much more
(persistence - this is what we wanted to implement 1,5 year ago). 

So my goal is before switching the project (which is quite big - 24
countries coverage) to see if I not face the same problem with job
management overhead from the MapReduce API in Ignite. And if it works we
would glad to use another very handy features like Off-heap memory, Ignite
Persistence, SQL engine and Data Streaming.

I hope it will help to understand what I'm going to achieve here.

Thx 



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/

Re: Task management - MapReduce & ForkJoin performance penalty

[Evgenii Zhuravlev <e....@gmail.com>]

Also, maybe it's better to compare your current solution with Ignite on
some real tasks? Or at least more approximate to the real use case

Evgenii

Re: Task management - MapReduce & ForkJoin performance penalty

[Evgenii Zhuravlev <e....@gmail.com>]

Hi,

I don't really understand, what you've tried to measure here?

If you run two nodes on the same machine you will have more context
switching of the CPU. In this case, your CPU run internal Ignite Threads
not from only one node, but from 2 nodes. Additionally, when you use more
that one node - some additional overhead for communicating between nodes
will be added.

If you want to benchmark a Compute Grid, I think it's better to run tasks
on nodes that hosted on different physical machines.

Maybe you could describe your case in detail, so we could suggest you a
better solution?

Evgenii



2017-08-30 18:21 GMT+03:00 ihorps <ih...@gmail.com>:

> It was tested on:
> - Windows 7 SP1
> - Intel I7-4700MQ 2.40GHz
> - 16GB RAM
> - SSD
> - java 1.8.0_112
> - Apache Ignite 2.1.0
>
>
>
> --
> Sent from: http://apache-ignite-users.70518.x6.nabble.com/
>

Re: Task management - MapReduce & ForkJoin performance penalty

[ihorps <ih...@gmail.com>]

It was tested on:
- Windows 7 SP1
- Intel I7-4700MQ 2.40GHz
- 16GB RAM
- SSD 
- java 1.8.0_112
- Apache Ignite 2.1.0



--
Sent from: http://apache-ignite-users.70518.x6.nabble.com/