쓰레드의 동기화(4)

volatile

싱글 코어 프로세서가 장착된 컴퓨터에서는 크게 문제가 없이 실행될 것이다. 

그러나 요즘 대부분 멀티코어 프로세서가 장착된 컴퓨터를 사용하기 때문에, 문제가 많아질 수 있다.

왜냐하면 멀티 코어 프로세서에서 코어마다 별도의 캐시를 가지고 있기 때문이다.

코어는 메모리에서 읽어온 값을 캐시에 저장하고 캐시에서 값을 읽어서 작업한다. 다시

같은 값을 읽어올때는 머넞 캐시에 있는지 확인하고 없을 때만 메모리에서 읽어온다.

그러다보니 도중에 메모리에 저장된 변수의 값이 변경되었는데도 캐시에 저장된 값이 갱신되지 않아서 

메모리에 저장된 값이 다른 경우가 발생한다.

그래서 변수 stopped의 값이 바뀌었는데도 쓰레드가 멈추지않고 계속 실행된다.

변수에 volatile을 붙이는 대신에 synchronized블럭을 사용해도 같은 효과를 얻을 수 있다. 

쓰레드가 synchronized블럭으로 들어갈 때와 나올 때, 캐시와 메모리간의 동기화가 이루어 지기 때문에 값의 불일치가 해소 되기 때문이다.

public void stop(){
stopped = true'
}
//->
public synchronized void stop()	{
stopped =true;
}

volatile로 long과 double을 원자화

JVM은 데이터를 4byte(=32bit)단위로 처리하기 때문에, int와 int보다 작은 타입들은 한 번에 읽거나 쓰는 것이 가능하다.

즉, 단 하나의 명령어로 읽거나 쓰기가 가능하다는 뜻이다. 하나의 명령어는 더 이상 나눌 수 없는 최소의 작업단위

이므로, 작업의 중간에 다른 쓰레드가 끼어들 틈이없다.

크기가 8바이트인 long과 double타입의 변수는 하나의 명령어로 값을 읽거나 쓸수 없기 때문에, 변수의 값을 읽는 과정에 다른 쓰레드가 끼어들 여지가 있다. 

다른쓰레드가 끼어들지 못하게 하려고 변수를 읽고 쓰는 모든 문장을 synchronized블럭으로 감쌀 수도 있지만, 

더 간단한 방법이있다. volatile을 붙이는 것이다.

volatile long sharedVal; //long타입의 변수를 원자화
volatile double sharedVal; //double타입의 변수를 원자화

volatile은 해당 변수에 대한 읽기나 쓰기가 원저화된다. 원자화라는 것은 작업을 더 이상 나눌 수 없게한다는 의미인데,

synchronized블럭도 일종의 원자화라고 할 수 있다.

즉, synchronized블럭은 여러 문장을 원자화 함으로써 쓰레드의 동기화를 구현한 것이라고 보면된다.

volatile은 변수의 읽거나 쓰기를 원자화 할 뿐, 동기화하는 것은 아니라는 점에 주의하자, 동기화가 필요 할 때 synchronized 블럭대신 volatile을 쓸 수 없다.

무슨 말인가?

volatile  long balace; //인스턴스 변수 balance를 원자화 한다.
synchronized int getBalance(){ //balance 의 값을 반환한다.
return balance;
}
synchronzized void withdraw(int money){ //balance의 값을 변경
	if(balance>=money){
	balance -=money;
	}
}

인스턴스 balance를 volatile로 원자화 했으니까, 이 값을 읽어서 반환하는 메서드 getBalance()를 동기화할 필요가 없다고 생각할 수 있으나, getBalance를 synchronized로 동화하지 않으면, withdraw가 호출되어 객체에 lock을 걸고 

작업을 수행하는 중인데도 getBalance가 호출되는 것이 가능해진다. 출금이 진행중일때는 기다렸다가

출금이 끝난 후 에 잔고를 조회할 수 있도록 하려면, getBalance에 synchronized를 붙여서 동기화를 해야한다.

---

fork & join 프레임웍

프로그래밍할때 멀티코어를 잘 활용할 수 있는 멀티쓰레드 프로그래밍이 점점 중요해지고 있다.

지금봐왔지만 솔직히 멀티쓰레드 프로그래밍이 쉽지않을 뿐더러 아직도 잘모르겠다..

fork & join 프레임웍이 추가되었고 , 이 프레임웍은 하나의 작업을 작은 단위로 나누어서 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어준다.

먼저 수행할 작업에 따라 RecursiveAction 과 RecursiveTask 두 클래스 중에서 하나를 상속 받아 구현해야한다.

RecurviceAction // 반환값이 없는 작업을 구현할할 때 사용
RecurviceTask // 반환값이 있는 작업을 구현할할 때 사용

두 클래스 모두 comput()라는 추상 메서드를 가지고 있는데, 우리는 상속을 통해 이 추상 메서드를 구현하기만 하면된다.

RecursiveActon클래스

public abstract class RecursiveAction extends ForkJoinTask<Void> {
    private static final long serialVersionUID = 5232453952276485070L;

    public RecursiveAction() {}

    protected abstract void compute();

    public final Void getRawResult() { return null; }

    protected final void setRawResult(Void mustBeNull) { }
}

RecursiveTask클래스

public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
    private static final long serialVersionUID = 5232453952276485270L;

    public RecursiveTask() {}

    @SuppressWarnings("serial") // Conditionally serializable
    V result;

    protected abstract V compute();
...

보는바와 같이 모두 상속을 통해 구현해주면된다.

예를 들어 1부터 n까지의 합을 계산한 결과를 돌려주는 작업의 구현은 다음과 같다.

class sumTask extned RecursiveTask<Long> //RecursiveTask
	long from, to;
    SumTask(long from, long to){
    	this.from = from;
        this.to = to;
	}
    public Long compute(){
    //계산할 식
    }
}

이 후에 쓰레드 풀과 수행할 작업을 생성하고, invoke()로 작업을 시작한다. 쓰레드를 시작할 때 run()이 아니라

start()를 호출하는 것처럼, fork & join프레임웍으로 수행할 작업도 comput()가 아닌 invoke()로 시작한다.

ForkJoinPool pool = new ForkJoinPool(); //쓰레드 풀을 생성
SumTask task = new SumTask(from, to);//수행할 작업 생성
Long result = pool.invoke(task); //invoke()를 호출해서 작업을 시작

ForkJoinPool은 fork & join 프레임웍에서 제공하는 쓰레드 풀로, 지정된 수의 쓰레드를 생성해서 미리 만들어 놓고

반복해서 재사용할 수 있게 한다. 그리고 쓰레드를 반복해서 생성하지 않아도 된다는 장점과 너무 많은 쓰레드가 생

성되어 성능이 저하되는 것을 막아준다.

쓰레드 풀은 쓰레드가 수행해야하는 작업이 담긴 큐를 제공하며, 각 쓰레드는 자신의 작업 큐에 담긴 작업을 순서대로 처리한다.

compute()의 구현

compute()를 구현할 때는 수행할 작업 외에도 작업을 어떻게 나눌 것인가에 대해서도 알려줘야한다.

	public void compute(){
		long size = to - from +1;//from <=i<=to
		if(size<=5) { // 더할 숫자가 5개 이하면
			return sum(); //숫자의 합을 반환, sum()은 from부터 to까지의 수를 더해서 반환
		}
		long harf = (from+to)/2; //범위를 반으로 나누어서 두개의 작업으 생성
		sumTask leftSum = new SumTask(from, half);
		sumTask rightSum = new SumTask(half+1, to);
		leftSum.fork(); //작업(leftSum)을 작업 큐에 넣는다.
		return reghtSum.comput() + leftSum.join(;)
	}

실제 수행할 작업은 sum뿐이고, 나머지는 수행할 작업의 범위를 반으로 나누어서 새로운 작업을 생성해서 실행시키기 위한 것이다. 좀 복잡해 보이지만, 작업의 범위를 어떻게 나눌 것인지만 정의해 주면 나머지는 항상 같은 패턴이다.

여기서는 지정된 범위를 절반으로 나누어서 나눠진 범위의 합을 계산하기 위한 새로운 SumTask를 생성하는데, 이 과정은 작업이 더 이상 나눠질 수 없을 때까지, size의 값이 5보다 작거나 같을 때까지 반복된다.

compute()구조는 일반적인 재귀호출 매서드와 동일하다.

이해를 돕기 위해 1~8까지의 숫자를 더하는 과정을 나타냈다.

이 그림에서는 작업의 size가 2가 될때까지 나눈다. compute()가 처음 호출되면, 더할 숫자의 범위를 반으로 나눠서 한 쪽에는 fork()를 호출해서 작업 큐에 저장한다. 

하나의 쓰레드는 compute를 재귀호출하면서 작업을 계속해서 반으로나누고, 다른 쓰레드는 fork()에 의해 작업 큐에 추가된 작업을 수행한다.

다른 쓰레드의 작업 훔쳐오기

fork()가 호출되어 작업 큐에 추가된 작업 역시, comput()에 의해 더 이상 나눌 수 없을 때까지 반복해서 나뉘고, 자신의 작업 큐가 비어있는 쓰레드는 다른 쓰레드의 작업 큐에서 작업을 가져와서 수행한다. 이것을 작업 훔쳐오기(work stealing)라고 하며, 이과정은 모드 쓰레드 풀에 의해 자동적으로 이루어진다.

이를 통해 한 쓰레드에 작업이 몰리지 않고, 여러 쓰레드가 골고루 작업을 나누어 처리하게 된다.

대신 작업의 크기를 충분히 작게해야 각 쓰레드가 골고루 작업을 나눠가질 수 있다...

 

---

fork()와 join()

fork() 는 작업을 쓰레드의 작업 큐에 넣는 것이고, 작업 큐에 들어간 작업은 더 이상 나눌 수 없을 때까지 나뉜다.

즉 compute()로 나누고, fork()로 작업 큐에 넣는 작업이 계속해서 반복된다. 그리고 나누어진 작업은 각 쓰레드가

골고루 나눠서 처리하고, 작업의 결과는 join()을 호출해서 얻을 수 있다.

fork(), join()의 중요한 차이점은 fork()는 비동기 메서드이고, join()은 동기 메서드이다.

 fork() // 해당 작업을 쓰레드 풀의 작업 큐에 넣는다. = 비동기 메서드
 join() // 해당 작업의 수행이 끝날 때까지 기다렸다가. 수행이 끝나면 그 결과를 반환한다.= 동기메서드

비동기 메서드는 일반적인 메서드와 달리 메서드를 호출만 할 뿐, 그 결과를 기다리지 않는다.

(내부적으로는 다른 쓰레드에게 작업을 수행하도록 지시만 하고, 결과를 기다리지 않고 돌아온다.)

그래서 아래의 코드처럼, fork()를 호출하면 결과를 기다리지 않고 다음 문장인 return으로 넘어간다.

return 문에서 compute()가 재귀 호출 될때, join()은 호출되지 않는다. 그러다가 작업을 더 이상 나눌 수 없게 되었을 때, compute() 의 재귀호출은 끝나고join()의 결과를 기다렸다가 더해서 결과를 반환한다. 재귀호출된 compute()가 모두 종료 될 때, 최종 결과를 얻는다.

publc Long compute(){
...
	SumTask leftSum = new SumTask(from, half);
	SumTask rightSum = new SumTask(half + 1, to);
	leftSUm.fork();
    return rightSum.compute() + leftSum.join();
}

예제를 통해 확인해보자

package ch13.Synchronized;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class SynchronizedEx6 {
	static final ForkJoinPool pool = new ForkJoinPool();

	public static void main(String[] args) {
		// TODO Auto-generated method stub
		long from = 1L, to = 100_000_000L;
		
		SumTask task = new SumTask(from, to);
		
		long start = System.currentTimeMillis();
		Long result = pool.invoke(task);
		System.out.println("Elapsed time(4core) : " + (System.currentTimeMillis() - start));
		
		System.out.printf("sum of %d~%d = %d%n", from, to, result);
		System.out.println();
		
		result = 0L;
		start = System.currentTimeMillis();
		for (long i = from; i <= to; i++) {
			result += i;
		}
		System.out.println("Elapsed time(1 Core) : " + (System.currentTimeMillis() - start));
		System.out.printf("sum of %d~%d =  %d%n", from, to, result);
	}
}

@SuppressWarnings("serial")
class SumTask extends RecursiveTask<Long> {
	long from, to;
	SumTask(Long from, long to) {
		this.from = from;
		this.to = to;
	}

	@Override
	protected Long compute() {
		// TODO Auto-generated method stub
		long size = to - from + 1;
		if (size <= 5) {
			return sum();
		}
		long half = (from + to) / 2;
		
		SumTask leftSum = new SumTask(from, half);
		SumTask rightSum = new SumTask(half + 1, to);
		leftSum.fork();
		return rightSum.compute()+leftSum.join();
	}

	long sum() {
		long tmp = 0L;
		for (long i = from; i <= to; i++) {
			tmp += i;
		}
		return tmp;
	}
}

1~100000000까지 더한 결과를 반영하도록 했다.

실행결과를 보면 fork&join프레임웍으로 게산한 결과보다 for문으로 계산한 결과가 시간이 덜걸린 것을 알 수있다. 

왜냐하면

작업을 나누고 다시 합치는데 걸리는 시간이 있기 떄문이다.

재귀 호출보다 for문이 더 빠른 것과 같은 이유이다.

항상 멀티쓰레드로 처리하는 것이 빠르다고 생각해서는 안된다.

반드시 테스트해보고 이득이 있을 때만 멀티쓰레드로 처리하도록 하자.

---

이로써 쓰레드의 학습을 마쳤다. 한번더 복습을 통해서 이해단계까지 끌어올리기위해 참고해야할 지식이나.

관련 자료들도 참고하면서 공부해야한다. 주말동안 하면될것같다.

그리고 칸반보드를 드디어 1주일만에 움직였다. 마지막까지 화이팅이다.!