kiến thức [Event Tặng Title] Concurrency/multi-threading in the nut shell

Fire Of Heart said:

các thím nghĩ sao về concurrency trên máy chỉ có 1 core?

Click to expand...

Fire Of Heart said:

Mình ko hiểu sao thím phải đi chứng minh cái này, có thể mình đang ko hiểu điều thím nói, nhưng ng ta định nghĩa:
1 giây là khoảng thời gian bằng 9 192 631 770 lần chu kỳ của bức xạ điện từ phát ra bởi nguyên tử Cs133 khi thay đổi trạng thái giữa hai mức năng lượng đáy siêu tinh vi. (wiki).
Vậy thì 1/6 tỷ giây của thím nó đâu đó là khoảng thời gian 1.5 chu kỳ còn gì?

Chưa hiểu điều thím đang băn khoăn ở đây là gì?

Click to expand...

Kurt Godel said:

Bởi vì các phép tính xảy ra trong 1 core nó tác động đến Physical memory có thể concurrency trong 1 thời điểm siêu vi hay không? Bởi vì con người chưa bao giờ loại bỏ hoàn toàn điện trở trong vật chất cấu tạo core nên nó (điện trở) vẫn tồn tại ảnh hưởng đến vận tốc trôi của electron => tác động đến xung nhịp. Hay nói cách khác là điều kiện để một xung nhịp tạo ra n phép tính, trong n phép tính/lần xung nhịp ấy để mà có các phép tính xảy ra đồng thời thì cần duy nhất 1 thời điểm hay cần n thời điểm?

Người ta phải chứng minh rằng siêu vi thời gian ấy trong vũ trụ có tồn tại một siêu vi thời gian như vậy hay không để quy chiếu về. Lúc này thì lại cần tới Các định lý bất toàn của Gödel để đi tìm chứng minh (mà chưa ai tìm ra).

Bất kỳ vật chất nào trong nó cũng có điện trở. Nên người ta đang sử dụng qubit (bit lượng tử) để đưa một phép tính xảy ra là gần như ->0 ở mặt thời gian. Hay nói cách khác họ cũng đang chứng minh ở mốc thời gian 1/100 tỷ s là concurrency với 1/100n tỷ s (n+1).

Click to expand...

Fire Of Heart said:

hmm đọc tầm vài lần thì tôi bắt đầu hình dung dc 1 cách mơ hồ rồi.
Hỏi tí trc thím học CS ở trường nào vậy, giờ đang làm mảng nào thế...

Click to expand...

recca2007 said:

Các bác cho hỏi Concurrency và Parallel thì với những CPU đời mới bây giờ cái nào hiệu quả hơn?
Thực thế áp dụng như nào thì hợp lý?

Click to expand...

QTQTQT said:

Về lý thuyết thì 1 cái công tắc chỉ có đóng hoặc mở tại 1 thời điểm, nên tại 1 thời điểm thì chỉ 1 phép tính. Có chăng phép tính đó cực nhanh và trong 1 khoảng thời gian rất ngắn thì có thể thực hiện 2 lệnh.

Tiếp theo nữa việc tối ưu từ hardware hay software cũng chỉ để tận dụng để hardware hoạt động hiệu quả nhất. Thread ở software hay intrusion hardware cũng thế thôi, trong lúc chờ các tác vụ i/o hoặc nạp/xoá thì đi làm việc khác.


Cpu mới thì nhiều core nên parallel và ở từng core thì vẫn concurrency thôi.

Click to expand...

Fire Of Heart said:

hmm đọc tầm vài lần thì tôi bắt đầu hình dung dc 1 cách mơ hồ rồi.
Hỏi tí trc thím học CS ở trường nào vậy, giờ đang làm mảng nào thế...

Click to expand...

LoserThapNghiep said:

Với cơ chế pipline thì concurency nó hoạt động đc trêb cả con pic16 luôn rồi ấy thím.

Click to expand...

Kurt Godel said:

Mình trước học ở VN sau học PhD tại Pháp, giờ đang làm tại pháp về Server System Infrastructure.

Click to expand...

QTQTQT said:

Thật ra em không hiểu sâu đâu thím. Với pipeline gồm 5 step kia. Thì vs em nó chỉ như 5 thằng công nhân trong dây chuyền. 1 core là 1 dây chuyền. Nếu làm trên N core thì sẽ có N thằng parallel step1, dây chuyền 1 xong step1 cũng có thể đẩy sang step2 ở dây chuyền khác nhưng nó sẽ phải đi đường dài hơn giữa các core vs nhau thay vì trong cùng core

Click to expand...

bribnt said:

Nhưng mà CPU Core bây giờ khác nhiều rồi. Nó không chỉ là cái nhà máy với 1 dây chuyền 5 bước cho 5 công nhân làm từng bước. Giờ với mỗi bước có thể lại là nhiều dây chuyền nhỏ giống hệt nhau, có thể hoạt động song song độc lập cho ra cùng lúc nhiều kết quả.

Lấy nhà máy đóng hộp làm ví dụ đi, một trong số các bước là "in vỏ" --> "cắt" --> "đóng hộp", sau khi đã in xong trên một tấm nền lớn thì có thể dùng nhiều dao để cắt cùng lúc ở các vị trí khác nhau, cắt xong rồi thì cũng có thể cùng lúc dùng nhiều máy đóng gói thành hộp. Tức là cùng một lúc có thể cho ra nhiều sản phẩm, mặc dù chỉ với 1 nhà máy.

Bạn có vẻ vẫn hiểu khái niệm core/thread hơi cứng nhắc, nó không phải là thứ gì đó cơ bản không phân chia được. Đúng là 1 core chỉ chạy được 1 thread tại 1 thời điểm (ko tính hyperthreading), nhưng các instruction trong thread đó nếu không phụ thuộc gì với nhau thì hoàn toàn có thể thực thi song song mà không sợ bị lỗi gì.
https://en.wikipedia.org/wiki/Instruction-level_parallelism

Click to expand...

QTQTQT said:

Em không hiểu quá sâu, nên nhiều cái nhận định có thể theo tư duy chủ quan của mình.

Thế giờ nếu giờ em gọi thằng excute từng instruction là instruction executor. Thằng executor này có thực hiện 2 execution unit cùng lúc không.

Ví dụ nhiều dao để cắt em nghĩ nó vẫn là 1 execution unit chỉ là được tối ưu để tận dụng excute các unit giống nhau dùng kết quả cho nhiều thằng sử dụng.

Click to expand...

Kurt Godel said:

Bởi vì các phép tính xảy ra trong 1 core nó tác động đến Physical memory có thể concurrency trong 1 thời điểm siêu vi hay không? Bởi vì con người chưa bao giờ loại bỏ hoàn toàn điện trở trong vật chất cấu tạo core nên nó (điện trở) vẫn tồn tại ảnh hưởng đến vận tốc trôi của electron => tác động đến xung nhịp. Hay nói cách khác là điều kiện để một xung nhịp tạo ra n phép tính, trong n phép tính/lần xung nhịp ấy để mà có các phép tính xảy ra đồng thời thì cần duy nhất 1 thời điểm hay cần n thời điểm?

Người ta phải chứng minh rằng siêu vi thời gian ấy trong vũ trụ có tồn tại một siêu vi thời gian như vậy hay không để quy chiếu về. Lúc này thì lại cần tới Các định lý bất toàn của Gödel để đi tìm chứng minh (mà chưa ai tìm ra).

Bất kỳ vật chất nào trong nó cũng có điện trở. Nên người ta đang sử dụng qubit (bit lượng tử) để đưa một phép tính xảy ra là gần như ->0 ở mặt thời gian. Hay nói cách khác họ cũng đang chứng minh ở mốc thời gian 1/100 tỷ s là concurrency với 1/100n tỷ s (n+1).

Click to expand...

sirhung1993 said:

Phần này thì tổ lái sang vật lý lượng tử 1 tẹo rồi. Các bit lượng tử di chuyển với tốc độ ~~ C , nên theo Einstein thì thời gian ko tồn tại ( 1s kéo vài tới vô cùng) . Thì việc chứng minh cái 1/100 tỷ là concurrency với 1/100n tỷ làm ntn nhỉ ?

Trong CPU thì đơn vị thời gian hay dùng là cycle, 1 rising-edge của Clock . Trước có làm giả lập 1 con MIPS đơn giản trên FPGA , thì đội kỹ sư đã tính toán mức xung làm sao để bảo đảm các electron kịp chạy qua các D-flipflop và kịp bật transistor cho đến rising-edge tiếp theo. Thường mức xung nhịp đã được tính toán trước và có 1 gap an toàn đủ để anh em ép xung lên 1 tẹo nữa.

Nên phần sai số do ảnh hưởng giữa các bộ phận với nhau được coi là ko đáng kể (vẫn có, nhưng thường có CRC đơn giản như parity check hoặc checksum). Nếu sai 1-2 bóng transistor thì nó tính lại. Parity check chi phát hiện được 1 bit sai lệch thôi.

Click to expand...

final Thread t1 = new Thread(() -> {
            // Do something
});
t1.start();

// Run a single task
// then wait it finish
final ExecutorService executorService = new ThreadPoolExecutor(
                // 1 core init
                1,
                // max core in runtime
                2,
                // keepAlive -> if task done, how long keep the thread in pool
                0L,
                TimeUnit.MILLISECONDS,
                // where to store the lambda function , limit 100 task
                new ArrayBlockingQueue<>(100));

// submit the task
final Future<Double> futureResult =
                executorService.submit(Math::random);

// wait 1 second to get the result , if not throw Exception
try {
    final double ret = futureResult.get(1L, Time)
} catch (ExecutionException | TimeoutException e) {
    System.out.println("timeout");
}

try {
    // Run a batch-job
            final int jobSize = 1000;
            final List<Callable<Double>> listJob = new ArrayList<>(jobSize);
            for (int i = 0; i < jobSize; i++) {
                listJob.add(Math::random);
            }

            // default get number parallel job = available cores
            final ExecutorService forkJoinPool = Executors.newWorkStealingPool();
            final BlockingQueue<Exception> exceptions = new ArrayBlockingQueue<>(128);
            final List<Double> resultBatchJob = forkJoinPool.invokeAll(listJob)
                    .stream().map(doubleFuture -> {
                        try {
                            return doubleFuture.get(1L, TimeUnit.SECONDS);
                        } catch (Exception e) {
                            try {
                                if (!exceptions.offer(e,1L, TimeUnit.SECONDS)) {
                                    System.err.println("Full queue");
                                }
                            } catch (InterruptedException ex) {
                                System.err.println("Exception, cannot push to queue");
                            }
                            return 0.0;
                        }
                    }).collect(Collectors.toList());

// Block until timeout or finish
if(!forkJoinPool.awaitTermination(1L, TimeUnit.HOURS)) {
   System.out.println("does not finish in time");
   forkJoinPool.shutdownNow();
}
            resultBatchJob.forEach(System.out::println);
            exceptions.forEach(System.out::println);
} catch (InterruptedException e) {
    System.out.println("timeout");
}

• Để main-thread biết được có lỗi xảy ra ở trong thread hay chưa, cần hứng exception vào 1 BlockingQueue ( hoặc print log và ignore)
• submit(Callable<Class>) dùng cho việc collect data về, có return
• run(Runnable) chạy và ko cần collect return ( có thể hứng vào 1 cấu trúc dữ liệu thread-safe)
• Main-thread bị block cho tới khi task được finish hết -> nếu ko set timeout cho awaitTermination thì đợi vô cực
• Các object share vào thread-scope khác bắt buộc phải thread-safe như BlockingQueue, Atomic, ConcurrentMap ...

# đây là 1 ví dụ về pipe / data pipe-line trên Shell
cat /proc/cpuinfo | grep processor | wc -l

try {
            final File fileCpuInfo = new File("/proc/cpuinfo");
            final long timeOut = 1L;
            final TimeUnit timeUnit = TimeUnit.SECONDS;
            final AtomicBoolean isPushingDone = new AtomicBoolean(false);
            final AtomicReference<Exception> exceptionInThreadScope = new AtomicReference<>(null);
            final int threadPoolSize = 2;
            final ExecutorService es = Executors.newFixedThreadPool(threadPoolSize);

            final BlockingQueue<String> fileInputQueue = new ArrayBlockingQueue<>(1024);
            final BlockingQueue<String> outputResult = new LinkedBlockingDeque<>();
            // Start while-loop in another thread
            for (int i = 0; i < threadPoolSize; i++) {
                es.execute(() -> {
                    while (!isPushingDone.get() ||
                            exceptionInThreadScope.get() == null ||
                            !fileInputQueue.isEmpty()) {
                        try {
                            final String line = fileInputQueue.poll(timeOut, timeUnit);
                            if (line != null && line.contains("processor")) {
                                // Push output to another queue -> input to next thread
                                outputResult.offer(line, timeOut, timeUnit);
                                System.out.println(line);
                            }
                        } catch (Exception e) {
                            exceptionInThreadScope.set(e);
                            System.err.println("Cannot push to queue in 1s");
                        }
                    }
                });
            }

            // Another thread-scope use output from outputResult
            final ExecutorService outputProcessor = Executors.newSingleThreadExecutor();
            outputProcessor.execute(() -> {
                while (!isPushingDone.get() || !outputResult.isEmpty()) {
                    try {
                        final String processorLine = outputResult.poll(timeOut,timeUnit);
                        if ( processorLine != null && !processorLine.isEmpty()) {
                            final String[] split = processorLine.split(":");
                            if (split.length > 0) {
                                System.out.printf("CoreId : %s%n", split[0]);  
                            }
                        }
                    } catch (Exception e) {
                        System.err.println("Output processing fail!");
                    }
                }
            });


            try (final FileReader fr = new FileReader(fileCpuInfo)) {
                try (final BufferedReader br = new BufferedReader(fr)) {
                    String currentLine = br.readLine();
                    while (currentLine != null) {
                        if (!fileInputQueue.offer(currentLine, timeOut, timeUnit)) {
                            System.err.println("fileInputQueue is full!");
                        }
                        currentLine = br.readLine();
                    }
                }
            } finally {
                isPushingDone.set(true);
            }

            // đợi các threadPool finish ở đây
           // như post trước awaitTermination


        } catch (Exception e) {
            System.err.println("File reading error!");
        }

• Ta có thể share blockingQueue trong object giữa các method hoặc chỉ khởi tạo trong 1 function-scope -> tùy vào logic xử lý.
• Set-timeout và try-catch là best-practice , exception throws trong thread-scope khác sẽ không try-catch được ở main-thread -> bị nuốt lỗi nếu ko xử lý exception trong thread-scope.
• Phần input thì thường là single-thread vì thường file hoặc db-connection chỉ cho 1 thread access ( nếu đọc file nén thì có thể 1 thread giải nén - 1 thread push data vào queue)
• Nên limit cho queue , nếu queue size ko limit -> OOM hoặc treo luôn JVM ( ở Java 11 vs GC1 thì JVM ko kill process khi OOM, mà sẽ stop-the-world vô cực -> set flag KillOnOOM để kill)
• ThreadPool cũng có 1 task-queue , bt là unbound-queue -> task push vào quá nhiều mà ko consume cũng gây OOM