source: doc/theses/colby_parsons_MMath/benchmarks/actors/akka/Repeat/AkkaRepeat.scala

Last change on this file was f945fa7, checked in by Peter A. Buhr <pabuhr@…>, 13 hours ago

fix spelling mistake in directory name

  • Property mode set to 100644
File size: 4.2 KB
Line 
1import akka.actor.typed.scaladsl.Behaviors
2import akka.actor.typed.scaladsl.LoggerOps
3import akka.actor.typed.{ ActorRef, ActorSystem, Behavior }
4import java.util.concurrent.atomic.AtomicInteger
5import akka.event.Logging
6import akka.actor.typed.DispatcherSelector
7import java.util.concurrent.Semaphore
8
9object Server {
10 def apply(): Behavior[Client.MainMessages] = Behaviors.receive { (context, message) =>
11 message match {
12 case Client.CharMsg( value ) =>
13 Client.system ! Client.CharMsg( 'X' )
14 case Client.IntMsg( value ) =>
15 Client.system ! Client.IntMsg( 7 )
16 case Client.State() =>
17 } // message
18 Behaviors.same
19 }
20}
21
22object Client {
23 sealed trait MainMessages
24 final case class State() extends MainMessages
25 final case class IntMsg( val value: Int ) extends MainMessages
26 final case class CharMsg( val value: Char ) extends MainMessages
27
28 var Messages = 100_000; var Times = 100; var Processors = 4; var Factor = 1 // default values
29 var times = 0
30 var results = 0
31
32 val servers = new Array[ActorRef[MainMessages]](Messages)
33
34 val actorCnt = new AtomicInteger(0)
35 val sem = new Semaphore(0)
36
37 var system : ActorSystem[MainMessages] = null
38 var startTime = System.nanoTime()
39
40 def process() = {
41
42 }
43
44 def apply(): Behavior[MainMessages] = Behaviors.setup { context =>
45 for ( id <- 0 until Messages ) { // create actors
46 servers(id) = context.spawn(Server(), "actor_" + id, DispatcherSelector.fromConfig("akka.dispatcher"))
47 } // for
48
49 Behaviors.receiveMessage { message =>
50 message match {
51 case State() =>
52 for ( id <- 0 until Messages ) { // start actors
53 servers(id) ! IntMsg( 0 )
54 servers(id) ! CharMsg( 0 )
55 } // for
56 Behaviors.same
57 case IntMsg( value ) =>
58 results += 1
59 if ( results == 2 * Messages ) {
60 times += 1
61 if ( times == Times ) {
62 sem.release()
63 Behaviors.stopped
64 } else {
65 results = 0
66 system ! State()
67 Behaviors.same
68 }
69 } else
70 Behaviors.same
71 case CharMsg( value ) =>
72 results += 1
73 if ( results == 2 * Messages ) {
74 times += 1
75 if ( times == Times ) {
76 sem.release()
77 Behaviors.stopped
78 } else {
79 results = 0
80 system ! State()
81 Behaviors.same
82 }
83 } else
84 Behaviors.same
85 } // match
86 }
87 }
88
89 def usage() = {
90 println( "Usage: " +
91 s"[ messages (> 0) | 'd' (default ${Messages}) ] " +
92 s"[ processors (> 0) | 'd' (default ${Processors}) ] " +
93 s"[ num times (> 0) | 'd' (default ${Times}) ]"
94 )
95 System.exit( 1 )
96 }
97
98 def main(args: Array[String]): Unit = {
99 if ( args.length > 5 ) usage() // process command-line arguments
100 if ( args.length == 4 ) { // fall through
101 if ( args(3) != "d" ) { // default ?
102 Factor = args(3).toInt
103 if ( Factor < 1 ) usage()
104 } // if
105 } // if
106 if ( args.length >= 3 ) { // fall through
107 if ( args(2) != "d" ) { // default ?
108 Times = args(2).toInt
109 if ( Times < 1 ) usage()
110 } // if
111 } // if
112 if ( args.length >= 2 ) { // fall through
113 if ( args(1) != "d" ) { // default ?
114 Processors = args(1).toInt
115 if ( Processors < 1 ) usage()
116 } // if
117 } // if
118 if ( args.length >= 1 ) { // fall through
119 if ( args(0) != "d" ) { // default ?
120 Messages = args(0).toInt
121 if ( Messages < 1 ) usage()
122 } // if
123 } // if
124
125 Times = Times / Factor
126
127 system = ActorSystem( Client(), "Executor" )
128 system ! State()
129 sem.acquire()
130
131 println( s"${Processors}\t" + f"${(System.nanoTime() - startTime) * Factor / 1_000_000_000.0}%1.2f" )
132 }
133}
134
135// Local Variables: //
136// tab-width: 4 //
137// mode: java //
138// compile-command: "sbt --warn -J-Xmx32g \"run\"" //
139// End: //
Note: See TracBrowser for help on using the repository browser.