Learn

Learn about latest technology

Build

Unleash your talent and coding!

Share

Let more people use it to improve!
 
Mostrando entradas con la etiqueta akka. Mostrar todas las entradas
Mostrando entradas con la etiqueta akka. Mostrar todas las entradas

Akka an intelligent solution for concurrency

domingo, 10 de marzo de 2019


I am here because we needed a solution to a concurrency problem, access to hundreds of thousands of websites, parse and process  all of them asynchronously. I tried many options always with a main idea: Futures.
So coding in that way I made something similar to the following code:



 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
//....................

//game_1: Int value indicating first game 

//game_n: Int value indicating game n   

val resultMatch:List[Future[List[Match]]] = (game_1 until game_n).map {
 matchnum =>
  Future{
   new ParserFootbalHTML("https://www.anyurl.com/..." + matchnum).finalMatch
    }}.toList
//..............
resultMatch foreach(future => future onComplete {
 Thread.sleep(1)
 doComplete
})
//...............
def doComplete: PartialFunction[Try[List[Match]],Unit] = { //
 case matches @ Success(_) =>{
  val resultMatch:List[Match] = matches.getOrElse(List.empty[Match])
    resultMatch.foreach(matches=>
//Do anything with matches
  )
 }
 case fail @ Failure(_) => println("error")
}
//....................

With minimal changes the main idea was to launch as many "Futures" as web sites we need to parse. Then after every computation finish(onComplete) I will do "anything" with  the result. So after several test in different escenarios checking how long time did it take I decided to explore Akka, for improve performance and reduce time of calculation.
The Core of Akka libraries is the Actor Model, I recommend understand it because is the base of this libraries. Akka has made its own implementation of this model.
Now that I have understood the Actor Model I can have a look to the actor model Akka implementation.



  • Actor and Actor-1 communicate exclusively by exchanging messages asynchronously which are placed into the recipient’s MailBox.
  • Actor and Actor 1 should avoid hogging resources.
  • Message SHOULD NOT be mutable.

ONE important thing from the official documentation:

The only meaningful way for a sender to know whether an interaction was successful is by receiving a business-level acknowledgement message, which is not something Akka could make up on its own (neither are we writing a “do what I mean” framework nor would you want us to).

So as we can see this is the real concept of encapsulation. Every thing happens inside the actor and if you want to delegate a task then It send a message to the actor and keep working. For that reason you ought granulate every task in actor as simple as possible and you should have at the end  that each actor processing be as simple as possible.
Split up the task and delegate it until they become small enough to be handle in One piece!


fig 1.3

As show in fig 1.3 when you invoke the ActorSystem there are several actors that have already been created.
Actor1 is a Supervisor of Actor1.1 and Actor 1.2.

Create only ONE ActorSystem per application is considered a BestPractice because it is a very heavy object and can be create by different ways:
The ActorSystem created can be in the same context or in any other execution context created by us and declared implicitly.
As we can see in fig 1.3 there are different ways of creating actors:
  • Actors that are children of user "guardian" are created by: system.actorOf and in this case  I use to create not many of this kind of actors due to those actors will be the own root actor of my app.
  • Actor that will be child of my own actors(created by us) are created by: context.actorOf , the context is an ActorContext implicit val that there is in the Actor trait.
The ActorContext exposes contextual info for the Actor:
  • self()
  • sender()
  • watch(): used for track actor activities and know when it stopped. Expected a "Terminated"
  • unwatch()
  • actorOf() 
  • become(): change the actor behavior. We decide what message we processed and how. An actor can start processing some type of message and the change its behavior for any reason.
  • actorSelection(actor path): We can select actor for its path (ref. fig 1.3 user/actor1, user/Actor1/Actor1.2)
When we create actors then we get an ActorRef, this is an inmutable reference and identify the actor until it terminate his life. If an actor restart its ActorRef will change.

Terminate(ActorRef  that is being watched, ref watch).  When it happen the Actor free up its resources.

When we have an ActorRef:
  • ! or tell: fire a message and forget.
  • ? or ask: send a message asynchronously and return a future.  
As best practice we should try to send message via ! or tell. [Explain it]

We indicate below the best way to create and invoke different types of actors:

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

object BootNotScheduled extends App with ScalaLogger{

  val moviePageBoundaries: List[Int] = Try(List(args(0).toInt, args(1).toInt)) match {
    case Success(expectedList) => expectedList
    case  Failure(e) => log.error("Error with the boundaries of your page numbers,reviews your parameters {}",e.toString)
      System.exit(1)
      Nil
  }

  // This actor will control the whole System ("IoTDaemon_Not_Scheduled")
  val system = ActorSystem(ConstantsObject.NotScheduledIoTDaemon)

// creating ioTAdmin actor  
val ioTAdmin = system.actorOf(IoTAdmin.props, ConstantsObject.NotScheduledIoTDaemonProcessing)
  val filteredAndOrderedList = moviePageBoundaries.toSet.toList.sorted
  val filteredAndOrderedListGamesBoundaries = (filteredAndOrderedList.head to filteredAndOrderedList.last).toList

// Sending a message to ioTAdmin actor
ioTAdmin!ParseUrlFilms(filteredAndOrderedListGamesBoundaries,Option(ioTAdmin))
}
ref. to BootNotScheduled code in my github

Props are config classes to specify options when you create actors. You can find below an snapshot configuring the actor ioTAdmin previously created in the above code:

........

object IoTAdmin {
  case class ParseUrlFilms(listTitles: List[moviePage], actorRef: Option[ActorRef]=None)
  case object ErrorStopActor
  case object StopProcess
  def props: Props = Props(new IoTAdmin)
}

...

override def receive = {
.....

}
ref. to IoTAdmin code in my github

The way how I deal  with the configuration is similar as used to be done in most libraries or framework or even in java language. I read the config info from application.conf but there are several way for do  that you can find it in Akka config information, it does not explain at all how deal with it.

What do you have to consider when deal with configuration files in akka:
  • Akka configuration values. We will use default values for the moment.
  • Where the configuration file should be saved and how name it.   
  • How the configuration files are loaded or read. [related class/interfaces: ConfigFactory, Config
The configuration files follow the HOCON specification and we can find a very simple examples on github that will let us create config files easily.

The main akka config files use to be application.conf and uses to be placed in the "resource" folder.
In our real life we usually need to work in different environments and some time we create our application.conf during the compilation process  other time we include in our specifics configurations depending on the environments in which we are working on. We have created a real world config example in which you can configure  a production(production.conf) or development(development.conf) environment. This is not the focus of this post  so if you want learn how work, setting and manage config file you can go to my post about how generating distributions for different environments.

We already have minimal information  to create an Akka App only using the actor module  anyway you can have a look to my github repository akka-quickstart-scala and check about how install and launching it(try the first option in your terminal[sbt "AkkaSchedulerPoc/runMain com.ldg.BootNotScheduled 1 5")]

We have talked in previous posts in this blog about the akka libraries, specifically about schedulers but in this case we post about  the first steps in akka and its core, the actor model. We gonna see other features in next posts  like:
Router
Dispatcher
Akka Test Kit
Akka HTTP




Scheduling in Akka with Quartz - a wise solution

lunes, 24 de septiembre de 2018

My problem arise when I needed to execute several task following specific schedule (day/time) indefinitely for long term. This tasks should have different patterns and the schedule will vary definitely in time, the task that we execute today will be different than the task that I am going to execute next week indeed and the same with its schedule.

So I am going to tell you my environment and what I want:
  • A process like linux cron services for running commands following pre-determined schedule.
  • Objects that should be able to process configuration files that contain lists of command lines and when its should be invoked.
  • Objects that should be able to process config files. This config files contain lists of command and when that can be invoked(in time).
  • Some of the aforementioned configurations files must be modified dynamically. That means without stop our application.
Why do not use linux cron + crontable and execute shell scripting files:
  1. For its configuration. 
  2. For the scalability that we need.
  3. For the shell scripting complexity that we ought need (in my real life I used to need data base connections, parse thousands urls, parallel processing on net, etc).
  4. It is our OS configuration task.
All of this in an environment where we need an intensive use of concurrency. So I have to deal with this scenario but akka libraries do not have solution for it. So the are several options on the market, some of them:
  1. Apache Camel Timer
  2. Quartz
  3. Akka-Quartz Scheduler
So in my case I going to use the last one, Akka-Quartz Scheduler, there are several reason, Apache Camel in my opinion is too complicated for just for use his Timer and related to Quartz have the same problem and too oriented to java community, so every kind of listener has to be implemented and we need to work in an akka environment.

I will not tell you here what is better o worst option for my “akka cron environment”. Scheduler and Timers are a very complicated aspects in programming so if you want a right opinion you will need a deep benchmarking between the three of them. It is not the intention of this document.

What are we going to do in our example:
  1. We need to execute several task following specific schedule (day/time) indefinitely until something goes wrong.
  2. We should  change the aforementioned schedule, if we need, for any other. All without stop any of our actors that are busy executing our task.
It is up to you don't change the scheduler with PAST dates.I do not make any validation about the date that you configure in your schedule.

As  Akka-Quartz Scheduler explain, its goal is to provide Akka with a scheduling system that is closer to what one would expect for Cron type jobs.

This is the example that I bring to readers today:

We have a platform whose target is to collect information from a web page that publishes everything related to movies. At the same time I know that every month is published the information about when(in the time) will be added new pages with the info related to new movies. All our code is based on AkkaScheduler module on my github, it is a multi-module project so you can follow the instruction indicated in the repository if you want to deploy it. 

Our project has a very  important aspect:

Configuration files: I have 2 configuration files.
One of them will indicate me that every month I have to do something, this configuration is internal and will be in an internal file. In our case will be in the akka configuration file(application.conf) on my github :

1
2
3
4
5
6
7
8
9
  quartz {
   defaultTimezone = "Europe/London"
   schedules {
    moviepages {
     description = "A cron job that fires off every month"
     expression = "0 30 2 2 * ? *"
    }
   }
  }
code 1.0

See this reference to the CronExpression. Because the info is published every first day in the month I  will check at the beginning of the second day every month. So this is an internal configuration ref: code 1.0 because it never change(I have to review every month for a schedule about the pages related to new films that will be added or updated)

The Second of them will indicate me when will be updated each page that has the information, during the whole month. This will be an external configuration file that indicate when will be updated or added any new page throughout the month.  So that day that the page be updated or added our actor will collect information about it. Every month this external file has to be changed with another new schedule.
An example of my external file (cronmoviepages.confthat you can find on my github is below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
## https://github.com/lightbend/config/blob/master/HOCON.md
## cronexpresssion = "Seconds Minutes Hours Day-of-month Month Day-of-Week Year(Optional)"

schedule {
  defaultTimezone = "Europe/London"
  moviepagemejorenvo = [
    {
      cronexpresssion = "0 4 2 20 9 ? *"
      moviepage = "40"
    }
    {
      cronexpresssion = "0 4 2 20 9 ? *"
      moviepage = "41"
    }
  ]
}
code 1.1

In code 1.1 cronexpression means when the actor has to update the associate page from the movie website. Internally the software will group all moviepage with the same cronexpression and will fire the jobs with each common group.

We have two main cores:
  • For initialize the process BootNotScheduled . In this case we know what pages we want to update so we do not use the schedule.
  • For start the schedule process BootScheduled . This is main program that fire the scheduler. 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
object BootScheduled extends App {
  //https://github.com/lambdista/config
  val system = ActorSystem(ConstantsObject.ScheduledIoTDaemon)
  // This actor will control the whole System ("IoTDaemon_Scheduled")
  val daemonScheduled = system.actorOf(IoTAdmin.props, ConstantsObject.ScheduledIoTDaemonProcessing)
  // This actor will control the reScheduling about time table for update new pages with films in original version
  val reeschedule = system.actorOf(ReSchedulingJob.props, ConstantsObject.DaemonReShedulingVoMoviePages)
  //Use system's dispatcher as ExecutionContext
  //QuartzSchedulerExtension is scoped to that ActorSystem and there will only ever be one instance of it per ActorSystem
  QuartzSchedulerExtension(system).schedule("moviepages", reeschedule, FireSchedule(daemonScheduled))
}
code 1.3

In the previous code 1.3 on line 10 It will use the internal configuration from our application.conf ref. code 1.0. At the same time this line is responsible of every second day of the month at 2.30 am fire a job that will read the schedule a new scheduler  ref. code 1.1(cronmoviepages.conf), the external file that can be read periodically and it indicates what jobs be re-scheduled every time with the new schedule.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
.....................

      cronExpreesionMatches.foreach(
        cronExpreesionMatch => {
          val (cronExpreesion, listOfMoviePages) = cronExpreesionMatch
          log.info(
            "Programming Schedule for cronexpression: {} including matches: {}",
            cronExpreesion,listOfMoviePages.mkString("-")
          )
          val headList = listOfMoviePages.head

          // TODO it is important to kill all Scheduled jobs that has been created ONCE time that the work has done

          /** This code generate a warning because the reschedule job never exist
            * any more because headList used for named it never is the same*/
          QuartzSchedulerExtension(system).rescheduleJob(
            s"Movie-Page-Scheduler$headList",
            scheduledDaemon,
            ParseUrlFilms(listOfMoviePages),
            Option("Scheduling "), cronExpreesion, None, defaultTimezone)
        }
      )
      log.info("schedule get new MoviePages")
  }

.....................
code 1.4

In the previous code 1.4 on line 16 we re-schedule an actor (IoTAdmin) to be fire following the external configuration and then it will launch concurrently so many actors(ProcessMoviePage) as many pages need to be updated. You can appreciate that in the code below.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
........

class IoTAdmin extends Actor with ActorLogging with MessageToTargetActors{
  override def preStart(): Unit = log.info("Start process in Akka Scheduler Example")
  override def postStop(): Unit = log.info("Stop process in Akka Schedule Example")
  // TODO checking perfromance: http://docs.scala-lang.org/overviews/collections/performance-characteristics.html

  var watched = Map.empty[ActorRef, Int]

  override def receive = {

    case ParseUrlFilms(urlPageNunmList, optActorRef) =>

      urlPageNunmList.foreach(
              urlpagenumber => {
                val currentActorRef: ActorRef = context.actorOf(ProcessMoviePage.props)
                watched += currentActorRef -> urlpagenumber
                // watcher for every actor that is created 'cause the actor need to know when the process have finished
                context.watch(currentActorRef)
                // TODO urlspatterns<=varconf
                val processUrlFilmMessage = ProccessPage(
                  s"http://mejorenvo.com/p${urlpagenumber}.html",
                  optActorRef.fold(self)(ref=>ref)
                )
                sendParserUrlMessage(currentActorRef, processUrlFilmMessage)
              })
.............
code 1.5

Every specific day that was configured in cronmoviepages.conf file the IoTAdmin actor will receive a ParseUrlFilms message to process the new page that should be updated.

This post and the project can be a base or skeleton for you if you want to create a process with the following specifications:
  • If you need to execute scheduled task and these tasks occur at a specific moment in time and not need to be executed periodically.
  • If you  need to execute several task following a customizable schedule (day/time), indefinitely until something goes wrong.

This is the idea of our cron but more in deep because some time we need to do more complex thing, external to our OS.
I have used Akka-Quartz Scheduler with docker container and it works pretty well. I have NOT tested the use of this libraries in an Akka Clustering Environment with the complexity that these kind of implementations entail. You can get access to Akka Documentation about  to try of configure seed nodes on any PaaS and run it. You can do it manually but when you are working on a PaaS you need to do it automatically.  The explanation in Akka Library documentation and specifically the reference to Cluster Bootstrap was not working at the time of writing this article. I will try to explain and implement it in next post.