9:45 p.m.
Hey guys,
As some of you might know we have received additional requirements from
community and internally to add a few things to dist.executors and
map/reduce API. On distributed executors front we need to enable
distributed executors to store results into cache directly rather than
returning them to invoker [1]. As soon as we introduce this API we also
need a asyc. mechanism to allow notifications of subtask
completion/failure. I was thinking we add a concept of
DistributedTaskExecutionListener which can be specified in
DistributedTaskBuilder:
DistributedTaskBuilder<T>
executionListener(DistributedTaskExecutionListener<K, T> listener);
We needed DistributedTaskExecutionListener anyway. All distributed tasks
might use some feedback about task progress, completion/failure and on.
My proposal is roughly:
public interface DistributedTaskExecutionListener<K, T> {
void subtaskSent(Address node, Set<K> inputKeys);
void subtaskFailed(Address node, Set<K> inputKeys, Exception e);
void subtaskSucceded(Address node, Set<K> inputKeys, T result);
void allSubtasksCompleted();
}
So much for that. If tasks do not use input keys these parameters would
be emply sets. Now for [1] we need to add additional methods to
DistributedExecutorService. We can not specify result cache in
DistributedTaskBuilder as we are still bound to only submit methods in
DistributedExecutorService that return futures and we don't want that.
We need two new void methods:
<T, K> void submitEverywhere(DistributedTask<T> task,
Cache<DistExecResultKey<K>, T> result);
<T, K > void submitEverywhere(DistributedTask<T> task,
Cache<DistExecResultKey<K>, T> result, K... input);
Now, why bother with DistExecResultKey? Well we have tasks that use
input keys and tasks that don't. So results cache could only be keyed by
either keys or execution address, or combination of those two.
Therefore, DistExecResultKey could be something like:
public interface DistExecResultKey<K> {
Address getExecutionAddress();
K getKey();
}
If you have a better idea how to address this aspect let us know. So
much for distributed executors.
For map/reduce we also have to enable storing of map reduce task results
into cache [2] and allow users to specify custom cache for intermediate
results[3]. Part of task [2] is to allow notification about map/reduce
task progress and completion. Just as in dist.executor I would add
MapReduceTaskExecutionListener interface:
public interface MapReduceTaskExecutionListener {
void mapTaskInitialized(Address executionAddress);
void mapTaskSucceeded(Address executionAddress);
void mapTaskFailed(Address executionTarget, Exception cause);
void mapPhaseCompleted();
void reduceTaskInitialized(Address executionAddress);
void reduceTaskSucceeded(Address executionAddress);
void reduceTaskFailed(Address address, Exception cause);
void reducePhaseCompleted();
}
while MapReduceTask would have an additional method:
public void execute(Cache<KOut, VOut> resultsCache);
MapReduceTaskExecutionListener could be specified using fluent
MapReduceTask API just as intermediate cache would be:
public MapReduceTask<KIn, VIn, KOut, VOut>
usingIntermediateCache(Cache<KOut, List<VOut>> tmpCache);
thus addressing issue [3].
Let me know what you think,
Vladimir
[1] https://issues.jboss.org/browse/ISPN-4030
[2] https://issues.jboss.org/browse/ISPN-4002
[3] https://issues.jboss.org/browse/ISPN-4021
6:57 p.m.
On Feb 19, 2014, at 8:45 PM, Vladimir Blagojevic <vblagoje(a)redhat.com> wrote:
Hey guys,
As some of you might know we have received additional requirements from
community and internally to add a few things to dist.executors and
map/reduce API. On distributed executors front we need to enable
distributed executors to store results into cache directly rather than
returning them to invoker [1]. As soon as we introduce this API we also
need a asyc. mechanism to allow notifications of subtask
completion/failure.
I think we need both in at the same time :-)
I was thinking we add a concept of
DistributedTaskExecutionListener which can be specified in
DistributedTaskBuilder:
DistributedTaskBuilder<T>
executionListener(DistributedTaskExecutionListener<K, T> listener);
We needed DistributedTaskExecutionListener anyway. All distributed tasks
might use some feedback about task progress, completion/failure and on.
My proposal is roughly:
public interface DistributedTaskExecutionListener<K, T> {
void subtaskSent(Address node, Set<K> inputKeys);
void subtaskFailed(Address node, Set<K> inputKeys, Exception e);
void subtaskSucceded(Address node, Set<K> inputKeys, T result);
void allSubtasksCompleted();
}
So much for that.
I think this it would make sense to add this logic for monitoring, + additional info such
as average execution time etc. I'm not sure if this is a generally useful API though,
unless there were people asking for it already?
If tasks do not use input keys these parameters would
be emply sets. Now for [1] we need to add additional methods to
DistributedExecutorService. We can not specify result cache in
DistributedTaskBuilder as we are still bound to only submit methods in
DistributedExecutorService that return futures and we don't want that.
We need two new void methods:
<T, K> void submitEverywhere(DistributedTask<T> task,
Cache<DistExecResultKey<K>, T> result);
<T, K > void submitEverywhere(DistributedTask<T> task,
Cache<DistExecResultKey<K>, T> result, K... input);
Now, why bother with DistExecResultKey? Well we have tasks that use
input keys and tasks that don't. So results cache could only be keyed by
either keys or execution address, or combination of those two.
Therefore, DistExecResultKey could be something like:
public interface DistExecResultKey<K> {
Address getExecutionAddress();
K getKey();
}
If you have a better idea how to address this aspect let us know. So
much for distributed executors.
For map/reduce we also have to enable storing of map reduce task results
into cache [2] and allow users to specify custom cache for intermediate
results[3]. Part of task [2] is to allow notification about map/reduce
task progress and completion. Just as in dist.executor I would add
MapReduceTaskExecutionListener interface:
public interface MapReduceTaskExecutionListener {
void mapTaskInitialized(Address executionAddress);
void mapTaskSucceeded(Address executionAddress);
void mapTaskFailed(Address executionTarget, Exception cause);
void mapPhaseCompleted();
void reduceTaskInitialized(Address executionAddress);
void reduceTaskSucceeded(Address executionAddress);
void reduceTaskFailed(Address address, Exception cause);
void reducePhaseCompleted();
}
IMO - in the first stage at leas - I would rather use a simpler (Notifying)Future, on
which the user can wait till the computation happens: it's simpler and more aligned
with the rest of our async API.
while MapReduceTask would have an additional method:
public void execute(Cache<KOut, VOut> resultsCache);
you could overload it with cache name only method.
MapReduceTaskExecutionListener could be specified using fluent
MapReduceTask API just as intermediate cache would be:
public MapReduceTask<KIn, VIn, KOut, VOut>
usingIntermediateCache(Cache<KOut, List<VOut>> tmpCache);
thus addressing issue [3]
+1
Let me know what you think,
Vladimir
[1] https://issues.jboss.org/browse/ISPN-4030
[2] https://issues.jboss.org/browse/ISPN-4002
[3] https://issues.jboss.org/browse/ISPN-4021
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
Cheers,
--
Mircea Markus
Infinispan lead (www.infinispan.org)
9:55 p.m.
See inline
On 2/24/2014, 12:57 PM, Mircea Markus wrote:
On Feb 19, 2014, at 8:45 PM, Vladimir Blagojevic
<vblagoje(a)redhat.com> wrote:
> Hey guys,
>
> As some of you might know we have received additional requirements from
> community and internally to add a few things to dist.executors and
> map/reduce API. On distributed executors front we need to enable
> distributed executors to store results into cache directly rather than
> returning them to invoker [1]. As soon as we introduce this API we also
> need a asyc. mechanism to allow notifications of subtask
> completion/failure.
I think we need both in at the same time :-)
Yes, that is what I actually meant.
Poor wording.
> I was thinking we add a concept of
> DistributedTaskExecutionListener which can be specified in
> DistributedTaskBuilder:
>
> DistributedTaskBuilder<T>
> executionListener(DistributedTaskExecutionListener<K, T> listener);
>
>
> We needed DistributedTaskExecutionListener anyway. All distributed tasks
> might use some feedback about task progress, completion/failure and on.
> My proposal is roughly:
>
>
> public interface DistributedTaskExecutionListener<K, T> {
>
> void subtaskSent(Address node, Set<K> inputKeys);
> void subtaskFailed(Address node, Set<K> inputKeys, Exception e);
> void subtaskSucceded(Address node, Set<K> inputKeys, T result);
> void allSubtasksCompleted();
>
> }
>
> So much for that.
I think this it would make sense to add this logic for monitoring, + additional info such
as average execution time etc. I'm not sure if this is a generally useful API though,
unless there were people asking for it already?
Ok, noted. If you remember any
references about this let me know and
I'll incorporate what people actually asked for rather than guess.
> If tasks do not use input keys these parameters would
> be emply sets. Now for [1] we need to add additional methods to
> DistributedExecutorService. We can not specify result cache in
> DistributedTaskBuilder as we are still bound to only submit methods in
> DistributedExecutorService that return futures and we don't want that.
> We need two new void methods:
>
> <T, K> void submitEverywhere(DistributedTask<T> task,
> Cache<DistExecResultKey<K>, T> result);
> <T, K > void submitEverywhere(DistributedTask<T> task,
> Cache<DistExecResultKey<K>, T> result, K... input);
>
>
> Now, why bother with DistExecResultKey? Well we have tasks that use
> input keys and tasks that don't. So results cache could only be keyed by
> either keys or execution address, or combination of those two.
> Therefore, DistExecResultKey could be something like:
>
> public interface DistExecResultKey<K> {
>
> Address getExecutionAddress();
> K getKey();
>
> }
>
> If you have a better idea how to address this aspect let us know. So
> much for distributed executors.
>
>
> For map/reduce we also have to enable storing of map reduce task results
> into cache [2] and allow users to specify custom cache for intermediate
> results[3]. Part of task [2] is to allow notification about map/reduce
> task progress and completion. Just as in dist.executor I would add
> MapReduceTaskExecutionListener interface:
>
>
> public interface MapReduceTaskExecutionListener {
>
> void mapTaskInitialized(Address executionAddress);
> void mapTaskSucceeded(Address executionAddress);
> void mapTaskFailed(Address executionTarget, Exception cause);
> void mapPhaseCompleted();
>
> void reduceTaskInitialized(Address executionAddress);
> void reduceTaskSucceeded(Address executionAddress);
> void reduceTaskFailed(Address address, Exception cause);
> void reducePhaseCompleted();
>
> }
IMO - in the first stage at leas - I would rather use a simpler (Notifying)Future, on
which the user can wait till the computation happens: it's simpler and more aligned
with the rest of our async API.
What do you mean? We already have futures in MapReduceTask API. This API
is more fine grained and allows monitoring/reporting of task progress.
Please clarify.
> while MapReduceTask would have an additional method:
>
> public void execute(Cache<KOut, VOut> resultsCache);
you could overload it with cache name only method.
Yeah, good idea. Same for
usingIntermediateCache? I actually asked you
this here https://issues.jboss.org/browse/ISPN-4021
Thanks Mircea!
Vladimir
1:33 p.m.
On Mon, Feb 24, 2014 at 10:55 PM, Vladimir Blagojevic
<vblagoje(a)redhat.com>wrote:
See inline
On 2/24/2014, 12:57 PM, Mircea Markus wrote:
> On Feb 19, 2014, at 8:45 PM, Vladimir Blagojevic <vblagoje(a)redhat.com>
wrote:
>
>> Hey guys,
>>
>> As some of you might know we have received additional requirements from
>> community and internally to add a few things to dist.executors and
>> map/reduce API. On distributed executors front we need to enable
>> distributed executors to store results into cache directly rather than
>> returning them to invoker [1]. As soon as we introduce this API we also
>> need a asyc. mechanism to allow notifications of subtask
>> completion/failure.
> I think we need both in at the same time :-)
Yes, that is what I actually meant. Poor wording.
Do we really need special support for distributed tasks to write results to
another cache? We already allow a task to do
cache.getCacheManager().getCache("outputCache").put(k, v)
>
>> I was thinking we add a concept of
>> DistributedTaskExecutionListener which can be specified in
>> DistributedTaskBuilder:
>>
>> DistributedTaskBuilder<T>
>> executionListener(DistributedTaskExecutionListener<K, T> listener);
>>
>>
>> We needed DistributedTaskExecutionListener anyway. All distributed tasks
>> might use some feedback about task progress, completion/failure and on.
>> My proposal is roughly:
>>
>>
>> public interface DistributedTaskExecutionListener<K, T> {
>>
>> void subtaskSent(Address node, Set<K> inputKeys);
>> void subtaskFailed(Address node, Set<K> inputKeys, Exception e);
>> void subtaskSucceded(Address node, Set<K> inputKeys, T result);
>> void allSubtasksCompleted();
>>
>> }
>>
>> So much for that.
> I think this it would make sense to add this logic for monitoring, +
additional info such as average execution time etc. I'm not sure if this is
a generally useful API though, unless there were people asking for it
already?
Ok, noted. If you remember any references about this let me know and
I'll incorporate what people actually asked for rather than guess.
Ok, let's wait until we get some actual requests from users then. TBH I
don't think distributed tasks with subtasks are something that users care
about. E.g. with Map/Reduce the reduce tasks are not subtasks of the
map/combine tasks, so this API wouldn't help.
Hadoop has a Reporter interface that allows you to report "ticks" and
increment counters, maybe we should add something like that instead?
>
>> If tasks do not use input keys these parameters would
>> be emply sets. Now for [1] we need to add additional methods to
>> DistributedExecutorService. We can not specify result cache in
>> DistributedTaskBuilder as we are still bound to only submit methods in
>> DistributedExecutorService that return futures and we don't want that.
>> We need two new void methods:
>>
>> <T, K> void submitEverywhere(DistributedTask<T> task,
>> Cache<DistExecResultKey<K>, T> result);
>> <T, K > void submitEverywhere(DistributedTask<T> task,
>> Cache<DistExecResultKey<K>, T> result, K... input);
>>
>>
>> Now, why bother with DistExecResultKey? Well we have tasks that use
>> input keys and tasks that don't. So results cache could only be keyed by
>> either keys or execution address, or combination of those two.
>> Therefore, DistExecResultKey could be something like:
>>
>> public interface DistExecResultKey<K> {
>>
>> Address getExecutionAddress();
>> K getKey();
>>
>> }
>>
>> If you have a better idea how to address this aspect let us know. So
>> much for distributed executors.
>>
I think we should allow each distributed task to deal with output in its
own way, the existing API should be enough.
>>
>> For map/reduce we also have to enable storing of map reduce task results
>> into cache [2] and allow users to specify custom cache for intermediate
>> results[3]. Part of task [2] is to allow notification about map/reduce
>> task progress and completion. Just as in dist.executor I would add
>> MapReduceTaskExecutionListener interface:
>>
>>
>> public interface MapReduceTaskExecutionListener {
>>
>> void mapTaskInitialized(Address executionAddress);
>> void mapTaskSucceeded(Address executionAddress);
>> void mapTaskFailed(Address executionTarget, Exception cause);
>> void mapPhaseCompleted();
>>
>> void reduceTaskInitialized(Address executionAddress);
>> void reduceTaskSucceeded(Address executionAddress);
>> void reduceTaskFailed(Address address, Exception cause);
>> void reducePhaseCompleted();
>>
>> }
> IMO - in the first stage at leas - I would rather use a simpler
(Notifying)Future, on which the user can wait till the computation happens:
it's simpler and more aligned with the rest of our async API.
>
What do you mean? We already have futures in MapReduceTask API. This API
is more fine grained and allows monitoring/reporting of task progress.
Please clarify.
I'm not sure about the usefulness of an API like this either... if the
intention is to allow the user to collect statistics about duration of
various phases, then I think exposing the durations via MapReduceTasks
would be better.
>> while MapReduceTask would have an additional method:
>>
>> public void execute(Cache<KOut, VOut> resultsCache);
> you could overload it with cache name only method.
Yeah, good idea. Same for usingIntermediateCache? I actually asked you
this here https://issues.jboss.org/browse/ISPN-4021
+1 to allow a cache name only. For the intermediate cache I don't think it
makes sense to allow a Cache version at all.
4:09 p.m.
On 2/25/2014, 7:33 AM, Dan Berindei wrote:
Do we really need special support for distributed tasks to write
results to another cache? We already allow a task to do
cache.getCacheManager().getCache("outputCache").put(k, v)
Yeah, very good
point Dan. Thanks for being sanity check. Mircea?
>
>> I was thinking we add a concept of
>> DistributedTaskExecutionListener which can be specified in
>> DistributedTaskBuilder:
>>
>> DistributedTaskBuilder<T>
>> executionListener(DistributedTaskExecutionListener<K, T> listener);
>>
>>
>> We needed DistributedTaskExecutionListener anyway. All
distributed tasks
>> might use some feedback about task progress, completion/failure
and on.
>> My proposal is roughly:
>>
>>
>> public interface DistributedTaskExecutionListener<K, T> {
>>
>> void subtaskSent(Address node, Set<K> inputKeys);
>> void subtaskFailed(Address node, Set<K> inputKeys,
Exception e);
>> void subtaskSucceded(Address node, Set<K> inputKeys, T result);
>> void allSubtasksCompleted();
>>
>> }
>>
>> So much for that.
> I think this it would make sense to add this logic for
monitoring, + additional info such as average execution time etc.
I'm not sure if this is a generally useful API though, unless
there were people asking for it already?
Ok, noted. If you remember any references about this let me know and
I'll incorporate what people actually asked for rather than guess.
Ok, let's wait until we get some actual requests from users then. TBH
I don't think distributed tasks with subtasks are something that users
care about. E.g. with Map/Reduce the reduce tasks are not subtasks of
the map/combine tasks, so this API wouldn't help.
Hadoop has a Reporter interface that allows you to report "ticks" and
increment counters, maybe we should add something like that instead?
The subtask I am referring to here is just to denote part of the
distributed task initiated using dist.executors. This interface (maybe
extended a bit with ideas from Reporter) could be used for both
monitoring and more application specific logic about task re-execution
and so on.
I think we should allow each distributed task to deal with output in
its own way, the existing API should be enough.
Yes, I can see your point. Mircea?
>> public interface MapReduceTaskExecutionListener {
>>
>> void mapTaskInitialized(Address executionAddress);
>> void mapTaskSucceeded(Address executionAddress);
>> void mapTaskFailed(Address executionTarget, Exception cause);
>> void mapPhaseCompleted();
>>
>> void reduceTaskInitialized(Address executionAddress);
>> void reduceTaskSucceeded(Address executionAddress);
>> void reduceTaskFailed(Address address, Exception cause);
>> void reducePhaseCompleted();
>>
>> }
> IMO - in the first stage at leas - I would rather use a simpler
(Notifying)Future, on which the user can wait till the computation
happens: it's simpler and more aligned with the rest of our async API.
>
What do you mean? We already have futures in MapReduceTask API.
This API
is more fine grained and allows monitoring/reporting of task progress.
Please clarify.
I'm not sure about the usefulness of an API like this either... if the
intention is to allow the user to collect statistics about duration of
various phases, then I think exposing the durations via MapReduceTasks
would be better.
How would you design that API Dan? Something other than
listener/callback interface?
>> while MapReduceTask would have an additional method:
>>
>> public void execute(Cache<KOut, VOut> resultsCache);
> you could overload it with cache name only method.
Yeah, good idea. Same for usingIntermediateCache? I actually asked you
this here https://issues.jboss.org/browse/ISPN-4021
+1 to allow a cache name only. For the intermediate cache I don't
think it makes sense to allow a Cache version at all.
Ok good. Deal.
Thanks,
Vladimir
5:30 p.m.
On Feb 25, 2014, at 3:09 PM, Vladimir Blagojevic <vblagoje(a)redhat.com> wrote:
On 2/25/2014, 7:33 AM, Dan Berindei wrote:
>
>
> Do we really need special support for distributed tasks to write results to another
cache? We already allow a task to do
>
> cache.getCacheManager().getCache("outputCache").put(k, v)
Yeah, very good point Dan. Thanks for being sanity check. Mircea?
+1
>
>
> >
> >> I was thinking we add a concept of
> >> DistributedTaskExecutionListener which can be specified in
> >> DistributedTaskBuilder:
> >>
> >> DistributedTaskBuilder<T>
> >> executionListener(DistributedTaskExecutionListener<K, T> listener);
> >>
> >>
> >> We needed DistributedTaskExecutionListener anyway. All distributed tasks
> >> might use some feedback about task progress, completion/failure and on.
> >> My proposal is roughly:
> >>
> >>
> >> public interface DistributedTaskExecutionListener<K, T> {
> >>
> >> void subtaskSent(Address node, Set<K> inputKeys);
> >> void subtaskFailed(Address node, Set<K> inputKeys, Exception e);
> >> void subtaskSucceded(Address node, Set<K> inputKeys, T result);
> >> void allSubtasksCompleted();
> >>
> >> }
> >>
> >> So much for that.
> > I think this it would make sense to add this logic for monitoring, + additional
info such as average execution time etc. I'm not sure if this is a generally useful
API though, unless there were people asking for it already?
> Ok, noted. If you remember any references about this let me know and
> I'll incorporate what people actually asked for rather than guess.
>
> Ok, let's wait until we get some actual requests from users then. TBH I don't
think distributed tasks with subtasks are something that users care about. E.g. with
Map/Reduce the reduce tasks are not subtasks of the map/combine tasks, so this API
wouldn't help.
>
> Hadoop has a Reporter interface that allows you to report "ticks" and
increment counters, maybe we should add something like that instead?
The subtask I am referring to here is just to denote part of the distributed task
initiated using dist.executors. This interface (maybe extended a bit with ideas from
Reporter) could be used for both monitoring and more application specific logic about task
re-execution and so on.
>
>
> I think we should allow each distributed task to deal with output in its own way, the
existing API should be enough.
Yes, I can see your point. Mircea?
+1 user driven features
>
>
> >> public interface MapReduceTaskExecutionListener {
> >>
> >> void mapTaskInitialized(Address executionAddress);
> >> void mapTaskSucceeded(Address executionAddress);
> >> void mapTaskFailed(Address executionTarget, Exception cause);
> >> void mapPhaseCompleted();
> >>
> >> void reduceTaskInitialized(Address executionAddress);
> >> void reduceTaskSucceeded(Address executionAddress);
> >> void reduceTaskFailed(Address address, Exception cause);
> >> void reducePhaseCompleted();
> >>
> >> }
> > IMO - in the first stage at leas - I would rather use a simpler
(Notifying)Future, on which the user can wait till the computation happens: it's
simpler and more aligned with the rest of our async API.
> >
> What do you mean? We already have futures in MapReduceTask API. This API
> is more fine grained and allows monitoring/reporting of task progress.
> Please clarify.
ah right, wasn't aware of MapReduceTask.executeAsynchronously() :-) That's what I
was after.
>
> I'm not sure about the usefulness of an API like this either... if the intention
is to allow the user to collect statistics about duration of various phases, then I think
exposing the durations via MapReduceTasks would be better.
How would you design that API Dan? Something other than listener/callback interface?
Functionally, what I was having in mind was JMX stats for the MapReduce tasks in general:
like average execution time, count etc. Also the ability to cancel a running task through
JMX/JON would be nice. I don't think we need to expose this to the user through the
MapReduceTaskExecutionListener above, though.
>
>
> >> while MapReduceTask would have an additional method:
> >>
> >> public void execute(Cache<KOut, VOut> resultsCache);
> > you could overload it with cache name only method.
> Yeah, good idea. Same for usingIntermediateCache? I actually asked you
> this here https://issues.jboss.org/browse/ISPN-4021
>
> +1 to allow a cache name only. For the intermediate cache I don't think it makes
sense to allow a Cache version at all.
Ok good. Deal.
Thanks,
Vladimir
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
Cheers,
--
Mircea Markus
Infinispan lead (www.infinispan.org)
8:31 p.m.
Hey,
I am starting to like this thread more and more :-) In conclusion, for
distributed executors we are not adding any new APIs because Callable
implementers can already write to cache using existing API. We don't
have to add any new elaborate callback/listener API either as users have
not requested but should investigate Hadoop Reporter like interface to
allow users some sense of task current execution phase.
For map/reduce we will add a new method:
public void execute(Cache<KOut, VOut> resultsCache);
Using fluent MapReduceTask API users would be able to specify an
intermediate cache:
public MapReduceTask<KIn, VIn, KOut, VOut> usingIntermediateCache(String
cacheName);
We are not adding MapReduceTaskExecutionListener but more like JMX stats
for the MapReduce tasks in general: like average execution time, count
etc. Also the ability to cancel a running task through JMX/JON would be
nice.
Regards,
Vladimir
9:44 p.m.
On Tue, Feb 25, 2014 at 9:31 PM, Vladimir Blagojevic <vblagoje(a)redhat.com>wrote:
Hey,
I am starting to like this thread more and more :-) In conclusion, for
distributed executors we are not adding any new APIs because Callable
implementers can already write to cache using existing API. We don't
have to add any new elaborate callback/listener API either as users have
not requested but should investigate Hadoop Reporter like interface to
allow users some sense of task current execution phase.
For map/reduce we will add a new method:
public void execute(Cache<KOut, VOut> resultsCache);
Using fluent MapReduceTask API users would be able to specify an
intermediate cache:
public MapReduceTask<KIn, VIn, KOut, VOut> usingIntermediateCache(String
cacheName);
We are not adding MapReduceTaskExecutionListener but more like JMX stats
for the MapReduce tasks in general: like average execution time, count
etc. Also the ability to cancel a running task through JMX/JON would be
nice.
For statistics, I was thinking of adding a getStatistics() method to
MapReduceTask that would return an object with the duration of each phase
and the number of keys processed on each node, after the M/R task is done.
This could probably be extended such that it gives the user in-progress
information as well.
The in-progress information would also tie in nicely with a progress
listener, but I feel the events you proposed are too coarse. If the user
wanted to display a progress bar in his application, and the cluster only
had 2 nodes, the progress bar would hover for half of the time around 0%
and for the other half of the time around 50%. So we'd need to keep
reporting something while a phase is in progress (e.g. by splitting a
node's keys to more than one mapping task, and reporting the end of each
subtask), otherwise the listener wouldn't be of much use.
Anyway, this would be something nice to have, but I don't think it's very
important, so supplying some global statistics via JMX should be enough for
now.
Cheers
Dan
3966
days inactive
3972
days old
infinispan-dev@lists.jboss.org
7 comments
3 participants
participants (3)
-
Dan Berindei -
Mircea Markus -
Vladimir Blagojevic