a developer: java concurrent

顯示具有 java concurrent 標籤的文章。顯示所有文章

Off Heap Cache - OHCache - Simple Example

Reference

https://docs.oracle.com/javase/8/docs/api/java/nio/MappedByteBuffer.html
https://github.com/shooeugenesea/study-practice/tree/ohc

What

OHC 是一個 off heap 的 cache library

Why

如果是用 map 做 cache, 資料都在 heap 裡面, heap 裡面的問題就是

會被 GC, GC 就會週期性的對 JVM 效能產生影響
如果要 persist 需要另外實作

如果是放在 off heap, 比方說用 mmap, 或是用 MappedByteBuffer access 的記憶體, 就是放在 heap memory 之外, 由作業系統控制著.

好處是記憶體量不用被 heap size 限制, 缺點是要自己小心控制, 不然會造成系統問題.'

How

Maven

<dependency>
 <groupId>org.caffinitas.ohc</groupId>
 <artifactId>ohc-core</artifactId>
 <version>0.4.4</version>
</dependency>

Codes

public class SimpleMain {

    public static void main(String[] params) {
        Scanner scanner = new Scanner(System.in);
        while(scanner.hasNextLine()) {
            if (scanner.nextLine().trim().equals("GO")) {
                System.out.println(Runtime.getRuntime().freeMemory()/1024/1024 + "/" + Runtime.getRuntime().totalMemory()/1024/1024);
                OHCache<String,String> c = OHCacheBuilder.<String,String>newBuilder()
                        .keySerializer(new StringSerializer())
                        .valueSerializer(new StringSerializer())
                        .build();
                c.put("A", "A");
                System.out.println(c.get("A"));
                IntStream.range(0, 1000_000).forEach(idx -> {
                    c.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
                });
                System.out.println(Runtime.getRuntime().freeMemory()/1024/1024 + "/" + Runtime.getRuntime().totalMemory()/1024/1024);
            } else {
                System.out.println("input 'GO' then click enter");
            }
        }
    }

    private static class StringSerializer implements CacheSerializer<String> {

        @Override
        public void serialize(String s, ByteBuffer byteBuffer) {
            byteBuffer.put(s.getBytes());
        }

        @Override
        public String deserialize(ByteBuffer byteBuffer) {
            return StandardCharsets.UTF_8.decode(byteBuffer).toString();
        }

        @Override
        public int serializedSize(String s) {
            return s.getBytes().length;
        }
    }

}

Output

這張是輸入了好幾次 GO, 每次都會輸入一百萬次 key value 到 cache.

jconsole 與系統資源使用圖可以看到 heap 只用不到 100MB, 但實際上已經用了 600MB.
就是都放在 off heap 中

ForkJoinPool - Thread Management

Reference

ForkJoinPoolManyThreadMain.java

ForkJoinPool Paper
ForkJoinPool - Error Handling

Question

原本以為 new ForkJoinPool(2) 像這樣的宣告, 是讓 ForkJoinPool 最多維持 2 個 thread 來執行.

可是當實際觀察的時候, 卻發現有非常多個 thread 被叫起來跑.

比方說原本 fork -> join 的範例, 如果把 thread name 印出來, 就會發現有很多個 thread 被叫起來.

而且每秒印出的 pool 也會發現 pool size 變大.

Codes: Thread size exceed parallel level

public class ForkJoinPoolManyThreadMain {

    static final CountDownLatch latch = new CountDownLatch(1);
    private static final AtomicInteger threadCount = new AtomicInteger();

    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> {
            System.out.println(pool); }, 0, 1, TimeUnit.SECONDS);
        pool.execute(new MainAction());
        latch.await();
        s.shutdown();
        pool.shutdown();
    }


    private static class MainAction extends RecursiveAction {

        @Override
        protected void compute() {
            List<SubAction> actions = IntStream.range(0,100).mapToObj(idx -> new SubAction()).collect(toList());
            actions.forEach(SubAction::fork);
            actions.forEach(SubAction::join);
            latch.countDown();
        }
    }

    private static class SubAction extends RecursiveAction {

        @Override
        protected void compute() {
            try {
                System.out.println(Thread.currentThread().getName() + " sleep 1 seconds");
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }


}

Output

從印出來的 log 就可以發現: thread 超過 2 個到達 14 個. pool size 實際上也到達 16

Requirement

發現這件事情的時候覺得無法理解, 同時也發現原本以為 main task 的 thread 是跟被 join 的 thread 合再一起了. 結果看起來是有很多新的 thread 被產生出來執行. 如此我們怎麼確保不會有過多的 thread 被建立出來?

Suggestion: Customized ThreadFactory? (Don't do this)

因為 ForkJoinPool 裡的這段程式, 網路上有人建議是提供自己的 thread factory, 如果超過指定的 thread 數量就回傳 null. 如此就被 ForkJoinPool 當成一個錯誤而繼續使用既有的 thread.

這個做法的確可以控制建立的 thread 數量 , 不過回傳 null 對 ForkJoinPool 來說是一個錯誤, 以 ForkJoinPool 的規則來說這個時候是要 createWorker 的, 不應該出現錯誤而回傳 null.

How: Join after done

突然想到會產生新的 thread 是不是因為 join 導致 MainAction 停住, 但又不是正在執行, 所以為了避免 starvation 使 ForkJoinPool 需要建立一個新的 thread. 所以試著先判斷 isDone, done 才 join. 結果就可以了. (但還不清楚原因)

Working Code

public class ForkJoinPoolFixedThreadMain {

    public static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        MainAction action = new MainAction();
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.execute(action);
        latch.await();
        s.shutdown();
    }

    private static class MainAction extends RecursiveAction {

        @Override
        protected void compute() {
            List<SubAction> actions = IntStream.range(0,100).mapToObj(idx -> new SubAction()).collect(toList());
            actions.forEach(SubAction::fork);
            while(!actions.isEmpty()) {
                for (Iterator<SubAction> i = actions.iterator(); i.hasNext(); ) {
                    SubAction action = i.next();
                    if (action.isDone()) {
                        action.join();
                        i.remove();
                    }
                }
            }
            latch.countDown();
        }
    }

    private static class SubAction extends RecursiveAction {

        @Override
        protected void compute() {
            long start = System.currentTimeMillis();
            while (true) {
                if (System.currentTimeMillis() - start > 1000) {
                    break;
                }
            }
        }
    }


}

Output

因為呼叫 new ForkJoinPool() 所以 pool size 最多就維持在 8 (我的電腦有 8 core)

ForkJoinPool - Error Handling

Reference

Git: ForkJoinPoolErrorMain.java
前篇: ForkJoinPool - Workstealing
Next: ForkJoinPool - Thread Management

Error Handling

如果 fork 出去的 task/action 有 exception, 呼叫 join 的時候要注意 catch exception.

不然 task/action 就死掉失控了.

而且可怕的是: 不會有 log.

Codes

這段程式是說每個 action 有個 id, 奇數 id 會報 exception.

Exception without catch when join

先是不要 catch 的情況

public class ForkJoinPoolErrorMain {

    static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.submit(new ErrorAction(MAIN_TASK_ID));
        latch.await(10, TimeUnit.SECONDS);
        pool.shutdown();
        s.shutdown();
    }

    static class ErrorAction extends RecursiveAction {

        public static final int MAIN_TASK_ID = -1;
        private final int id;

        ErrorAction(int id) {
            this.id = id;
        }

        private boolean isSubTask() {
            return id >= 0;
        }

        @Override
        protected void compute() {
            if (isSubTask()) {
                if (id % 2 == 1) {
                    throw new IllegalStateException("Error when id is odd number");
                }
            } else {
                List<ErrorAction> actions = IntStream.range(0,10).mapToObj(idx -> new ErrorAction(idx)).collect(Collectors.toList());
                actions.forEach(ErrorAction::fork);
                actions.forEach(action -> {
                    action.join();
                });
                latch.countDown();
            }
            System.out.println("compute done. id=" + id);
        }
    }

}

輸出如下, 不過一個 exception 也沒有.
main action 也沒有完成.(main action id = -1)

Exception with catch statement

有 catch 的程式如下

public class ForkJoinPoolErrorMain {

    static final CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] params) throws InterruptedException {
        ForkJoinPool pool = new ForkJoinPool(2);
        ScheduledExecutorService s = Executors.newScheduledThreadPool(1);
        s.scheduleAtFixedRate(() -> System.out.println(pool), 0, 1, TimeUnit.SECONDS);
        pool.submit(new ErrorAction(MAIN_TASK_ID));
        latch.await(10, TimeUnit.SECONDS);
        pool.shutdown();
        s.shutdown();
    }

    static class ErrorAction extends RecursiveAction {

        public static final int MAIN_TASK_ID = -1;
        private final int id;

        ErrorAction(int id) {
            this.id = id;
        }

        private boolean isSubTask() {
            return id >= 0;
        }

        @Override
        protected void compute() {
            if (isSubTask()) {
                if (id % 2 == 1) {
                    throw new IllegalStateException("Error when id is odd number");
                }
            } else {
                List<ErrorAction> actions = IntStream.range(0,10).mapToObj(idx -> new ErrorAction(idx)).collect(Collectors.toList());
                actions.forEach(ErrorAction::fork);
                actions.forEach(action -> {
                    try {
                        action.join();
                    } catch (Exception ex) {
                        System.err.println(ex + ", id=" + action.id);
                    }
                });
                latch.countDown();
            }
            System.out.println("compute done. id=" + id);
        }
    }

}

這個程式就可以印 error, 而且 maintask 可以執行完成

Exception with ExecutorService

如果是 ExecutorService, 就算是丟 RuntimeException.

ExecutorService 還是會把 exception 印到 console 上.

public class ExecutorServiceWithExceptionMain {

    public static void main(String[] params) {
        ExecutorService ex = Executors.newSingleThreadExecutor();
        ex.execute(() -> {
            throw new IllegalStateException("test throw illegalState");
        });
        ex.shutdown();
    }


}

即使是直接丟 exception 也會印到 console.

ForkJoinPool - Work stealing

Reference

https://en.wikipedia.org/wiki/Work_stealing

前篇: Java Concurrency - ForkJoinPool 的 deadlock
下篇: ForkJoinPool - Error Handling

Introduction

ForkJoinPool 實作 work stealing 的概念, 簡單說明一下

Work-stealing

每個 thread 會有一個 queue, queue 裡面放的是 CPU bound 的工作
queue 裡面的工作可能會產生新的能平行作業的工作
新產生的工作會放在 queue 裡面
如果有 thread 把 queue 裡面的工作做完了, 就會去別的 queue 拿工作來處理 (steal)

Java Concurrent - ForkJoinPool

ExecutorService and Problem

用一個 ExecutorService submit task 後, 如果 task 會 submit 其他的 task, 這些 task 都會放進 ExecutorService 的 queue 中.

這樣就沒辦法做到 "第一個 task 與其產生的 subtask 完成後就印一行 log"

ForkJoinPool and Purpose

使用了 ForkJoinPool 就可以透過 task 的 fork 與 join 讓 task 變成樹狀結構.

建立 ForkJoinPool
用 ForkJoinPool submit task
task 建立幾個新的 subtask
呼叫 subtask 的 fork
呼叫 subtask 的 join
執行到 parent 的 task 結束

如此就能讓 task 自然分群

Reference and Sample

Code

public class ForkJoinPoolMain {

    private static AtomicInteger actionCounter = new AtomicInteger();
    public static void main(String[] params) throws ExecutionException, InterruptedException {
        ForkJoinPool pool = new ForkJoinPool();
        System.out.println("main:" + pool.submit(new MyAction(new int[]{1,2,3,4,5,6,7,8,9,10,11})).get());
    }

    private static class MyAction extends RecursiveTask<Integer> {
        // Extends RecursiveAction if no return value required

        private int count;
        private int[] numbers;

        public MyAction(int[] numbers) {
            this.count = actionCounter.incrementAndGet();
            this.numbers = numbers;
        }

        @Override
        protected Integer compute() {
            if (numbers.length > 2) {
                System.out.println(count + " split " + Arrays.toString(numbers) + " to 2 arrays");
                List<MyAction> actions = new ArrayList<>();
                actions.add(new MyAction(Arrays.copyOfRange(numbers, 0, numbers.length / 2)));
                actions.add(new MyAction(Arrays.copyOfRange(numbers, numbers.length / 2, numbers.length)));
                actions.forEach(MyAction::fork);
                return actions.stream().mapToInt(MyAction::join).sum();
            } else if (numbers.length == 2) {
                System.out.println(count + " add " + numbers[0] + " and " + numbers[1]);
                return numbers[0] + numbers[1];
            }
            System.out.println(count + " just return " + numbers[0]);
            return numbers[0];
        }
    }

}
下篇：ForkJoinPool 沒處理好也會 deadlock

Java Concurrent CyclicBarrier

Introduction

CyclicBarrier 很久沒用都忘了, 來複習一下..

Description

簡單說就是很多個 thread 去作事情, 作完之後就 await.

CyclicBarrier 等作完的 thread 達到指定的個數後, await 就結束.

Code

public class CyclicBarrierMain {

    public static void main(String[] params) throws InterruptedException {
        int runner = 5;
        CountDownLatch gameOver = new CountDownLatch(runner);
        CyclicBarrier b = new CyclicBarrier(runner, () -> {
            System.out.println("barrier done");
        });
        IntStream.range(0,runner).forEach(idx -> {
            new Thread() {
                @Override
                public void run() {
                    try {
                        System.out.println("wait " + idx);
                        b.await();
                        System.out.println("count down " + idx);
                        gameOver.countDown();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }.start();
        });
        gameOver.await();
        System.out.println("game over");
    }

}

Java Concurrent Phaser

Introduction

今天才知道 Java 7 有提供一個 Phaser class 來處理更複雜的 CountDownLatch 需求.

Simple Phaser

想用 Phaser, 首先你的 threads 工作要有分階段, 一個階段沒做完就不能進行下個階段.
有了這個限制後就可以照下面的流程做事情

先註冊, 看有幾個 thread 要做事情.
=> 透過呼叫 Phaser#register 去註冊
做完之後呼叫 arrive, 告訴 Phaser 有事情做完了
當每個 thread 都把事情做完, Phaser 就會進入下一個階段, 稱做 advance
當進入下一階段, Phaser 的 onAdvance method 就會被呼叫,
這個 method 是留給我們 developer 去實做

以上. 簡單的概念, 覺得看好久都沒看懂, 希望看這篇的人可以簡單看懂.

接著是範例, 這範例是

假設有幾個人賽跑
首先要到達起跑線
都到了以後才開跑
每個人都到達終點舊比賽結束.

public class PhaserRacingGame {

    public static void main(String[] params) {
        Random r = new Random();
        Phaser perGamePhase = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("phase:" + phase + ", registeredParties:" + registeredParties + ", arrived:" + getArrivedParties());
                switch (phase) {
                    case 0:
                        System.out.println("start!");
                        break;
                    case 1:
                        System.out.println("game finish");
                        break;
                    default:
                        System.out.println("unexpected:" + phase);
                }
                return super.onAdvance(phase, registeredParties);
            }
        };
        IntStream.range(0,5).forEach(idx -> {
            perGamePhase.register();
            new Thread(() -> {
                System.out.println(idx + " arrive starting line and wait for start");
                perGamePhase.arriveAndAwaitAdvance();
                System.out.println(idx + " run!! startTime:" + System.currentTimeMillis());
                long spendTime = r.nextInt(1000);
                sleep(spendTime);
                perGamePhase.arrive();
            }).start();
        });
        sleep(5000);
    }

    private static void sleep(long ms) {
        try {
            TimeUnit.MILLISECONDS.sleep(ms);
        } catch (InterruptedException e) {
        }
    }

}

Parent Phaser

再來比較複雜的是 Parent Phaser, Parent Phaser 的目的是減少 synchronization 的時候搶 lock 的碰撞.

使用 Parent Phaser 的時候要注意的是: advance 是發生在 Parent Phaser 而不是 Child Phaser,

使用 Parent Phaser 的時候, 是可以確保每個 Child Phaser 準備號都進入下一個 phase 的時候, 已經先到達的 Child Phaser 才可以更進一步 arrive.

比方說世足十六強賽, 要等所有能進入八強賽的隊伍都出線之後才可以開始進行八強賽.

那 Parent Phaser 就是賽程, 十六強賽則是有八個 Child Phaser.

以下範例是看出特性, 方便 copy paste 觀察

public class ParentPhaser {

    public static void main(String[] params) {
        Phaser parent = new Phaser() {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("parent phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        Phaser child1 = new Phaser(parent, 4) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("child phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        Phaser child2 = new Phaser(parent, 3) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("child phase:" + phase + ", registeredParties:" + registeredParties);
                return super.onAdvance(phase, registeredParties);
            }
        };
        IntStream.range(0,10).forEach(idx -> {
            child1.arrive();
            child1.arrive();
            child1.arrive();
            child1.arrive();
//            child1.arrive(); // cause error, child1 need wait for child2 finish
            child2.arrive();
            child2.arrive();
            child2.arrive();
        });
    }


}

訂閱：文章 (Atom)