[프로그래밍 클로저] 6장 병행성

락의 문제 (lester.thx)

병행 프로그래밍에서 가장 큰 어려움
- 변화하는 상태를 다루는 것
오늘날 대부분의 프로그래밍 언어
- 모든 상태가 변경 가능하다.
- 병행성은 스레드라고 불리는 독립된 실행 단위를 통해 구현된다.
- 락을 통해 한 번에 한 스레드만 그 상태에 접근하도록 보호해야 한다.
- race condition, deadlock...
Clojure
- 데이터의 상태가 기본적으로 변경 불가능하다.
- 상태 변경이 꼭 필요한 아주 작은 부분의 코드만이 명시적으로 병행성 API를 사용해 상태 변화를 꾀하게 된다.
- 프로그램의 모델
  - 상태 변화가 없는 함수형 모델
    - 대부분의 코드가 이 계층에 속할 것이다.
    - 가독성이 높다.
    - 테스트하기 쉬우며, 병행 실행할 수 있다.
  - 상태 변화가 존재하는 모델
    - 불이익이 생긴다.
    - 더 편리할 경우가 있다.

ref와 소프트트웨어 트랜젝션 메모리(STM) (lester.thx)

클로저 개체는 대부분 상태 변화가 불가능하다.
레퍼런스(ref)를 선언함으로써 상태 변화가 가능해진다.

(ref initial-state)

; 지금 재생되는 노래를 가리키는 레퍼런스
(def current-track (ref "Mars, the Bringer of War"))
-> #'user/current-track

; ref는 그 내부 상태를 감싸서 보호하기 때문에, 레퍼런스가 가리키는 내용을 보호하려면 defref를...
(defref reference)

; deref 함수는 @ 리더 매크로로 간단하게 표현할 수도 있다.
(deref current-track)
-> "Mars, the Bringer of War"
@current-track
-> "Mars, the Bringer of War"

; 생각해 보면 클로저가 현실 세계를 정확하게 반영하고 있다.

ref-set

; ref-set를 이용하면 레퍼런스가 다른 대상을 가리키도록 바꿀 수 있다.
(ref-set reference new-value)

; ref-set를 이용해 다른 트랙을 들어보자.
(ref-set current-track "Venus, the Bringer of Peace")
-> java.lang.IllegalStateException: No transaction running

; ref는 변경 가능하기 때문에, 갱신할 때 적절한 보호가 필요하다.
; 클로저에서는 트랜잭션을 이용할 수 있다.
; 트랜잭션이 일어나는 부분을 dosync로 둘러싸 주면 된다.

(dosync & exprs)

; ref-set를 dosync로 감싸 주면, 정상적으로 동작한다.
(dosync (ref-set current-track "Venus, the Bringer of Peace"))
-> "Venus, the Bringer of Peace"

트랜잭션의 속성

DB의 트랜잭션과 마찬가지로 STM 트랜잭션 역시 몇 가지 중요한 속성을 보장한다.
- 원자성(atomicity): 트랜잭션 내에서 ref를 하나 이상 갱신하는 경우, 트랜잭션 바깥에서 볼 때 그 모든 변경이 여러 번에 나눠서가 아니라 동시에 일어나는 것으로 보인다.
- 일관성(consistency): ref는 유효성 확인 함수(validation function)를 가질 수 있는데, 이 함수 가운데 하나가 실패하면 트랜잭션 전체가 취소된다.
- 고립성(isolation): 실행 중인 트랜잭션의 부분적인 결과를 다른 트랜잭션에서 접근하지 못한다.
- DB에서는 영구성(durability)이라는 속성을 추가로 지원한다. 하지만 클로저의 트랜잭션은 메모리에서 일어나기 때문에, 트랜잭션의 결과가 영구 유지된다고 보장하지 못한다.
  - 클로저의 STM은 ACI만 지원한다.

; current-track과 current-composer의 값이 동시에 갱신되도록

(def current-track (ref "Venus, the Bringer of Peace"))
-> #'user/current-track

(def current-composer (ref "Holst"))
-> #'user/current-composer

(dosync
  (ref-set current-track "Credo")
  (ref-set current-composer "Byrd"))
-> "Byrd"

alter

; 간단한 채팅 프로그램

(defstruct message :sender :text)

; 메시지 생성
(struct message "stu" "test message")
-> {:sender "stu", :text "text message"}

; 빈 리스트를 가리키는 messages 레퍼런스
(def messages (ref ()))

; 새 메시지를 추가할 함수가 필요하다.

; 좋지 않은 접근방식
(defn naive-add-message [msg]
  (dosync (ref-set messages (cons msg @messages))))


; alter는 레퍼런스가 가리키는 객체에 업데이트 함수를 트랜잭션 안에서 적용한다.  
(alter ref update-fn & args...)

; alter는 트랜잭션 안에서 갱신된 ref의 새 값을 반환한다.

(defn add-message [msg]
  (dosync (alter messages conj msg)))
; cons가 아니라 conj라는 것에 주목하자.
; conj가 인자를 받는 순서가 alter를 적용하기에 좋기 때문이다.
(cons item sequence)
(conj sequence item)

; alter 함수는 ref가 가리키는 값을 update-fn의 첫 인자로 넘기기 때문에,
; 이 경우에는 시퀀스를 첫 인자로 받는 conj를 쓰는 것이 맞다.

(your-func thing-that-gets-updated & optional-other-args)

(add-message (struct message "user 1" "hello"))
-> ({:sender "user 1", :text "hello"})

(add-message (struct message "user 2" "howdy"))
-> ({:sender "user 2", :text "howdy"}
    {:sender "user 1", :text "hello"})

STM의 작동 원리: MVCC

클로저의 STM은 다중 버전 병행성 제어(Multiversion Concurrency Control, MVCC)라는 기법을 사용한다.

commute

; commute는 병행적인 특성이 강한 alter라고 할 수 있다.
(commute ref update-fn & args...)

; commute는 트랜잭션의 순서가 교환가능(commutative)하다는 뜻에서 붙여진 이름이다.

(defn add-message-commute [msg]
  (dosync (commute messages conj msg)))

; 이 함수를 사용하면, 마지막으로 변경된 트랜잭션 내부 값이 최종 ref의 값이 되리라는 보장이 없다.
; STM이 갱신 순서를 바꿔 버릴 수 있기 때문이다.

alter가 나을 경우도 있다

갱신 순서가 늘 상관이 없는 것은 아니다.

; 숫자가 하나씩 증가하는 카운터
(def counter (ref 0))

; 이때에는 카운터를 갱신하기 위해 commute를 사용해서는 안된다.
; 이런 경우 alter를 사용해야 한다.

(defn next-counter [] (dosync (alter counter inc)))

(next-counter)
->1
(next-counter)
->2

; commute보다는 alter를 흔히 사용하게 된다.
; alter를 사용한 코드는 이해하기 쉽고,
; 예측 불가능한 오류를 일으키지 않기 때문이다.
; 하지만 commute의 경우에는 사용하기 전에 트랜잭션의 의미를 곰곰히 생각해...
; commute로 충분한 곳에 alter를 사용할 경우, 수행 성능만 손해 볼 뿐이다.

ref에 유효성 확인(validation) 추가

(ref initial-state options*)
; options는 다음과 같은 키/값 쌍으로 이루어진다.
;  :validator validate-fn
;  :meta metadata-map

; options는 맵이 아니라 키/값 쌍의 시퀀스라는 점에 주의하자.

(def validate-message-list
  (partial every? #(and (:sender %) (:text %))))

(def messages (ref () :validator validate-message-list))

; 이러한 유효성 확인 함수는 DB의 테이블에 대한 키 제약조건과도 같다.

(add-message "not a valid message")
-> java.lang.IllegalStateException: Invalid reference state

@messages
-> ()

; 제약 조건에 맞는 메시지의 경우에는 아무 문제가 없다.
(add-message (struct message "stu" "legit message"))
-> ({:sender "stu", :text "legit message"})

joony 궁금한 것 임시 단락..;

184에서, track : 현실 세계 반영하려면 ref로 가르키는게 계속 변해야하는 것 아닌가? 그걸 refset에서 하는 구나.. ㅋㅋ 갑자기 composer..이건 또 왜 alter를 쓰는지 와닿지 않음!

제어가 필요없는 동기화된 갱신에 애텀 사용하기 (joony.l)

atom

(atom initial-state options?)
; option은 다음과 같읕 키/값 쌍으로 이뤄진다
; :validator validate-fn
; :meta metadata-map

ref는 공유되는 상태에 대한 접근을 제어할 필요가 있을 때 유용하게 사용.
공유되지 않는 데이터를 갱신하는 경우라면, atom을 사용하는 것이 낫다. (번역이 잘못된듯
ref보다 가벼운 접근 방식.

위 예에서 ref대신 atom

(def current-track (atom "Venus, the Bringer of peace"))
-> #'user/current-track

;값을 알아내는 것은 deref로 알아내듯 동일

(deref current-track)
-> "Venus, the Bringer of peace"

@current-track
-> "Venus, the Bringer of peace"

reset

(reset! an-atom newval)

atom은 트랜잭션 내에서 변경되는 것이 아니기 때문에, dosync를 사용할 필요가 없다.
atom의 값을 갱신하기 위해서는 reset!을 사용하면 그만.

(reset! current-track "Credo")

ref와 atom의 차이는 atom을 사용하면서 current-track과 current-composer가 동시에 변경되도록 할 수 없는 점.
할 수 있는 방법이 있긴하지만, 이것은 문제를 모델링하는 방식을 바꾸는 경우에 한함. 트랙 제목과 작곡가로 이루어진 맵을 만들고 한 atom에 저장하면 되지 않을까?
값들이 동시에 갱신되도록 제어하려면 ref를 사용해야한다.

(def current-track (atom {:title "credo" :composer "Byrd"}))

(reset! current-track {:title "Spem in Alium" :composer "Tallis"})
-> {:title "Spem in Alium", :composer "Tallis"}

swap!

작곡가는 그대로고, 트랙만 변경하고 싶다면?

(swap! an-atom f & args))

함수 f에 an-atom의 현재 값과 args들을 넘겨, 거기서 반환된 값으로 an-atom을 갱신한다.

swap!가 :title키에 대헤 assoc함수를 적용하도록 만들어서 트랙제목만 변경하기
```
(swap! current-track assoc :title "Sancte Dues")
-> {:title "Sancte Dues", :composer "Tallis"}
```
swap!이 새로운 값을 반환한다.
만약 다른 스레드가 같은 atom의 값을 변경하려고 시도중이라면
swap은 재시도 된다.
따라서 swap!에 넘기는 함수는 부수효과를 가져선 안된다.

<<<<<<< HEAD swap!이 새로운 값을 반환한다. 만약 다른 스레드가 같은 atom의 값을 변경하려고 시도중이라면 swap은 재시도 된다. 따라서 swap!에 넘기는 함수는 부수효과를 가져선 안된다.

ref와 atom은 모두 동기화된 갱신을 위한 것. 갱신 함수가 값을 반환하는 시점에, ref나 atom이 가리키는 값 역시 갱신되어 있다. 값이 나중에 변경될 수도 있는 비동기적 갱신으로도 충분한 경우라면, 에이젼트를 사용하는 게 낫다.

ref와 atom은 모두 동기화된 갱신을 위한 것.
갱신 함수가 값을 반환하는 시점에, ref나 atom이 가리키는 값 역시 갱신되어 있다.
값이 나중에 변경될 수도 있는 비동기적 갱신으로도 충분한 경우라면, agent를 사용하는 게 낫다.

3220fa811acb624721ca94b999227d8ebd887b5e

비동기적 갱신을 위한 에이전트 (joony.l)

agent

(agent initial-state)

task가 대체로 독립적으로 수행되서, 작업사이 제어가 별로 필요 없는 애플리케이션도 있다.
- 이럴 땐 클로저의 agent를 주로 사용한다.
- ref와 비슷!
ref처럼 초기 상태를 인자로 넘겨 agent 생성

(def counter (agent 0))
-> #'user/counter

send

(send agent update-fn & args)

agent가 생성되고 나면, 상태 갱신함수 send를 이용해 agent에 보낼 수 있다.
send는 나중에 쓰레드 풀에 있는 스레드에 적용할 update-fn을 대기열에 집어 넣는다.
agent에 send를 사용하는 것은 ref에 commute를 사용하는 것과 비슷.

counter에 inc를 적용한 예.

(def counter (agent 0))

(send counter inc)
->#<Agent@5030b604: 0>

send를 호출하면 agent의 새 값이 반환되는 것이 아니라 agent 그 자체가 반환.
- 새 값이 반환 되지 않는 이유는 send가 새 값이 무엇인지 알지 못하기 때문.
- send는 inc를 대기열에 집어 넣어 나중에 수행되도록 해놓고,자신은 바로 반환됨.
send는 애이전트의 새로운 값을 알지 못하지만, repl은 알고 있을 지 모름.
애이전트 스레드와 repl쓰레드 가운데 무엇이 먼저 실행되느냐에 따라서 반환 값의 콜론뒤의 값이 0또는 1이 나올것

@counter; 또는 (deref counter)
-> 1 ;내 경우 1

await

(await & agents)
(await-for timeout-millis & agents)

repl과 agent 스레드 사이에서 경쟁상태(race condition)가 발생해 곤란한 경우라면, await이나 await-for를 호출해, agent에 보낸 작업이 먼저 완료되도록 할 수 있다.

agent에 대한 작업이 완료될 떄 까지 현재 수행되는 스레드를 block함 await-for은 timeout-millis만큼의 시간이 지나도 작업이 완료되지 않으면 nil을 반환, 그렇지 않으면 Nil이 아닌 값을 반환 await은 시간제한은 없지만, 스레드들이 무한이 대기 할 수 있기 때문에 주의.

에이전트에 대한 유효성 확인과 에러처리

agent가 ref와 또 다른 유사한 점은, agent역시 유효성확인 함수를 가질수 있는 것.

(agent initail-state options*)
; options는 다음과 같은 키/값 쌍을 외뤄짐
; :validator validate-fn
; :meta metadata-map

counter가 숫자 값만 받도록 유효성 확인 함수를 추가해 다시 생성해보자.

(def counter (agent 0 :validator number?))
-> #'user/counter
(send counter (fn [_] "boo"))
-> #<Agent@1bc1d189: 0>

@counter
-> 0;여기서 에러가 나야하는데 안나넹;; 유효하지 않으면 값을 변경 안하는 듯

구체적 에러 알아내기 위해선 agent-errors를 호출하면 됨. agent-erros는 agent에 대한 작업동안 발생한 에러의 시퀀스를 반환 (시퀀스를 반환하는 건가.. 한개인 거 같은데..)

(agent-errors counter)
-> (#<IllegalStateException java.lang.IllegalStateException: Invalid reference state>)

일단 agent에 에러가 발생하고 나면, 이어서 에이전트에 날려지는 쿼리도 모두 에러를 반환하게 된다. agent를 다시 사용가능하게 만들려면 clear-agent-errors를 호출하면 된다.

(clear-agent-errors counter)
-> 0

@counter
-> 0

(agent-errors counter)
-> nil

트랜잭션에 에이전트를 포함하기

클로저의 트랜잭션은 상황에 따라 몇 번이고 재시도 될 수 있기 때문에, 트랜잭션이 부수효과를 가져서는 안됨. but 트랜잭션이 성공했을 때 부수효과를 일으켜야하는 경우도 때때로 존재.

agent를 이용하면 이게 가능해짐

트랜잭션 내부에서 agent에 작업을 보내면, 그 작업은 트랜잭션이 성공하는 경우에만 딱 한번 에이전트에 보내지게 됨.

(def backup-agent (agent "./test.clj"))


(defn add-message-with-backup [msg]
  (dosync
   (let [snapshot (commute messages conj msg)]
     (send-off backup-agent (fn [filename]
                              (spit filename snapshot)
                              filename))
     snapshot)))

;use는 필요 없음
;


(add-message-with-backup (struct message "joony" "message"))
-> ({:sender "joony", :text "message"} {:sender "user 2", :text "hi"} {:sender "user 1", :text "hello"})

;use는 필요 없음
;send대신 send-off

send-off는 파일에 쓰는 것과 같은, 블록이 필요한 작업을 에이전트에 보낼 때 사용. send-off를 통해 보내지는 작업들은 별도으 스레드 풀에서 처리 이 경우 또 다른 작업을 블록하는 함수는 send-off를 통해 보내지 말자. 그렇지 않으면 다른 에이전트들에 대한 작업이 불필요하게 지연될 수 있다.

repl이 보여주는 메모리속의 message와 message-backup 파일의 내용이 같은지 볼것.

전략은 다른 방식으로 개선해도 됨. 백업 빈도를 낮추거나, 정보의 변화가 있을 때만 백업하게 만들기.

클로저의 stm의 acid가운데 aci만 지원하고, 파일에 기록하는 것으로 d(영구성)을 얻을 수 있기 때문에 db와 똑같다고 착각할 수 있지만 클로저의 트랜잭션은 에이전트에 작업을 보낸 것일 뿐, 그 작업이 Aci속성을 가지고 수행되는 것은 아니다.

예를들어 트랜잭션은 완료됐지만, 에이전트가 파일에 내용을 쓰기전에 누군가 전원 코드를 뽑아 버릴 수도 있다. 디비가 필요하면 진짜 디비를 써라!

일관성 있는 갱신 모델

지금까지 봤던 ref, atom, agent의 공통된 매커니즘은 함수를 인자로 받고, 그 함수를 기존상태에 적용해 새로운 상태로 변화시키는 것 이같은 모델은 클로저에서 공유된 상태를 다루는데 핵심이 되는 개념

https://www.dropbox.com/s/f20qp1kry7c3zah/%EC%8A%A4%ED%81%AC%EB%A6%B0%EC%83%B7%202014-08-25%2023.29.05.png?dl=0

var를 이용해 스레드별 상태 관리하기 (todd.krause)

def나 defn으로 생성한 바인딩은 var의 루트 바인딩으로 모든 스레드에서 공유된다.
```
(defn foo 10)

(.start (Thread. (fn [] (println foo))))
-> nil
10
```

binding 매크로를 사용하면 스레드에 한정된 var의 바인딩을 만들 수 있다.

(def ^:dynamic foo 10) ; 1.4부터 ^:dynamic을 써줘야한다.

(binding [foo 42] foo) ; let 처럼 사용한다.
-> 42

(defn print-foo [] (println foo))

(let [foo "let foo"] (print-foo))
-> 10
nil

(binding [foo "bound foo"] (print-foo))
-> bound foo
nil

binding을 이용해서 Thread Local을 만들 수 있다.

(def ^{:dynamic true} *counter*)

(defn inc-counter []
	(set! *counter* (inc *counter*)))

(map (fn [i] (.start (Thread. (fn []
	(binding [*counter* 0]
		(println (str "start " i))
		(doall (repeatedly 5 inc-counter))
		(println (str "thread " i " " *counter*)))))))
	(range 10))

Provide feedback

Saved searches

Use saved searches to filter your results more quickly