12:42 p.m.
On 17 Dec 2010, at 11:41, Vladimir Blagojevic wrote:
Even better this way. I would like to hear more about your reasoning
behind using DT on per-cache basis. Yes, it would be simpler and easier API for the users
but we do not cover uses cases when distributed task execution needs access to more than
one cache during its execution....
I was just wondering whether such a use case exists or whether we're just inventing
stuff. :-) It would lead to a much more cumbersome API since you'd need to provide a
map of cache names to keys, etc.
On 10-12-16 9:07 AM, Manik Surtani wrote:
>
>
> Hmm. Maybe it is better to not involve an API on the CacheManager at all. Following
JSR166y [1], we could do:
>
> DistributedForkJoinPool p = DisributedForkJoinPool.newPool(cache); // I still think
it should be on a per-cache basis
>
> DistributedTask<MyResultType, K, V> dt = new DistributedTask<MyResultType,
K, V>() {
>
> public void map(Map.Entry<K, V> entry, Map<K, V> context) {
> // select the entries you are interested in. Transform if needed and store
in context
> }
>
> public MyResultType reduce(Map<Address, Map<K, V>> contexts) {
> // aggregate from context and return value.
> };
>
> };
>
> MyResultType result = p.invoke(dt, key1, key2, key3); // keys are optional.
>
> What I see happening is:
>
> * dt is broadcast to all nodes that hold either of {key1, key2, key3}. If keys are
not provided, broadcast to all.
> * dt.map() is called on each node, for each key specified (if it exists on the local
node).
> * Contexts are sent back to the calling node and are passed to dt.reduce()
> * Result of dt.reduce() passed to the caller of p.invoke()
>
> What do you think?
>
>
> [1] http://gee.cs.oswego.edu/dl/jsr166/dist/jsr166ydocs/index.html
> --
> Manik Surtani
> manik(a)jboss.org
> twitter.com/maniksurtani
>
> Lead, Infinispan
> http://www.infinispan.org
>
>
>
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
12:56 p.m.
New subject: Distributed tasks - specifying task input
On 10-12-17 8:42 AM, Manik Surtani wrote:
I was just wondering whether such a use case exists or whether we're
just inventing stuff. :-) It would lead to a much more cumbersome
API since you'd need to provide a map of cache names to keys, etc.
Ok agreed. Unless there is a way to solve this nicely lets go with cache
based DT. I'll think more about this over the weekend.
Also, why would you call DT#map for each key specified? I am thinking it
should be called once when task is migrated to execution node, context
should be filled with entries on that cache node and passed to map. So
call map once per node with all keys already in context.
I'd also try to use our own interfaces here rather than pure Map so we
can evolve them if we need to - DistributedTaskContext or whatever.....
3:01 p.m.
New subject: Distributed tasks - specifying task input
On 17 Dec 2010, at 11:56, Vladimir Blagojevic wrote:
On 10-12-17 8:42 AM, Manik Surtani wrote:
>
> I was just wondering whether such a use case exists or whether we're
> just inventing stuff. :-) It would lead to a much more cumbersome
> API since you'd need to provide a map of cache names to keys, etc.
Ok agreed. Unless there is a way to solve this nicely lets go with cache
based DT. I'll think more about this over the weekend.
Hmm, just saw Sanne's message. Ok, so if we were to support multiple caches... how
could this look?
interface DistributedTask<K, V, R> {
Collection<String> cacheNames();
Collection<K> keys(String cache);
void map(String cacheName, Map.Entry<K, V> e, DistributedTaskContext <K, V,
R> ctx);
R reduce(Map<Address, DistributedTaskContext<K, V, R>> contexts);
}
interface DistributedTaskContext<K, V, R> {
String getCacheName();
// ... map-like methods on K, V
}
DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
result = p.invoke(new DistributedTask() { ... });
and let the DistributedTask impl provide the names of caches and keys as needed? Some
semantics - a null collection of cache names = all caches, a null collection of keys = all
keys? Or is that a bit shit? :-)
Cheers
Manik
Also, why would you call DT#map for each key specified? I am thinking it
should be called once when task is migrated to execution node, context
should be filled with entries on that cache node and passed to map. So
call map once per node with all keys already in context.
Just saves the implementation having to iterate over the collection. We do the iterating
and callback each time. :-) Minor point.
I'd also try to use our own interfaces here rather than pure Map
so we
can evolve them if we need to - DistributedTaskContext or whatever.....
+1.
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
3:58 p.m.
New subject: Distributed tasks - specifying task input
About>
void map(String cacheName, Map.Entry<K, V> e, DistributedTaskContext
<K, V, R> ctx);
To satisfy a Lucene query it should be able to at least read multiple
caches at the same time.
It would still do map/reduce on a single cache, just I need to be able
to invoke getCache() during both
the map and reduce operations - you might think that would defeat the
purpose, but the situation is that one cache is
distributed and the other replicated, so I won't do remote get
operations as we obviously use the keys of the
distributed one.
So the API seems fine - I hope as I don't have a prototype - if there
is some means to have a reference to the
EmdeddableCacheManager.
Wouldn't it be more usefull to have a reference to the Cache object
instead of it's name.
Finally, I might consider it very useful to be able to declare usage
on keys which don't exist yet, hope that's not a problem?
So in case of Lucene I might know that I'm going to write and rewrite
several times a file called "A",
so I send the operation to the node which is going to host A, though
this key A might exist, or be created
by the task during the process.
Sanne
2010/12/17 Manik Surtani <manik(a)jboss.org>:
On 17 Dec 2010, at 11:56, Vladimir Blagojevic wrote:
> On 10-12-17 8:42 AM, Manik Surtani wrote:
>>
>> I was just wondering whether such a use case exists or whether we're
>> just inventing stuff. :-) It would lead to a much more cumbersome
>> API since you'd need to provide a map of cache names to keys, etc.
>
> Ok agreed. Unless there is a way to solve this nicely lets go with cache
> based DT. I'll think more about this over the weekend.
Hmm, just saw Sanne's message. Ok, so if we were to support multiple caches... how
could this look?
interface DistributedTask<K, V, R> {
Collection<String> cacheNames();
Collection<K> keys(String cache);
void map(String cacheName, Map.Entry<K, V> e, DistributedTaskContext <K, V,
R> ctx);
R reduce(Map<Address, DistributedTaskContext<K, V, R>> contexts);
}
interface DistributedTaskContext<K, V, R> {
String getCacheName();
// ... map-like methods on K, V
}
DistributedForkJoinPool p = cacheManager.newDistributedForkJoinPool();
result = p.invoke(new DistributedTask() { ... });
and let the DistributedTask impl provide the names of caches and keys as needed? Some
semantics - a null collection of cache names = all caches, a null collection of keys = all
keys? Or is that a bit shit? :-)
Cheers
Manik
>
> Also, why would you call DT#map for each key specified? I am thinking it
> should be called once when task is migrated to execution node, context
> should be filled with entries on that cache node and passed to map. So
> call map once per node with all keys already in context.
Just saves the implementation having to iterate over the collection. We do the iterating
and callback each time. :-) Minor point.
> I'd also try to use our own interfaces here rather than pure Map so we
> can evolve them if we need to - DistributedTaskContext or whatever.....
+1.
--
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
4:10 p.m.
New subject: Distributed tasks - specifying task input
On 17 Dec 2010, at 14:58, Sanne Grinovero wrote:
About>
void map(String cacheName, Map.Entry<K, V> e, DistributedTaskContext
<K, V, R> ctx);
To satisfy a Lucene query it should be able to at least read multiple
caches at the same time.
It would still do map/reduce on a single cache, just I need to be able
to invoke getCache() during both
the map and reduce operations - you might think that would defeat the
purpose, but the situation is that one cache is
distributed and the other replicated, so I won't do remote get
operations as we obviously use the keys of the
distributed one.
So the API seems fine - I hope as I don't have a prototype - if there
is some means to have a reference to the
EmdeddableCacheManager.
Wouldn't it be more usefull to have a reference to the Cache object
instead of it's name.
Makes sense. Perhaps we could pass in a Set<Cache> of all caches declared as
required in cacheNames().
Finally, I might consider it very useful to be able to declare usage
on keys which don't exist yet, hope that's not a problem?
So in case of Lucene I might know that I'm going to write and rewrite
several times a file called "A",
so I send the operation to the node which is going to host A, though
this key A might exist, or be created
by the task during the process.
Should be fine, since the CH works on non-existent keys just as well. But then it would
mean that we can't call map() for each entry if the entry doesn't as yet exist.
Perhaps then to satisfy this case, Vladimir's suggestion of calling map() once per
node makes more sense. E.g.,
void map(Set<Cache<K, V>> caches, DistributedTaskContext <K, V, R>
ctx);
Cheers
Manik
--
Manik Surtani
manik(a)jboss.org
twitter.com/maniksurtani
Lead, Infinispan
http://www.infinispan.org
8:50 p.m.
New subject: Distributed tasks - specifying task input
On 10-12-17 12:10 PM, Manik Surtani wrote:
> Finally, I might consider it very useful to be able to declare
usage
> on keys which don't exist yet, hope that's not a problem?
> So in case of Lucene I might know that I'm going to write and rewrite
> several times a file called "A",
> so I send the operation to the node which is going to host A, though
> this key A might exist, or be created
> by the task during the process.
Should be fine, since the CH works on non-existent keys just as well. But then it would
mean that we can't call map() for each entry if the entry doesn't as yet exist.
Perhaps then to satisfy this case, Vladimir's suggestion of calling map() once per
node makes more sense. E.g.,
void map(Set<Cache<K, V>> caches, DistributedTaskContext<K, V, R>
ctx);
Cool. Except I would possibly "stuff" Set<Cache<K,V>> into
DistributedTaskContext since they may be needed for reduce call as well.
In order to use caches in reduce phase users might be tempted to "save"
reference to these caches into a field of DistributedTask
implementation. This would create problems as we migrate instances of DT
across JVMs. Having access to Set<Cache<K,V>> from DistributedTask we
better signal to users the general life cycle and proper use of these
references.
Cheers
9:24 p.m.
New subject: Distributed tasks - specifying task input
On 10-12-17 8:56 AM, Vladimir Blagojevic wrote:
Also, why would you call DT#map for each key specified? I am thinking
it
should be called once when task is migrated to execution node, context
should be filled with entries on that cache node and passed to map. So
call map once per node with all keys already in context.
I might have caused a
confusion with above statements! I believe Sanne
and Manik thought of map as being explicit instruction where to migrate
execution tasks?
4880
days inactive
4883
days old
infinispan-dev@lists.jboss.org
6 comments
3 participants
participants (3)
-
Manik Surtani -
Sanne Grinovero -
Vladimir Blagojevic