Problem with unique name when reusing actor

[ccirim]

Hi

I have a strange problem when reuse an actor, (the error is random!)
This is my scenario:
- send 40 async request (messages) to my principal actor.
- principal actor send every single request (message) to 10 secondary actors.
- secondary actors reply to an assembler actor who collect the results and send the final result.

I resuse every time same actor in this way:

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else
        return getContext().actorOf(SpringExtProvider.get(getContext().system()).props(actorName).withRouter(new SmallestMailboxRouter(4)), actorName);
}

(this is "like" store actors in map or cache, its correct?!)

and this is the error that return to me in random way at execution time:

[ERROR] [02/03/2014 18:40:19.248] [OrchestrationApplication-akka.actor.default-dispatcher-2] [akka.dispatch.Dispatcher] actor name [actorName1] is not unique!

akka.actor.InvalidActorNameException: actor name [actorName1] is not unique!     

at akka.actor.dungeon.ChildrenContainer$NormalChildrenContainer.reserve(ChildrenContainer.scala:130)

...

[ERROR] [02/03/2014 18:40:19.248] [OrchestrationApplication-akka.actor.default-dispatcher-2] [akka.dispatch.Dispatcher] actor name [actorName2] is not unique!

akka.actor.InvalidActorNameException: actor name [actorName2] is not unique!     

at akka.actor.dungeon.ChildrenContainer$NormalChildrenContainer.reserve(ChildrenContainer.scala:130)

...

etc etc...

why this error?  if the actor is defines (also actor name) why system try to recreate actor?

I think that if an actor dead or is stopped also his name is removed of actopr system, or not?

ps:

my dispatcher is default dispatcher, and I have a 4 core processor (i5)
teh default config for dispatcher is:

 fork-join-executor {
    parallelism-min = 8
    parallelism-factor = 3.0
    parallelism-max = 64
  }
  throughput = 10

its correct?


tnx!
carlo

[√iktor Ҡlang]

Please minimize the code to reproduce, and what version are you using?

--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ: http://akka.io/faq/
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/groups/opt_out.

--

Cheers,

√

———————

Viktor Klang

Chief Architect - Typesafe

Twitter: @viktorklang

[ccirim]

OK!
code sample:
..............

        for(String nomeName : listActorName){
               
                final ActorRef actor = getActor(nomeName);           
                       
                Patterns.ask(actor, inputMessage, Conf.sectionTimeout).onComplete(new OnComplete<Object>() {

                    @Override
                    public void onComplete(Throwable t, Object result) throws Throwable {
                      
                        if(t != null){
                            do something!....
                        }
                        else{
                            do something else!....
                        }
                       
                    }

                }, getContext().dispatcher());
               
            }

................

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else
        return getContext().actorOf(

SpringExtProvider.get(getContext().system()).props(actorName).withRouter(new SmallestMailboxRouter(4)), actorName);
}

........

Ciao
Carlo

[√iktor Ҡlang]

Hi Carlo,

Can you reproduce without the SpringExt etc (i.e. is this fully minimized or not)?

[ccirim]

        for(String nomeName : listActorName){
               
                final ActorRef actor = getActor(actorName);           
                       
                Patterns.ask(actor, inputMessage, Conf.sectionTimeout).

onComplete(new OnComplete<Object>() {

                    @Override
                    public void onComplete(Throwable t, Object result) throws Throwable {
                      
                        if(t != null){
                            do something!....
                        }
                        else{
                            do something else!....
                        }
                       
                    }

                }, getContext().dispatcher());
               
            }

................

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else

        return getContext().actorOf(Props.create(MyServiceActor.class).withRouter(new SmallestMailboxRouter(5)), actorName);

}

........

ok?

tnx!
carlo

[√iktor Ҡlang]

On Wed, Feb 5, 2014 at 1:22 PM, ccirim <carlo.c...@gmail.com> wrote:

        for(String nomeName : listActorName){

What is a listActorName and what does it contain?

 

               
                final ActorRef actor = getActor(actorName);           
                       
                Patterns.ask(actor, inputMessage, Conf.sectionTimeout).

What is an inputMessage and how long is a sectionTimeout?

 

onComplete(new OnComplete<Object>() {

                    @Override
                    public void onComplete(Throwable t, Object result) throws Throwable {
                      
                        if(t != null){
                            do something!....
                        }
                        else{
                            do something else!....
                        }

Why does this section need to do an if-check on the "t" for this problem to manifest?

 

                       
                    }

                }, getContext().dispatcher());
               
            }

................

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else

        return getContext().actorOf(Props.create(MyServiceActor.class).withRouter(new SmallestMailboxRouter(5)), actorName);

Does this require a MyServiceActor for the problem to manifest and what does a MyServiceActor do?

Does this also require a SmallestMailboxRouter or even a Router at all to manifest?

[Oleg Zhurakousky]

Carlo

Can you please include the rest of the stack trace that follows:

at akka.actor.dungeon.ChildrenContainer$NormalChildrenContainer.reserve(ChildrenContainer.scala:130)

I want to see the origin of it and since you have "do something. . .' comments I am not sure if there is some other code that may attempt to create actors. In other words the only way I was able to reproduce this is by attempting to create an Actor with the same name, which brings me to another question. If you remove Patterns.ask(actor, inputMessage...) block all together, do you still see the problem?

Cheers

Oleg

[ccirim]

in text....

Il giorno mercoledì 5 febbraio 2014 13:38:28 UTC+1, √ ha scritto:

On Wed, Feb 5, 2014 at 1:22 PM, ccirim <carlo.c...@gmail.com> wrote:

        for(String nomeName : listActorName){

What is a listActorName and what does it contain?

is the list of actors'name who execute a single external job (external service)

 

 

               
                final ActorRef actor = getActor(actorName);           
                       
                Patterns.ask(actor, inputMessage, Conf.sectionTimeout).

What is an inputMessage and how long is a sectionTimeout?

The input message contains the parameters for execute external job, time out = 300 s
 

 

onComplete(new OnComplete<Object>() {

                    @Override
                    public void onComplete(Throwable t, Object result) throws Throwable {
                      
                        if(t != null){
                            do something!....
                        }
                        else{
                            do something else!....
                        }

Why does this section need to do an if-check on the "t" for this problem to manifest?

there are logic to say if a result job is necessary or not. after this result is send to aggregator.
 

 

                       
                    }

                }, getContext().dispatcher());
               
            }

................

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else

        return getContext().actorOf(Props.create(MyServiceActor.class).withRouter(new SmallestMailboxRouter(5)), actorName);

Does this require a MyServiceActor for the problem to manifest and what does a MyServiceActor do?

Does this also require a SmallestMailboxRouter or even a Router at all to manifest?

sorry I have write wrong my code....

MyServiceActor.class is

MyServiceActor1.class
MyServiceActor2.class
MyServiceActor3.class
MyServiceActor4.class
....

one for every external service that need to run...

sample explain of my application:
my application compose same books (Principal Actor).
every book (P.A.) call an actor (secondary actor) that produce a single chapter.
... aggregator collect chapters and return complete book
I reuse secondary actor for create chapters.

[√iktor Ҡlang]

Please just send a minimized (no code that isn't absolutely required to reproduce) and self-contained, context-free example to reproduce.

[ccirim]

this is the original code:

        for(String sectionName : sectionListName){

                final ActorRef sectionActor = getActor(sectionName);                       
                       
                Patterns.ask(sectionActor, inputSectionMessage, 300s).onComplete(new OnComplete<Object>() {

                    @Override
                    public void onComplete(Throwable t, Object result) throws Throwable {
                       

                        SectionOutput<?> section = null;
                       
                        if(t != null){
                            String nomeWrapper = sectionActor.path().name();
                            section = new SectionOutput<Object>();
                            section.setName(nomeWrapper);
                            section.setError(t.toString());
                            section.setExecutionTime(Conf.sectionTimeout.duration().toMillis());
                        }
                        else{
                            section = (SectionOutput<?>) result;
                        }
                       
                        //... send section to aggregator
                       
                    }

                }, getContext().dispatcher());

               
            }

....

private ActorRef getActor(String actorName) {
    Option<ActorRef> child = getContext().child(actorName);
    if(child.isDefined())
        return child.get();
    else

        //return getContext().actorOf(Props.create(MyServiceActor.class).withRouter(new SmallestMailboxRouter(4)), actorName);
          return getContext().actorOf(SpringExtProvider.get(getContext().system()).props(nomeActor).withRouter(new SmallestMailboxRouter (Conf.NUMBER_ROUTERS)), nomeActor);


Conf.NUMBER_ROUTERS = 4


and this the ST:

"ngb.ProtestiWrapper" is one of  list actor name,

[ERROR] [02/05/2014 15:23:46.102] [OrchestrationApplication-akka.actor.default-dispatcher-12] [akka.dispatch.Dispatcher] actor name [ngb.ProtestiWrapper] is not unique!
akka.actor.InvalidActorNameException: actor name [ngb.ProtestiWrapper] is not unique!
    at akka.actor.dungeon.ChildrenContainer$NormalChildrenContainer.reserve(ChildrenContainer.scala:130)
    at akka.actor.dungeon.Children$class.reserveChild(Children.scala:77)
    at akka.actor.ActorCell.reserveChild(ActorCell.scala:338)
    at akka.actor.dungeon.Children$class.makeChild(Children.scala:186)
    at akka.actor.dungeon.Children$class.actorOf(Children.scala:38)
    at akka.actor.ActorCell.actorOf(ActorCell.scala:338)
    at com.cervedgroup.orchidea.orchestrator.actor.engine.DocumentActor.getActor(DocumentActor.java:377)
    at com.cervedgroup.orchidea.orchestrator.actor.engine.DocumentActor.executeSezioni(DocumentActor.java:181)
    at com.cervedgroup.orchidea.orchestrator.actor.engine.DocumentActor.access$2(DocumentActor.java:171)
    at com.cervedgroup.orchidea.orchestrator.actor.engine.DocumentActor$1.onComplete(DocumentActor.java:149)
    at akka.dispatch.OnComplete.internal(Future.scala:247)
    at akka.dispatch.OnComplete.internal(Future.scala:244)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:174)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:171)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:29)
    at akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.processBatch$1(BatchingExecutor.scala:67)
    at akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:82)
    at akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
    at akka.dispatch.BatchingExecutor$Batch$$anonfun$run$1.apply(BatchingExecutor.scala:59)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72)
    at akka.dispatch.BatchingExecutor$Batch.run(BatchingExecutor.scala:58)
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:42)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)
    at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
    at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
    at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
    at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

[ccirim]

MyServiceActor1.class = ProtestiWrapper.class
MyServiceActor2.class = CaseWrapper.class
MyServiceActor3.class = OfferteWrapper.class
MyServiceActor4.class = RegaliWrapper.class
etc etc

name of ProtestiWrapper.class is ngb.ProtestiWrapper

public class ProtestiWrapper extends UntypedActor

[Oleg Zhurakousky]

What version of Akka you are using?

Looking at your stack trace I am puzzled with this:

         at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)

I can't seem to find method AkkaForkJoinTask.exec(..) which takes any input parameters.

Oleg

[ccirim]

I use 2.2.3

C:\DR\com\typesafe\akka\akka-actor_2.10\2.2.3\akka-actor_2.10-2.2.3.jar
C:\DR\org\scala-lang\scala-library\2.10.1\scala-library-2.10.1.jar
C:\DR\com\typesafe\config\1.0.2\config-1.0.2.jar

[ccirim]

I use 2.2.3

com\typesafe\akka\akka-actor_2.10\2.2.3\akka-actor_2.10-2.2.3.jar

org\scala-lang\scala-library\2.10.1\scala-library-2.10.1.jar

com\typesafe\config\1.0.2\config-1.0.2.jar

ciao
carlo

[Oleg Zhurakousky]

OK, so I am confused now. 

Can you post a fully functional sample with configuration which reproduces the issue as Victor suggested earlier.

Oleg

--

[Oleg Zhurakousky]

Sorry, I've misinterpreted the stack trace.

Anyway, it would be nice if you were to post "fully functional sample with configuration which reproduces the issue" as I suspect the bits and pieces that we see do not represent the whole picture.

The good news is I figured it out, or at least I was able to reproduce it with the stack trace resembling yours.

The issue is here (my stack trace with 2.2.3 now looks the same as yours):

     . . .

     at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:42)
     at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386)

     . . .

The above implies that processing is delegated to a separate thread within the actor (e.g., invocation of getActor() is done on a separate thread) and therefore creates a classic race condition where two+ threads trying to create actor with the same name before one knows that its already been created.

Anyway, as I said while I was able to reproduce it, no one can't tell you for sure what's wrong with your code until we can see it.

Cheers

Oleg

[ccirim]

OK! I try to reassume my application... It's a big application! :D
As soon as send you my demo application classes.

[Patrik Nordwall]

I think Oleg has found the issue. You probably access the actor context from a Future callback, or a scheduled Runnable, within the actor.

/Patrik

Patrik Nordwall
Typesafe -  Reactive apps on the JVM
Twitter: @patriknw

[ccirim]

Oh YES! I have found my error! I call a method that call getActor in Future callback...
Tnx so much for all support and congratulation for all yours professionality and competence! :)

Ciao ciao!
Carlo

[Roland Kuhn]

You’re welcome, and happy hakking!

Dr. Roland Kuhn
Akka Tech Lead
Typesafe – Reactive apps on the JVM.
twitter: @rolandkuhn

[Krishna Kadam]

Hi All, 

 I got the same problem while reusing the akka actors by their names, you have mentioned accessing of Future object returned by using getSender.tell() method. I understood the cause of problem, but you have not mentioned solution for this problem. 

1. What configuration should be used to avoid this problem?

2. Is there any other way to avoid this problem?

Thanks & Regards

Krishna Kadam

[Akka Team]

Hi,

On Wed, Apr 1, 2015 at 8:12 AM, Krishna Kadam <shrikrish...@gmail.com> wrote:

Hi All, 

 I got the same problem while reusing the akka actors by their names, you have mentioned accessing of Future object returned by using getSender.tell() method. I understood the cause of problem, but you have not mentioned solution for this problem. 

The solution is that you should not close over mutable state in Future callbacks. The sender() call accesses a mutable field. The solution is simply to not do that.

 

1. What configuration should be used to avoid this problem?

This is not related to any configuration, this is a user bug.

 

2. Is there any other way to avoid this problem?

Don't close over mutable state in asynchronous callbacks.

-Endre

 

Thanks & Regards

Krishna Kadam

--

>>>>>>>>>> Read the docs: http://akka.io/docs/

>>>>>>>>>> Check the FAQ: http://doc.akka.io/docs/akka/current/additional/faq.html

>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.

For more options, visit https://groups.google.com/d/optout.

--

Akka Team

Typesafe - Reactive apps on the JVM

Blog: letitcrash.com
Twitter: @akkateam