10:25 a.m.
Hey,
I spent the last week working on concrete API proposals for distributed
execution framework. I believe that we are close to finalize the
proposal and your input and feedback is important now! Here are the main
ideas where I think we made progress since we last talked.
Access to multiple caches during task execution
While we have agreed to allow access to multiple caches during task
execution including this logic into task API complicates it greatly. The
compromise I found is to focus all API on to a one specific cache but
allow access to other caches through DistributedTaskContext API. The
focus on one specific cache and its input keys will allows us to
properly CH map task units across Infinispan cluster and will cover most
of the use cases. DistributedTaskContext can also easily be mapped to a
single cache. See DistributedTask and DistributedTaskContext for more
details.
DistributedTask and DistributedCallable
I found it useful to separate task characteristics in general and actual
work/computation details. Therefore the main task characteristics are
specified through DistributedTask API and details of actual task
computation are specified through DistributedCallable API.
DistributedTask specifies coarse task details, the failover policy, the
task splitting policy, cancellation policy and so on while in
DistributedCallable API implementers focus on actual details of a
computation/work unit.
I have updated the original document [1] to reflect API update. You can
see the actual proposal in git here [2] and I have also included the
variation of this approach [3] that separates map and reduce task phases
with separate interfaces and removes DistributedCallable interaface. I
have also kept Trustin's ideas in another proposal [4] since I would
like to include them as well if possible.
Regards,
Vladimir
[1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
[2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
[3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
[4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
3:16 a.m.
On Dec 27, 2010, at 5:25 PM, Vladimir Blagojevic wrote:
Hey,
I spent the last week working on concrete API proposals for distributed
execution framework. I believe that we are close to finalize the
proposal and your input and feedback is important now! Here are the main
ideas where I think we made progress since we last talked.
Access to multiple caches during task execution
While we have agreed to allow access to multiple caches during task
execution including this logic into task API complicates it greatly. The
compromise I found is to focus all API on to a one specific cache but
allow access to other caches through DistributedTaskContext API. The
focus on one specific cache and its input keys will allows us to
properly CH map task units across Infinispan cluster and will cover most
of the use cases. DistributedTaskContext can also easily be mapped to a
single cache. See DistributedTask and DistributedTaskContext for more
details.
Maybe I'm reading this wrong but are you saying that multiple caches cause problem
with mapping of task units to nodes in cluster?
Or are you just doing it not to clutter the API?
I think DistributedTaskContext extending CacheContainer is rather confusing, particularly
when DistributedTaskContext has K,V parameters that generally are associated with Cache
rather than CacheContainer.
Also, why is a context iterable? Iterates the contents of a CacheContainer? extends
generally means that "is something". AFAIK, you'd be able to iterate a Map
or Cache, but not a CacheContainer.
DistributedTask and DistributedCallable
I found it useful to separate task characteristics in general and actual
work/computation details. Therefore the main task characteristics are
specified through DistributedTask API and details of actual task
computation are specified through DistributedCallable API.
DistributedTask specifies coarse task details, the failover policy, the
task splitting policy, cancellation policy and so on while in
DistributedCallable API implementers focus on actual details of a
computation/work unit.
Personally, I think DistributedTask has too many generics (K, V, T, R) and it's hard
to read. IMO, only T and R should only exist. I would also try to stick to Callable
conventions that takes a V.
I don't like to see things like this, reminds me of EJB 2.1 where you were forced to
implement a method to simply get hold of a ctx. There're much nicer ways to do things
like this, if completely necessary (see EJB3) :
@Override
public void mapped(DistributedTaskContext<String, String> ctx) {
this.ctx = ctx;
}
Looking at the example provided, it seems to me that all DistributedTaskContext is used
for is to navigate the Cache contents from a user defined callable, in which case I would
limit its scope.
Finally, what is the aim of the write() methods in DTC? I don't see them in use in the
given example. If they're not meant to be used by the user and are only for internal
fwk use, I would not leak them.
I have updated the original document [1] to reflect API update. You can
see the actual proposal in git here [2] and I have also included the
variation of this approach [3] that separates map and reduce task phases
with separate interfaces and removes DistributedCallable interaface. I
have also kept Trustin's ideas in another proposal [4] since I would
like to include them as well if possible.
Regards,
Vladimir
[1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
[2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
[3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
[4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Galder Zamarreño
Sr. Software Engineer
Infinispan, JBoss Cache
5:37 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-03 6:16 AM, Galder Zamarreño wrote:
Maybe I'm reading this wrong but are you saying that multiple
caches cause problem with mapping of task units to nodes in cluster?
Or are you just doing it not to clutter the API?
Clutter of API. If you did not like K,V,T,R imagine dealing with
multiple cache confusion! It would be horrible.
I think DistributedTaskContext extending CacheContainer is rather
confusing, particularly when DistributedTaskContext has K,V parameters that generally are
associated with Cache rather than CacheContainer.
Yes, true but DistributedTaskContext is primarily geared towards one
cache while providing opportunity to read data from other caches as
well. Hence K,V for the primary cache. Any suggestions how to deal with
this in a more elegant way? Maybe pass DistributedTaskContext and
CacheContainer as separate parameters?
Also, why is a context iterable? Iterates the contents of a
CacheContainer? extends generally means that "is something". AFAIK, you'd be
able to iterate a Map or Cache, but not a CacheContainer.
True.
Personally, I think DistributedTask has too many generics (K, V, T,
R) and it's hard to read. IMO, only T and R should only exist. I would also try to
stick to Callable conventions that takes a V.
I don't like to see things like this, reminds me of EJB 2.1 where you were forced to
implement a method to simply get hold of a ctx. There're much nicer ways to do things
like this, if completely necessary (see EJB3) :
You mean injection? There is a proposal 2 that essentially does this.
@Override
public void mapped(DistributedTaskContext<String, String> ctx) {
this.ctx = ctx;
}
Looking at the example provided, it seems to me that all DistributedTaskContext is used
for is to navigate the Cache contents from a user defined callable, in which case I would
limit its scope.
What do you mean - "limit its scope"?
1:30 p.m.
Why not?
public interface DistributedTaskContext {
<K, V> Cache<K, V> getCache(String cacheName);
Address getExecutionNode();
}
-- Eduardo
..............................................
http://emmartins.blogspot.com
http://redhat.com/solutions/telco
On Mon, Jan 3, 2011 at 11:37 AM, Vladimir Blagojevic
<vblagoje(a)redhat.com> wrote:
On 11-01-03 6:16 AM, Galder Zamarreño wrote:
> Maybe I'm reading this wrong but are you saying that multiple caches cause
problem with mapping of task units to nodes in cluster?
>
> Or are you just doing it not to clutter the API?
Clutter of API. If you did not like K,V,T,R imagine dealing with
multiple cache confusion! It would be horrible.
> I think DistributedTaskContext extending CacheContainer is rather confusing,
particularly when DistributedTaskContext has K,V parameters that generally are associated
with Cache rather than CacheContainer.
Yes, true but DistributedTaskContext is primarily geared towards one
cache while providing opportunity to read data from other caches as
well. Hence K,V for the primary cache. Any suggestions how to deal with
this in a more elegant way? Maybe pass DistributedTaskContext and
CacheContainer as separate parameters?
> Also, why is a context iterable? Iterates the contents of a CacheContainer? extends
generally means that "is something". AFAIK, you'd be able to iterate a Map
or Cache, but not a CacheContainer.
True.
> Personally, I think DistributedTask has too many generics (K, V, T, R) and it's
hard to read. IMO, only T and R should only exist. I would also try to stick to Callable
conventions that takes a V.
>
> I don't like to see things like this, reminds me of EJB 2.1 where you were forced
to implement a method to simply get hold of a ctx. There're much nicer ways to do
things like this, if completely necessary (see EJB3) :
You mean injection? There is a proposal 2 that essentially does this.
> @Override
> public void mapped(DistributedTaskContext<String, String> ctx) {
> this.ctx = ctx;
> }
>
> Looking at the example provided, it seems to me that all DistributedTaskContext is
used for is to navigate the Cache contents from a user defined callable, in which case I
would limit its scope.
What do you mean - "limit its scope"?
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
3:21 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-03 4:30 PM, Eduardo Martins wrote:
Why not?
public interface DistributedTaskContext {
<K, V> Cache<K, V> getCache(String cacheName);
Address getExecutionNode();
}
-- Eduardo
..............................................
http://emmartins.blogspot.com
http://redhat.com/solutions/telco
Yes, that seems to be a good approach. Galder already suggested this
instead of extending CacheContainer. I already corrected the original
proposal.
Cheers,
Vladimir
11:14 a.m.
New subject: Distributed execution framework - API proposal(s)
Hmm - I'm looking at your prop-1 branch, somewhat confusing. Is this the
correct/up-to-date branch?
On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
On 11-01-03 6:16 AM, Galder Zamarreño wrote:
> Maybe I'm reading this wrong but are you saying that multiple caches cause
problem with mapping of task units to nodes in cluster?
>
> Or are you just doing it not to clutter the API?
Clutter of API. If you did not like K,V,T,R imagine dealing with
multiple cache confusion! It would be horrible.
> I think DistributedTaskContext extending CacheContainer is rather confusing,
particularly when DistributedTaskContext has K,V parameters that generally are associated
with Cache rather than CacheContainer.
Yes, true but DistributedTaskContext is primarily geared towards one
cache while providing opportunity to read data from other caches as
well. Hence K,V for the primary cache. Any suggestions how to deal with
this in a more elegant way? Maybe pass DistributedTaskContext and
CacheContainer as separate parameters?
> Also, why is a context iterable? Iterates the contents of a CacheContainer? extends
generally means that "is something". AFAIK, you'd be able to iterate a Map
or Cache, but not a CacheContainer.
True.
> Personally, I think DistributedTask has too many generics (K, V, T, R) and it's
hard to read. IMO, only T and R should only exist. I would also try to stick to Callable
conventions that takes a V.
>
> I don't like to see things like this, reminds me of EJB 2.1 where you were forced
to implement a method to simply get hold of a ctx. There're much nicer ways to do
things like this, if completely necessary (see EJB3) :
You mean injection? There is a proposal 2 that essentially does this.
> @Override
> public void mapped(DistributedTaskContext<String, String> ctx) {
> this.ctx = ctx;
> }
>
> Looking at the example provided, it seems to me that all DistributedTaskContext is
used for is to navigate the Cache contents from a user defined callable, in which case I
would limit its scope.
What do you mean - "limit its scope"?
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
12:47 p.m.
New subject: Distributed execution framework - API proposal(s)
Also, I think we need to be clear about these 2 (map and reduce) functions. Map
doesn't mean "pick node to run task on" in map/reduce speak. Map means
select /transform data for inclusion into a result set. Perhaps it also makes sense to
use smaller/simpler interfaces. I know this breaks away from the F/J API, but I'm
beginning to wonder if there is a proper alignment of purpose here in the first place -
going back on my original plans here. How's this for an alternate API:
Mapper<K, V, T> {
// just "maps" entries on a remote node. Map = filter and transform. Invoked
once for each entry on a remote cache.
// null responses are ignored/filtered out.
T map(K, V);
}
Reducer<T, R> {
// incrementally reduces a transformed entry. Called once for each T produced by the
mapper.
// previously reduced value passed in each time.
R reduce(T, R);
}
Collator<R> {
// Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
add(Address origin, R remote);
// collates all results added so far.
R collate();
}
And the API could do something like
MapReduceContext c = new MapReduceContext(cache);
// 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V pair.
Stores result T for each invocation if T != null.
// 2) For each T, reducer.reduce() is called. Each time, the previous value of R is
passed back in to reduce().
// 3) Final value of R is sent back as a RPC result. For each result, address and R is
passed collator.add()
// 4) Once all remote RPCs have responded, collator.collate() is called, pass result back
to caller.
R r = c.invoke(mapper, reducer, collator);
Variants may include:
Filtering nodes:
// restricts the set of nodes where RPCs are sent, based on the subset of the cluster that
contain one or more of K.
// question: does this mean only K/V pairs that are in K... are passed in to the mapper?
R r = c.invoke(mapper, reducer, collator, K...);
Using futures:
NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
Example: implementing a word count. but only for keys that start with "text" :
Mapper<String, String, Integer> mapper = new Mapper<String, String, Integer>
() {
Integer map(String k, String v) {
return k.startsWith("text") ? v.length() : null;
}
}
Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
Integer reduce(Integer transformed, Integer prevReduced) {return transformed +
prevReduced;}
}
Collator<Integer> collator = Collator<Integer>() {
int collated = 0;
void add(Address origin, Integer result) {collated += result;}
Integer collate() {return collated;}
}
WDYT? :-)
Cheers
Manik
On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
On 11-01-03 6:16 AM, Galder Zamarreño wrote:
> Maybe I'm reading this wrong but are you saying that multiple caches cause
problem with mapping of task units to nodes in cluster?
>
> Or are you just doing it not to clutter the API?
Clutter of API. If you did not like K,V,T,R imagine dealing with
multiple cache confusion! It would be horrible.
> I think DistributedTaskContext extending CacheContainer is rather confusing,
particularly when DistributedTaskContext has K,V parameters that generally are associated
with Cache rather than CacheContainer.
Yes, true but DistributedTaskContext is primarily geared towards one
cache while providing opportunity to read data from other caches as
well. Hence K,V for the primary cache. Any suggestions how to deal with
this in a more elegant way? Maybe pass DistributedTaskContext and
CacheContainer as separate parameters?
> Also, why is a context iterable? Iterates the contents of a CacheContainer? extends
generally means that "is something". AFAIK, you'd be able to iterate a Map
or Cache, but not a CacheContainer.
True.
> Personally, I think DistributedTask has too many generics (K, V, T, R) and it's
hard to read. IMO, only T and R should only exist. I would also try to stick to Callable
conventions that takes a V.
>
> I don't like to see things like this, reminds me of EJB 2.1 where you were forced
to implement a method to simply get hold of a ctx. There're much nicer ways to do
things like this, if completely necessary (see EJB3) :
You mean injection? There is a proposal 2 that essentially does this.
> @Override
> public void mapped(DistributedTaskContext<String, String> ctx) {
> this.ctx = ctx;
> }
>
> Looking at the example provided, it seems to me that all DistributedTaskContext is
used for is to navigate the Cache contents from a user defined callable, in which case I
would limit its scope.
What do you mean - "limit its scope"?
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
5:20 p.m.
As a data-oriented API, why should I deal with details such as Address?
Could we avoid that?
Also I didn't see a Collator in other examples; I've never used M/R seriusly
so I might have forgotten the more complex examples, but my impression was
that problems should be entirely expressed in Mapper and Reducer only.
Sanne
On 4 Jan 2011 18:47, "Manik Surtani" <manik(a)jboss.org> wrote:
Also, I think we need to be clear about these 2 (map and reduce) functions.
Map doesn't mean "pick node to run task on" in map/reduce speak. Map
means
select /transform data for inclusion into a result set. Perhaps it also
makes sense to use smaller/simpler interfaces. I know this breaks away from
the F/J API, but I'm beginning to wonder if there is a proper alignment of
purpose here in the first place - going back on my original plans here.
How's this for an alternate API:
Mapper<K, V, T> {
// just "maps" entries on a remote node. Map = filter and transform.
Invoked once for each entry on a remote cache.
// null responses are ignored/filtered out.
T map(K, V);
}
Reducer<T, R> {
// incrementally reduces a transformed entry. Called once for each T
produced by the mapper.
// previously reduced value passed in each time.
R reduce(T, R);
}
Collator<R> {
// Adds reduced results from remote nodes. Called once for each R
returned by a RemoteReducer.
add(Address origin, R remote);
// collates all results added so far.
R collate();
}
And the API could do something like
MapReduceContext c = new MapReduceContext(cache);
// 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V
pair. Stores result T for each invocation if T != null.
// 2) For each T, reducer.reduce() is called. Each time, the previous value
of R is passed back in to reduce().
// 3) Final value of R is sent back as a RPC result. For each result,
address and R is passed collator.add()
// 4) Once all remote RPCs have responded, collator.collate() is called,
pass result back to caller.
R r = c.invoke(mapper, reducer, collator);
Variants may include:
Filtering nodes:
// restricts the set of nodes where RPCs are sent, based on the subset of
the cluster that contain one or more of K.
// question: does this mean only K/V pairs that are in K... are passed in to
the mapper?
R r = c.invoke(mapper, reducer, collator, K...);
Using futures:
NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
Example: implementing a word count. but only for keys that start with
"text" :
Mapper<String, String, Integer> mapper = new Mapper<String, String, Integer>
() {
Integer map(String k, String v) {
return k.startsWith("text") ? v.length() : null;
}
}
Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
Integer reduce(Integer transformed, Integer prevReduced) {return
transformed + prevReduced;}
}
Collator<Integer> collator = Collator<Integer>() {
int collated = 0;
void add(Address origin, Integer result) {collated += result;}
Integer collate() {return collated;}
}
WDYT? :-)
Cheers
Manik
On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
On 11-01-03 6:16 AM, Galder Zamarreño wrote:
> Maybe I'm reading this wrong but are you saying th...
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.or...
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/...
7 p.m.
New subject: Distributed execution framework - API proposal(s)
The Address is purely additional info if the collator cares to filter out stuff from
certain nodes.
And the whole concept of the collator is just another reduce - couldn't think of a
better name. Often it makes sense to reduce once remotely and once again locally.
Hope this makes sense. :)
Sent from my iPhone
On 4 Jan 2011, at 23:20, Sanne Grinovero <sanne.grinovero(a)gmail.com> wrote:
As a data-oriented API, why should I deal with details such as
Address? Could we avoid that?
Also I didn't see a Collator in other examples; I've never used M/R seriusly so I
might have forgotten the more complex examples, but my impression was that problems should
be entirely expressed in Mapper and Reducer only.
Sanne
> On 4 Jan 2011 18:47, "Manik Surtani" <manik(a)jboss.org> wrote:
>
> Also, I think we need to be clear about these 2 (map and reduce) functions. Map
doesn't mean "pick node to run task on" in map/reduce speak. Map means
select /transform data for inclusion into a result set. Perhaps it also makes sense to
use smaller/simpler interfaces. I know this breaks away from the F/J API, but I'm
beginning to wonder if there is a proper alignment of purpose here in the first place -
going back on my original plans here. How's this for an alternate API:
>
> Mapper<K, V, T> {
> // just "maps" entries on a remote node. Map = filter and
transform. Invoked once for each entry on a remote cache.
> // null responses are ignored/filtered out.
> T map(K, V);
> }
>
> Reducer<T, R> {
> // incrementally reduces a transformed entry. Called once for each T produced
by the mapper.
> // previously reduced value passed in each time.
> R reduce(T, R);
> }
>
> Collator<R> {
> // Adds reduced results from remote nodes. Called once for each R returned by
a RemoteReducer.
> add(Address origin, R remote);
>
> // collates all results added so far.
> R collate();
> }
>
>
> And the API could do something like
>
> MapReduceContext c = new MapReduceContext(cache);
>
> // 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V
pair. Stores result T for each invocation if T != null.
> // 2) For each T, reducer.reduce() is called. Each time, the previous value of R is
passed back in to reduce().
> // 3) Final value of R is sent back as a RPC result. For each result, address and R
is passed collator.add()
> // 4) Once all remote RPCs have responded, collator.collate() is called, pass result
back to caller.
> R r = c.invoke(mapper, reducer, collator);
>
> Variants may include:
>
> Filtering nodes:
> // restricts the set of nodes where RPCs are sent, based on the subset of the cluster
that contain one or more of K.
> // question: does this mean only K/V pairs that are in K... are passed in to the
mapper?
> R r = c.invoke(mapper, reducer, collator, K...);
>
> Using futures:
> NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
>
> Example: implementing a word count. but only for keys that start with
"text" :
>
> Mapper<String, String, Integer> mapper = new Mapper<String, String,
Integer> () {
> Integer map(String k, String v) {
> return k.startsWith("text") ? v.length() : null;
> }
> }
>
> Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
> Integer reduce(Integer transformed, Integer prevReduced) {return transformed +
prevReduced;}
> }
>
> Collator<Integer> collator = Collator<Integer>() {
> int collated = 0;
> void add(Address origin, Integer result) {collated += result;}
> Integer collate() {return collated;}
> }
>
>
> WDYT? :-)
>
> Cheers
> Manik
>
>
>
> On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
>
> > On 11-01-03 6:16 AM, Galder Zamarreño wrote:
> >> Maybe I'm reading this wrong but are you saying th...
>
> --
> Manik Surtani
> manik(a)jboss.org
> twitter.com/maniksurtani
>
> Lead, Infinispan
> http://www.infinispan.or...
>
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/...
>
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
9:46 p.m.
New subject: Distributed execution framework - API proposal(s)
Manik,
Of course we could go this direction as well. This is very similar to
Hadoop approach and close to original map/reduce paradigm. I
intentionally changed this paradigm into a simpler one because I read a
lot of criticism how it is very hard for "regular" developers to adapt
to original map/reduce paradigm. Hence simpler approach of mapping
runnable to execution nodes and collating results - unfortunately I
named them map and reduce as well.
Anyone else has an opinion while I think about this a bit more?
Regards,
Vladimir
On 11-01-04 3:47 PM, Manik Surtani wrote:
Also, I think we need to be clear about these 2 (map and reduce)
functions. Map doesn't mean "pick node to run task on" in map/reduce speak.
Map means select /transform data for inclusion into a result set. Perhaps it also makes
sense to use smaller/simpler interfaces. I know this breaks away from the F/J API, but
I'm beginning to wonder if there is a proper alignment of purpose here in the first
place - going back on my original plans here. How's this for an alternate API:
Mapper<K, V, T> {
// just "maps" entries on a remote node. Map = filter and transform. Invoked
once for each entry on a remote cache.
// null responses are ignored/filtered out.
T map(K, V);
}
Reducer<T, R> {
// incrementally reduces a transformed entry. Called once for each T produced by the
mapper.
// previously reduced value passed in each time.
R reduce(T, R);
}
Collator<R> {
// Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
add(Address origin, R remote);
// collates all results added so far.
R collate();
}
And the API could do something like
MapReduceContext c = new MapReduceContext(cache);
// 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V pair.
Stores result T for each invocation if T != null.
// 2) For each T, reducer.reduce() is called. Each time, the previous value of R is
passed back in to reduce().
// 3) Final value of R is sent back as a RPC result. For each result, address and R is
passed collator.add()
// 4) Once all remote RPCs have responded, collator.collate() is called, pass result back
to caller.
R r = c.invoke(mapper, reducer, collator);
Variants may include:
Filtering nodes:
// restricts the set of nodes where RPCs are sent, based on the subset of the cluster
that contain one or more of K.
// question: does this mean only K/V pairs that are in K... are passed in to the mapper?
R r = c.invoke(mapper, reducer, collator, K...);
Using futures:
NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
Example: implementing a word count. but only for keys that start with "text"
:
Mapper<String, String, Integer> mapper = new Mapper<String, String, Integer>
() {
Integer map(String k, String v) {
return k.startsWith("text") ? v.length() : null;
}
}
Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
Integer reduce(Integer transformed, Integer prevReduced) {return transformed +
prevReduced;}
}
Collator<Integer> collator = Collator<Integer>() {
int collated = 0;
void add(Address origin, Integer result) {collated += result;}
Integer collate() {return collated;}
}
WDYT? :-)
Cheers
Manik
On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
> On 11-01-03 6:16 AM, Galder Zamarreño wrote:
>> Maybe I'm reading this wrong but are you saying that multiple caches cause
problem with mapping of task units to nodes in cluster?
>>
>> Or are you just doing it not to clutter the API?
> Clutter of API. If you did not like K,V,T,R imagine dealing with
> multiple cache confusion! It would be horrible.
>
>> I think DistributedTaskContext extending CacheContainer is rather confusing,
particularly when DistributedTaskContext has K,V parameters that generally are associated
with Cache rather than CacheContainer.
> Yes, true but DistributedTaskContext is primarily geared towards one
> cache while providing opportunity to read data from other caches as
> well. Hence K,V for the primary cache. Any suggestions how to deal with
> this in a more elegant way? Maybe pass DistributedTaskContext and
> CacheContainer as separate parameters?
>
>
>> Also, why is a context iterable? Iterates the contents of a CacheContainer?
extends generally means that "is something". AFAIK, you'd be able to iterate
a Map or Cache, but not a CacheContainer.
> True.
>
>> Personally, I think DistributedTask has too many generics (K, V, T, R) and
it's hard to read. IMO, only T and R should only exist. I would also try to stick to
Callable conventions that takes a V.
>>
>> I don't like to see things like this, reminds me of EJB 2.1 where you were
forced to implement a method to simply get hold of a ctx. There're much nicer ways to
do things like this, if completely necessary (see EJB3) :
> You mean injection? There is a proposal 2 that essentially does this.
>
>> @Override
>> public void mapped(DistributedTaskContext<String, String> ctx) {
>> this.ctx = ctx;
>> }
>>
>> Looking at the example provided, it seems to me that all DistributedTaskContext
is used for is to navigate the Cache contents from a user defined callable, in which case
I would limit its scope.
> What do you mean - "limit its scope"?
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
9:21 a.m.
New subject: Distributed execution framework - API proposal(s)
These two approaches are not mutually exclusive. Here is what we can do.
We can have a very simple "execute this runnable on remote nodes and
collect results" model as an infrastructure for a higher layer
map/reduce you proposed Manik. That way we cover both ends - a simple
execution model on remote nodes for a given input data set and a more
sophisticated, original map/reduce model built on top of it. Users can
choose what fit their needs best. I can definitely see a need arising
for both of these approaches.
WDYT?
On 11-01-05 12:46 AM, Vladimir Blagojevic wrote:
Manik,
Of course we could go this direction as well. This is very similar to
Hadoop approach and close to original map/reduce paradigm. I
intentionally changed this paradigm into a simpler one because I read a
lot of criticism how it is very hard for "regular" developers to adapt
to original map/reduce paradigm. Hence simpler approach of mapping
runnable to execution nodes and collating results - unfortunately I
named them map and reduce as well.
Anyone else has an opinion while I think about this a bit more?
Regards,
Vladimir
On 11-01-04 3:47 PM, Manik Surtani wrote:
> Also, I think we need to be clear about these 2 (map and reduce) functions. Map
doesn't mean "pick node to run task on" in map/reduce speak. Map means
select /transform data for inclusion into a result set. Perhaps it also makes sense to
use smaller/simpler interfaces. I know this breaks away from the F/J API, but I'm
beginning to wonder if there is a proper alignment of purpose here in the first place -
going back on my original plans here. How's this for an alternate API:
>
> Mapper<K, V, T> {
> // just "maps" entries on a remote node. Map = filter and transform.
Invoked once for each entry on a remote cache.
> // null responses are ignored/filtered out.
> T map(K, V);
> }
>
> Reducer<T, R> {
> // incrementally reduces a transformed entry. Called once for each T produced by
the mapper.
> // previously reduced value passed in each time.
> R reduce(T, R);
> }
>
> Collator<R> {
> // Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
> add(Address origin, R remote);
>
> // collates all results added so far.
> R collate();
> }
>
>
> And the API could do something like
>
> MapReduceContext c = new MapReduceContext(cache);
>
> // 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V
pair. Stores result T for each invocation if T != null.
> // 2) For each T, reducer.reduce() is called. Each time, the previous value of R is
passed back in to reduce().
> // 3) Final value of R is sent back as a RPC result. For each result, address and R
is passed collator.add()
> // 4) Once all remote RPCs have responded, collator.collate() is called, pass result
back to caller.
> R r = c.invoke(mapper, reducer, collator);
>
> Variants may include:
>
> Filtering nodes:
> // restricts the set of nodes where RPCs are sent, based on the subset of the cluster
that contain one or more of K.
> // question: does this mean only K/V pairs that are in K... are passed in to the
mapper?
> R r = c.invoke(mapper, reducer, collator, K...);
>
> Using futures:
> NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
>
> Example: implementing a word count. but only for keys that start with
"text" :
>
> Mapper<String, String, Integer> mapper = new Mapper<String, String,
Integer> () {
> Integer map(String k, String v) {
> return k.startsWith("text") ? v.length() : null;
> }
> }
>
> Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
> Integer reduce(Integer transformed, Integer prevReduced) {return transformed +
prevReduced;}
> }
>
> Collator<Integer> collator = Collator<Integer>() {
> int collated = 0;
> void add(Address origin, Integer result) {collated += result;}
> Integer collate() {return collated;}
> }
>
>
> WDYT? :-)
>
> Cheers
> Manik
>
>
>
> On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
>
>> On 11-01-03 6:16 AM, Galder Zamarreño wrote:
>>> Maybe I'm reading this wrong but are you saying that multiple caches
cause problem with mapping of task units to nodes in cluster?
>>>
>>> Or are you just doing it not to clutter the API?
>> Clutter of API. If you did not like K,V,T,R imagine dealing with
>> multiple cache confusion! It would be horrible.
>>
>>> I think DistributedTaskContext extending CacheContainer is rather confusing,
particularly when DistributedTaskContext has K,V parameters that generally are associated
with Cache rather than CacheContainer.
>> Yes, true but DistributedTaskContext is primarily geared towards one
>> cache while providing opportunity to read data from other caches as
>> well. Hence K,V for the primary cache. Any suggestions how to deal with
>> this in a more elegant way? Maybe pass DistributedTaskContext and
>> CacheContainer as separate parameters?
>>
>>
>>> Also, why is a context iterable? Iterates the contents of a CacheContainer?
extends generally means that "is something". AFAIK, you'd be able to iterate
a Map or Cache, but not a CacheContainer.
>> True.
>>
>>> Personally, I think DistributedTask has too many generics (K, V, T, R) and
it's hard to read. IMO, only T and R should only exist. I would also try to stick to
Callable conventions that takes a V.
>>>
>>> I don't like to see things like this, reminds me of EJB 2.1 where you
were forced to implement a method to simply get hold of a ctx. There're much nicer
ways to do things like this, if completely necessary (see EJB3) :
>> You mean injection? There is a proposal 2 that essentially does this.
>>
>>> @Override
>>> public void mapped(DistributedTaskContext<String, String>
ctx) {
>>> this.ctx = ctx;
>>> }
>>>
>>> Looking at the example provided, it seems to me that all
DistributedTaskContext is used for is to navigate the Cache contents from a user defined
callable, in which case I would limit its scope.
>> What do you mean - "limit its scope"?
>>
>> _______________________________________________
>> infinispan-dev mailing list
>> infinispan-dev(a)lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
> --
> Manik Surtani
> manik(a)jboss.org
> twitter.com/maniksurtani
>
> Lead, Infinispan
> http://www.infinispan.org
>
>
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
10:19 a.m.
New subject: Distributed execution framework - API proposal(s)
On 5 Jan 2011, at 15:21, Vladimir Blagojevic wrote:
These two approaches are not mutually exclusive. Here is what we can
do.
We can have a very simple "execute this runnable on remote nodes and
collect results" model as an infrastructure for a higher layer
map/reduce you proposed Manik. That way we cover both ends - a simple
execution model on remote nodes for a given input data set and a more
sophisticated, original map/reduce model built on top of it. Users can
choose what fit their needs best. I can definitely see a need arising
for both of these approaches.
Ok, so something like what I proposed, plus:
DistributedCallable<T> extends Callable<T> () {
setEmbeddedCacheManager(ecm);
setCache(c);
}
DistributedRemoteTask dt = new DistributedRemoteTask(cache, new
DistributedCallable<T>() {... });
List<Future<T>> res = dt.invoke(); // executes on all nodes
List<Future<T>> res = dt.invoke(K...); // executes on all nodes containing
some or more of K
and expect the user to iterate over whatever is needed, performing any mapping,
transforming, reducing, etc in DistributedCallable.call()?
If so, then we need to define the following APIs (along with an example of using each, so
we are clear on how each API works).
M/R API:
Mapper<K, V, T>
Reducer<T, R>
Collator<R>
MapReduceTask<T, R>
DistributedCallable API:
DistributedCallable<K, V, R>
DistributedRemoteTask<R>
right?
MapReduceTask could extend DistributedRemoteTask... (maybe) ... and would need the
following overloaded invocation methods, perhaps:
R invoke(); //invoke on all nodes
R invoke(K...); // invoke on a subset of nodes
R invoke(Address...); // invoke on a subset of nodes
Future<R> invokeAsync(); //invoke on all nodes
Future<R> invokeAsync(K...); // invoke on a subset of nodes
Future<R> invokeAsync(Address...); // invoke on a subset of nodes
Or, perhaps, making use of a more fluent API:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
Future<R> result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collateAsynchronously(collator);
List<R> result = new
DistributedRemoteTask(cache).on(K...).execute(distributedCallable);
List<Future<R>> result = new
DistributedRemoteTask(cache).on(K...).executeAsync(distributedCallable);
WDYT? I think the fluent API makes a lot of sense, and it is very clear/user friendly.
Cheers
Manik
WDYT?
On 11-01-05 12:46 AM, Vladimir Blagojevic wrote:
> Manik,
>
> Of course we could go this direction as well. This is very similar to
> Hadoop approach and close to original map/reduce paradigm. I
> intentionally changed this paradigm into a simpler one because I read a
> lot of criticism how it is very hard for "regular" developers to adapt
> to original map/reduce paradigm. Hence simpler approach of mapping
> runnable to execution nodes and collating results - unfortunately I
> named them map and reduce as well.
>
> Anyone else has an opinion while I think about this a bit more?
>
> Regards,
> Vladimir
>
>
>
>
> On 11-01-04 3:47 PM, Manik Surtani wrote:
>> Also, I think we need to be clear about these 2 (map and reduce) functions. Map
doesn't mean "pick node to run task on" in map/reduce speak. Map means
select /transform data for inclusion into a result set. Perhaps it also makes sense to
use smaller/simpler interfaces. I know this breaks away from the F/J API, but I'm
beginning to wonder if there is a proper alignment of purpose here in the first place -
going back on my original plans here. How's this for an alternate API:
>>
>> Mapper<K, V, T> {
>> // just "maps" entries on a remote node. Map = filter and transform.
Invoked once for each entry on a remote cache.
>> // null responses are ignored/filtered out.
>> T map(K, V);
>> }
>>
>> Reducer<T, R> {
>> // incrementally reduces a transformed entry. Called once for each T produced
by the mapper.
>> // previously reduced value passed in each time.
>> R reduce(T, R);
>> }
>>
>> Collator<R> {
>> // Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
>> add(Address origin, R remote);
>>
>> // collates all results added so far.
>> R collate();
>> }
>>
>>
>> And the API could do something like
>>
>> MapReduceContext c = new MapReduceContext(cache);
>>
>> // 1) distributes 'mapper' cluster wide. Calls mapper.map() for each K/V
pair. Stores result T for each invocation if T != null.
>> // 2) For each T, reducer.reduce() is called. Each time, the previous value of R
is passed back in to reduce().
>> // 3) Final value of R is sent back as a RPC result. For each result, address
and R is passed collator.add()
>> // 4) Once all remote RPCs have responded, collator.collate() is called, pass
result back to caller.
>> R r = c.invoke(mapper, reducer, collator);
>>
>> Variants may include:
>>
>> Filtering nodes:
>> // restricts the set of nodes where RPCs are sent, based on the subset of the
cluster that contain one or more of K.
>> // question: does this mean only K/V pairs that are in K... are passed in to the
mapper?
>> R r = c.invoke(mapper, reducer, collator, K...);
>>
>> Using futures:
>> NotifyingFuture<R> f = c.invokeFuture(mapper, reducer, collator)
>>
>> Example: implementing a word count. but only for keys that start with
"text" :
>>
>> Mapper<String, String, Integer> mapper = new Mapper<String, String,
Integer> () {
>> Integer map(String k, String v) {
>> return k.startsWith("text") ? v.length() : null;
>> }
>> }
>>
>> Reducer<Integer, Integer> reducer = Reducer<Integer, Integer>() {
>> Integer reduce(Integer transformed, Integer prevReduced) {return transformed +
prevReduced;}
>> }
>>
>> Collator<Integer> collator = Collator<Integer>() {
>> int collated = 0;
>> void add(Address origin, Integer result) {collated += result;}
>> Integer collate() {return collated;}
>> }
>>
>>
>> WDYT? :-)
>>
>> Cheers
>> Manik
>>
>>
>>
>> On 3 Jan 2011, at 11:37, Vladimir Blagojevic wrote:
>>
>>> On 11-01-03 6:16 AM, Galder Zamarreño wrote:
>>>> Maybe I'm reading this wrong but are you saying that multiple caches
cause problem with mapping of task units to nodes in cluster?
>>>>
>>>> Or are you just doing it not to clutter the API?
>>> Clutter of API. If you did not like K,V,T,R imagine dealing with
>>> multiple cache confusion! It would be horrible.
>>>
>>>> I think DistributedTaskContext extending CacheContainer is rather
confusing, particularly when DistributedTaskContext has K,V parameters that generally are
associated with Cache rather than CacheContainer.
>>> Yes, true but DistributedTaskContext is primarily geared towards one
>>> cache while providing opportunity to read data from other caches as
>>> well. Hence K,V for the primary cache. Any suggestions how to deal with
>>> this in a more elegant way? Maybe pass DistributedTaskContext and
>>> CacheContainer as separate parameters?
>>>
>>>
>>>> Also, why is a context iterable? Iterates the contents of a
CacheContainer? extends generally means that "is something". AFAIK, you'd be
able to iterate a Map or Cache, but not a CacheContainer.
>>> True.
>>>
>>>> Personally, I think DistributedTask has too many generics (K, V, T, R)
and it's hard to read. IMO, only T and R should only exist. I would also try to stick
to Callable conventions that takes a V.
>>>>
>>>> I don't like to see things like this, reminds me of EJB 2.1 where you
were forced to implement a method to simply get hold of a ctx. There're much nicer
ways to do things like this, if completely necessary (see EJB3) :
>>> You mean injection? There is a proposal 2 that essentially does this.
>>>
>>>> @Override
>>>> public void mapped(DistributedTaskContext<String, String>
ctx) {
>>>> this.ctx = ctx;
>>>> }
>>>>
>>>> Looking at the example provided, it seems to me that all
DistributedTaskContext is used for is to navigate the Cache contents from a user defined
callable, in which case I would limit its scope.
>>> What do you mean - "limit its scope"?
>>>
>>> _______________________________________________
>>> infinispan-dev mailing list
>>> infinispan-dev(a)lists.jboss.org
>>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>> --
>> Manik Surtani
>> manik(a)jboss.org
>> twitter.com/maniksurtani
>>
>> Lead, Infinispan
>> http://www.infinispan.org
>>
>>
>>
>>
>> _______________________________________________
>> infinispan-dev mailing list
>> infinispan-dev(a)lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
11:05 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-05 1:19 PM, Manik Surtani wrote:
Ok, so something like what I proposed, plus:
DistributedCallable<T> extends Callable<T> () {
setEmbeddedCacheManager(ecm);
setCache(c);
}
DistributedRemoteTask dt = new DistributedRemoteTask(cache, new
DistributedCallable<T>() {... });
List<Future<T>> res = dt.invoke(); // executes on all nodes
List<Future<T>> res = dt.invoke(K...); // executes on all nodes containing
some or more of K
and expect the user to iterate over whatever is needed, performing any mapping,
transforming, reducing, etc in DistributedCallable.call()?
Essentially yes. Lets merge it with
https://github.com/vblagoje/infinispan/commit/b6cd39fe56e1cced96926ed09e6...
Forget for a moment about misnamed map/reduce. We give access to users
to choose execution nodes they want or let CH handle it for them by
default. We could do invoke on task that returns futures as you
suggested and/or keep reduce and have invoke return R. Those are just
detail variations of the same concept. Furthermore, all distribution and
execution related policies would be set on DistributedTask level. There
would be one DistributedCallable per Infinispan node and that
DistributedCallable would be prepared for execution with
DistributedTaskContext given to that Callable would be setup to reflect
all K,V pairs residing on that node. Also users have access to other
caches if needed. Finally, see WordCount example to get a feel.
If so, then we need to define the following APIs (along with an
example of using each, so we are clear on how each API works).
M/R API:
Mapper<K, V, T>
Reducer<T, R>
Collator<R>
MapReduceTask<T, R>
DistributedCallable API:
DistributedCallable<K, V, R>
DistributedRemoteTask<R>
right?
MapReduceTask could extend DistributedRemoteTask... (maybe) ... and would need the
following overloaded invocation methods, perhaps:
R invoke(); //invoke on all nodes
R invoke(K...); // invoke on a subset of nodes
R invoke(Address...); // invoke on a subset of nodes
Future<R> invokeAsync(); //invoke on all nodes
Future<R> invokeAsync(K...); // invoke on a subset of nodes
Future<R> invokeAsync(Address...); // invoke on a subset of nodes
Or, perhaps, making use of a more fluent API:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
Future<R> result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collateAsynchronously(collator);
List<R> result = new
DistributedRemoteTask(cache).on(K...).execute(distributedCallable);
List<Future<R>> result = new
DistributedRemoteTask(cache).on(K...).executeAsync(distributedCallable);
Awesome! I love this fluent API proposal!
Excellent! Anyone else?
WDYT? I think the fluent API makes a lot of sense, and it is very
clear/user friendly.
Cheers
Manik
12:39 p.m.
New subject: Distributed execution framework - API proposal(s)
On 5 Jan 2011, at 17:05, Vladimir Blagojevic wrote:
> Or, perhaps, making use of a more fluent API:
>
> R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
>
> Future<R> result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collateAsynchronously(collator);
>
> List<R> result = new
DistributedRemoteTask(cache).on(K...).execute(distributedCallable);
>
> List<Future<R>> result = new
DistributedRemoteTask(cache).on(K...).executeAsync(distributedCallable);
>
Awesome! I love this fluent API proposal!
Cool. Then if this is the case lets stick with the fluent API - I think it would make
life much easier for developers.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
8:51 a.m.
You probably already discussed this, sorry if that's the case, but for
someone new (but interested) at the discussion, why not start with an
ExecutiveService impl that executes in the Infinispan cloud?
-- Eduardo
..............................................
http://emmartins.blogspot.com
http://redhat.com/solutions/telco
On Wed, Jan 5, 2011 at 6:39 PM, Manik Surtani <manik(a)jboss.org> wrote:
On 5 Jan 2011, at 17:05, Vladimir Blagojevic wrote:
Or, perhaps, making use of a more fluent API:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
Future<R> result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collateAsynchronously(collator);
List<R> result = new
DistributedRemoteTask(cache).on(K...).execute(distributedCallable);
List<Future<R>> result = new
DistributedRemoteTask(cache).on(K...).executeAsync(distributedCallable);
Awesome! I love this fluent API proposal!
Cool. Then if this is the case lets stick with the fluent API - I think it
would make life much easier for developers.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
1:28 p.m.
On 11-01-05 3:39 PM, Manik Surtani wrote:
>
>
> Awesome! I love this fluent API proposal!
>
Cool. Then if this is the case lets stick with the fluent API - I
think it would make life much easier for developers.
--
Manik Surtani
Proposal Manik and I talked about is here
https://github.com/vblagoje/infinispan/commit/59b0c3cb78bce70fdcd49d2d36f...
I have slightly changed it to better match map/reduce paradigm described
here http://labs.google.com/papers/mapreduce.html
Therefore, we now have both map/reduce and a more simple "execute this
task unit on remote nodes and return results" proposal.
Users might not like number of type parameters in MapReduceTask. I
wanted to be as precise as possible to begin with and we can discuss how
and if to reduce them.
Example for map/reduce is WordCountExample.java. Have a look.
I put a proposal in distexec package. I wanted to have parent package to
branch from. We need to accommodate simple execution model, map/reduce,
various execution policies and so on. They could not fit all in
mapreduce package and its children. Hence distexec package. If you think
you have better name - speak up.
Regards,
Vladimir
7:28 a.m.
Hi there.
Lets take the 2 API sets (M/R and distributed exec) separately:
M/R
* Why does MapReduceTask extend Set? It probably shouldn't even extend
DistributedRemoteTask, since some methods such as execute(Callable) doesn't make sense
for MapReduceTask. Perhaps a common super-interface? That alone should simplify the
generics...
* Why does Mapper.map() return a Map? [1] Shouldn't it return a single transformation
for a given K and V?
* Reducer.reduce()'s return type should not necessarily be the same type as the mapped
transformation. Also, does the reducer really need the key? Surely just the transformed
version of each K/V pair.
* Similarly, Collator.add() should just need the address and the reduced result from each
node (each node would only produce 1 result!)
DistExec
* Why do you have execute() and executeAsync() with no params? What do these methods do?
* Not sure what DistributedTaskContext is all about. What does read() and write() do? Is
this meant to be a "layer" in front of the cache in question? Why not just pass
in a ref to the cache?
* What does Factory do?
Almost there! :-)
Cheers
Manik
[1] How many times can I say "map" in one sentence? :)
On 10 Jan 2011, at 19:28, Vladimir Blagojevic wrote:
On 11-01-05 3:39 PM, Manik Surtani wrote:
>
>
>>
>>
>> Awesome! I love this fluent API proposal!
>>
>
> Cool. Then if this is the case lets stick with the fluent API - I think it would
make life much easier for developers.
>
> --
> Manik Surtani
>
Proposal Manik and I talked about is here
https://github.com/vblagoje/infinispan/commit/59b0c3cb78bce70fdcd49d2d36f...
I have slightly changed it to better match map/reduce paradigm described here
http://labs.google.com/papers/mapreduce.html
Therefore, we now have both map/reduce and a more simple "execute this task unit on
remote nodes and return results" proposal.
Users might not like number of type parameters in MapReduceTask. I wanted to be as
precise as possible to begin with and we can discuss how and if to reduce them.
Example for map/reduce is WordCountExample.java. Have a look.
I put a proposal in distexec package. I wanted to have parent package to branch from. We
need to accommodate simple execution model, map/reduce, various execution policies and so
on. They could not fit all in mapreduce package and its children. Hence distexec package.
If you think you have better name - speak up.
Regards,
Vladimir
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
8:18 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-11 10:28 AM, Manik Surtani wrote:
Hi there.
Lets take the 2 API sets (M/R and distributed exec) separately:
M/R
* Why does MapReduceTask extend Set? It probably shouldn't even
extend DistributedRemoteTask, since some methods such as
execute(Callable) doesn't make sense for MapReduceTask. Perhaps a
common super-interface? That alone should simplify the generics...
Leftover that I did not catch. I meant Map<K1,K2> instead of
Set<Map.Entry<K2, V2>>; it does not extend Set it just constraints one
of the type parameters :-)
Completely agree on common parent that has the common setters for
various task aspects.
* Why does Mapper.map() return a Map? [1] Shouldn't it return a
single
transformation for a given K and V?
Yes, I thought so too but how would we then match intermediate keys for
reduce phase. Our plumbing has to find all the same
transformed/intermediate keys returned from a map phase and then invoke
the reduce phase.
* Reducer.reduce()'s return type should not necessarily be the
same
type as the mapped transformation. Also, does the reducer really need
the key? Surely just the transformed version of each K/V pair.
Yeah, when we solve the above this should fall into place.
* Similarly, Collator.add() should just need the address and the
reduced result from each node (each node would only produce 1 result!)
DistExec
* Why do you have execute() and executeAsync() with no params? What
do these methods do?
In case user provided Factory for DistributedCallable. We have to handle
cases where simple no parameter constructor for callables is not
sufficient - hence factory. Now that I think about it maybe adding K...
input as a parameter to call function of DistributedCallable makes more
sense, or no? WDYT?
* Not sure what DistributedTaskContext is all about. What does
read()
and write() do? Is this meant to be a "layer" in front of the cache
in question? Why not just pass in a ref to the cache?
True.
* What does Factory do?
Creates DistributedCallables. See above.
Almost there! :-)
Cheers
Manik
[1] How many times can I say "map" in one sentence? :)
Thanks Manik. Almost there :-)
8:37 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11 Jan 2011, at 14:18, Vladimir Blagojevic wrote:
On 11-01-11 10:28 AM, Manik Surtani wrote:
> Hi there.
>
> Lets take the 2 API sets (M/R and distributed exec) separately:
>
> M/R
> * Why does MapReduceTask extend Set? It probably shouldn't even
> extend DistributedRemoteTask, since some methods such as
> execute(Callable) doesn't make sense for MapReduceTask. Perhaps a
> common super-interface? That alone should simplify the generics...
Leftover that I did not catch. I meant Map<K1,K2> instead of
Set<Map.Entry<K2, V2>>; it does not extend Set it just constraints one
of the type parameters :-)
Completely agree on common parent that has the common setters for
various task aspects.
> * Why does Mapper.map() return a Map? [1] Shouldn't it return a single
> transformation for a given K and V?
Yes, I thought so too but how would we then match intermediate keys for
reduce phase. Our plumbing has to find all the same
transformed/intermediate keys returned from a map phase and then invoke
the reduce phase.
What's an intermediate key? :-) For each K, V pair, we call map(K, V) which
transforms V to a T. So all we expect from map() should be T.
> * Reducer.reduce()'s return type should not necessarily be
the same
> type as the mapped transformation. Also, does the reducer really need
> the key? Surely just the transformed version of each K/V pair.
Yeah, when we solve the above this should fall into place.
Yep. So we then just call reduce(T) for each value of T that we get from the mapping
phase (along with the previous, reduced result of R from the previous invocation). Just
maybe, we may also need to mention the key for which the mapping took place. In which
case, reduce(T, R) would look like reduce(K, T, R). But TBH I don't really see the
use of passing in K.
> * Similarly, Collator.add() should just need the address and the
> reduced result from each node (each node would only produce 1 result!)
>
> DistExec
> * Why do you have execute() and executeAsync() with no params? What
> do these methods do?
In case user provided Factory for DistributedCallable. We have to handle
cases where simple no parameter constructor for callables is not
sufficient - hence factory.
Why are we constructing DistributedCallables? Surely the user passes in an instance?
Now that I think about it maybe adding K...
input as a parameter to call function of DistributedCallable makes more
sense, or no? WDYT?
Not sure this is needed either.
List<ResultType> res = new DistributedRemoteTask(cache).on(key1,
key2).execute(myCallableInstance);
> * Not sure what DistributedTaskContext is all about. What does
read()
> and write() do? Is this meant to be a "layer" in front of the cache
> in question? Why not just pass in a ref to the cache?
True.
> * What does Factory do?
Creates DistributedCallables. See above.
Also, isn't the name DistributedRemoteTask redundant? :-)
Cheers
Manik
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
9:02 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-11 11:37 AM, Manik Surtani wrote:
What's an intermediate key? :-) For each K, V pair, we call
map(K, V) which transforms V to a T. So all we expect from map() should be T.
>> * Reducer.reduce()'s return type should not necessarily be the same
>> type as the mapped transformation. Also, does the reducer really need
>> the key? Surely just the transformed version of each K/V pair.
> Yeah, when we solve the above this should fall into place.
Yep. So we then just call reduce(T) for each value of T that we get from the mapping
phase (along with the previous, reduced result of R from the previous invocation). Just
maybe, we may also need to mention the key for which the mapping took place. In which
case, reduce(T, R) would look like reduce(K, T, R). But TBH I don't really see the
use of passing in K.
They way I understood map/reduce is that for every map(k1,v2) we get
transformed result k2,v2. Then we match all k2's that are identical and
invoke reduce(k2, collection(v2)) once for each identical k2. If we
follow your suggestion then number of invocations in map phase is the
same as in reduce phase which does not make sense - that is not map/reduce.
Would you please have a look at chapter 2 of
http://labs.google.com/papers/mapreduce.html
Maybe I completely lost it? :-)
>> * Similarly, Collator.add() should just need the address and
the
>> reduced result from each node (each node would only produce 1 result!)
>>
>> DistExec
>> * Why do you have execute() and executeAsync() with no params? What
>> do these methods do?
> In case user provided Factory for DistributedCallable. We have to handle
> cases where simple no parameter constructor for callables is not
> sufficient - hence factory.
Why are we constructing DistributedCallables? Surely the user passes in an instance?
Ok, so we need to serialize/migrate across JVMs this instance of
DistributedCallable and invoke it on a target Infinispan node, right? If
so, then we need to add K...input as a parameter to call function since
each migrated instance (on execution node) needs to have different input
for invocation.
Not sure this is needed either.
List<ResultType> res = new DistributedRemoteTask(cache).on(key1,
key2).execute(myCallableInstance);
What do you mean?
Also, isn't the name DistributedRemoteTask redundant? :-)
Just DistributedTask should suffice.
Let me know,
Vladimir
9:27 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11 Jan 2011, at 15:02, Vladimir Blagojevic wrote:
On 11-01-11 11:37 AM, Manik Surtani wrote:
> What's an intermediate key? :-) For each K, V pair, we call map(K, V) which
transforms V to a T. So all we expect from map() should be T.
>
>>> * Reducer.reduce()'s return type should not necessarily be the same
>>> type as the mapped transformation. Also, does the reducer really need
>>> the key? Surely just the transformed version of each K/V pair.
>> Yeah, when we solve the above this should fall into place.
> Yep. So we then just call reduce(T) for each value of T that we get from the mapping
phase (along with the previous, reduced result of R from the previous invocation). Just
maybe, we may also need to mention the key for which the mapping took place. In which
case, reduce(T, R) would look like reduce(K, T, R). But TBH I don't really see the
use of passing in K.
>
They way I understood map/reduce is that for every map(k1,v2) we get
transformed result k2,v2. Then we match all k2's that are identical and
invoke reduce(k2, collection(v2)) once for each identical k2. If we
follow your suggestion then number of invocations in map phase is the
same as in reduce phase which does not make sense - that is not map/reduce.
Would you please have a look at chapter 2 of
http://labs.google.com/papers/mapreduce.html
Maybe I completely lost it? :-)
You can do that but, even as the paper suggests, you usually end up with just 1 result.
Essentially it is a case of how simple we want to make the API. I think for a large
number of the use cases we've come across, mapping a K and V to a single T works fine
(this is a simple value mapping you see in many functional languages). The alternative is
to map an entire tuple (e.g., k1, v1 becomes k2, v2) but as Java doesn't support
multiple return values, so then you end up with Mapper.map() returning a tuple
(Map.Entry<K2, V2> in Java) which I find clunky.
Perhaps, if there is adequate demand, we could extend Mapper with TupleMapper, which would
support this.
>>> * Similarly, Collator.add() should just need the address
and the
>>> reduced result from each node (each node would only produce 1 result!)
>>>
>>> DistExec
>>> * Why do you have execute() and executeAsync() with no params? What
>>> do these methods do?
>> In case user provided Factory for DistributedCallable. We have to handle
>> cases where simple no parameter constructor for callables is not
>> sufficient - hence factory.
> Why are we constructing DistributedCallables? Surely the user passes in an
instance?
>
Ok, so we need to serialize/migrate across JVMs this instance of
DistributedCallable and invoke it on a target Infinispan node, right? If
so, then we need to add K...input as a parameter to call function since
each migrated instance (on execution node) needs to have different input
for invocation.
Yeah good point. My bad. :) We'd need to document this appropriately in the
DistributedCallable javadocs.
Also, isn't the name DistributedRemoteTask redundant? :-)
>
Just DistributedTask should suffice.
+1.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
10:16 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-11 12:27 PM, Manik Surtani wrote:
You can do that but, even as the paper suggests, you usually end up
with just 1 result. Essentially it is a case of how simple we want to make the API. I
think for a large number of the use cases we've come across, mapping a K and V to a
single T works fine (this is a simple value mapping you see in many functional languages).
The alternative is to map an entire tuple (e.g., k1, v1 becomes k2, v2) but as Java
doesn't support multiple return values, so then you end up with Mapper.map() returning
a tuple (Map.Entry<K2, V2> in Java) which I find clunky.
Perhaps, if there is adequate demand, we could extend Mapper with TupleMapper, which
would support this.
Ok noted, but I still need a clarification about your proposal:
Reducer<T, R> {
// incrementally reduces a transformed entry. Called once for each T produced by the
mapper.
// previously reduced value passed in each time.
R reduce(T, R);
}
Collator<R> {
// Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
add(Address origin, R remote);
// collates all results added so far.
R collate();
}
So far I understand that we invoke reduce (T) for each returned T map(K,V) invocation.
Now, where is this previously reduced R parameter coming from? Are you invoking reduce
phase serially across a list of all returned results of T map(K,V) passing R from last
invocation of reduce in a new invocation of reduce?
>>>> * Similarly, Collator.add() should just need the
address and the
>>>> reduced result from each node (each node would only produce 1 result!)
>>>>
>>>> DistExec
>>>> * Why do you have execute() and executeAsync() with no params? What
>>>> do these methods do?
>>> In case user provided Factory for DistributedCallable. We have to handle
>>> cases where simple no parameter constructor for callables is not
>>> sufficient - hence factory.
>> Why are we constructing DistributedCallables? Surely the user passes in an
instance?
>>
> Ok, so we need to serialize/migrate across JVMs this instance of
> DistributedCallable and invoke it on a target Infinispan node, right? If
> so, then we need to add K...input as a parameter to call function since
> each migrated instance (on execution node) needs to have different input
> for invocation.
Yeah good point. My bad. :) We'd need to document this appropriately in the
DistributedCallable javadocs.
:-) What is more elegant and accommodating for interface implementers -
passing parameter K to call of DistributedCallable or using factories
for DistributedCallable?
Cheers,
Vladimir
11:42 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11 Jan 2011, at 16:16, Vladimir Blagojevic wrote:
On 11-01-11 12:27 PM, Manik Surtani wrote:
> You can do that but, even as the paper suggests, you usually end up with just 1
result. Essentially it is a case of how simple we want to make the API. I think for a
large number of the use cases we've come across, mapping a K and V to a single T works
fine (this is a simple value mapping you see in many functional languages). The
alternative is to map an entire tuple (e.g., k1, v1 becomes k2, v2) but as Java
doesn't support multiple return values, so then you end up with Mapper.map() returning
a tuple (Map.Entry<K2, V2> in Java) which I find clunky.
>
> Perhaps, if there is adequate demand, we could extend Mapper with TupleMapper, which
would support this.
>
Ok noted, but I still need a clarification about your proposal:
Reducer<T, R> {
// incrementally reduces a transformed entry. Called once for each T produced by the
mapper.
// previously reduced value passed in each time.
R reduce(T, R);
}
Collator<R> {
// Adds reduced results from remote nodes. Called once for each R returned by a
RemoteReducer.
add(Address origin, R remote);
// collates all results added so far.
R collate();
}
So far I understand that we invoke reduce (T) for each returned T map(K,V) invocation.
Now, where is this previously reduced R parameter coming from? Are you invoking reduce
phase serially across a list of all returned results of T map(K,V) passing R from last
invocation of reduce in a new invocation of reduce?
How about this pseudocode:
On each node:
mapped = list()
for entry in cache.entries:
t = mapper.map(entry.key, entry.value)
mapped.add(t)
r = null
for t in mapped:
r = reducer.reduce(t, r)
return r to calling node
On calling node:
reduced_results = invoke map reduce task on all nodes, retrieve map{address: result}
for r in reduced_results.entries:
remote_address = r.key
remote_reduced_result = r.value
collator.add(remote_address, remote_reduced_result)
return collator.collate()
* Similarly, Collator.add() should just need the address and the
>>>>> reduced result from each node (each node would only produce 1
result!)
>>>>>
>>>>> DistExec
>>>>> * Why do you have execute() and executeAsync() with no params? What
>>>>> do these methods do?
>>>> In case user provided Factory for DistributedCallable. We have to handle
>>>> cases where simple no parameter constructor for callables is not
>>>> sufficient - hence factory.
>>> Why are we constructing DistributedCallables? Surely the user passes in an
instance?
>>>
>> Ok, so we need to serialize/migrate across JVMs this instance of
>> DistributedCallable and invoke it on a target Infinispan node, right? If
>> so, then we need to add K...input as a parameter to call function since
>> each migrated instance (on execution node) needs to have different input
>> for invocation.
> Yeah good point. My bad. :) We'd need to document this appropriately in the
DistributedCallable javadocs.
>
:-) What is more elegant and accommodating for interface implementers -
passing parameter K to call of DistributedCallable or using factories
for DistributedCallable?
Hmm. I think both approaches are best, to allow for the most flexibility. So perhaps
overload execute() with the following:
execute(DistributedCallable c); // Javadoc says DistributedCallable must have a no arg
ctor
execute(DistributedCallableFactory cf); // Javadoc says DistributedCallable must have a
no arg ctor
execute(Class<? extends DistributedCallable> cc);
execute(Class<? extends DistributedCallableFactory> cfc);
Also, wrt. names and methods on the API on the Distributed Exec part of things, wdyt about
aligning it with JDK executors (see separate email from me)?
And - not trying to confuse you, promise! - here's some scope for additional work
later on as well: network classloaders to deal with non-homogenous deployments with remote
tasks like these. ;)
Cheers
Manik
Cheers,
Vladimir
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
6:17 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-11 2:42 PM, Manik Surtani wrote:
How about this pseudocode:
On each node:
mapped = list()
for entry in cache.entries:
t = mapper.map(entry.key, entry.value)
mapped.add(t)
r = null
for t in mapped:
r = reducer.reduce(t, r)
return r to calling node
On calling node:
reduced_results = invoke map reduce task on all nodes, retrieve
map{address: result}
for r in reduced_results.entries:
remote_address = r.key
remote_reduced_result = r.value
collator.add(remote_address, remote_reduced_result)
return collator.collate()
Interesting Manik! Where did you come up with this model? The examples I
tried to write using this model (WordCount etc) would work yet your
model is definitely different than the original proposed map/reduce!
Hmm. I think both approaches are best, to allow for the most
flexibility. So perhaps overload execute() with the following:
execute(DistributedCallable c); // Javadoc says DistributedCallable
must have a no arg ctor
execute(DistributedCallableFactory cf); // Javadoc says
DistributedCallable must have a no arg ctor
execute(Class<? extends DistributedCallable> cc);
execute(Class<? extends DistributedCallableFactory> cfc);
Also, wrt. names and methods on the API on the Distributed Exec part
of things, wdyt about aligning it with JDK executors (see separate
email from me)?
Sure no problem!
And - not trying to confuse you, promise! - here's some scope
for
additional work later on as well: network classloaders to deal with
non-homogenous deployments with remote tasks like these. ;)
Cheers
Manik
6:47 a.m.
New subject: Distributed execution framework - API proposal(s)
On 12 Jan 2011, at 12:17, Vladimir Blagojevic wrote:
On 11-01-11 2:42 PM, Manik Surtani wrote:
> How about this pseudocode:
>
> On each node:
> mapped = list()
> for entry in cache.entries:
> t = mapper.map(entry.key, entry.value)
> mapped.add(t)
>
> r = null
> for t in mapped:
> r = reducer.reduce(t, r)
>
> return r to calling node
>
> On calling node:
> reduced_results = invoke map reduce task on all nodes, retrieve
> map{address: result}
> for r in reduced_results.entries:
> remote_address = r.key
> remote_reduced_result = r.value
> collator.add(remote_address, remote_reduced_result)
>
> return collator.collate()
>
Interesting Manik! Where did you come up with this model? The examples I
tried to write using this model (WordCount etc) would work yet your
model is definitely different than the original proposed map/reduce!
I just made it up. :) Keep in mind that the original Google paper was designed to be very
generic in terms of use case as well as implementation.
Cheers
Manik
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
11:42 a.m.
New subject: Distributed execution framework - API proposal(s)
Also, wrt. names and methods on the API on the Distributed Exec part
of things, wdyt about aligning it with JDK executors (see separate
email from me)?
What exactly are you referring to? API like this one:
DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
result = p.invoke(new DistributedTask() { ... });
Do we need an executor service? We invoke tasks directly with execute
and executeAsync while MapReduceTask is invoked with:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
And - not trying to confuse you, promise! - here's some scope
for
additional work later on as well: network classloaders to deal with
non-homogenous deployments with remote tasks like these. ;)
There are tons of other things to finish!
Cheers,
Vladimir
10:40 a.m.
New subject: Distributed execution framework - API proposal(s)
Hey Manik and all,
https://github.com/vblagoje/infinispan/tree/2ce49c49ea2cb8ba8d0e7d3e4829b...
Let me know if you have comments. The plan is to go forward with this
proposal for ALPHA2 tomorrow.
Regards,
Vladimir
On 11-01-12 2:42 PM, Vladimir Blagojevic wrote:
> Also, wrt. names and methods on the API on the Distributed Exec
part
> of things, wdyt about aligning it with JDK executors (see separate
> email from me)?
>
What exactly are you referring to? API like this one:
DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
result = p.invoke(new DistributedTask() { ... });
Do we need an executor service? We invoke tasks directly with execute
and executeAsync while MapReduceTask is invoked with:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
11:55 a.m.
New subject: Distributed execution framework - API proposal(s)
I had a look, still have a few concerns:
* Why do we still have DistributedTaskContext?
* Factory should probably not be genericised. E.g.,
interface DistributedCallableFactory {
DistributedCallable<K, V, T> createDistributedCallable(Cache c,
EmbeddedCacheManager ecm);
}
but then again that will depend on what you think re: my proposal to make this look like
an ExecutorService (see separate email)
The M/R stuff looks good. :)
Cheers
Manik
On 13 Jan 2011, at 16:40, Vladimir Blagojevic wrote:
Hey Manik and all,
https://github.com/vblagoje/infinispan/tree/2ce49c49ea2cb8ba8d0e7d3e4829b...
Let me know if you have comments. The plan is to go forward with this
proposal for ALPHA2 tomorrow.
Regards,
Vladimir
On 11-01-12 2:42 PM, Vladimir Blagojevic wrote:
>> Also, wrt. names and methods on the API on the Distributed Exec part
>> of things, wdyt about aligning it with JDK executors (see separate
>> email from me)?
>>
> What exactly are you referring to? API like this one:
>
> DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
>
> result = p.invoke(new DistributedTask() { ... });
>
>
>
> Do we need an executor service? We invoke tasks directly with execute
> and executeAsync while MapReduceTask is invoked with:
>
> R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
1:42 p.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-13 2:55 PM, Manik Surtani wrote:
I had a look, still have a few concerns:
* Why do we still have DistributedTaskContext?
We don't need it but it is extra layer of indirection and isolation from
API like EmbeddeCacheManager. DistributedTaskContext is a handle to
obtain mapped cache and other caches, a protection against
EmbeddeCacheManager becoming too complex, being removed and/or similar....
* Factory should probably not be genericised. E.g.,
interface DistributedCallableFactory {
DistributedCallable<K, V, T> createDistributedCallable(Cache c,
EmbeddedCacheManager ecm);
}
but then again that will depend on what you think re: my proposal to make this look like
an ExecutorService (see separate email)
The M/R stuff looks good. :)
Ok let me see.
3:47 a.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-13 2:55 PM, Manik Surtani wrote:
I had a look, still have a few concerns:
* Why do we still have DistributedTaskContext?
* Factory should probably not be genericised. E.g.,
interface DistributedCallableFactory {
DistributedCallable<K, V, T> createDistributedCallable(Cache c,
EmbeddedCacheManager ecm);
}
but then again that will depend on what you think re: my proposal to make this look like
an ExecutorService (see separate email)
The M/R stuff looks good. :)
Cheers
Manik
On 13 Jan 2011, at 16:40, Vladimir Blagojevic wrote:
Having DistributedExecutorService does not remove a need for
DistributedTask. I would keep both and have them co-exist in our
framework. Users really need settings for distributed execution policies
(cancellation, fail-over, etc etc) provided by DistributedTask.
But yes there is also a need for simple execute and forget API provided
by ExecutorService. We can have distributed executor that executes
Callable and DistributedCalllable and assumes default distributed
execution policies.
Cheers,
Vladimir
11:50 a.m.
New subject: Distributed execution framework - API proposal(s)
I resent the email about this as I think it wasn't delivered for some reason.
On 12 Jan 2011, at 17:42, Vladimir Blagojevic wrote:
>
> Also, wrt. names and methods on the API on the Distributed Exec part
> of things, wdyt about aligning it with JDK executors (see separate
> email from me)?
>
What exactly are you referring to? API like this one:
DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
result = p.invoke(new DistributedTask() { ... });
Do we need an executor service? We invoke tasks directly with execute
and executeAsync while MapReduceTask is invoked with:
R result = new
MapReduceTask(cache).on(K...).mappedWith(mapper).reducedWith(reducer).collate(collator);
> And - not trying to confuse you, promise! - here's some scope for
> additional work later on as well: network classloaders to deal with
> non-homogenous deployments with remote tasks like these. ;)
>
There are tons of other things to finish!
Cheers,
Vladimir
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
8:45 a.m.
New subject: Distributed execution framework - API proposal(s)
And for the distributed executor API, how's this as an alternative (suggested by
Eduardo earlier)?
DistributedExecutorService extends ExecutorService [1], adds overloaded methods of
submit(Callable) with submit(Callable, K...). In keeping with the ExecutorService
contract, submit() will ensure the Callable is executed just once, on any one node. We
could add a submitEverywhere(Callable) and submitEverywhere(Callable, K...) to mean
execution on _all_ nodes.
Also ships with a DistributedCallable which extends Callable, adds setCache() and
setEmbeddedCacheManager() setters. If the callable is a DistributedCallable, then these
setters are called before Callable.call(); otherwise just do Callable.call().
WDYT?
[1]
http://download.oracle.com/javase/6/docs/api/java/util/concurrent/Executo...
On 11 Jan 2011, at 14:18, Vladimir Blagojevic wrote:
On 11-01-11 10:28 AM, Manik Surtani wrote:
> Hi there.
>
> Lets take the 2 API sets (M/R and distributed exec) separately:
>
> M/R
> * Why does MapReduceTask extend Set? It probably shouldn't even
> extend DistributedRemoteTask, since some methods such as
> execute(Callable) doesn't make sense for MapReduceTask. Perhaps a
> common super-interface? That alone should simplify the generics...
Leftover that I did not catch. I meant Map<K1,K2> instead of
Set<Map.Entry<K2, V2>>; it does not extend Set it just constraints one
of the type parameters :-)
Completely agree on common parent that has the common setters for
various task aspects.
> * Why does Mapper.map() return a Map? [1] Shouldn't it return a single
> transformation for a given K and V?
Yes, I thought so too but how would we then match intermediate keys for
reduce phase. Our plumbing has to find all the same
transformed/intermediate keys returned from a map phase and then invoke
the reduce phase.
> * Reducer.reduce()'s return type should not necessarily be the same
> type as the mapped transformation. Also, does the reducer really need
> the key? Surely just the transformed version of each K/V pair.
Yeah, when we solve the above this should fall into place.
> * Similarly, Collator.add() should just need the address and the
> reduced result from each node (each node would only produce 1 result!)
>
> DistExec
> * Why do you have execute() and executeAsync() with no params? What
> do these methods do?
In case user provided Factory for DistributedCallable. We have to handle
cases where simple no parameter constructor for callables is not
sufficient - hence factory. Now that I think about it maybe adding K...
input as a parameter to call function of DistributedCallable makes more
sense, or no? WDYT?
> * Not sure what DistributedTaskContext is all about. What does read()
> and write() do? Is this meant to be a "layer" in front of the cache
> in question? Why not just pass in a ref to the cache?
True.
> * What does Factory do?
Creates DistributedCallables. See above.
>
> Almost there! :-)
>
> Cheers
> Manik
>
> [1] How many times can I say "map" in one sentence? :)
>
>
Thanks Manik. Almost there :-)
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
11:49 a.m.
New subject: Fwd: Distributed execution framework - API proposal(s)
Resending, since for some reason I can't find this message in the archives and wonder
if it has been delivered. :/
Vladimir, this is what I meant by a more JDK-like API for distributed executors.
Cheers
Manik
Begin forwarded message:
From: Manik Surtani <manik(a)jboss.org>
Date: 11 January 2011 14:45:11 GMT
To: infinispan -Dev List <infinispan-dev(a)lists.jboss.org>
Subject: Re: [infinispan-dev] Distributed execution framework - API proposal(s)
Reply-To: infinispan -Dev List <infinispan-dev(a)lists.jboss.org>
And for the distributed executor API, how's this as an alternative (suggested by
Eduardo earlier)?
DistributedExecutorService extends ExecutorService [1], adds overloaded methods of
submit(Callable) with submit(Callable, K...). In keeping with the ExecutorService
contract, submit() will ensure the Callable is executed just once, on any one node. We
could add a submitEverywhere(Callable) and submitEverywhere(Callable, K...) to mean
execution on _all_ nodes.
Also ships with a DistributedCallable which extends Callable, adds setCache() and
setEmbeddedCacheManager() setters. If the callable is a DistributedCallable, then these
setters are called before Callable.call(); otherwise just do Callable.call().
WDYT?
[1]
http://download.oracle.com/javase/6/docs/api/java/util/concurrent/Executo...
On 11 Jan 2011, at 14:18, Vladimir Blagojevic wrote:
> On 11-01-11 10:28 AM, Manik Surtani wrote:
>> Hi there.
>>
>> Lets take the 2 API sets (M/R and distributed exec) separately:
>>
>> M/R
>> * Why does MapReduceTask extend Set? It probably shouldn't even
>> extend DistributedRemoteTask, since some methods such as
>> execute(Callable) doesn't make sense for MapReduceTask. Perhaps a
>> common super-interface? That alone should simplify the generics...
>
> Leftover that I did not catch. I meant Map<K1,K2> instead of
> Set<Map.Entry<K2, V2>>; it does not extend Set it just constraints one
> of the type parameters :-)
>
> Completely agree on common parent that has the common setters for
> various task aspects.
>
>> * Why does Mapper.map() return a Map? [1] Shouldn't it return a single
>> transformation for a given K and V?
>
> Yes, I thought so too but how would we then match intermediate keys for
> reduce phase. Our plumbing has to find all the same
> transformed/intermediate keys returned from a map phase and then invoke
> the reduce phase.
>
>> * Reducer.reduce()'s return type should not necessarily be the same
>> type as the mapped transformation. Also, does the reducer really need
>> the key? Surely just the transformed version of each K/V pair.
>
> Yeah, when we solve the above this should fall into place.
>
>> * Similarly, Collator.add() should just need the address and the
>> reduced result from each node (each node would only produce 1 result!)
>>
>> DistExec
>> * Why do you have execute() and executeAsync() with no params? What
>> do these methods do?
>
> In case user provided Factory for DistributedCallable. We have to handle
> cases where simple no parameter constructor for callables is not
> sufficient - hence factory. Now that I think about it maybe adding K...
> input as a parameter to call function of DistributedCallable makes more
> sense, or no? WDYT?
>
>> * Not sure what DistributedTaskContext is all about. What does read()
>> and write() do? Is this meant to be a "layer" in front of the cache
>> in question? Why not just pass in a ref to the cache?
>
> True.
>
>> * What does Factory do?
>
> Creates DistributedCallables. See above.
>
>>
>> Almost there! :-)
>>
>> Cheers
>> Manik
>>
>> [1] How many times can I say "map" in one sentence? :)
>>
>>
> Thanks Manik. Almost there :-)
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
1:48 p.m.
New subject: Distributed execution framework - API proposal(s)
On 11-01-11 11:45 AM, Manik Surtani wrote:
And for the distributed executor API, how's this as an
alternative
(suggested by Eduardo earlier)?
DistributedExecutorService extends ExecutorService [1], adds
overloaded methods of submit(Callable) with submit(Callable, K...).
In keeping with the ExecutorService contract, submit() will ensure
the Callable is executed just once, on any one node. We could add a
submitEverywhere(Callable) and submitEverywhere(Callable, K...) to
mean execution on _all_ nodes.
Also ships with a DistributedCallable which extends Callable, adds
setCache() and setEmbeddedCacheManager() setters. If the callable is
a DistributedCallable, then these setters are called before
Callable.call(); otherwise just do Callable.call().
WDYT?
[1]
http://download.oracle.com/javase/6/docs/api/java/util/concurrent/Executo...
Ok no problem but I'd prefer to have one setter
setDistributedTaskContext that has access to caches and input keys. WDYT?
public interface DistributedTaskContext<K, V> {
public Collection<K> inputKeys();
/**
* Retrieves the default cache associated with this context.
* <p/>
*
* @return the associated cache.
*/
Cache<K, V> getCache();
/**
* Allows access to an arbitrary cache where this
DistributedTaskContext is being provided
*
* @param <T>
* @param <R>
* @param cacheName
* @return
*/
<T, R> Cache<T, R> getCache(String cacheName);
/**
* Returns Infinispan address where this DistributedTaskContext is
being provided
*
* @return
*/
Address getExecutionNode();
}
4:30 a.m.
Something else to add here. Earlier today I was looking at GridGain and the functional way
to define map/reduce functions and seemed quite appealing, particularly cos they had
managed to do it for Java as well:
"Native Java & Scala Support" section in
http://www.gridgain.com/product.html
Having contracted the Scala virus last year, this way of coding functions looks more
appealing (to me) than the way they did in 2.0 which is fairly close to what is currently
being proposed.
Maybe something not for the first version, but something definitely worth keeping in
mind.
On Dec 27, 2010, at 5:25 PM, Vladimir Blagojevic wrote:
Hey,
I spent the last week working on concrete API proposals for distributed
execution framework. I believe that we are close to finalize the
proposal and your input and feedback is important now! Here are the main
ideas where I think we made progress since we last talked.
Access to multiple caches during task execution
While we have agreed to allow access to multiple caches during task
execution including this logic into task API complicates it greatly. The
compromise I found is to focus all API on to a one specific cache but
allow access to other caches through DistributedTaskContext API. The
focus on one specific cache and its input keys will allows us to
properly CH map task units across Infinispan cluster and will cover most
of the use cases. DistributedTaskContext can also easily be mapped to a
single cache. See DistributedTask and DistributedTaskContext for more
details.
DistributedTask and DistributedCallable
I found it useful to separate task characteristics in general and actual
work/computation details. Therefore the main task characteristics are
specified through DistributedTask API and details of actual task
computation are specified through DistributedCallable API.
DistributedTask specifies coarse task details, the failover policy, the
task splitting policy, cancellation policy and so on while in
DistributedCallable API implementers focus on actual details of a
computation/work unit.
I have updated the original document [1] to reflect API update. You can
see the actual proposal in git here [2] and I have also included the
variation of this approach [3] that separates map and reduce task phases
with separate interfaces and removes DistributedCallable interaface. I
have also kept Trustin's ideas in another proposal [4] since I would
like to include them as well if possible.
Regards,
Vladimir
[1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
[2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
[3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
[4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Galder Zamarreño
Sr. Software Engineer
Infinispan, JBoss Cache
4:42 a.m.
On Jan 5, 2011, at 11:30 AM, Galder Zamarreño wrote:
Something else to add here. Earlier today I was looking at GridGain
and the functional way to define map/reduce functions and seemed quite appealing,
particularly cos they had managed to do it for Java as well:
"Native Java & Scala Support" section in
http://www.gridgain.com/product.html
Having contracted the Scala virus last year, this way of coding functions looks more
appealing (to me) than the way they did in 2.0 which is fairly close to what is currently
being proposed.
Maybe something not for the first version, but something definitely worth keeping in
mind.
A few years this might have been a niche, but since Scala and Java 7 (later this year), I
see this way of coding becoming more and more ubiquitous, so maybe we should do this right
from the start? I could certainly give a hand with this part.
On Dec 27, 2010, at 5:25 PM, Vladimir Blagojevic wrote:
> Hey,
>
> I spent the last week working on concrete API proposals for distributed
> execution framework. I believe that we are close to finalize the
> proposal and your input and feedback is important now! Here are the main
> ideas where I think we made progress since we last talked.
>
>
> Access to multiple caches during task execution
>
> While we have agreed to allow access to multiple caches during task
> execution including this logic into task API complicates it greatly. The
> compromise I found is to focus all API on to a one specific cache but
> allow access to other caches through DistributedTaskContext API. The
> focus on one specific cache and its input keys will allows us to
> properly CH map task units across Infinispan cluster and will cover most
> of the use cases. DistributedTaskContext can also easily be mapped to a
> single cache. See DistributedTask and DistributedTaskContext for more
> details.
>
>
> DistributedTask and DistributedCallable
>
> I found it useful to separate task characteristics in general and actual
> work/computation details. Therefore the main task characteristics are
> specified through DistributedTask API and details of actual task
> computation are specified through DistributedCallable API.
> DistributedTask specifies coarse task details, the failover policy, the
> task splitting policy, cancellation policy and so on while in
> DistributedCallable API implementers focus on actual details of a
> computation/work unit.
>
>
> I have updated the original document [1] to reflect API update. You can
> see the actual proposal in git here [2] and I have also included the
> variation of this approach [3] that separates map and reduce task phases
> with separate interfaces and removes DistributedCallable interaface. I
> have also kept Trustin's ideas in another proposal [4] since I would
> like to include them as well if possible.
>
> Regards,
> Vladimir
>
>
> [1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
> [2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
> [3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
> [4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
>
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Galder Zamarreño
Sr. Software Engineer
Infinispan, JBoss Cache
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Galder Zamarreño
Sr. Software Engineer
Infinispan, JBoss Cache
7:11 a.m.
While I agree on the importance of a nice API, I'd polish the Java one first.
This article from May 2008 is already showing off a pretty nice example:
http://www.infoq.com/news/2008/03/fork_join
IMHO even the Java-coded version is very nice, and I don't foresee
people writing hundreds of F/J functions, so clarity is far more
important than brevity and syntax coolness.
I'm not against a nice Scala API, always nice to have, but the Java
one should be made understandable first. (And sorry it might be my
lack of field experience, but I didn't understand how I should code a
simple exercise with the current proposals, while the F/J example is
self-speaking coolness).
Sanne
2011/1/5 Galder Zamarreño <galder(a)redhat.com>:
On Jan 5, 2011, at 11:30 AM, Galder Zamarreño wrote:
> Something else to add here. Earlier today I was looking at GridGain and the
functional way to define map/reduce functions and seemed quite appealing, particularly cos
they had managed to do it for Java as well:
> "Native Java & Scala Support" section in
http://www.gridgain.com/product.html
>
> Having contracted the Scala virus last year, this way of coding functions looks more
appealing (to me) than the way they did in 2.0 which is fairly close to what is currently
being proposed.
>
> Maybe something not for the first version, but something definitely worth keeping in
mind.
A few years this might have been a niche, but since Scala and Java 7 (later this year), I
see this way of coding becoming more and more ubiquitous, so maybe we should do this right
from the start? I could certainly give a hand with this part.
>
> On Dec 27, 2010, at 5:25 PM, Vladimir Blagojevic wrote:
>
>> Hey,
>>
>> I spent the last week working on concrete API proposals for distributed
>> execution framework. I believe that we are close to finalize the
>> proposal and your input and feedback is important now! Here are the main
>> ideas where I think we made progress since we last talked.
>>
>>
>> Access to multiple caches during task execution
>>
>> While we have agreed to allow access to multiple caches during task
>> execution including this logic into task API complicates it greatly. The
>> compromise I found is to focus all API on to a one specific cache but
>> allow access to other caches through DistributedTaskContext API. The
>> focus on one specific cache and its input keys will allows us to
>> properly CH map task units across Infinispan cluster and will cover most
>> of the use cases. DistributedTaskContext can also easily be mapped to a
>> single cache. See DistributedTask and DistributedTaskContext for more
>> details.
>>
>>
>> DistributedTask and DistributedCallable
>>
>> I found it useful to separate task characteristics in general and actual
>> work/computation details. Therefore the main task characteristics are
>> specified through DistributedTask API and details of actual task
>> computation are specified through DistributedCallable API.
>> DistributedTask specifies coarse task details, the failover policy, the
>> task splitting policy, cancellation policy and so on while in
>> DistributedCallable API implementers focus on actual details of a
>> computation/work unit.
>>
>>
>> I have updated the original document [1] to reflect API update. You can
>> see the actual proposal in git here [2] and I have also included the
>> variation of this approach [3] that separates map and reduce task phases
>> with separate interfaces and removes DistributedCallable interaface. I
>> have also kept Trustin's ideas in another proposal [4] since I would
>> like to include them as well if possible.
>>
>> Regards,
>> Vladimir
>>
>>
>> [1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
>> [2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
>> [3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
>> [4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
>>
>>
>>
>> _______________________________________________
>> infinispan-dev mailing list
>> infinispan-dev(a)lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
> --
> Galder Zamarreño
> Sr. Software Engineer
> Infinispan, JBoss Cache
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Galder Zamarreño
Sr. Software Engineer
Infinispan, JBoss Cache
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
9:48 a.m.
I agree with Sanne. A top-level Scala API may be nice, but definitely lower in prio when
compared to a more traditional Java-centric API using Java paradigms. The latter is still
what most folks will use for some time to come.
On 5 Jan 2011, at 13:11, Sanne Grinovero wrote:
While I agree on the importance of a nice API, I'd polish the
Java one first.
This article from May 2008 is already showing off a pretty nice example:
http://www.infoq.com/news/2008/03/fork_join
IMHO even the Java-coded version is very nice, and I don't foresee
people writing hundreds of F/J functions, so clarity is far more
important than brevity and syntax coolness.
I'm not against a nice Scala API, always nice to have, but the Java
one should be made understandable first. (And sorry it might be my
lack of field experience, but I didn't understand how I should code a
simple exercise with the current proposals, while the F/J example is
self-speaking coolness).
Sanne
2011/1/5 Galder Zamarreño <galder(a)redhat.com>:
>
> On Jan 5, 2011, at 11:30 AM, Galder Zamarreño wrote:
>
>> Something else to add here. Earlier today I was looking at GridGain and the
functional way to define map/reduce functions and seemed quite appealing, particularly cos
they had managed to do it for Java as well:
>> "Native Java & Scala Support" section in
http://www.gridgain.com/product.html
>>
>> Having contracted the Scala virus last year, this way of coding functions looks
more appealing (to me) than the way they did in 2.0 which is fairly close to what is
currently being proposed.
>>
>> Maybe something not for the first version, but something definitely worth keeping
in mind.
>
> A few years this might have been a niche, but since Scala and Java 7 (later this
year), I see this way of coding becoming more and more ubiquitous, so maybe we should do
this right from the start? I could certainly give a hand with this part.
>
>>
>> On Dec 27, 2010, at 5:25 PM, Vladimir Blagojevic wrote:
>>
>>> Hey,
>>>
>>> I spent the last week working on concrete API proposals for distributed
>>> execution framework. I believe that we are close to finalize the
>>> proposal and your input and feedback is important now! Here are the main
>>> ideas where I think we made progress since we last talked.
>>>
>>>
>>> Access to multiple caches during task execution
>>>
>>> While we have agreed to allow access to multiple caches during task
>>> execution including this logic into task API complicates it greatly. The
>>> compromise I found is to focus all API on to a one specific cache but
>>> allow access to other caches through DistributedTaskContext API. The
>>> focus on one specific cache and its input keys will allows us to
>>> properly CH map task units across Infinispan cluster and will cover most
>>> of the use cases. DistributedTaskContext can also easily be mapped to a
>>> single cache. See DistributedTask and DistributedTaskContext for more
>>> details.
>>>
>>>
>>> DistributedTask and DistributedCallable
>>>
>>> I found it useful to separate task characteristics in general and actual
>>> work/computation details. Therefore the main task characteristics are
>>> specified through DistributedTask API and details of actual task
>>> computation are specified through DistributedCallable API.
>>> DistributedTask specifies coarse task details, the failover policy, the
>>> task splitting policy, cancellation policy and so on while in
>>> DistributedCallable API implementers focus on actual details of a
>>> computation/work unit.
>>>
>>>
>>> I have updated the original document [1] to reflect API update. You can
>>> see the actual proposal in git here [2] and I have also included the
>>> variation of this approach [3] that separates map and reduce task phases
>>> with separate interfaces and removes DistributedCallable interaface. I
>>> have also kept Trustin's ideas in another proposal [4] since I would
>>> like to include them as well if possible.
>>>
>>> Regards,
>>> Vladimir
>>>
>>>
>>> [1] http://community.jboss.org/wiki/InfinispanDistributedExecutionFramework
>>> [2] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop1
>>> [3] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop2
>>> [4] https://github.com/vblagoje/infinispan/tree/t_ispn-39_master_prop3
>>>
>>>
>>>
>>> _______________________________________________
>>> infinispan-dev mailing list
>>> infinispan-dev(a)lists.jboss.org
>>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>>
>> --
>> Galder Zamarreño
>> Sr. Software Engineer
>> Infinispan, JBoss Cache
>>
>>
>> _______________________________________________
>> infinispan-dev mailing list
>> infinispan-dev(a)lists.jboss.org
>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
> --
> Galder Zamarreño
> Sr. Software Engineer
> Infinispan, JBoss Cache
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
4884
days inactive
4902
days old
infinispan-dev@lists.jboss.org
37 comments
6 participants
participants (6)
-
Eduardo Martins -
Galder Zamarreño -
Manik Surtani -
Manik Surtani -
Sanne Grinovero -
Vladimir Blagojevic