zakgof / actr Goto Github PK

In ForkJoinPoolScheduler#schedule, requires finally block (line 45):

mailbox.locked.set(false);

Otherwise, it can be skipped due to a runtime error, so a mailbox is locked forever.

#4 is a particular case

Hi, i am reading your project, why do you choose Akka Actor for working as High concurrency, can u tell me the reason, the pros and cons of Actor Akka, why don't you choose another reactive stream like RxJava, Reactive...

Hello,
Nice clean implementation of Actors!
Is there an example of Supervisor Actor -> Child Actor error handling?

Scenario:
A Supervisor is sending parsing commands to multiple Child actors to parse a particular piece of data.
If any child fails, the Supervisor should be aware but not escalate, but all the results should be collated to present a say 90% complete solution.
This could be done by wrapping calls etc, but I would like to know if there is a more elegant way of doing that fits this model.

Any tips or suggestions you have that turn out to be implementable, I'll gladly submit as a PR for inclusion.

Thanks, and great work!

#2 is a particular case

First of all, thanks for creating this library. It's great because it's simple (I like simplicity) and TypeSafe (I took a look at akka, but didn't like it so much because its weak typeness). I'm going to use actr in an upcoming project.

I think it would be great to have future compatibility, for instance:

primeNumberDetectorActor.ask(
        /*action*/ primeNumberDetector -> primeNumberDetector.isNumberPrime(5),
        /*consumer*/ isPrime -> System.out.println("5 is prime? "+isPrime));

If I want to handle success / error states, I have to wrap the action in try - catch:

public static class ActorResult<SUCCESS_TYPE>
{
    public SUCCESS_TYPE successValue = null;
    public Throwable error = null;
}

primeNumberDetectorActor.ask(
    /*action*/ primeNumberDetector -> {
        ActorResult<Boolean> result = new ActorResult<>();

        try {
            result.successValue = primeNumberDetector.isNumberPrime(5);
        } catch (Throwable ex) {
            result.error = ex;
        }

        return result;
    },
    /*consumer*/ result -> {
        if (result.error!=null)
            result.error.printStackTrace();
        else System.out.println("5 is prime? "+result.successValue);
    });

It would be nice to have this wrapping done automatically by the library, and to be able to use futures to handle success / error cases.

In the following example, I'm proposing a method with the following signature:

public <ACTOR, SUCCESS_RESULT_TYPE> CompletableFuture<SUCCESS_RESULT_TYPE> askAnsweringFuture(Function<ACTOR, SUCCESS_RESULT_TYPE> action);

Usage:

primeNumberDetectorActor.askAnsweringFuture(primeNumberDetector -> primeNumberDetector.isNumberPrime(5))
                .thenApply(isPrime -> System.out.println("5 is prime? "+isPrime))
                .exceptionally(ex -> ex.printStackTrace());

Since I'm going to use Kotlin, I already made a similar extension method. But since Java doesn't have extension methods, I think this would be a nice feature.

I really appreciate your idea of pojo classes. It makes it easier to integrate with existing applications.

It is really interesting to have an actor system that can communicate with the external world using standard protocol.

Did you already thing of building module for

• rest server (spark/jetty integration for example)
• graphql
• database persistence actors(redis, sql, mongo, ...)

And example to show how it could be used. What about creating predefined actor/connector to help better understand the power of actr.
Like:

• PersistenceRedisActr(redis connector with some fonctions),
• PersistenceHibernateActr(works with hibernate objects)
• RestApiActr(handling http api request)
• GraphqlApiActr
• WebsocketActr
• EmailActr
• ...

Each of them with own ExecutorServive and a parameter to increase the number of threads

When certain single actr or a set of actrs receive messages with a high rate, thread starvation is highly possible.
This will happen in the loop (lines 36-44):

			for(;;) {
				Runnable runnable = mailbox.queue.peek();
				if (runnable != null) {
					...
					runnable.run();
					mailbox.queue.remove();
				} else

When the message rate is high, a certain 'hot' thread wan't 'yield` the execution to others, that can cause a starvation.

In Akka, this is solved with the throughput parameter, that cops the contents of the mailbox into pieces (batches).

ActorSystem#shutdown() can miss to destroy the actors if they are created 'in-between'

here is my code example
a class handling http request with spark java

public class Http {
    void start() {
        Service app = Service.ignite().port(82);
        app.get("/", (req, res)-> {
            System.out.println("request from web: "+Instant.now(Clock.systemUTC()).toString());
            return "hello ..."+new Date().toString();
        });
        System.out.println("start spark server on port :"+app.port());
    }
}

and my main method

 public static void main(String[] args) {
        //final IActorSystem systemBlocking = Actr.newSystem("blocking-System", Schedulers.newThreadPerActorScheduler());
        final IActorSystem system = Actr.newSystem("Example-System");
        final IActorRef<Printer> printerActor1 = system.actorOf(Printer::new, "Printer 1");
        final IActorRef<Printer> printerActor2 = system.actorOf(Printer::new, "Printer 2");
        final IActorRef<Http> server = system.actorOf(Http::new, "Printer 2");
        printerActor1.tell(s -> s.print("first hello"));
        // infinite works
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            printerActor2.tell(s -> s.print("waoh ...."+ Instant.now()));
        }, 1, 5, TimeUnit.SECONDS);
        // start server
        Executors.newFixedThreadPool(5).execute(()-> server.tell(s -> s.start()));
        printerActor2.tell(s -> s.print("second hello"));
        system.shutdownCompletable().join();
        //system.shutdownCompletable().join();
    }

this code is working as expected with this pom

<dependency>
            <groupId>com.github.zakgof</groupId>
            <artifactId>actr</artifactId>
            <version>0.4.0</version>
</dependency>
<dependency>
            <groupId>com.sparkjava</groupId>
            <artifactId>spark-core</artifactId>
            <version>2.9.3</version>
</dependency>

does it make sense ?
how can i start different Actorsystem in parallel ? my ideas was to create a different system for handling http request but i didn't works if i just call shutdownCompletble() twice

In ForkJoinPoolScheduler#schedule(..), an actor can intermittently skip messages.

Possible scenarios (others possible):
1.

• The single actr created;
• Thread A adds a runnable to the actr's mailbox;
• Thread A locks the mailbox and runs the runnable;
• Thread B adds a runnable, the actor's mailbox is still locked, so the 'runnable.run()' is skipped.
• Thread A unlocks the mailbox;

• The single actr created;
• Thread A adds a runnable to the actr's mailbox;
• Thread A locks the malebox and runs the runnable;
• Thread B adds a runnable to the same mailbox;
• Thread A removes the last runnable (from thread B) without execution.

The result: The message (runnable) from thread B is lost.

after adding this dependency to my mvn project

<dependency>
            <groupId>com.github.zakgof</groupId>
            <artifactId>actr</artifactId>
            <version>0.4.0</version>
</dependency>

i get thie error.

Could not find artifact com.github.zakgof:actr:pom:0.4.0 in central (https://repo.maven.apache.org/maven2)

does anyone has an ideas why. it seems like it is no more awailaible in maven repo

Use slf4j or something

DedicatedThreadScheduler#schedule(..) can schedule a task on the executor, that is being shut down, that will cause a runtime error.

Unsafe publication in ActorImpl constructor.
Instance internals exposed with the inconsistent state to the outer scope.
The 'outer' consumer could see the ActorImpl when its construction is not complete.
Will produce different kinds of 'quantum' effects in the multi-threaded application.

ActorImpl(...) {
    ...
    Actr.setCurrent(this); //Fixme: Unsafe publication!
    ...
    actorSystem.add(this); //Fixme: Unsafe publication!
}

In ForkJoinPoolScheduler#schedule(..), an actor can intermittently duplicate messages.

A possible scenario (one of ?):

• The single actr created;
• Thread A adds a runnable to the actr's mailbox;
• Thread A locks the mailbox and runs the runnable;
• Thread B adds a runnable to the same mailbox;
• Thread A removes the last runnable from thread B (which is lost !!!);
• The mailbox is still not empty, so the first runnable is executed again;