Akkaによるスレッドの使い方

Transcript

1. "LLBʹΑΔ
   εϨουͷ࢖͍ํ ϦνϟʔυɹҏਅԬ
2. "LLBͱ͍͑͹
3. https://akka.io/
4. https://akka.io/
5. > Akka is a toolkit for building highly concurrent, distributed, and resilient message-driven applications for Java and Scala
6. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling
7. શ෦ؾʹ͠ͳ͕Β
   ίʔυΛॻ͍͍ͯͩ͘͞
8. ແཧήʔ
9. ͕ॿ͚ͯ͘Ε·͢
10. "DUPS
    ʹΑΔ
    ϝοηʔδύογϯά 
    "LLB
11. import scala.collection class MyActor extends Actor { //mutable state val state1: mutable.Map[String, String] = … val state2: mutable.ListBuffer[String] = … def receive = { //immutable message case MyMessage(msg) => procedure(msg) } def procedure(msg: String) { //࣍ͷϝοηʔδ͔ΒׂΓࠐ·Εͳ͍ } }
12. import scala.collection class MyActor extends Actor { //mutable state val state1: mutable.Map[String, String] = … val state2: mutable.ListBuffer[String] = … def receive = { //immutable message case MyMessage(msg) => procedure(msg) } def procedure(msg: String) { //࣍ͷϝοηʔδ͔ΒׂΓࠐ·Εͳ͍ } }
13. import scala.collection class MyActor extends Actor { //mutable state val state1: mutable.Map[String, String] = … val state2: mutable.ListBuffer[String] = … def receive = { //immutable message case MyMessage(msg) => procedure(msg) } def procedure(msg: String) { //࣍ͷϝοηʔδ͔ΒׂΓࠐ·Εͳ͍ } }
14. import scala.collection class MyActor extends Actor { //mutable state val state1: mutable.Map[String, String] = … val state2: mutable.ListBuffer[String] = … def receive = { //immutable message case MyMessage(msg) => procedure(msg) } def procedure(msg: String) { //࣍ͷϝοηʔδ͔ΒׂΓࠐ·Εͳ͍ } }
15. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling ๨Ε͍͍ͯ
16. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling ๨Ε͍͍ͯ $PODVSSFODZಛ༗ͷ໰୊͸Ͳ͏ͯ͠΋
    ى͜Γ·͢
17. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling ๨Ε͍͍ͯ Ұ෦ͷ௿Ϩϕϧͳॲཧ͔Β͸
    ղ์͞ΕΔ
18. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling "LLB͕ड͚࣋ͬͯ͘ΕΔ Ұ෦ͷ௿Ϩϕϧͳॲཧ͔Β͸
    ղ์͞ΕΔ
19. ͋͑ͯ
20. •Thread •join •Runnable/Callable •Lock •synchronized •Atomic variables •volatile •ThreadLocal •DynamicVariable •ConcurrentHashMap •ConcurrentLinkedQueue •ForkJoin •ThreadPool •Executor/ExecutorService •ExecutionContext •Future •Thread intereference •Memory Consistency error •Dead lock •Live lock •Starvation •InterruptedException •Scheduling ͦͷ෦෼Λݟ͍͖ͯ·͠ΐ͏ "LLB͕ड͚࣋ͬͯ͘ΕΔ
21. Կ͔஌ݟ͕͋Δ͔΋͠Εͳ͍
22. ͱ͍͏ͷ͕ࠓ೔ͷτʔΫ
23. ͭ·Γ
25. ! ϝιουΛݺͿͱ ͭ·Γ Կ͕ى͜Δͷ͔
26. ·ͣ͸TFOEFSଆͷ࿩
28. ͳΔ΄ͲΘ͔ΒΜ
29. ҰาͣͭίʔυΛ௥͍͖ͬͯ·͢
30. class LocalActorRef override def !(message: Any) (implicit sender: ActorRef = Actor.noSender): Unit = actorCell.sendMessage(message, sender)
31. class LocalActorRef override def !(message: Any) (implicit sender: ActorRef = Actor.noSender): Unit = actorCell.sendMessage(message, sender)
32. trait Dispatch def sendMessage(msg: Envelope): Unit = try { val msgToDispatch = if (system.settings.SerializeAllMessages) serializeAndDeserialize(msg) else msg dispatcher.dispatch(this, msgToDispatch) } catch handleException
33. trait Dispatch def sendMessage(msg: Envelope): Unit = try { val msgToDispatch = if (system.settings.SerializeAllMessages) serializeAndDeserialize(msg) else msg dispatcher.dispatch(this, msgToDispatch) } catch handleException
34. Dispatcher
35. EJTQBUDI
    
    
    ʤॻྨɾՙ෺ͳͲΛʥૹΔɺ
    ૹΓग़͢ɺൃૹ͢Δɺٸૹ͢Δ https://eow.alc.co.jp/search?q=dispatch εϖʔεΞϧΫ
36. EJTQBUDIFS
    
    㲈
    TFOEFS
    
    https://eow.alc.co.jp/search?q=dispatch εϖʔεΞϧΫ
37. ͪΐͬͱ͕͍ͪ·͢
38. TFOEFS ૹ৴ݩ ͱ͍͏ΑΓ͸
    %JTQBUDIFSΛ௨ա
39. Dispatcher
40. ΋͏Ұͭେࣄͳಛ௃
41. Dispatcher
42. trait Dispatch def sendMessage(msg: Envelope): Unit = try { val msgToDispatch = if (system.settings.SerializeAllMessages) serializeAndDeserialize(msg) else msg dispatcher.dispatch(this, msgToDispatch) } catch handleException
43. class Dispatcher def dispatch( receiver: ActorCell, invocation: Envelope ): Unit = { val mbox = receiver.mailbox mbox.enqueue(receiver.self, invocation) registerForExecution(mbox, true, false) }
44. class Dispatcher def dispatch( receiver: ActorCell, invocation: Envelope ): Unit = { val mbox = receiver.mailbox mbox.enqueue(receiver.self, invocation) registerForExecution(mbox, true, false) }
45. abstract class Mailbox def enqueue( receiver: ActorRef, msg : Envelope ): Unit = messageQueue.enqueue(receiver, msg)
46. abstract class Mailbox def enqueue( receiver: ActorRef, msg : Envelope ): Unit = messageQueue.enqueue(receiver, msg)
47. MessageQueue object UnboundedMailbox { class MessageQueue extends ConcurrentLinkedQueue[Envelope] with UnboundedQueueBasedMessageQueue { final def queue: Queue[Envelope] = this } }
48. MessageQueue object UnboundedMailbox { class MessageQueue extends ConcurrentLinkedQueue[Envelope] with UnboundedQueueBasedMessageQueue { final def queue: Queue[Envelope] = this } }
49. Class ConcurrentLinkedQueue<E>
50. Class ConcurrentLinkedQueue<E>
51. http://www.cs.rochester.edu/~scott/papers/ 1996_PODC_queues.pdf
52. http://www.cs.rochester.edu/~scott/papers/ 1996_PODC_queues.pdf
53. /PO#MPDLJOH
57. -PPQ
    
    $"4
    $PNQBSF
    BOE
    4XBQ
58. $"4
    DPNQBSF
    BOE
    TXBQ
    BUPNJDͳૢ࡞
60. ૢ࡞͕੒ޭ͢Δ·Ͱ
    $"4
    ϦτϥΠ
61. ͳΔ΄Ͳ
    /PO#MPDLJOH
63. sender/receiver͔Β
    concurrentʹΞΫηε
64. ଟ਺ͷsender͔Β
    concurrentʹΞΫηε
65. https://doc.akka.io/docs/akka/2.5/mailboxes.html SingleConsumerOnlyUnboundedMailbox This queue may or may not be faster than the default one depending on your use-case— be sure to benchmark properly! NonBlockingBoundedMailbox Backed by a very efficient Multiple-Producer Single-Consumer queue …etc ଞʹ΋MessageQueue৭ʑ͋Γ·͢
66. MessageQueue ʹೖΕͨ͋ͱͷ࿩
67. class Dispatcher def dispatch( receiver: ActorCell, invocation: Envelope ): Unit = { val mbox = receiver.mailbox mbox.enqueue(receiver.self, invocation) registerForExecution(mbox, true, false) }
68. class Dispatcher def dispatch( receiver: ActorCell, invocation: Envelope ): Unit = { val mbox = receiver.mailbox mbox.enqueue(receiver.self, invocation) registerForExecution(mbox, true, false) }
69. class Dispatcher override def registerForExecution( mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean ): Boolean = { if (mbox.canBeScheduledForExecution( hasMessageHint, hasSystemMessageHint)) { if (mbox.setAsScheduled()) { … executorService execute mbox … … }
70. class Dispatcher override def registerForExecution( mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean ): Boolean = { if (mbox.canBeScheduledForExecution( hasMessageHint, hasSystemMessageHint)) { if (mbox.setAsScheduled()) { … executorService execute mbox … … }
71. ͜͜Ͱ;Γ͔͑Γ
72. Dispatcher
73. ExecutorService
74. Java
    ͔Β
    ThreadΑΓந৅౓ͷߴ͍ඇಉظ༻JOUFSGBDF ExecutorService
75. &YFDVUPS4FSWJDF
76. class Dispatcher override def registerForExecution( mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean ): Boolean = { if (mbox.canBeScheduledForExecution( hasMessageHint, hasSystemMessageHint)) { if (mbox.setAsScheduled()) { … executorService execute mbox … … }
77. abstract class Mailbox abstract class Mailbox( val messageQueue: MessageQueue ) extends ForkJoinTask[Unit] with SystemMessageQueue with Runnable { ... }
79. ͜͜·Ͱ͸senderଆ
80. ͔͜͜Β͸receiverଆ
81. class Dispatcher override def registerForExecution( mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean ): Boolean = { if (mbox.canBeScheduledForExecution( hasMessageHint, hasSystemMessageHint)) { if (mbox.setAsScheduled()) { … executorService execute mbox … … }
82. abstract class Mailbox abstract class Mailbox( val messageQueue: MessageQueue ) extends ForkJoinTask[Unit] with SystemMessageQueue with Runnable { ... }
83. ForkJoinTask
84. ForkJoinTask join() join() fork() fork()
85. ForkJoinTask join() join() fork() fork() Α͘ݟΔਤ
86. ForkJoinTask join() join() fork() fork()
87. Fork΋Join΋͠ͳ͍
    ForkJoinTask
88. Fork΋Join΋͠ͳ͍ͳΒ
    ͳΜͰForkJoinTaskʁ
89. ౴͑
    ύϑΥʔϚϯε
90. Class ForkJoinTask<V>
91. Class ForkJoinTask<V>
92. Class ForkJoinPool
93. Class ForkJoinPool
94. AkkaͷچΦϑΟγϟϧϒϩά LET IT CRASH blog
95. Scalability of Fork Join Pool LET IT CRASH blog 2012
96. LET IT CRASH blog 2012
97. ʮAkkaͷϕϯνϚʔΫऔΖ͏ͥʯ LET IT CRASH blog 2012
98. ʮ΋͏Ϛγϯങͬͨʯ LET IT CRASH blog 2012
99. LET IT CRASH blog 2012
100. LET IT CRASH blog 2012 ʮͰ͔ͬʂ$16ίΞ͍ͭ͋͘Δͷʁʯ
101. ʮʯ LET IT CRASH blog 2012
102. ʮ
     ",#͔ΑXXXʯ LET IT CRASH blog 2012
103. ʮϕϯν૸ΒͤΔΑʔʂʂʯ LET IT CRASH blog 2012 ·Ͱݟࣄʹεέʔϧ͢ΔͰ͠ΐ
104. LET IT CRASH blog 2012
105. LET IT CRASH blog 2012 ʂʁʁʁ
106. LET IT CRASH blog 2012 ·Ͱεέʔϧ͠ͳ͍ͷʂʁʁ
107. LET IT CRASH blog 2012
108. LET IT CRASH blog 2012
109. LET IT CRASH blog 2012
110. LET IT CRASH blog 2012 7JLUPS
     ݱ%FQVUZ ෭ʁ
     $50
     BU
     -JHIUCFOE
111. LET IT CRASH blog 2012
112. LET IT CRASH blog 2012 %PVH
     -FB
113. https://en.wikipedia.org/wiki/Doug_Lea
114. https://en.wikipedia.org/wiki/Doug_Lea
115. +BWBͷGPSL
     KPJO
     ࡞ͬͨਓ
116. LET IT CRASH blog 2012 … resulted in improved implementation of fork join pool … randomized queing and stealing
117. LET IT CRASH blog 2012
118. LET IT CRASH blog 2012 ۙ͘·Ͱ
     ΄ͲΑ͘εέʔϧ
119. ͔ͩΒForkJoinPool͕
     "LLB
     DispatcherͷσϑΥϧτ
120. receiverଆͷ
     ࿩͠ʹ໭Γ·͢
121. abstract class Mailbox abstract class Mailbox( val messageQueue: MessageQueue ) extends ForkJoinTask[Unit] with SystemMessageQueue with Runnable { ... } ExecutorServiceͷ
     executeϝιουΛ࢖͏ͱ
     Runnableͷrunϝιου͕ݺ͹ΕΔ
     
122. abstract class Mailbox override final def run(): Unit = { try { if (!isClosed) { //Volatile read, needed here processAllSystemMessages() //First, deal with any system messages processMailbox() //Then deal with messages } } finally { setAsIdle() //Volatile write, needed here dispatcher.registerForExecution(this, false, false) } }
123. abstract class Mailbox override final def run(): Unit = { try { if (!isClosed) { //Volatile read, needed here processAllSystemMessages() //First, deal with any system messages processMailbox() //Then deal with messages } } finally { setAsIdle() //Volatile write, needed here dispatcher.registerForExecution(this, false, false) } }
124. abstract class Mailbox @tailrec private final def processMailbox( left: Int = java.lang.Math.max(dispatcher.throughput, 1), deadlineNs: Long = …): Unit = if (shouldProcessMessage) { val next = dequeue() if (next ne null) { … actor invoke next … processMailbox(left - 1, deadlineNs) } }
125. abstract class Mailbox @tailrec private final def processMailbox( left: Int = java.lang.Math.max(dispatcher.throughput, 1), deadlineNs: Long = …): Unit = if (shouldProcessMessage) { val next = dequeue() if (next ne null) { … actor invoke next … processMailbox(left - 1, deadlineNs) } }
126. abstract class Mailbox @tailrec private final def processMailbox( left: Int = java.lang.Math.max(dispatcher.throughput, 1), deadlineNs: Long = …): Unit = if (shouldProcessMessage) { val next = dequeue() if (next ne null) { … actor invoke next … processMailbox(left - 1, deadlineNs) } } //receiveϝιουΛݺͿ class MyActor extends Actor { def receive = { … } }
127. abstract class Mailbox @tailrec private final def processMailbox( left: Int = java.lang.Math.max(dispatcher.throughput, 1), deadlineNs: Long = …): Unit = if (shouldProcessMessage) { val next = dequeue() if (next ne null) { … actor invoke next … processMailbox(left - 1, deadlineNs) } }
128. recursive processMailbox() ͳͥʁʁ
129. ෆཁͳεϨου੾Γସ͑Λ๷͙
130. ྫ
     ճͷrunͰͭͷϝοηʔδΛϓϩηε
132. Ͱ΋
133. MessageQueueʹϝοηʔδ͋ͬͨΒ
     ಉ͡εϨου͕ԆʑͱprocessMailbox
134. ͦ͏͡Όͳ͍
135. throughput
136. rerence.conf # Throughput defines the number of messages # that are processed in a batch # before the thread is returned to the pool. # Set to 1 for as fair as possible. throughput = 5
137. abstract class Mailbox @tailrec private final def processMailbox( left: Int = java.lang.Math.max(dispatcher.throughput, 1), deadlineNs: Long = …): Unit = if (shouldProcessMessage) { val next = dequeue() if (next ne null) { … actor invoke next … processMailbox(left - 1, deadlineNs) } }
138. σϑΥϧτͰͭ·Ͱ͔͠
     ࿈ଓͰϝοηʔδΛॲཧ͠ͳ͍
139. receiverଆ΋ऴΘͬͨ
140. શମͷ͓͞Β͍
141. "DUPS
     ʹΑΔ
     ϝοηʔδύογϯά 
     "LLB
143. ௿Ϩϕϧ͔ΒݟΔͱ
148. ͞Βʹ௿Ϩϕϧ͔ΒݟΔͱ
150. +BWBͷ'PSL+PJO5BTLΛ
     ܁Γฦ͠FYFDVUF͢Δ
     ΞϓϦέʔγϣϯ 
     "LLB
151. ͓͠·͍