Actor mailbox size limit
Shimi
If it is, how can it be done using the JavaAPI?
I want to prevent a case of OOME when there is a slow consumer and a
fast producer.
Shimi
Viktor Klang
What is your desired behavior when the mailbox is full?
Shimi
--
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To post to this group, send email to akka...@googlegroups.com.
To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.
--
Viktor Klang
| "A complex system that works is invariably
| found to have evolved from a simple system
| that worked." - John Gall
Akka - the Actor Kernel: Akkasource.org
Twttr: twitter.com/viktorklang
Shimi
any new messages.
Shimi
On Jun 28, 4:48 pm, Viktor Klang <viktor.kl...@gmail.com> wrote:
> > Is it possible to limit the mailbox size without blocking the sender?
> > If it is, how can it be done using the JavaAPI?
> > I want to prevent a case of OOME when there is a slow consumer and a
> > fast producer.
>
> What is your desired behavior when the mailbox is full?
>
>
>
> > Shimi
>
> > --
> > You received this message because you are subscribed to the Google Groups
> > "Akka User List" group.
> > To post to this group, send email to akka...@googlegroups.com.
> > To unsubscribe from this group, send email to
> > .
Shimi
ThreadPoolExecutor.DiscardPolicy to the dispatcher?
Shimi
Jonas Bonér
> Will it solve my problem if I will pass
> ThreadPoolExecutor.DiscardPolicy to the dispatcher?
Yes. Either:
java.util.concurrent.ThreadPoolExecutor.AbortPolicy
or
java.util.concurrent.ThreadPoolExecutor.DiscardPolicy
or
java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy
Read more here:
http://java.sun.com/javase/6/docs/api/index.html?java/util/concurrent/RejectedExecutionHandler.html
> To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
> For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.
>
>
--
Jonas Bonér
work: http://jayway.com
code: http://akkasource.com
blog: http://jonasboner.com
twitter: @jboner
Jonas Bonér
You can also set the rejection policy that should be used, e.g. what should be done if the dispatcher (e.g. the Actor) can't keep up and the mailbox is growing up to the limit defined. You can choose between four different rejection policies:
- java.util.concurrent.ThreadPoolExecutor.CallerRuns - will run the message processing in the caller's thread as a way to slow him down and balance producer/consumer
- java.util.concurrent.ThreadPoolExecutor.AbortPolicy - rejected messages by throwing a 'RejectedExecutionException'
- java.util.concurrent.ThreadPoolExecutor.DiscardPolicy - discards the message (throws it away)
- java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy - discards the oldest message in the mailbox (throws it away)
You cane read more about these policies http://java.sun.com/javase/6/docs/api/index.html?java/util/concurrent/RejectedExecutionHandler.html
Here is an example:
class MyActor extends Actor { self.dispatcher = Dispatchers.newExecutorEventBasedThreadPoolDispatcher(name) dispatcher .withNewThreadPoolWithBoundedBlockingQueue(100) .setCorePoolSize(16) .setMaxPoolSize(128) .setKeepAliveTimeInMillis(60000) .setRejectionPolicy(new CallerRunsPolicy) .buildThreadPool ... }
Shimi
Shimi
On Jun 28, 5:45 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> Added it to the docs herehttp://doc.akkasource.org/dispatchers:
> You can also set the rejection policy that should be used, e.g. what should
> be done if the dispatcher (e.g. the Actor) can't keep up and the mailbox is
> growing up to the limit defined. You can choose between four different
> rejection policies:
>
> balance producer/consumer
> by throwing a 'RejectedExecutionException'
> - java.util.concurrent.ThreadPoolExecutor.DiscardPolicy - discards the
> message (throws it away)
> - java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy - discards
>
> Here is an example:
>
> class MyActor extends Actor {
> self.dispatcher = Dispatchers.newExecutorEventBasedThreadPoolDispatcher(name)
> dispatcher
> .withNewThreadPoolWithBoundedBlockingQueue(100)
> .setCorePoolSize(16)
> .setMaxPoolSize(128)
> .setKeepAliveTimeInMillis(60000)
> .setRejectionPolicy(new CallerRunsPolicy)
> .buildThreadPool
> ...}
>
> >> Will it solve my problem if I will pass
> >> ThreadPoolExecutor.DiscardPolicy to the dispatcher?
>
> > Yes. Either:
> > java.util.concurrent.ThreadPoolExecutor.AbortPolicy
> > or
> > java.util.concurrent.ThreadPoolExecutor.DiscardPolicy
> > or
> > java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy
>
> > Read more here:
>
Jonas Bonér
When defining a BlockingQueue in the fluent API. See the doc snippet I sent.
--
Jonas Bonér
http://jayway.com
http://akkasource.com
twitter: jboner
On 28 Jun 2010 17:34, "Shimi" <shi...@gmail.com> wrote:
Where do you define the mailbox limit?
Shimi
On Jun 28, 5:45 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> Added it to the docs herehttp://doc.akkasource.org/dispatchers:
>
> You can also set the rejection policy that should be used, e.g. what should
> be done if the dis...
> You cane read more about these policieshttp://java.sun.com/javase/6/docs/api/index.html?java/util/concurrent...</here>
>
> Here is an example:
>
> class MyActor extends Actor {
> self.dispatcher = Dispatchers.newExecu...
> On 28 June 2010 16:38, Jonas Bonér <jo...@jonasboner.com> wrote:> On 28 June 2010 16:00, Shimi <sh...
> http://java.sun.com/javase/6/docs/api/index.html?java/util/concurrent...
>
>
>
>
>
> >> Shimi
>
> >> On Jun 28, 4:52 pm, Shimi <shim...@gmail.com> wrote:
> >>> In my case it...
> > Jonas Bonér
>
> > work: http://jayway.com
> > code: http://akkasource.com
> > blog: http://...
--
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To...
Shimi
I am a little bit confused. I thought the mailbox is set to
ConcurrentLinkedDeque in the ActorRef class and the Dispacther defines
a queue of Actors.
Shimi
On Jun 28, 6:48 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> When defining a BlockingQueue in the fluent API. See the doc snippet I sent.
>
> --
> twitter: jboner
Jonas Bonér
> this withNewThreadPoolWithBoundedBlockingQueue(100) ?
> I am a little bit confused. I thought the mailbox is set to
> ConcurrentLinkedDeque in the ActorRef class and the Dispacther defines
> a queue of Actors.
Yes. There is effectively two queues. One is the mailbox on the
ActorRef and one is the used by the dispatcher. The latter one can be
used for throttling and is highly configurable. Would you need to
override the mailbox queue as well? You can't set a rejection policy
for that one, you would have to implement such a thing yourself.
> To post to this group, send email to akka...@googlegroups.com.
> To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
> For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.
>
>
--
Jonas Bonér
work: http://jayway.com
code: http://akkasource.com
blog: http://jonasboner.com
twitter: @jboner
Shimi
"You can also set the rejection policy that should be used, e.g. what
should
mailbox is
growing up to the limit defined."
mailbox.
When I write:
dispatcher
.withNewThreadPoolWithBoundedBlockingQueue(100)
.setRejectionPolicy(new DiscardPolicy)
Do I limit the number of Actors or the number of the items in their
mailbox (like you wrote above)?
Shimi
On Jun 28, 7:20 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
Jonas Bonér
> I still do not understand. You wrote:
> "You can also set the rejection policy that should be used, e.g. what
> should
> be done if the dispatcher (e.g. the Actor) can't keep up and the
> mailbox is
> growing up to the limit defined."
>
> This seems more like limiting the number of Actors and not their
> mailbox.
> When I write:
> dispatcher
> .withNewThreadPoolWithBoundedBlockingQueue(100)
> .setRejectionPolicy(new DiscardPolicy)
>
> Do I limit the number of Actors or the number of the items in their
> mailbox (like you wrote above)?
>
I see your point. You do limit the # of messages if you use one
dispatcher per Actor, but not if it is shared.
Currently the mailbox is defined as:
protected[akka] val _mailbox: Deque[MessageInvocation] = new
ConcurrentLinkedDeque[MessageInvocation]
You'd like to be able to plug in other mailbox impls?
> For more options, visit this group at http://groups.google.com/group/akka-user?hl=en.
Jonas Bonér
> On 28 June 2010 19:21, Shimi <shi...@gmail.com> wrote:
>> I still do not understand. You wrote:
>> "You can also set the rejection policy that should be used, e.g. what
>> should
>> be done if the dispatcher (e.g. the Actor) can't keep up and the
>> mailbox is
>> growing up to the limit defined."
>>
>> This seems more like limiting the number of Actors and not their
>> mailbox.
>> When I write:
>> dispatcher
>> .withNewThreadPoolWithBoundedBlockingQueue(100)
>> .setRejectionPolicy(new DiscardPolicy)
>>
>> Do I limit the number of Actors or the number of the items in their
>> mailbox (like you wrote above)?
>>
>
> I see your point. You do limit the # of messages if you use one
> dispatcher per Actor, but not if it is shared.
>
> Currently the mailbox is defined as:
> protected[akka] val _mailbox: Deque[MessageInvocation] = new
> ConcurrentLinkedDeque[MessageInvocation]
>
> You'd like to be able to plug in other mailbox impls?
E.g. a http://gee.cs.oswego.edu/dl/jsr166/dist/jsr166xdocs/jsr166x/LinkedBlockingDeque.html
Jonas Bonér
// if mailboxSize is negative then use unbounded, if positive create a
BlockingDeque with the size
def actorOf[T <: Actor: Manifest](mailboxSize = -1): ActorRef =
new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <:
Actor]], mailboxSize)
val actor = actorOf[MyActor](mailboxSize = 100)
Or should we do as we do with the new STM stuff? E.g. accept an
implicit ActorRefConfig that can do more?
def actorOf[T <: Actor: Manifest](implicit config: ActorRefConfig =
DefaultActorRefConfig()): ActorRef =
new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <: Actor]], config)
Thoughts?
Viktor Klang
Should we solve it by adding this?
// if mailboxSize is negative then use unbounded, if positive create a
BlockingDeque with the size
def actorOf[T <: Actor: Manifest](mailboxSize = -1): ActorRef =
new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <:
Actor]], mailboxSize)
val actor = actorOf[MyActor](mailboxSize = 100)
Or should we do as we do with the new STM stuff? E.g. accept an
implicit ActorRefConfig that can do more?
def actorOf[T <: Actor: Manifest](implicit config: ActorRefConfig =
DefaultActorRefConfig()): ActorRef =
new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <: Actor]], config)
Glad you bring this up, I've also (since last night) been thinking of the current actor/actorref initialization/configuration strategy.
From the library consumer perspective, is there any difference between the two? Is that difference obvious?
All config goes into Actor.init or parts go into actorOf?
Food for thought!
--
Jonas Bonér
>
>
> On Mon, Jun 28, 2010 at 11:49 PM, Jonas Bonér <jo...@jonasboner.com> wrote:
>>
>> Should we solve it by adding this?
>>
>> // if mailboxSize is negative then use unbounded, if positive create a
>> BlockingDeque with the size
>> def actorOf[T <: Actor: Manifest](mailboxSize = -1): ActorRef =
>> new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <:
>> Actor]], mailboxSize)
>>
>> val actor = actorOf[MyActor](mailboxSize = 100)
>>
>> Or should we do as we do with the new STM stuff? E.g. accept an
>> implicit ActorRefConfig that can do more?
>>
>> def actorOf[T <: Actor: Manifest](implicit config: ActorRefConfig =
>> DefaultActorRefConfig()): ActorRef =
>> new LocalActorRef(manifest[T].erasure.asInstanceOf[Class[_ <: Actor]],
>> config)
>
> Glad you bring this up, I've also (since last night) been thinking of the
> current actor/actorref initialization/configuration strategy.
> From the library consumer perspective, is there any difference between the
> two? Is that difference obvious?
>
> All config goes into Actor.init or parts go into actorOf?
In this case (and many other) I do not want to make mailbox a 'var' to
change in 'init' since then it can be changed at any point in time. It
should be a 'val' so only at ActorRef creation.
So I think we need both for sure.
Jonas Bonér
But what else could be put into the ActorRefConfig? Just one config
element does not pull its weight.
Viktor Klang
dispatcher?
timeout?
fault-handler?
trapExit?
lifeCycle?
isTransactor?
Jonas Bonér
Right. We can put all these there.
Shimi
> dispatcher per Actor, but not if it is shared.
of Actors. If I understood right, if you set:
dispatcher
.withNewThreadPoolWithBoundedBlockingQueue(1)
.setCorePoolSize(16)
.setMaxPoolSize(128)
.setKeepAliveTimeInMillis(60000)
.setRejectionPolicy(new ThreadPoolExecutor.DiscardPolicy)
.buildThreadPool
You will be able to use only one Actor with this dispatcher but the #
of messages to that Actor will be unlimited.
> Currently the mailbox is defined as:
> protected[akka] val _mailbox: Deque[MessageInvocation] = new
> ConcurrentLinkedDeque[MessageInvocation]
>
> You'd like to be able to plug in other mailbox impls?
LinkedBlockingDeque with limit and ThreadPoolExecutor.DiscardPolicy.
By the way the Dispatcher fluent API does not work. It should be
newExecutorBasedEventDrivenDispatcher and not
newExecutorEventBasedThreadPoolDispatcher.
Any way it throws IllegalStateException(
"Is not in the process of building a thread pool, start building
one by invoking one of the 'newThreadPool*' methods")
Shimi
On Jun 28, 11:31 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
Jonas Bonér
The number of concurrent actors is limited by max thread pool size (if blocking queue else min thread pool size).
I have added configuring mailbox to the backlog.
Don't get what the problem with the fluent API is. I use it without problems.
--
Jonas Bonér
http://jayway.com
http://akkasource.com
twitter: jboner
On 29 Jun 2010 15:26, "Shimi" <shi...@gmail.com> wrote:
> I see your point. You do limit the # of messages if you use one
> dispatcher per Actor, but not if...
It seems to me that you don't limit the # of messages but the number
of Actors. If I understood right, if you set:
dispatcher
.withNewThreadPoolWithBoundedBlockingQueue(1)
.setCorePoolSize(16)
.setRejectionPolicy(new ThreadPoolExecutor.DiscardPolicy)
.setMaxPoolSize(128)
.setKeepAliveTimeInMillis(60000)
.buildThreadPool
You will be able to use only one Actor with this dispatcher but the #
of messages to that Actor will be unlimited.
> Currently the mailbox is defined as:
> protected[akka] val _mailbox: Deque[MessageInvocation] =...
It will be nice to plugin a different Queue. In my case it can be a
LinkedBlockingDeque with limit and ThreadPoolExecutor.DiscardPolicy.
By the way the Dispatcher fluent API does not work. It should be
newExecutorBasedEventDrivenDispatcher and not
newExecutorEventBasedThreadPoolDispatcher.
Any way it throws IllegalStateException(
"Is not in the process of building a thread pool, start building
one by invoking one of the 'newThreadPool*' methods")
Shimi
On Jun 28, 11:31 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> On 28 June 2010 19:21, Shimi <shim...@gmail.com> wrote:
>
>
>
>
>
> > I still do not understand. Y...
Jonas Bonér
I see there is a typo in the docs. Wrong name in fluent API. But apart from that?
--
Jonas Bonér
http://jayway.com
http://akkasource.com
twitter: jboner
On 29 Jun 2010 16:30, "Jonas Bonér" <jo...@jonasboner.com> wrote:
The number of concurrent actors is limited by max thread pool size (if blocking queue else min thread pool size).
I have added configuring mailbox to the backlog.
Don't get what the problem with the fluent API is. I use it without problems.
--
Jonas Bonér
http://jayway.com
http://akkasource.com
twitter: jboner
>
> On 29 Jun 2010 15:26, "Shimi" <shi...@gmail.com> wrote:
>
> > I see your point. You do limit the # of messages if you use one
> dispatcher per Actor, but not if...
>
> It seems to me that you don't limit the # of messages but the number
> of Actors. If I understo...
>
> protected[akka] val _mailbox: Deque[MessageInvocation] =...
>
> > Currently the mailbox is defined as:
>
> It will be nice to plugin a different Queue. In my case it can be a
> LinkedBlockingDeque with ...
Shimi
> Don't get what the problem with the fluent API is. I use it without
> problems.
For example:
dispatcher.withNewThreadPoolWithBoundedBlockingQueue(1)
.setMaxPoolSize(128)
.setKeepAliveTimeInMillis(60000)
.setRejectionPolicy(new ThreadPoolExecutor.DiscardPolicy)
.buildThreadPool
to .setCorePoolSize(16) .
Do you have a snippet of working code that I can try?
Shimi
On Jun 29, 5:33 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> I see there is a typo in the docs. Wrong name in fluent API. But apart from
> that?
>
> --
>
> On 29 Jun 2010 16:30, "Jonas Bonér" <jo...@jonasboner.com> wrote:
>
> The number of concurrent actors is limited by max thread pool size (if
> blocking queue else min thread pool size).
>
> I have added configuring mailbox to the backlog.
>
> Don't get what the problem with the fluent API is. I use it without
> problems.
>
> --
Shimi
Do you think there is some kind of workaround I can use in the mean
time until this feature will be implemented? Only something that will
limit the number of messages in the queue.
Shimi
On Jun 29, 5:30 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> The number of concurrent actors is limited by max thread pool size (if
> blocking queue else min thread pool size).
>
> I have added configuring mailbox to the backlog.
>
> Don't get what the problem with the fluent API is. I use it without
> problems.
>
> --
> twitter: jboner
Shimi
Is it possible to do some work around using the API?
Shimi
Jonas Bonér
> I have added configuring mailbox to the backlog.Since you know the code much better then me :)
Do you think there is some kind of workaround I can use in the mean
time until this feature will be implemented? Only something that will
limit the number of messages in the queue.
--
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To post to this group, send email to akka...@googlegroups.com.
To unsubscribe from this group, send email to akka-user+...@googlegroups.com.
Jonas Bonér
Shimi
I was hoping there is a clean way to do it with akka API.
On Jun 30, 6:01 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> Track progress here:https://www.assembla.com/spaces/akka/tickets/298-configure-actor-mail...
> >> .
Jonas Bonér
Since it is a val it is hard.
--
Jonas Bonér
http://jayway.com
http://akkasource.com
twitter: jboner
On 30 Jun 2010 19:02, "Shimi" <shi...@gmail.com> wrote:
Thanks.
I was hoping there is a clean way to do it with akka API.
On Jun 30, 6:01 pm, Jonas Bonér <jo...@jonasboner.com> wrote:
> Track progress here:https://www.assembla.com/spaces/akka/tickets/298-configure-actor-mail...
>
> On 30 June 2010 16:56, Jonas Bonér <jo...@jonasboner.com> wrote:
>
>
>
>
>
>
>
> > On 29 June 2010 19:01, Shimi <shim...@gmail.com> wrote:
>
> >> > I have added conf...
> >> akka-user+...@googlegroups.com<akka-user%2Bunsubscribe@googlegroups .com>
> >> .
> >> For more options, visit this group at
> >>http://groups.google.com/group/akka-user?hl=en...
--
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To...