6:40 p.m.
Hey guys,
I've been working on a prototype of integrating Infinispan into our app.
We do a lot of distributed processing across a small cluster, so I've
played with Infinispan's existing distributed execution framework (which is
nice), as well as using Infinispan alongside a normal message queue to
distribute tasks. But I've also put together a prototype of a new
distributed execution framework using fork-join pools that you all might be
interested in. If it sounds like something that would be worthwhile for
Infinispan, I can raise a Jira and submit a pull request with what I have
so far. I'd need to get the CA and company policy stuff finalized; that
might take a couple days. Meanwhile, in case there is any interest, I've
described the approach I've taken below.
First, a little background:
A while back I worked on a side project that integrated a distributed
work-stealing algorithm into the standard JDK fork-join queue. It used
JGroups for communication, because it was quick and easy for prototyping.
So this week I thought i'd take a stab at porting that over to Infinispan.
The algorithm I came up with for Infinispan is a bit less of a
work-stealing algorithm, to take advantage of Infinispan's built-in
distribution capabilities, but I think it's still fairly efficient.
My basic approach was to take in a cache in the constructor, much like the
existing distributed executor, and then create a parallel, DIST-mode cache
that uses the same hash & grouping configuration as the original cache.
That new parallel cache is the "task cache", and we use that to distribute
available tasks across the cluster. It's a distributed cache so that tasks
are partitioned across a large cluster, and it uses the hashing config of
the original cache and a KeyAffinityService to attempt to distribute the
tasks to the same nodes that contain the data being worked on. Nodes use
cache listeners to be notified when there is new work available, and the
atomic replace() to "check out" the tasks for execution, and "check
in" the
results.
The basic algorithm is something like this:
For a refresher, a normal FJ pool has a fork() method that takes in a task,
and then places that task on an internal queue (actually, one of several
queues). When threads are idle, they look to the nearest work queue for
work. If that work queue does not have work, they "steal" work from another
thread's queue. So in the best case, tasks remain on the same thread as
the task that spawned them, so tasks that process the same data as their
parents may still have that data in the CPU's cache, etc. There's more to
it than that, but that's the basic idea.
This distributed algorithm just adds an extra layer on top for tasks that
are marked "distributable" (by extending DistributedFJTask instead of the
normal ForkJoinTask). When you call fork() with a DistributedFJTask, it
first checks to see if the local pool's work queue is empty. If so, we just
go ahead and submit it locally; there's no reason to distribute it. If not,
we put the task in the task cache, and let Infinispan distribute it. When a
node has no more work to do in its internal fork-join queues, it looks at
the task cache and tries to pull work from there.
So, it isn't really a "work-stealing" algorithm, per se; the distributable
tasks are being distributed eagerly using Infinispan's normal cache
distribution. But I'm hoping that doing that also makes it easier to handle
node failure, since nodes collectively share a common picture of the work
to be done.
This approach required one change to the actual FJ classes themselves
(in org.infinispan.util.concurrent.jdk8backported).
That's probably the most controversial change. I had to make the original
ForkJoinTask's fork() method non-final in order to extend it cleanly.
There's probably a way around that, but that's the cleanest option I have
thought of thus far.
And lastly, it's not done yet: basic task distribution is working, but I
haven't tackled failover to any real extent yet. The biggest questions,
though, are around what to do with the existing distributed execution
interfaces. For example, DistributedTask has a getCallable() method because
it assumes it's wrapping a Callable. But ForkJoinTasks don't extend
Callable. I could put in a shim to wrap the DistributedFJTasks into
Callables for the sake of that method, but I don't know if it's worth it.
Similarly, the DistributedExecutorService interface exposes a lot of
submit-to-specific-address or submit-to-all-addresses methods, which are an
odd fit here since tasks are distributed via their own cache. Even if I
used a KeyAffinityService to target the task to the given Address, it might
get picked up by another node that shares that same hash. But I can add in
a direct-to-single-Address capability in if that seems worthwhile.
Alternately, I can just use entirely different interfaces
(DistributedFJExecutorService, DistributedFJTask?).
Thoughts? Concerns? Glaring issues?
10:15 a.m.
Hi Matt,
my view is that if you use Infinispan to distribute tasks to be
processed (master-slave processing) to workers, then Infinispan is the
wrong tech for this.
Assuming I understood what you wrote below, you're distributing tasks to
be executed to nodes identified by the consistent hash over the task
data you insert into the cache.
Workers get a notification when tasks are added to their cache, process
them and remove the tasks when done. This is a system where you have 90%
writes (task insertions and removals) and 10% reads (peeking into the
cache to see if tasks are available).
Infinispan was designed to be a cache for frequently used data, in which
reads outnumber writes (perhaps 80%/20% w/r), so with your task
distribution you use Infinispan in a way that's opposite to what it was
designed for. It will cerianly work, but I don't think you'll get great
perf.
I've seen similar abuses before where people used Infinispan/JBossCache
as a notification bus.
For the scenario you describe below I recommend a messaging system like
JGroups or JMS. I've implemented a peer-to-peer task distribution system
[1] in JGroups before, take a look, perhaps this is closer to what you
have in kind...
[1] http://www.jgroups.org/taskdistribution.html
On 3/2/13 6:40 PM, matt hoffman wrote:
Hey guys,
I've been working on a prototype of integrating Infinispan into our
app. We do a lot of distributed processing across a small cluster, so
I've played with Infinispan's existing distributed execution framework
(which is nice), as well as using Infinispan alongside a normal message
queue to distribute tasks. But I've also put together a prototype of a
new distributed execution framework using fork-join pools that you all
might be interested in. If it sounds like something that would be
worthwhile for Infinispan, I can raise a Jira and submit a pull request
with what I have so far. I'd need to get the CA and company policy stuff
finalized; that might take a couple days. Meanwhile, in case there is
any interest, I've described the approach I've taken below.
First, a little background:
A while back I worked on a side project that integrated a distributed
work-stealing algorithm into the standard JDK fork-join queue. It used
JGroups for communication, because it was quick and easy for
prototyping. So this week I thought i'd take a stab at porting that over
to Infinispan. The algorithm I came up with for Infinispan is a bit less
of a work-stealing algorithm, to take advantage of Infinispan's built-in
distribution capabilities, but I think it's still fairly efficient.
My basic approach was to take in a cache in the constructor, much like
the existing distributed executor, and then create a parallel, DIST-mode
cache that uses the same hash & grouping configuration as the original
cache. That new parallel cache is the "task cache", and we use that to
distribute available tasks across the cluster. It's a distributed cache
so that tasks are partitioned across a large cluster, and it uses the
hashing config of the original cache and a KeyAffinityService to attempt
to distribute the tasks to the same nodes that contain the data being
worked on. Nodes use cache listeners to be notified when there is new
work available, and the atomic replace() to "check out" the tasks for
execution, and "check in" the results.
The basic algorithm is something like this:
For a refresher, a normal FJ pool has a fork() method that takes in a
task, and then places that task on an internal queue (actually, one of
several queues). When threads are idle, they look to the nearest work
queue for work. If that work queue does not have work, they "steal" work
from another thread's queue. So in the best case, tasks remain on the
same thread as the task that spawned them, so tasks that process the
same data as their parents may still have that data in the CPU's cache,
etc. There's more to it than that, but that's the basic idea.
This distributed algorithm just adds an extra layer on top for tasks
that are marked "distributable" (by extending DistributedFJTask instead
of the normal ForkJoinTask). When you call fork() with a
DistributedFJTask, it first checks to see if the local pool's work queue
is empty. If so, we just go ahead and submit it locally; there's no
reason to distribute it. If not, we put the task in the task cache, and
let Infinispan distribute it. When a node has no more work to do in its
internal fork-join queues, it looks at the task cache and tries to pull
work from there.
So, it isn't really a "work-stealing" algorithm, per se; the
distributable tasks are being distributed eagerly using Infinispan's
normal cache distribution. But I'm hoping that doing that also makes it
easier to handle node failure, since nodes collectively share a common
picture of the work to be done.
This approach required one change to the actual FJ classes themselves
(in org.infinispan.util.concurrent.jdk8backported). That's probably the
most controversial change. I had to make the original ForkJoinTask's
fork() method non-final in order to extend it cleanly. There's probably
a way around that, but that's the cleanest option I have thought of thus
far.
And lastly, it's not done yet: basic task distribution is working, but I
haven't tackled failover to any real extent yet. The biggest questions,
though, are around what to do with the existing distributed execution
interfaces. For example, DistributedTask has a getCallable() method
because it assumes it's wrapping a Callable. But ForkJoinTasks don't
extend Callable. I could put in a shim to wrap the DistributedFJTasks
into Callables for the sake of that method, but I don't know if it's
worth it. Similarly, the DistributedExecutorService interface exposes a
lot of submit-to-specific-address or submit-to-all-addresses methods,
which are an odd fit here since tasks are distributed via their own
cache. Even if I used a KeyAffinityService to target the task to the
given Address, it might get picked up by another node that shares that
same hash. But I can add in a direct-to-single-Address capability in if
that seems worthwhile. Alternately, I can just use entirely different
interfaces (DistributedFJExecutorService, DistributedFJTask?).
Thoughts? Concerns? Glaring issues?
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Bela Ban, JGroups lead (http://www.jgroups.org)
5:09 a.m.
Bela,
Thanks a lot for the response. I really appreciate your input.
Some comments inline:
On Mar 3, 2013 4:16 AM, "Bela Ban" <bban(a)redhat.com> wrote:
Hi Matt,
my view is that if you use Infinispan to distribute tasks to be
processed (master-slave processing) to workers, then Infinispan is the
wrong tech for this.
Assuming I understood what you wrote below, you're distributing tasks to
be executed to nodes identified by the consistent hash over the task
data you insert into the cache.
Workers get a notification when tasks are added to their cache, process
them and remove the tasks when done. This is a system where you have 90%
writes (task insertions and removals) and 10% reads (peeking into the
cache to see if tasks are available).
Infinispan was designed to be a cache for frequently used data, in which
reads outnumber writes (perhaps 80%/20% w/r), so with your task
distribution you use Infinispan in a way that's opposite to what it was
designed for. It will cerianly work, but I don't think you'll get great
perf.
Your understanding is basically right, except workers don't remove the
tasks when done. But it is certainly write-heavy. Tasks are inserted, read
by workers, updated when a worker starts working on them, updated again
when complete, and read again by the task originator. Perhaps 60/40 w/r,
but still write-heavy.
I've seen similar abuses before where people used
Infinispan/JBossCache
as a notification bus.
For the scenario you describe below I recommend a messaging system like
JGroups or JMS. I've implemented a peer-to-peer task distribution system
[1] in JGroups before, take a look, perhaps this is closer to what you
have in kind...
[1] http://www.jgroups.org/taskdistribution.html
My first version of this distribution system did use JGroups directly (and
drew some inspiration from your paper, actually). I can definitely go back
to a JGroups-based approach if you think it is worthwhile (or a
Infinispan-command-based approach, if there's interest from the Infinispan
community… I'm assuming that using Infinispan's command pattern and
RPCManager would be comparable to vanilla JGroups, performance-wise, but
would be more consistent with Infinispan's architecture).
The differences between this algorithm and the one you propose in your task
distribution paper, for whatever it's worth:
1.) tasks are broadcast to just a subset of nodes, not all. The subset
hopefully corresponds to the subset of nodes that contain the data being
processed, if the task involves data in cache.
2.) The algorithm is pull-based, instead of push-based, so that workers
only "pull" more work when they're idle (or nearly idle). The hope is that
this allows workers to remain more evenly loaded.
3.) the algorithm attempts to keep tasks on the originating node if
possible, for efficiency.
Now, those could all be done with vanilla JGroups with just a couple tweaks
to your design. There are some downsides to this algorithm vs. your design,
too -- it requires one more exchange between master & worker (to "claim" a
task), for example. I could implement something more like your design, with
a push-based distribution scheme that pins tasks to nodes to see what the
quantifiable effect is.
The thing that originally made me want to go with a task cache: in your
paper (and in any design that supports node failure), each node is holding
a view of other tasks in the system. When a new node joins, they receive a
list of tasks. There's a protocol for keeping the task lists synchronized
(multicast add and remove, for example), and if we want to scale either in
# of tasks or # of nodes, we need to partition the tasks so that each node
contains a subset of tasks, while still ensuring that each task is stored
on some number of nodes and that when one of those nodes goes down, that
data is assigned to another node… This all sounds very much like a
distributed cache. It handles distributing data to new peers, making sure
data is stored on X nodes at all times, and ensuring that each node's view
of the data remains in sync. So on the one hand, using an actual cache for
that seems simpler: you're letting the cache do what it does best
(distribute data and keep it consistent amongst peers). It's not just using
it as a notification bus, as you suggest; it's the actual data distribution
I'm interested in.
On the other hand, I think you're arguing that a task cache is doing more
than just distributing the data, and a simpler task broadcast is more
appropriate. I don't know all the things that Infinispan is doing under
the covers to optimize for a read-heavy workload, so I'll have to take your
word for it.
When I find some more time, I can dust off my JGroups-based prototype of
the same algorithm and run some more performance benchmarks. In very
preliminary tests, the task-cache-based design and the JGroups-based
design were very similar in performance. I'll need to run more extensive
tests before I can really quantify the overhead, though. And I'm sure the
JGroups-based design could be made more efficient.
Thanks a lot for the feedback -- I really appreciate it.
Matt
On 3/2/13 6:40 PM, matt hoffman wrote:
> Hey guys,
>
> I've been working on a prototype of integrating Infinispan into our
> app. We do a lot of distributed processing across a small cluster, so
> I've played with Infinispan's existing distributed execution framework
> (which is nice), as well as using Infinispan alongside a normal message
> queue to distribute tasks. But I've also put together a prototype of a
> new distributed execution framework using fork-join pools that you all
> might be interested in. If it sounds like something that would be
> worthwhile for Infinispan, I can raise a Jira and submit a pull request
> with what I have so far. I'd need to get the CA and company policy stuff
> finalized; that might take a couple days. Meanwhile, in case there is
> any interest, I've described the approach I've taken below.
>
> First, a little background:
>
> A while back I worked on a side project that integrated a distributed
> work-stealing algorithm into the standard JDK fork-join queue. It used
> JGroups for communication, because it was quick and easy for
> prototyping. So this week I thought i'd take a stab at porting that over
> to Infinispan. The algorithm I came up with for Infinispan is a bit less
> of a work-stealing algorithm, to take advantage of Infinispan's built-in
> distribution capabilities, but I think it's still fairly efficient.
>
> My basic approach was to take in a cache in the constructor, much like
> the existing distributed executor, and then create a parallel, DIST-mode
> cache that uses the same hash & grouping configuration as the original
> cache. That new parallel cache is the "task cache", and we use that to
> distribute available tasks across the cluster. It's a distributed cache
> so that tasks are partitioned across a large cluster, and it uses the
> hashing config of the original cache and a KeyAffinityService to attempt
> to distribute the tasks to the same nodes that contain the data being
> worked on. Nodes use cache listeners to be notified when there is new
> work available, and the atomic replace() to "check out" the tasks for
> execution, and "check in" the results.
>
> The basic algorithm is something like this:
>
> For a refresher, a normal FJ pool has a fork() method that takes in a
> task, and then places that task on an internal queue (actually, one of
> several queues). When threads are idle, they look to the nearest work
> queue for work. If that work queue does not have work, they "steal" work
> from another thread's queue. So in the best case, tasks remain on the
> same thread as the task that spawned them, so tasks that process the
> same data as their parents may still have that data in the CPU's cache,
> etc. There's more to it than that, but that's the basic idea.
>
> This distributed algorithm just adds an extra layer on top for tasks
> that are marked "distributable" (by extending DistributedFJTask instead
> of the normal ForkJoinTask). When you call fork() with a
> DistributedFJTask, it first checks to see if the local pool's work queue
> is empty. If so, we just go ahead and submit it locally; there's no
> reason to distribute it. If not, we put the task in the task cache, and
> let Infinispan distribute it. When a node has no more work to do in its
> internal fork-join queues, it looks at the task cache and tries to pull
> work from there.
>
> So, it isn't really a "work-stealing" algorithm, per se; the
> distributable tasks are being distributed eagerly using Infinispan's
> normal cache distribution. But I'm hoping that doing that also makes it
> easier to handle node failure, since nodes collectively share a common
> picture of the work to be done.
>
> This approach required one change to the actual FJ classes themselves
> (in org.infinispan.util.concurrent.jdk8backported). That's probably the
> most controversial change. I had to make the original ForkJoinTask's
> fork() method non-final in order to extend it cleanly. There's probably
> a way around that, but that's the cleanest option I have thought of thus
> far.
>
> And lastly, it's not done yet: basic task distribution is working, but I
> haven't tackled failover to any real extent yet. The biggest questions,
> though, are around what to do with the existing distributed execution
> interfaces. For example, DistributedTask has a getCallable() method
> because it assumes it's wrapping a Callable. But ForkJoinTasks don't
> extend Callable. I could put in a shim to wrap the DistributedFJTasks
> into Callables for the sake of that method, but I don't know if it's
> worth it. Similarly, the DistributedExecutorService interface exposes a
> lot of submit-to-specific-address or submit-to-all-addresses methods,
> which are an odd fit here since tasks are distributed via their own
> cache. Even if I used a KeyAffinityService to target the task to the
> given Address, it might get picked up by another node that shares that
> same hash. But I can add in a direct-to-single-Address capability in if
> that seems worthwhile. Alternately, I can just use entirely different
> interfaces (DistributedFJExecutorService, DistributedFJTask?).
>
> Thoughts? Concerns? Glaring issues?
>
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
--
Bela Ban, JGroups lead (http://www.jgroups.org)
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
9:11 a.m.
Hi Matt,
there are subtle differences between map-reduce scenarios and task
distribution: map-reduce works mostly on cached data (lots of reads with
few writes), whereas task distribution produces mostly new data (lots of
writes).
If I misunderstood you, and you're actually doing map-reduce, then using
Infinispan is fine.
Comments inline ...
On 3/4/13 5:09 AM, matt hoffman wrote:
Bela,
Thanks a lot for the response. I really appreciate your input.
Some comments inline:
On Mar 3, 2013 4:16 AM, "Bela Ban" <bban(a)redhat.com
<mailto:bban@redhat.com>> wrote:
>
> Hi Matt,
>
> my view is that if you use Infinispan to distribute tasks to be
> processed (master-slave processing) to workers, then Infinispan is the
> wrong tech for this.
>
> Assuming I understood what you wrote below, you're distributing tasks to
> be executed to nodes identified by the consistent hash over the task
> data you insert into the cache.
>
> Workers get a notification when tasks are added to their cache, process
> them and remove the tasks when done. This is a system where you have 90%
> writes (task insertions and removals) and 10% reads (peeking into the
> cache to see if tasks are available).
>
> Infinispan was designed to be a cache for frequently used data, in which
> reads outnumber writes (perhaps 80%/20% w/r), so with your task
> distribution you use Infinispan in a way that's opposite to what it was
> designed for. It will cerianly work, but I don't think you'll get great
> perf.
>
Your understanding is basically right, except workers don't remove the
tasks when done. But it is certainly write-heavy. Tasks are inserted,
read by workers, updated when a worker starts working on them, updated
again when complete, and read again by the task originator. Perhaps
60/40 w/r, but still write-heavy.
So if we use DIST, we have a write (insert) to N nodes, a local read,
one or more writes (updates) and a final write (done), followed by a
(potentially remote) read by the coordinator.
I'm wondering whether the intermediate writes (updates) are needed, as
they trigger messages to other nodes: if the intermediate steps are
relatively small, then there's no need to back them up to other nodes,
as others could simply restart the work from scratch if a node crashes.
Most of the time, nodes don't crash, so the update-writes might not be
amortized by crashes, unless it's a lot of work to recreate those
intermediate steps.
Anyway, yes, this is write-heavy, and there are a few disadvantages
using DIST:
- Unneeded writes for intermediate work
- DIST reshuffles state on a view change. This means that even if a
given worker doesn't crash, it might have to surrender its work to a
different worker if the view changes in the middle of a task. Actually,
your system can probably deal with this, but even if a worker continues
processing a task during a view change, chances are its writes are now
mostly remote, which is inefficient
- While work stealing is a good thing in general, in Infinispan it leads
to contention on the same locks, ie. the original worker and the worker
which stole the task.
> I've seen similar abuses before where people used
Infinispan/JBossCache
> as a notification bus.
>
> For the scenario you describe below I recommend a messaging system like
> JGroups or JMS. I've implemented a peer-to-peer task distribution system
> [1] in JGroups before, take a look, perhaps this is closer to what you
> have in kind...
>
> [1] http://www.jgroups.org/taskdistribution.html
>
>
My first version of this distribution system did use JGroups directly
(and drew some inspiration from your paper, actually). I can definitely
go back to a JGroups-based approach if you think it is worthwhile (or a
Infinispan-command-based approach, if there's interest from the
Infinispan community… I'm assuming that using Infinispan's command
pattern and RPCManager would be comparable to vanilla JGroups,
performance-wise, but would be more consistent with Infinispan's
architecture).
While I'm a big proponent of Infinispan, there are cases where it's the
wrong tool for the job, e.g. for messaging systems, notification buses
and master-slave work distribution.
The differences between this algorithm and the one you propose in
your
task distribution paper, for whatever it's worth:
1.) tasks are broadcast to just a subset of nodes, not all. The subset
hopefully corresponds to the subset of nodes that contain the data being
processed, if the task involves data in cache.
Got it, using DIST. You could use a similar mechanism to send tasks to a
subset, using a consistent hash over the task ID, for example. In a
large system, this would certainly be better as the tasks are now not
stored by everyone, but only by a small subset. My system used
multicasts because it was a prototype and mcasts were simpler to implement.
2.) The algorithm is pull-based, instead of push-based, so that
workers
only "pull" more work when they're idle (or nearly idle). The hope is
that this allows workers to remain more evenly loaded.
OK, makes senses. But this gets back to using the right tool. If I has
given the task of implementing such a system, I'd intuitively had used
JMS queues, or a peer-to-peer system like JGroups.
3.) the algorithm attempts to keep tasks on the originating node if
possible, for efficiency.
Now, those could all be done with vanilla JGroups with just a couple
tweaks to your design. There are some downsides to this algorithm vs.
your design, too -- it requires one more exchange between master &
worker (to "claim" a task), for example. I could implement something
more like your design, with a push-based distribution scheme that pins
tasks to nodes to see what the quantifiable effect is.
Yes, in my design, a task is only rescheduled to a different node if the
original worker node *died*, ie. was removed from the view, and not if
the consistent hash over the task ID pointed to a different node.
The thing that originally made me want to go with a task cache: in
your
paper (and in any design that supports node failure), each node is
holding a view of other tasks in the system. When a new node joins, they
receive a list of tasks. There's a protocol for keeping the task lists
synchronized (multicast add and remove, for example), and if we want to
scale either in # of tasks or # of nodes, we need to partition the tasks
so that each node contains a subset of tasks, while still ensuring that
each task is stored on some number of nodes and that when one of those
nodes goes down, that data is assigned to another node…
Agreed. But as I said above, this could be done by computing a hash over
the task and sending the work to a subset of the cluster (with N
unicasts) rather than a multicast.
Or you could multicast the task and only the nodes which match the
hash(task_id) would store the task... the former is probably more
efficient, the latter simpler to implement. Although either case doesn't
sound like it's overly complicated to implement.
On the other hand, I think you're arguing that a task cache is
doing
more than just distributing the data, and a simpler task broadcast is
more appropriate. I don't know all the things that Infinispan is doing
under the covers to optimize for a read-heavy workload, so I'll have to
take your word for it.
I wrote JBossCache (Infinispan's predecessor) and my design was geared
towards read-heavy workloads. TBH, I just *assumed* but didn't really
know whether this is still the case with Infinispan, perhaps someone
from the Infinispan team can chime in ?
When I find some more time, I can dust off my JGroups-based prototype
of
the same algorithm and run some more performance benchmarks. In very
preliminary tests, the task-cache-based design and the JGroups-based
design were very similar in performance. I'll need to run more extensive
tests before I can really quantify the overhead, though. And I'm sure
the JGroups-based design could be made more efficient.
Yes, see above
Thanks a lot for the feedback -- I really appreciate it.
NP, cheers,
--
Bela Ban, JGroups lead (http://www.jgroups.org)
1:07 p.m.
This sounds really interesting Matt. In the Cloud-TM project
(www.cloudtm.eu) we are currently developing a parallel graph-analysis
algorithm on top of Infinispan's DEF. I would be really curious to take
a look at the framework you developed, and see how it could be exploited
in our application.
Regards,
Paolo
--
Paolo Romano, PhD
Coordinator of the Cloud-TM ICT FP7 Project (www.cloudtm.eu)
Senior Researcher @ INESC-ID (www.inesc-id.pt)
Assistant Professor @ Instituto Superior Tecnico (www.ist.utl.pt)
Rua Alves Redol, 9
1000-059, Lisbon Portugal
Tel. + 351 21 3100300
Fax + 351 21 3145843
Webpage http://www.gsd.inesc-id.pt/~romanop
On 3/2/13 5:40 PM, matt hoffman wrote:
Hey guys,
I've been working on a prototype of integrating Infinispan into our
app. We do a lot of distributed processing across a small cluster, so
I've played with Infinispan's existing distributed execution framework
(which is nice), as well as using Infinispan alongside a normal
message queue to distribute tasks. But I've also put together a
prototype of a new distributed execution framework using fork-join
pools that you all might be interested in. If it sounds like
something that would be worthwhile for Infinispan, I can raise a Jira
and submit a pull request with what I have so far. I'd need to get the
CA and company policy stuff finalized; that might take a couple days.
Meanwhile, in case there is any interest, I've described the approach
I've taken below.
First, a little background:
A while back I worked on a side project that integrated a distributed
work-stealing algorithm into the standard JDK fork-join queue. It
used JGroups for communication, because it was quick and easy for
prototyping. So this week I thought i'd take a stab at porting that
over to Infinispan. The algorithm I came up with for Infinispan is a
bit less of a work-stealing algorithm, to take advantage of
Infinispan's built-in distribution capabilities, but I think it's
still fairly efficient.
My basic approach was to take in a cache in the constructor, much like
the existing distributed executor, and then create a parallel,
DIST-mode cache that uses the same hash & grouping configuration as
the original cache. That new parallel cache is the "task cache", and
we use that to distribute available tasks across the cluster. It's a
distributed cache so that tasks are partitioned across a large
cluster, and it uses the hashing config of the original cache and a
KeyAffinityService to attempt to distribute the tasks to the same
nodes that contain the data being worked on. Nodes use cache listeners
to be notified when there is new work available, and the atomic
replace() to "check out" the tasks for execution, and "check in" the
results.
The basic algorithm is something like this:
For a refresher, a normal FJ pool has a fork() method that takes in a
task, and then places that task on an internal queue (actually, one of
several queues). When threads are idle, they look to the nearest work
queue for work. If that work queue does not have work, they "steal"
work from another thread's queue. So in the best case, tasks remain on
the same thread as the task that spawned them, so tasks that process
the same data as their parents may still have that data in the CPU's
cache, etc. There's more to it than that, but that's the basic idea.
This distributed algorithm just adds an extra layer on top for tasks
that are marked "distributable" (by extending DistributedFJTask
instead of the normal ForkJoinTask). When you call fork() with a
DistributedFJTask, it first checks to see if the local pool's work
queue is empty. If so, we just go ahead and submit it locally; there's
no reason to distribute it. If not, we put the task in the task cache,
and let Infinispan distribute it. When a node has no more work to do
in its internal fork-join queues, it looks at the task cache and tries
to pull work from there.
So, it isn't really a "work-stealing" algorithm, per se; the
distributable tasks are being distributed eagerly using Infinispan's
normal cache distribution. But I'm hoping that doing that also makes
it easier to handle node failure, since nodes collectively share a
common picture of the work to be done.
This approach required one change to the actual FJ classes themselves
(in org.infinispan.util.concurrent.jdk8backported). That's probably
the most controversial change. I had to make the original
ForkJoinTask's fork() method non-final in order to extend it cleanly.
There's probably a way around that, but that's the cleanest option I
have thought of thus far.
And lastly, it's not done yet: basic task distribution is working, but
I haven't tackled failover to any real extent yet. The biggest
questions, though, are around what to do with the existing distributed
execution interfaces. For example, DistributedTask has a getCallable()
method because it assumes it's wrapping a Callable. But ForkJoinTasks
don't extend Callable. I could put in a shim to wrap the
DistributedFJTasks into Callables for the sake of that method, but I
don't know if it's worth it. Similarly, the
DistributedExecutorService interface exposes a lot of
submit-to-specific-address or submit-to-all-addresses methods, which
are an odd fit here since tasks are distributed via their own cache.
Even if I used a KeyAffinityService to target the task to the given
Address, it might get picked up by another node that shares that same
hash. But I can add in a direct-to-single-Address capability in if
that seems worthwhile. Alternately, I can just use entirely different
interfaces (DistributedFJExecutorService, DistributedFJTask?).
Thoughts? Concerns? Glaring issues?
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
2:50 a.m.
Resurrecting this topic: I've put some sample code on my fork here:
https://github.com/matthoffman/infinispan/tree/dfj
There's a README there that offers an overview. I hope to do some
performance comparisons over the weekend, but in the meantime, the code is
there if you're curious.
I'm mainly interested whether this is the type of thing that Infinispan may
be interested in, as an alternative to the current distributed executor
(much like the ForkJoinPool in the JDK is an alternative to a traditional
thread pool).
If so, the next steps would be to do some performance tests, pick one
implementation to move forward with (out of the 3 prototypes on that
branch) and clean it up sufficiently to be considered for a pull request.
If its not something that Infinispan is interested in, then I'll change
approaches and make it more generic, so that it isn't
Infinispan-specific and can be used with other transports. I'm open to
either option... I think there are pros and cons either way.
Thanks!
matt
On Mon, Mar 4, 2013 at 7:07 AM, Paolo Romano <romano(a)inesc-id.pt> wrote:
This sounds really interesting Matt. In the Cloud-TM project (
www.cloudtm.eu) we are currently developing a parallel graph-analysis
algorithm on top of Infinispan's DEF. I would be really curious to take a
look at the framework you developed, and see how it could be exploited in
our application.
Regards,
Paolo
--
Paolo Romano, PhD
Coordinator of the Cloud-TM ICT FP7 Project (www.cloudtm.eu)
Senior Researcher @ INESC-ID (www.inesc-id.pt)
Assistant Professor @ Instituto Superior Tecnico (www.ist.utl.pt)
Rua Alves Redol, 9
1000-059, Lisbon Portugal
Tel. + 351 21 3100300
Fax + 351 21 3145843
Webpage http://www.gsd.inesc-id.pt/~romanop
On 3/2/13 5:40 PM, matt hoffman wrote:
Hey guys,
I've been working on a prototype of integrating Infinispan into our app.
We do a lot of distributed processing across a small cluster, so I've
played with Infinispan's existing distributed execution framework (which is
nice), as well as using Infinispan alongside a normal message queue to
distribute tasks. But I've also put together a prototype of a new
distributed execution framework using fork-join pools that you all might be
interested in. If it sounds like something that would be worthwhile for
Infinispan, I can raise a Jira and submit a pull request with what I have
so far. I'd need to get the CA and company policy stuff finalized; that
might take a couple days. Meanwhile, in case there is any interest, I've
described the approach I've taken below.
First, a little background:
A while back I worked on a side project that integrated a distributed
work-stealing algorithm into the standard JDK fork-join queue. It used
JGroups for communication, because it was quick and easy for prototyping.
So this week I thought i'd take a stab at porting that over to Infinispan.
The algorithm I came up with for Infinispan is a bit less of a
work-stealing algorithm, to take advantage of Infinispan's built-in
distribution capabilities, but I think it's still fairly efficient.
My basic approach was to take in a cache in the constructor, much like the
existing distributed executor, and then create a parallel, DIST-mode cache
that uses the same hash & grouping configuration as the original cache.
That new parallel cache is the "task cache", and we use that to distribute
available tasks across the cluster. It's a distributed cache so that tasks
are partitioned across a large cluster, and it uses the hashing config of
the original cache and a KeyAffinityService to attempt to distribute the
tasks to the same nodes that contain the data being worked on. Nodes use
cache listeners to be notified when there is new work available, and the
atomic replace() to "check out" the tasks for execution, and "check
in" the
results.
The basic algorithm is something like this:
For a refresher, a normal FJ pool has a fork() method that takes in a
task, and then places that task on an internal queue (actually, one of
several queues). When threads are idle, they look to the nearest work queue
for work. If that work queue does not have work, they "steal" work from
another thread's queue. So in the best case, tasks remain on the same
thread as the task that spawned them, so tasks that process the same data
as their parents may still have that data in the CPU's cache, etc. There's
more to it than that, but that's the basic idea.
This distributed algorithm just adds an extra layer on top for tasks that
are marked "distributable" (by extending DistributedFJTask instead of the
normal ForkJoinTask). When you call fork() with a DistributedFJTask, it
first checks to see if the local pool's work queue is empty. If so, we just
go ahead and submit it locally; there's no reason to distribute it. If not,
we put the task in the task cache, and let Infinispan distribute it. When a
node has no more work to do in its internal fork-join queues, it looks at
the task cache and tries to pull work from there.
So, it isn't really a "work-stealing" algorithm, per se; the distributable
tasks are being distributed eagerly using Infinispan's normal cache
distribution. But I'm hoping that doing that also makes it easier to handle
node failure, since nodes collectively share a common picture of the work
to be done.
This approach required one change to the actual FJ classes themselves (in
org.infinispan.util.concurrent.jdk8backported).
That's probably the most controversial change. I had to make the original
ForkJoinTask's fork() method non-final in order to extend it cleanly.
There's probably a way around that, but that's the cleanest option I have
thought of thus far.
And lastly, it's not done yet: basic task distribution is working, but I
haven't tackled failover to any real extent yet. The biggest questions,
though, are around what to do with the existing distributed execution
interfaces. For example, DistributedTask has a getCallable() method because
it assumes it's wrapping a Callable. But ForkJoinTasks don't extend
Callable. I could put in a shim to wrap the DistributedFJTasks into
Callables for the sake of that method, but I don't know if it's worth it.
Similarly, the DistributedExecutorService interface exposes a lot of
submit-to-specific-address or submit-to-all-addresses methods, which are an
odd fit here since tasks are distributed via their own cache. Even if I
used a KeyAffinityService to target the task to the given Address, it might
get picked up by another node that shares that same hash. But I can add in
a direct-to-single-Address capability in if that seems worthwhile.
Alternately, I can just use entirely different interfaces
(DistributedFJExecutorService, DistributedFJTask?).
Thoughts? Concerns? Glaring issues?
_______________________________________________
infinispan-dev mailing
listinfinispan-dev@lists.jboss.orghttps://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
11:52 a.m.
Hi Matt.
Thanks for resurrecting and sorry for not responding on the original thread.
Interesting discussions. I would intuitively agree with Bela were it not for the fact
that, as you said, you're actually storing data, require high availability of that
data during server failure, and may possibly chain tasks (as per a local fork-join) and
want to maintain server affinity across related tasks.
You could achieve all of this with JGroups but would end up re-implementing certain bits
of Infinispan, specifically:
- task failover
- task cancelling
- consistent hash based node selection for delivery, ensuring affinity across data and
tasks
So I think it is a pretty valid use case to make use of what Infinispan already has.
I'll have a look at your sources and comment more. I'm sure Paolo and others at
CloudTM will probably do the same.
Cheers
Manik
On 9 May 2013, at 01:50, Matt Hoffman <matt(a)mhoffman.org> wrote:
Resurrecting this topic: I've put some sample code on my fork
here: https://github.com/matthoffman/infinispan/tree/dfj
There's a README there that offers an overview. I hope to do some performance
comparisons over the weekend, but in the meantime, the code is there if you're
curious.
I'm mainly interested whether this is the type of thing that Infinispan may be
interested in, as an alternative to the current distributed executor (much like the
ForkJoinPool in the JDK is an alternative to a traditional thread pool).
If so, the next steps would be to do some performance tests, pick one implementation to
move forward with (out of the 3 prototypes on that branch) and clean it up sufficiently to
be considered for a pull request.
If its not something that Infinispan is interested in, then I'll change approaches
and make it more generic, so that it isn't Infinispan-specific and can be used with
other transports. I'm open to either option... I think there are pros and cons either
way.
Thanks!
matt
On Mon, Mar 4, 2013 at 7:07 AM, Paolo Romano <romano(a)inesc-id.pt> wrote:
This sounds really interesting Matt. In the Cloud-TM project (www.cloudtm.eu) we are
currently developing a parallel graph-analysis algorithm on top of Infinispan's DEF. I
would be really curious to take a look at the framework you developed, and see how it
could be exploited in our application.
Regards,
Paolo
--
Paolo Romano, PhD
Coordinator of the Cloud-TM ICT FP7 Project (www.cloudtm.eu)
Senior Researcher @ INESC-ID (www.inesc-id.pt)
Assistant Professor @ Instituto Superior Tecnico (www.ist.utl.pt)
Rua Alves Redol, 9
1000-059, Lisbon Portugal
Tel. + 351 21 3100300
Fax + 351 21 3145843
Webpage http://www.gsd.inesc-id.pt/~romanop
On 3/2/13 5:40 PM, matt hoffman wrote:
> Hey guys,
>
> I've been working on a prototype of integrating Infinispan into our app. We do a
lot of distributed processing across a small cluster, so I've played with
Infinispan's existing distributed execution framework (which is nice), as well as
using Infinispan alongside a normal message queue to distribute tasks. But I've also
put together a prototype of a new distributed execution framework using fork-join pools
that you all might be interested in. If it sounds like something that would be worthwhile
for Infinispan, I can raise a Jira and submit a pull request with what I have so far.
I'd need to get the CA and company policy stuff finalized; that might take a couple
days. Meanwhile, in case there is any interest, I've described the approach I've
taken below.
>
> First, a little background:
>
> A while back I worked on a side project that integrated a distributed work-stealing
algorithm into the standard JDK fork-join queue. It used JGroups for communication,
because it was quick and easy for prototyping. So this week I thought i'd take a stab
at porting that over to Infinispan. The algorithm I came up with for Infinispan is a bit
less of a work-stealing algorithm, to take advantage of Infinispan's built-in
distribution capabilities, but I think it's still fairly efficient.
>
> My basic approach was to take in a cache in the constructor, much like the existing
distributed executor, and then create a parallel, DIST-mode cache that uses the same hash
& grouping configuration as the original cache. That new parallel cache is the
"task cache", and we use that to distribute available tasks across the cluster.
It's a distributed cache so that tasks are partitioned across a large cluster, and it
uses the hashing config of the original cache and a KeyAffinityService to attempt to
distribute the tasks to the same nodes that contain the data being worked on. Nodes use
cache listeners to be notified when there is new work available, and the atomic replace()
to "check out" the tasks for execution, and "check in" the results.
>
> The basic algorithm is something like this:
>
> For a refresher, a normal FJ pool has a fork() method that takes in a task, and then
places that task on an internal queue (actually, one of several queues). When threads are
idle, they look to the nearest work queue for work. If that work queue does not have work,
they "steal" work from another thread's queue. So in the best case, tasks
remain on the same thread as the task that spawned them, so tasks that process the same
data as their parents may still have that data in the CPU's cache, etc. There's
more to it than that, but that's the basic idea.
>
> This distributed algorithm just adds an extra layer on top for tasks that are marked
"distributable" (by extending DistributedFJTask instead of the normal
ForkJoinTask). When you call fork() with a DistributedFJTask, it first checks to see if
the local pool's work queue is empty. If so, we just go ahead and submit it locally;
there's no reason to distribute it. If not, we put the task in the task cache, and let
Infinispan distribute it. When a node has no more work to do in its internal fork-join
queues, it looks at the task cache and tries to pull work from there.
>
> So, it isn't really a "work-stealing" algorithm, per se; the
distributable tasks are being distributed eagerly using Infinispan's normal cache
distribution. But I'm hoping that doing that also makes it easier to handle node
failure, since nodes collectively share a common picture of the work to be done.
>
> This approach required one change to the actual FJ classes themselves (in
org.infinispan.util.concurrent.jdk8backported). That's probably the most controversial
change. I had to make the original ForkJoinTask's fork() method non-final in order to
extend it cleanly. There's probably a way around that, but that's the cleanest
option I have thought of thus far.
>
> And lastly, it's not done yet: basic task distribution is working, but I
haven't tackled failover to any real extent yet. The biggest questions, though, are
around what to do with the existing distributed execution interfaces. For example,
DistributedTask has a getCallable() method because it assumes it's wrapping a
Callable. But ForkJoinTasks don't extend Callable. I could put in a shim to wrap the
DistributedFJTasks into Callables for the sake of that method, but I don't know if
it's worth it. Similarly, the DistributedExecutorService interface exposes a lot of
submit-to-specific-address or submit-to-all-addresses methods, which are an odd fit here
since tasks are distributed via their own cache. Even if I used a KeyAffinityService to
target the task to the given Address, it might get picked up by another node that shares
that same hash. But I can add in a direct-to-single-Address capability in if that seems
worthwhile. Alternately, I can just use entirely different interfaces
(DistributedFJExecutorService, DistributedFJTask?).
>
> Thoughts? Concerns? Glaring issues?
>
>
>
> _______________________________________________
> 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
Platform Architect, JBoss Data Grid
http://red.ht/data-grid
1:23 a.m.
Very interesting stuff Matt!
I find the design that leverages on Infinispan for distributing the
tasks - while maximizing locality and exploiting Infinispan's
fault-tolerance capabilities - modular and elegant. It will be
interesting to evaluate if there is a significant overhead with respect
to a purely JGroups-based mechanism.
We will be looking at your code, and may pheraps come up with some
use-cases.
Cheers,
Paolo
On 5/9/13 10:52 AM, Manik Surtani wrote:
Hi Matt.
Thanks for resurrecting and sorry for not responding on the original
thread.
Interesting discussions. I would intuitively agree with Bela were it
not for the fact that, as you said, you're actually storing data,
require high availability of that data during server failure, and may
possibly chain tasks (as per a local fork-join) and want to maintain
server affinity across related tasks.
You could achieve all of this with JGroups but would end up
re-implementing certain bits of Infinispan, specifically:
- task failover
- task cancelling
- consistent hash based node selection for delivery, ensuring affinity
across data and tasks
So I think it is a pretty valid use case to make use of what
Infinispan already has.
I'll have a look at your sources and comment more. I'm sure Paolo and
others at CloudTM will probably do the same.
Cheers
Manik
On 9 May 2013, at 01:50, Matt Hoffman <matt(a)mhoffman.org
<mailto:matt@mhoffman.org>> wrote:
> Resurrecting this topic: I've put some sample code on my fork here:
> https://github.com/matthoffman/infinispan/tree/dfj
> There's a README there that offers an overview. I hope to do some
> performance comparisons over the weekend, but in the meantime, the
> code is there if you're curious.
>
> I'm mainly interested whether this is the type of thing that
> Infinispan may be interested in, as an alternative to the current
> distributed executor (much like the ForkJoinPool in the JDK is an
> alternative to a traditional thread pool).
> If so, the next steps would be to do some performance tests, pick one
> implementation to move forward with (out of the 3 prototypes on that
> branch) and clean it up sufficiently to be considered for a pull
> request.
>
> If its not something that Infinispan is interested in, then I'll
> change approaches and make it more generic, so that it isn't
> Infinispan-specific and can be used with other transports. I'm open
> to either option... I think there are pros and cons either way.
>
>
> Thanks!
>
>
> matt
>
>
>
> On Mon, Mar 4, 2013 at 7:07 AM, Paolo Romano <romano(a)inesc-id.pt
> <mailto:romano@inesc-id.pt>> wrote:
>
> This sounds really interesting Matt. In the Cloud-TM project
> (www.cloudtm.eu <http://www.cloudtm.eu/>) we are currently
> developing a parallel graph-analysis algorithm on top of
> Infinispan's DEF. I would be really curious to take a look at the
> framework you developed, and see how it could be exploited in our
> application.
>
> Regards,
>
> Paolo
>
>
> --
>
> Paolo Romano, PhD
> Coordinator of the Cloud-TM ICT FP7 Project (www.cloudtm.eu
<http://www.cloudtm.eu/>)
> Senior Researcher @ INESC-ID (www.inesc-id.pt <http://www.inesc-id.pt/>)
> Assistant Professor @ Instituto Superior Tecnico (www.ist.utl.pt
<http://www.ist.utl.pt/>)
> Rua Alves Redol, 9
> 1000-059, Lisbon Portugal
> Tel.+ 351 21 3100300 <tel:%2B%20351%2021%203100300>
> Fax+ 351 21 3145843 <tel:%2B%20351%2021%203145843>
> Webpagehttp://www.gsd.inesc-id.pt/~romanop
<http://www.gsd.inesc-id.pt/%7Eromanop>
>
>
>
> On 3/2/13 5:40 PM, matt hoffman wrote:
>>
>> Hey guys,
>>
>> I've been working on a prototype of integrating Infinispan into
>> our app. We do a lot of distributed processing across a small
>> cluster, so I've played with Infinispan's existing distributed
>> execution framework (which is nice), as well as using Infinispan
>> alongside a normal message queue to distribute tasks. But I've
>> also put together a prototype of a new distributed execution
>> framework using fork-join pools that you all might be interested
>> in. If it sounds like something that would be worthwhile for
>> Infinispan, I can raise a Jira and submit a pull request with
>> what I have so far. I'd need to get the CA and company policy
>> stuff finalized; that might take a couple days. Meanwhile, in
>> case there is any interest, I've described the approach I've
>> taken below.
>>
>> First, a little background:
>>
>> A while back I worked on a side project that integrated a
>> distributed work-stealing algorithm into the standard JDK
>> fork-join queue. It used JGroups for communication, because it
>> was quick and easy for prototyping. So this week I thought i'd
>> take a stab at porting that over to Infinispan. The algorithm I
>> came up with for Infinispan is a bit less of a work-stealing
>> algorithm, to take advantage of Infinispan's built-in
>> distribution capabilities, but I think it's still fairly efficient.
>>
>> My basic approach was to take in a cache in the constructor,
>> much like the existing distributed executor, and then create a
>> parallel, DIST-mode cache that uses the same hash & grouping
>> configuration as the original cache. That new parallel cache is
>> the "task cache", and we use that to distribute available tasks
>> across the cluster. It's a distributed cache so that tasks are
>> partitioned across a large cluster, and it uses the hashing
>> config of the original cache and a KeyAffinityService to attempt
>> to distribute the tasks to the same nodes that contain the data
>> being worked on. Nodes use cache listeners to be notified when
>> there is new work available, and the atomic replace() to "check
>> out" the tasks for execution, and "check in" the results.
>>
>> The basic algorithm is something like this:
>>
>> For a refresher, a normal FJ pool has a fork() method that takes
>> in a task, and then places that task on an internal queue
>> (actually, one of several queues). When threads are idle, they
>> look to the nearest work queue for work. If that work queue does
>> not have work, they "steal" work from another thread's queue.
>> So in the best case, tasks remain on the same thread as the task
>> that spawned them, so tasks that process the same data as their
>> parents may still have that data in the CPU's cache, etc.
>> There's more to it than that, but that's the basic idea.
>>
>> This distributed algorithm just adds an extra layer on top for
>> tasks that are marked "distributable" (by extending
>> DistributedFJTask instead of the normal ForkJoinTask). When you
>> call fork() with a DistributedFJTask, it first checks to see if
>> the local pool's work queue is empty. If so, we just go ahead
>> and submit it locally; there's no reason to distribute it. If
>> not, we put the task in the task cache, and let Infinispan
>> distribute it. When a node has no more work to do in its
>> internal fork-join queues, it looks at the task cache and tries
>> to pull work from there.
>>
>> So, it isn't really a "work-stealing" algorithm, per se; the
>> distributable tasks are being distributed eagerly using
>> Infinispan's normal cache distribution. But I'm hoping that
>> doing that also makes it easier to handle node failure, since
>> nodes collectively share a common picture of the work to be done.
>>
>> This approach required one change to the actual FJ classes
>> themselves (in org.infinispan.util.concurrent.jdk8backported).
>> That's probably the most controversial change. I had to make the
>> original ForkJoinTask's fork() method non-final in order to
>> extend it cleanly. There's probably a way around that, but
>> that's the cleanest option I have thought of thus far.
>>
>> And lastly, it's not done yet: basic task distribution is
>> working, but I haven't tackled failover to any real extent yet.
>> The biggest questions, though, are around what to do with the
>> existing distributed execution interfaces. For example,
>> DistributedTask has a getCallable() method because it assumes
>> it's wrapping a Callable. But ForkJoinTasks don't extend
>> Callable. I could put in a shim to wrap the DistributedFJTasks
>> into Callables for the sake of that method, but I don't know if
>> it's worth it. Similarly, the DistributedExecutorService
>> interface exposes a lot of submit-to-specific-address or
>> submit-to-all-addresses methods, which are an odd fit here since
>> tasks are distributed via their own cache. Even if I used a
>> KeyAffinityService to target the task to the given Address, it
>> might get picked up by another node that shares that same
>> hash. But I can add in a direct-to-single-Address capability in
>> if that seems worthwhile. Alternately, I can just use entirely
>> different interfaces (DistributedFJExecutorService,
>> DistributedFJTask?).
>>
>> Thoughts? Concerns? Glaring issues?
>>
>>
>>
>> _______________________________________________
>> infinispan-dev mailing list
>> infinispan-dev(a)lists.jboss.org
<mailto:infinispan-dev@lists.jboss.org>
>> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org
> <mailto:infinispan-dev@lists.jboss.org>
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
>
>
> _______________________________________________
> infinispan-dev mailing list
> infinispan-dev(a)lists.jboss.org <mailto:infinispan-dev@lists.jboss.org>
> https://lists.jboss.org/mailman/listinfo/infinispan-dev
--
Manik Surtani
manik(a)jboss.org <mailto:manik@jboss.org>
twitter.com/maniksurtani <http://twitter.com/maniksurtani>
Platform Architect, JBoss Data Grid
http://red.ht/data-grid
_______________________________________________
infinispan-dev mailing list
infinispan-dev(a)lists.jboss.org
https://lists.jboss.org/mailman/listinfo/infinispan-dev
4010
days inactive
4079
days old
infinispan-dev@lists.jboss.org
7 comments
5 participants
participants (5)
-
Bela Ban -
Manik Surtani -
matt hoffman -
Matt Hoffman
-
Paolo Romano