Download pdf - Akka - Monadische Komposition im Workflow

[email protected]. Oktober 2014Hendrik Schöneberg

●●●●●


● Digitalisierungszentrum der Universitätsbibliothek Würzburg

Wer

Was

Funktionale Anforderungen

● Retrodigitalisierung● Metadaten-Anreicherung und -Pflege● Export*) und Archivierung

● Qualität maximieren● automatisiertes Qualitätsmanagement● Metadaten gewinnen● direktes Feedback● hohe Fehlertoleranz● hohe Performance (viel IO, hohe CPU-

Lasten für Analysen)

*) http://vb.uni-wuerzburg.de/


Iacobus de Therinis (M.p.th.q.35, ca 1450)http://vb.uni-wuerzburg.de/ub/mpthq35/ueber.html


Defensor Locogiacensis (M.p.th.f.13, ca 750)http://vb.uni-wuerzburg.de/ub/mpthf13/ueber.html


● Digitalisierungszentrum der Universitätsbibliothek Würzburg

Wer

Was

Funktionale Anforderungen

● Retrodigitalisierung● Metadaten-Anreicherung und -Pflege● Export*) und Archivierung

*) http://vb.uni-wuerzburg.de/

● Qualität maximieren● automatisiertes Qualitätsmanagement● Metadaten gewinnen● direktes Feedback● hohe Fehlertoleranz● hohe Performance (viel IO, hohe CPU-

Lasten für Analysen)


●●●●●●

●●●●

➢





...

EIP: “Pipes and Filters”Beispiel (Handschrift)

analysis modules information retrievalmodules

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

QuelleZiel

...Zeitachse


...

EIP: “Pipes and Filters”Beispiel (Handschrift)

analysis modules information retrievalmodules

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

QuelleZiel

Funktionale Anforderungen:● Performance● Robustheit

...Zeitachse


a: AKlassischer Workflow


a: A b: Bf: A → BKlassischer Workflow


a: A b: Bf: A → BKlassischer Workflow

c: Cg: B → C ...


a: A b: Bf: A → B

ma: M<A>

return

Klassischer Workflow

Monadischer Workflow

c: Cg: B → C ...

Monade return → bind → →


a: A b: Bf: A → B

ma: M<A>

return



f’: (M<A> x (A → M)) → M

mb: Mbind

c: Cg: B → C ...



a: A b: Bf: A → B

ma: M<A>

return



f’: (M<A> x (A → M)) → M

mb: Mbind

c: Cg: B → C ...



a: A b: Bf: A → B

ma: M<A>

return



f’: (M<A> x (A → M)) → M

mb: Mbind

returnreturn

g’: (M x (B → M<C>)) → M<C>

ma: M<C>bind

c: Cg: B → C ...

...



...

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

Quelle

Ziel

...Zeitachse

information retrievalmodules

analysis modules


...

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

Quelle

Ziel

...Zeitachse

returnM<A>


analysis modules


...

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

Quelle

Ziel

...Zeitachse

returnM<A>

M

bind


analysis modules


...

exportmodules

blurdetection

angleanalysis

ToCextraction

onlinepublishing

archive

Quelle

Ziel

...Zeitachse

returnM<A>

M

bind


analysis modules

bind

M<X>


●●●●●●


●●●●●●


●●●●●●


●●●●●●


●●●●●●


●●●●●●


●○○

●○

●○○


●○○

●○

●○○





Fork-Join-Executor Threadpool-Executor














●

● …●

○●


●●●●

●●●●



●●●●

●●●●


Quelle: https://www.packtpub.com/books/content/dispatchers-and-routers


Quelle: http://letitcrash.com/post/17607272336/scalability-of-fork-join-pool


Quelle: http://letitcrash.com/post/17607272336/scalability-of-fork-join-pool


// dispatcher for p2p file transferfile-transfer {

executor = "thread-pool-executor"thread-pool-executor {

core-pool-size-min = 1core-pool-size-factor = 0.0core-pool-size-max = 1

}type = Dispatchertroughput = 5000 // give tcp slow start a chance …

}

...remote {

netty.tcp {port = 1337 maximum-frame-size = 10 MiBsend-buffer-size = 5 MiBreceive-buffer-size = 5 MiB

}}

val myActor = context.actorOf(Props[MyActor].withDispatcher("file-transfer"), "myactor2")


// dispatcher for fine-grained parallelism incl. work stealingimage-analysis { type = Dispatcher executor = "fork-join-executor" fork-join-executor { parallelism-min = 4 parallelism-factor = 1 parallelism-max = 16 }}

val myActor = context.actorOf(Props[MyActor].withDispatcher("image-analysis"), "myactor2")

// don’t do that → will run in default dispatcherval dummyActor = context.actorOf(Props[MyActor], "dummyActor")


●○○

●○○

●○

●○


●●●●●


● …val probe = new TestProbe(system) {

def expectUpdate(x: Int) = {expectMsgPF() {case Update(id, _) if id == x => true

}sender() ! "ACK"}

}

val probe = TestProbe()probe.setAutoPilot( new TestActor.AutoPilot { def run(sender: ActorRef, msg: Any): TestActor.AutoPilot = msg match { case "stop" ⇒ TestActor.NoAutoPilot case x ⇒ testActor.tell(x, sender); TestActor.KeepRunning }})


repository.createIndex( config, context ).andThen( doTestQuery( repository, config.getLabel() ), context ).andThen( shutdown(context), context );


repository.createIndex( config, context ).andThen( doTestQuery( repository, config.getLabel() ), context ).andThen( shutdown(context), context );

public Future<Boolean> createIndex(final IndexSetup setup, finalExecutionContextExecutorService ec) {

return future(new Callable<Boolean>(){

@Overridepublic Boolean call() throws Exception{

return createIndex(setup);}

}, ec);}


owner.tell( sr, getSelf() );

monitor.tell( new Update(status) );

worker.tell( PoisonPill.getInstance() );


●●●●


public class TarWorker extends Worker {

...

@Overridepublic void doWork(ActorRef workSender, Object work, final Promise<BoxedUnit> p)

throws Exception {if (work instanceof Tar) {

try {tar((Tar) work, workSender);p.success(BoxedUnit.UNIT);

} catch (Exception e) {p.failure(e);

}}

else {log().error("TarWorker#doWork(): can not compute message {}", work);p.failure(new IllegalStateException("Can not compute message: "

+ work));}

}


●○○

●

●

●

●●

http://letitcrash.com/

http://doc.akka.io/docs/akka/snapshot/scala/howto.html



http://doc.akka.io/docs/akka/snapshot/dev/multi-jvm-testing.html



http://doc.akka.io/docs/akka/snapshot/common/circuitbreaker.html



http://www.manning.com/roestenburg/

http://www.akkaessentials.in/