멀티코어컴퓨팅 - 5. Java Concurrency Utilities

Java Concurrency Utilities

concurrent application을 만드는데 도움이 되는 java 패키지들이 잘 정리되어 있다.

참고)   https://jenkov.com/tutorials/java-util-concurrent/index.html

• Java BlockingQueue : 이 인터페이스를 상속해서 ArrayblockingQueue, LinkedBlockingQueue등이 만들어진다.
• Java ArrayBlockingQueue : 큐가 꽉 찼거나 비어있으면 en/dequeue 하려는 쓰레드들을 block하는 등의 기능이 구현되어 있다. (array기반으로 된 큐이다.), 동시에 접근하는 operation에 대해서 concurrency를 보장한다. add()대신 put()을 사용하면 큐가 꽉 차있을때 enqueue하려는 쓰레드는 기다린다. take()를 사용하면 큐가 비어있을 때 dequeue하려는 쓰레드는 기다린다.
• Java DelayQueue
• Java LinkedBlockingQueue : linked list로 구현된 큐이다.
• Java PriorityBlockingQueue : 우선순위 큐로 구현되어 있다.
• Java SynchronousQueue
• Java BlockingDeque : 덱 인터페이스
• Java LinkedBlockingDeque : 구현된 concurrent 덱
• Java ConcurrentMap : 인터페이스이다. 실제로 구현한 클래스는 아래 나열되어 있다.
• Java ConcurrentNavigableMap : 구현된 concurrent 맵
• Java CountDownLatch : countdownlatch는 어떤 쓰레드가 다른 쓰레드에서 작업이 완료될 때 까지 기다릴 수 있도록해주는 클래스이다.
• Java CyclicBarrier : 쓰레드들이 멈춰야 하는 barrier를 설정해서 멈추게할 수 있다. 그리고 다시 동시에 시작하게도 할 수 있다.
• Java Exchanger
• Java Semaphore : 다음 장에서 자세히 설명함
• Java ExecutorService
• Java Callable
• Java Future
• Java ThreadPoolExecutor
• Java ScheduledExecutorService
• Java ForkJoinPool
• Java Lock : 공유하는 자원이 lock을 한개 소유하고 자원을 점유한 쓰레드에게 lock을 준다. 일을 마치면 unlock으로 lock을 반환한다. 그래야 다음 thread가 lock을 소유할 수 있게 된다.
• Java ReadWriteLock : 여러 reader는 writer가 없을 때 자원에 모두 접근 가능. 한명이라도 writer가 있으면 reader 전부 block
• Java AtomicInteger : 여러 쓰레드가 공유하는 integer 타입 자원에 대해 접근하는 함수들의 synchronized가 보장되어 있다.
• Java AtomicLong
• Java AtomicReference
• Java AtomicStampedReference
• Java AtomicIntegerArray
• Java AtomicLongArray
• Java AtomicReferenceArray

 

volatile

https://nesoy.github.io/articles/2018-06/Java-volatile

Java volatile이란?

volatile keyword는 Java 변수를 Main Memory에 저장하겠다라는 것을 명시하는 것이다.

volatile로 선언한 변수에 대해서 thread들은 값을 read/write할 때 main memory까지 가서 작업한다.

보통 multithread 환경에서는 성능 향상을 위해 메인 메모리에서 가져온 값을 각각 캐쉬에 저장해두고 쓰레드들이 작업을 한다. 즉 변수의 값을 읽을 때는 cahce에서 값을 불러오고 값을 쓴다. 이 때 값 불일치 문제가 발생할 수 있다.

public class SharedObject {
    public volatile int counter = 0;
}

이렇게 변수를 메인메모리에 선언해두면 모든 thread가 메인 메모리에서 값을 가져오고 쓴다.

보통은 Multi Thread 환경에서 하나의 Thread만 read & write하고 나머지 Thread가 read하는 상황에서 가장 최신의 값을 보장한다.

메모리의 값을 업데이트하는 쓰레드가 여러 개일 경우는 추가로 synchronization(atomicity)이 필요하다.

 

Exercise 1

class C extends Thread {
  int i;
  C(int i) { this.i = i; }
  public void run() {
    System.out.println("Thread " + i + " says hi");
    try { 
      sleep(500);
    } catch (InterruptedException e) {}
    System.out.println("Thread " + i + " says bye");
  }
}

public class ex1 {
   private static final int NUM_THREAD = 10;
  public static void main(String[] args) {
    System.out.println("main thread start!");
    C[] c = new C[NUM_THREAD];
    for(int i=0; i < NUM_THREAD; ++i) {
      c[i] = new C(i);
      c[i].start();
    }
    System.out.println("main thread calls join()!");
    for(int i=0; i < NUM_THREAD; ++i) {
      try { 
        c[i].join();
      } catch (InterruptedException e) {}
    }
    System.out.println("main thread ends!");
  }
}

C쓰레드 10개를 만들고 실행시킨다. 각 스레드는 hi 쉬고 bye를 출력한다.

메인쓰레드가 10개의 c 쓰레드들이 끝나기를 join()으로 기다리고 실험 종료 메시지를 출력할 수 있도록 한다.

 

Exercise 2

package test;


public class Test {
  public static void main(String[] args) {
    int[] int_arr = new int [10000]; 
    int i,s;
    for (i=0;i<10000;i++) int_arr[i]=i+1;
    s=sum(int_arr);
    System.out.println("sum=" + s +"\n");
  }

  static int sum(int[] arr) {
    int i;
    int s=0;
    for (i=0;i<arr.length;i++) s+=arr[i];
    return s;
  }
}

1부터 10000까지 합을 구하는 코드이다. 병렬처리로 빠르게 계산을 해보자.

class SumThread extends Thread {
  int lo; // fields for communicating inputs
  int hi;
  int[] arr;
  int ans = 0; // for communicating result
  SumThread(int[] a, int l, int h) {
    lo=l; hi=h; arr=a;
  }
  public void run() { // overriding, must have this type
    for(int i=lo; i<hi; i++)
      ans += arr[i];
  }
}

class ex2 {

  private static int NUM_END=10000;
  private static int NUM_THREAD=4;

  public static void main(String[] args) {
    if (args.length==2) {
      NUM_THREAD = Integer.parseInt(args[0]);
      NUM_END = Integer.parseInt(args[1]);
    } 
    System.out.println("number of threads:"+NUM_THREAD);
    System.out.println("sum from 1 to "+NUM_END+"=");

    int[] int_arr = new int [NUM_END]; 
    int i,s;
    
    for (i=0;i<NUM_END;i++) int_arr[i]=i+1;
    s=sum(int_arr);
    System.out.println(s);
  }

  static int sum(int[] arr) {
    int len = arr.length;
    int ans = 0;
    SumThread[] ts = new SumThread[NUM_THREAD];
    for(int i=0; i < NUM_THREAD; i++) {
      ts[i] = new SumThread(arr,(i*len)/NUM_THREAD,((i+1)*len)/NUM_THREAD);
      ts[i].start();
    }
    try {
      for(int i=0; i < NUM_THREAD; i++) {
        ts[i].join();
        ans += ts[i].ans;
      }
    } catch (InterruptedException IntExp) {
    }

    return ans;
  }
}

1부터 10000까지 범위를 나눠서 총 4개의 쓰레드가 각각 합을 구한다.

그다음 main쓰레드는 join()으로 모든 쓰레드들이 작업을 마치고 각자의 결과를 완전히 구할때까지 기다린다.

그 결과들을 다 더해서 최종 합을 반환한다.

Exercise 3

적분 면적을 병렬처리로 빠르게 구해보자.

class IntThread extends Thread {
  int my_id; // fields for communicating inputs
  int num_steps;
  int num_threads;
  double sum;

  IntThread(int id, int numSteps, int numThreads) {
    my_id=id; num_steps=numSteps; num_threads=numThreads;
    sum=0.0;
  }
  public void run() { 
    double x;
    int i;
    int i_start = my_id * (num_steps/num_threads);
    int i_end = i_start + (num_steps/num_threads);
    double step = 1.0/(double)num_steps;
    for (i=i_start;i<i_end;i++) { 
      x=(i+0.5)*step;
      sum=sum+4.0/(1.0+x*x);
    }
    sum = sum*step;
    System.out.println("myid"+my_id+", sum=" + sum);
  }
  public double getSum() { return sum; }
}

class ex3 {
  private static int NUM_THREAD = 4;
  private static int NUM_STEP = 1000000000;

  public static void main(String[] args) {
    int i;
    double sum=0.0;

    if (args.length==2) {
      NUM_THREAD = Integer.parseInt(args[0]);
      NUM_STEP = Integer.parseInt(args[1]);
    } 

    long start = System.currentTimeMillis();
    IntThread[] ts = new IntThread[NUM_THREAD];
    for(i=0; i < NUM_THREAD; i++) {
      ts[i] = new IntThread(i,NUM_STEP,NUM_THREAD);
      ts[i].start();
    }
     
    try {
      for (i=0;i<NUM_THREAD;i++) {
        ts[i].join();
        sum += ts[i].getSum();
      }
    } catch (InterruptedException e) {}
    long finish = System.currentTimeMillis();

    System.out.println("integration result=" + sum +"\n");
    System.out.println("Execution time (" + NUM_THREAD + "): "+(finish-start)+"ms");
  }

}