scala_multi_thread.pdf

1. SCALAͰͷϚϧνεϨου ॲཧʹ͍ͭͯ 1
2. ࠓ೔ͷൃදΛฉ͍ͯΘ͔Δ͜ͱ ‣ Java͕ఏڙ͢ΔඇಉظϞδϡʔϧͷಛ௃ɾϢʔεέʔε ‣ ActorϞσϧ͕ͲͷΑ͏ʹεϨουηʔϑΛ࣮ݱ͍ͯ͠Δ͔ 2
3. 3 ϚϧνεϨουͷॏཁੑ ‣ ϜʔΞͷ๏ଇͷ่յ CPUΫϩοΫ਺ͷ಄ଧͪ ‣ ϚϧνίΞϓϩηοα࣌୅ͷ౸དྷ ฒߦ/ฒྻʹϓϩάϥϜΛ࣮ߦ͢Δඞཁ͕͋Δ ҰͭͷϓϩάϥϜͰෳ਺ͷεϨουΛ࣮ߦ͢Δ͜ͱ εϨου̍ ॲཧ εϨου2 ॲཧ εϨου3 ॲཧ ෳ਺ͷॲཧ͕ɺॱෆಉ΋͘͠͸ಉ࣌ʹى͜Γ͏Δ
4. 4 ϚϧνεϨουͷ೉͠͞ • ಡΈॻ͖ͷڝ߹ͷͨΊʹɺλΠϛϯάʹΑͬͯ݁Ռ͕ҟͳΔ৔߹͕͋Δ • σουϩοΫ ‣ εϨουؒͷಉظॲཧ/ഉଞ੍ޚ ‣ ࢥ͏Α͏ʹύϑΥʔϚϯε͕޲্͠ͳ͍͜ͱ΋ • ্ख͘ഉଞ੍ޚͰ͖ͳ͔ͬͨΓɺ༨ܭͳಉظॲཧ͕૸ΔͱٯʹύϑΥʔϚϯε͕Լ͕Δ
5. 5 ϚϧνεϨουؔ࿈༻ޠ ‣ ഉଞ੍ޚ ‣ εϨουηʔϑ ‣ ΞτϛοΫʢෆՄ෼ૢ࡞ʣ ෳ਺ͷτϥϯβΫγϣϯ͔Βಉ͡Ϧιʔεʹରͯ͠ɺ ಉ࣌ʹߋ৽ॲཧ͕ߦΘΕΔࡍʹɺσʔλͷ੔߹ੑΛอͭͨΊʹߦ͏ॲཧ ϚϧνεϨουͰಈ࡞ͤͯ͞΋໰୊ͷͳ͍ϓϩάϥϜͷ͜ͱ(σουϩοΫɺσʔλڝ߹͕ى͖ͳ͍) ଞͷεϨουʹ͸ׂΓࠐΈΛͤͣ͞ɺݱࡏߦ͍ͬͯΔมߋૢ࡞͕࣮֬ʹऴΘͬͨޙʹ࣮ߦͰ͖Δ͜ͱ
6. 6 ࠓ೔ग़ͯ͘ΔϞδϡʔϧ 1. Volatileम০ࢠ 2. Synchronizedम০ࢠ 3. concurrent.atomicύοέʔδ 4. ActorϞσϧ
7. 7 ࠓ೔ग़ͯ͘ΔϞδϡʔϧ 1. Volatileम০ࢠ 2. Synchronizedम০ࢠ 3. concurrent.atomicύοέʔδ 4. ActorϞσϧ
8. 8 Volatileम০ࢠ 1.ίϯύΠϥͷ࠷దԽʢஔ͖׵͑ʣΛ๷ࢭ͢Δ 2.ίϯύΠϥͷ࠷దԽʢϦΦʔμʔʣΛ๷ࢭ͢Δ 3.volatileम০ࢠΛ෇༩͞Εͨม਺͸ɺ௚઀ϝΠϯϝϞϦʹอଘ͞Εɺ CPUΩϟογϡ΁ͷίϐʔͳͲ͕ى͖ͳ͍ @volatile var flag = false // ௚઀ϝΠϯϝϞϦʹอଘ͞ΕΔ [ৄ͍͠ࢿྉ] http://kannokanno.hatenablog.com/entry/20120528/1338229536
9. 9 ϚϧνεϨουͷ࢓૊Έ εϨου1 ϝΠϯϝϞϦ εϨου2 ಡΈࠐΈ ಡΈࠐΈ ॻ͖ࠐΈ Ճࢉ • ࣮ߦޮ཰Λ͋͛ΔͨΊʹɺ֤εϨου͸࡞ۀ༻ϝϞϦΛ͍࣋ͬͯΔ • جຊ͸࡞ۀ༻ϝϞϦʹରͯ͠ಡΈॻ͖͕ߦΘΕɺ ద౰ͳλΠϛϯάͰϝΠϯϝϞϦ΁ಉظ͢Δ ίϐʔ ಉظ ಉظ ίϐʔ
10. 10 Volatileม਺͕ղܾ͢Δ͜ͱ ϝΠϯϝϞϦ΁ͷಉظ͕஗Ε͍ͯΔ৔߹ɺ ผεϨου͕ࢀর͍ͯ͠Δ஋͸ݹ͍Մೳੑ͕͋Δ Volatileम০ࢠʹΑͬͯ௚઀ϝΠϯϝϞϦʹ อଘ͢Δ͜ͱʹΑͬͯɺ๷͙ • ࣮ߦޮ཰Λ͋͛ΔͨΊʹɺ֤εϨου͸࡞ۀ༻ϝϞϦΛ͍࣋ͬͯΔ • جຊ͸࡞ۀ༻ϝϞϦʹରͯ͠ಡΈॻ͖͕ߦΘΕɺ ద౰ͳλΠϛϯάͰϝΠϯϝϞϦ΁ಉظ͢Δ
11. 11 Volatileम০ࢠ͕ղܾ͢Δ͜ͱ εϨου1 ϝΠϯϝϞϦ εϨου2 ಡΈࠐΈ ಡΈࠐΈ ॻ͖ࠐΈ ॻ͖ࠐΈ Ճࢉ Ճࢉ Volatileम০ࢠʹΑͬͯ௚઀ϝΠϯϝϞϦʹ อଘ͢Δ͜ͱʹΑͬͯɺ๷͙
12. 12 VolatileϢʔεέʔε ҎԼͷ৚݅Λຬͨ͢৔߹ʹ࢖͑Δ 1.ͦͷม਺΁ͷॻ͖ࠐΈ͕ɺͦͷม਺ͷݱࡏͷ஋ʹґଘ͠ͳ͍ 2.ͦͷม਺͕ɺଞͷม਺ͱͷෆมࣜʹ࢖ΘΕ͍ͯͳ͍ ྫ 1 ౓͚ͩൃੜ͢ΔॏཁͳϥΠϑαΠΫϧɾΠϕϯτͷൃੜΛࣔ͢ɺεςʔλεϑϥά private[akka] abstract class Mailbox(val messageQueue: MessageQueue) extends ForkJoinTask[Unit] with SystemMessageQueue with Runnable { @volatile var actor: ActorCell = _ … } Actor಺෦Ͱ΋࢖ΘΕ͍ͯΔ
13. 13 Volatileम০ࢠͷ໰୊఺ ++΍+=ͱ͍ͬͨԋࢉ(ಡΈऔΓɺมߋɺ௥ՃͷҰ࿈ͷԋࢉ)͸ΞτϛοΫͰ͸ͳ͍ (εϨουΛϩοΫ͠ͳ͍ͨΊ) ΞτϛοΫͳԋࢉʹ͸java.concurrent.atomicΛ࢖͏ ϝιου΍ΦϒδΣΫτ୯ҐͰ෇༩Ͱ͖ͳ͍ synchronizedͳΒͰ͖Δ
14. 14 ࠓ೔ग़ͯ͘ΔϞδϡʔϧ 1. Volatileम০ࢠ 2. Synchronizedम০ࢠ 3. concurrent.atomicύοέʔδ 4. ActorϞσϧ
15. 15 Synchronizedम০ࢠ Volatileम০ࢠʹ͸ͳ͍ϩοΫػߏΛ༻͍ͯɺ ΦϒδΣΫτ΍ϝιουʹରͯ͠ϩοΫΛ഑ஔ͢Δ͜ͱʹΑͬͯɺ ಉ࣌ʹෳ਺ͷεϨου͕ॲཧΛߦΘͳ͍Α͏ʹഉଞ੍ޚΛߦ͏ def process() = synchronized { // ϝιουʹର͢Δഉଞ੍ޚ } def process() = obj.synchronized { // ΦϒδΣΫτʹର͢Δഉଞ੍ޚ } ࢖͍͜ͳͤΕ͹ɺෳࡶͳॲཧ΋҆શʹߦ͑Δ
16. 16 ϩοΫػߏ ͋ΔεϨου͕ΦϒδΣΫτͷϩοΫΛऔಘ͢ΔͱɺͦͷϩοΫΛղ์͢Δ·Ͱ ଞͷεϨου͸ͦͷΦϒδΣΫτͷϩοΫΛऔಘ͢Δ͜ͱ͕Ͱ͖ͣɺ଴ͨͳ͚Ε͹ͳΒͳ͍ ഉଞ੍ޚΛ࣮ݱ͢Δػߏ
17. 17 SynchronizedϢʔεέʔε ෳ਺ͷεϨου͕ಉ࣌ʹॻ͖ࠐΈʹߦ͘ def add(account: Account, n: Int) = { account.money += n Thread.sleep(1000) println(s"thread: ${Thread.currentThread().getId()}, money: ${account.money}") } val jane = new Account("Jane", 100) val t1 = thread { add(jane, 10) } // εϨου1 val t2 = thread { add(jane, 70) } // εϨου2 t1.join(); t2.join() class Account(val name: String, var money: Int) ————————————————————————— thread: 1, money: 180 thread: 2, money: 180 εϨου1͕moneyΛදࣔ͢Δલʹɺ εϨου2͕moneyΛߋ৽͍ͯ͠Δ
18. 18 accountΦϒδΣΫτʹରͯ͠ϩοΫΛ͔͚ͯɺ ̍εϨου୯ҐͰ͔ͦ͠ͷaccountΦϒδΣΫτʹ ΞΫηεͰ͖ͳ͍Α͏ʹ͍ͯ͠Δ def add(account: Account, n: Int) = account.synchronized { account.money += n Thread.sleep(1000) println(s"thread: ${Thread.currentThread().getId()}, money: ${account.money}") } val jane = new Account("Jane", 100) val t1 = thread { add(jane, 10) } // εϨου1 val t2 = thread { add(jane, 70) } // εϨου2 t1.join(); t2.join() class Account(val name: String, var money: Int) —————————————————————————- thread: 1, money: 110 thread: 2, money: 180 ֤εϨου͕ஞ࣍తʹॲཧΛߦ͍ͬͯΔ SynchronizedϢʔεέʔε
19. 19 Synchronizedͷ໰୊఺ def send(a: Account, b: Account, n: Int) = a.synchronized { b.synchronized { a.money -= n b.money += n } } val a = new Account("Jack", 1000) val b = new Account("Jill", 2000) val t1 = thread { for(i <- 0 until 100) send(a, b, 1) } // εϨου1 val t2 = thread { for(i <- 0 until 100) send(b, a, 1) } // εϨου2 t1.join(); t2.join() σουϩοΫ ͓ޓ͍͕ϩοΫΛղ์͞ΕΔͷΛ଴ͬͯ͠·͏
20. 20 Synchronizedͷ໰୊఺ ύϑΥʔϚϯεͷ௿Լ synchronizedΛଟ༻͠ա͗Δͱɺ ฒྻͰಈ࡞͢΂͖ϝιουʹ΋ϩοΫ͕͔͔ͬͯɺ γϯάϧεϨουॲཧʹͳͬͯ͠·͏͜ͱ͕͋Δɻ ৔߹ʹΑͬͯ͸ɺγϯάϧεϨουॲཧΑΓ΋஗͘ͳΔՄೳੑ΋͋Δ
21. 21 ϩοΫػߏͷ໰୊఺ 1. ϩοΫ͸Ճࢉɾ݁߹Ͱ͖ͳ͍ 2. ϩοΫΛ෇͚๨ΕΔ 3. ϩοΫΛෆ༻ҙʹ࢖͍͗ͯ͢஗͘ͳΔ 4. ؒҧͬͨϩοΫΛೖΕͯ૝ఆ֎ͷॲཧ͕ൃੜ͢Δ 5. ϩοΫͷ഑ஔͷॱংΛؒҧ͑Δ 6. Τϥʔͷൃݟ͕೉͍͠ɺΤϥʔ͔Βͷ෮ؼ͕ࠔ೉
22. 22 ϩοΫػߏ͸σουϩοΫ ͳͲͷةݥੑΛؚΜͰ͍Δ ϩοΫΛઃܭ͔Βഉআʹ͢Δ͜ͱʹΑͬͯɺ ϚϧνεϨουϓϩάϥϛϯάΛ؆қʹ͢Δઃܭࢥ૝͕ ෯Λ͖͔͖͍ͤͯͯΔ
23. 23 ࠓ೔ग़ͯ͘ΔϞδϡʔϧ 1. Volatileम০ࢠ 2. Synchronizedम০ࢠ 3. concurrent.atomicύοέʔδ 4. ActorϞσϧ
24. 24 atomicύοέʔδ ୯Ұͷม਺ʹର͢ΔϩοΫϑϦʔͰεϨουηʔϑͳϓϩάϥϛϯά Λαϙʔτ͢ΔΫϥεΛؚΉύοέʔδ • AtomicBoolean: BooleanܕΛѻ͏ • AtomicInteger: IntegerܕΛѻ͏ • AtomicLong: LongܕΛѻ͏ • AtomicReference<V>: ΦϒδΣΫτܕΛѻ͏ • …
25. 25 Atomicม਺ Volatileम০ࢠͷ͍ͭͨม਺ʹର͢Δ ΠϯΫϦϝϯτॲཧ͸ΞτϛοΫʹͳΒͳ͍ private val LOOP = 100000000 // 1ԯճϧʔϓ @volatile private var counter = 0L def count() = { for { i <- 1 to LOOP } counter += 1 } val t1 = thread { count() } // εϨου1 val t2 = thread { count() } // εϨου2 t1.join(); t2.join() println(s"counter: ${counter}”) ———————————————————— counter: 101280698 private val LOOP = 100000000 // 1ԯճϧʔϓ private val counter = new AtomicLong(0L) def count() = { for { i <- 1 to LOOP } counter.getAndIncrement() } val t1 = thread { count() } // εϨου1 val t2 = thread { count() } // εϨου2 t1.join(); t2.join() println(s"counter: ${counter}") ———————————————————— counter: 200000000 Atomicม਺͸ɺΞτϛοΫʹ ΠϯΫϦϝϯτॲཧ͕Մೳ
26. 26 Atomicม਺ private val uid = new AtomicLong(0L) @tailrec def getUniqueId(): Long = { val oldUid = uid.get val newUid = oldUid + 1 // uid͕oldUidͱ౳͚͠Ε͹ɺಉظతʹuidʹnewUidΛઃఆ͢Δ if(uid.compareAndSet(oldUid, newUid)) newUid // ࣦഊͨ͠৔߹͸ɺϦτϥΠ͢Δ else getUniqueId() } Atomicม਺ͷϝιου͸ϩοΫϑϦʔͳͷͰɺσουϩοΫ͕ى͖ͳ͍ ϩοΫΛͤͣʹɺଞͷεϨου͕ม਺Λมߋ͢Δͱɺ CAS(CompareAndSwap)͕ݕग़ʢ·ͨ͸ࣦഊʣ͠ɺͦΕͷૢ࡞Λ܁Γฦ͢͜ͱ͕Ͱ͖Δ
27. 27 Atomicม਺Ϣʔεέʔε • ҆શੑͱ଱ো֐ੑͰධՁ͕ߴ͍ (CASϕʔεͷฒߦΞϧΰϦζϜ͸ϩοΫϑϦʔͱݺ͹ΕΔ) • ύϑΥʔϚϯεʹ͓͍ͯ͸ɺsynchronizedͱ΄ͱΜͲҧ͍͕ͳ͍ͱݴΘΕ͍ͯΔ • ෳ਺ͷม਺ͷߋ৽ʹ͸޲͍͍ͯͳ͍ ྫ ୯ҰͷΧ΢ϯλʔॲཧ Future಺෦Ͱ΋࢖ΘΕͯΔ object Future { def find[T](@deprecatedName('futurestravonce) futures: TraversableOnce[Future[T]]) (@deprecatedName('predicate) p: T => Boolean) (implicit executor: ExecutionContext): Future[Option[T]] = { …… val ref = new AtomicInteger(futuresBuffer.size) …… if (ref.decrementAndGet == 0) { ??? } …… } }
28. 28 ࠓ೔ग़ͯ͘ΔϞδϡʔϧ 1. Volatileम০ࢠ 2. Synchronizedम০ࢠ 3. concurrent.atomicύοέʔδ 4. ActorϞσϧ
29. 29 ActorϞσϧ
30. 30 ActorϞσϧ ΞΫλʔͱ͸ɺܰྔεϨου AkkaͰ͸1GBͷϝϞϦʔͰɺΞΫλʔΛ270ສݸੜ੒͢Δ͜ͱ͕Ͱ͖Δ ΞΫλʔ͸ϝοηʔδΛड৴͢ΔϝʔϧϘοΫεΛ࣋ͭ εϨου1 εϨου2 ϝʔϧϘοΫε ϝʔϧϘοΫε
31. 31 ActorϞσϧ͸εϨουηʔϑ “ϝοηʔδΛૹΔਓ͸ड͚औΔਓͷঢ়ଶʹؔΘΒͣͲΜͲΜϝοηʔδΛૹ৴Ͱ͖Δɻ ϝοηʔδΛड͚औΔਓ͸Ωϡʔʹཷ·ͬͨϝοηʔδͷ಺༰Λॱ࣍ॲཧ͢ΔͨΊɺ ΞΫλʔ͕ॴ༗͍ͯ͠ΔΠϯελϯεʹඇಉظʹΞΫηε͞ΕΔ͜ͱ͸ͳ͘εϨουηʔϑʹͳΔɻ”
32. 32 ActorϞσϧ͸εϨουηʔϑ ΞΫλʔ͸಺෦ʹϛϡʔλϒϧͳঢ়ଶΛ࣋ͭ͜ͱ͕Ͱ͖ɺ ͦΕΛ҆શʹߋ৽͢Δ͜ͱ͕Ͱ͖Δ class GreetActor(val who: String) extends Actor { var greeting = "" def receive = { case `who` => greeting = s"hello, $who" case msg => sender ! msg } }
33. 33 ΞΫλʔ͸಺෦ʹϛϡʔλϒϧͳঢ়ଶΛ࣋ͭ͜ͱ͕Ͱ͖ɺ ͦΕΛ҆શʹߋ৽͢Δ͜ͱ͕Ͱ͖Δ class GreetActor(val who: String) extends Actor { var greeting = "" def receive = { case `who` => greeting = s"hello, $who" case msg => sender ! msg } } ଞͷεϨου͔Βvarͷมߋ͸ผͷεϨου͔Βݟ͍͑ͯΔͱ͸ݶΒͳ͍ͷʹɺ ͳͥvarม਺ʹ@volatileએݴΛ͠ͳͯ͘΋େৎ෉ͳͷ͔ʁ ActorϞσϧ͸εϨουηʔϑ
34. 34 Akka͕อূ͍ͯ͠Δϧʔϧ [Ҿ༻] https://doc.akka.io/docs/akka/2.4.7/general/jmm.html#Actors_and_the_Java_Memory_Model 1. The actor send rule 2. The actor subsequent processing rule
35. 35 1. The actor send rule ΞΫλʔ΁ͷϝοηʔδૹ৴͸ɺ ͦͷΞΫλʔ͕ͦͷϝοηʔδΛड৴͢Δલʹى͖Δ “Ξϯό΢ϯσουϝʔϧϘοΫεͷ৔߹ConcurrentLinkedQueueΛ࢖͏͜ͱͰvolatile writeΛߦ͍ɺ ό΢ϯσουϝʔϧϘοΫεͷ৔߹͸LinkedBlockingQueueΛ࢖͏͜ͱͰϩοΫΛߦ͏”
36. 36 2. The actor subsequent processing rule ΞΫλʔͰͷ͋Δϝοηʔδͷॲཧ͸ɺ ͦͷΞΫλʔͰͷ࣍ͷϝοηʔδͷॲཧͷલʹى͜Δ “Ҏલͷϝοηʔδॲཧ࣌ʹߦͬͨॻ͖ࠐΈ͸ಉظ͞Εɺ ࣍ͷϝοηʔδΛॲཧ͢Δͱ͖ʹ؍ଌͰ͖Δ͜ͱ͕อূ͞ΕΔ”
37. 37 Akka͕อূ͍ͯ͠Δϧʔϧ [Ҿ༻] https://doc.akka.io/docs/akka/2.4.7/general/jmm.html#Actors_and_the_Java_Memory_Model 1. The actor send rule ΞΫλʔ΁ͷϝοηʔδૹ৴͸ɺͦͷΞΫλʔ͕ͦͷϝοηʔδΛड৴͢Δલʹى͖Δ 2. The actor subsequent processing rule ΞΫλʔͰͷ͋Δϝοηʔδͷॲཧ͸ɺͦͷΞΫλʔͰͷ࣍ͷϝοηʔδͷॲཧͷલʹى͜Δ ϝʔϧϘοΫεʹΑ࣮ͬͯݱ͍ͯ͠Δ
38. 38 ·ͱΊ • volatileએݴ͞Εͨม਺͸ɺৗʹϝϞϦ͔Β࠷৽ͷ஋Λऔಘ͢Δ͕ɺ ΞτϛοΫͳԋࢉ͸Ͱ͖ͳ͍ • synchronized͸ϓϩάϥϚ্͕खʹ࢖Θͳ͍ͱɺ σουϩοΫ΍ύϑΥʔϚϯε௿ԼΛҾ͖ى͜͢ • ϚϧνεϨου؀ڥͰͷ୯Ұม਺΁ͷΞτϛοΫͳԋࢉ͸ concurrent.atomicύοέʔδʹؚ·ΕΔΦϒδΣΫτ͕ద͍ͯ͠Δ • ActorϞσϧ͸ϝʔϧϘοΫεʹΑͬͯεϨουηʔϑΛ࣮ݱ͍ͯ͠Δ