Sito Eratostenesa współbieżnie.

Cześć,

Chciałem porównać czasy obliczania sita Eratostenesa dla różnej liczby wątków. Napisałem następujący program, ale nie działa tak jak trzeba.

import java.util.HashMap;

public class Sieve implements Runnable{
	private HashMap<Integer,Boolean> array = new HashMap<Integer,Boolean>();
	private int counter = 1;
	private int x;
	public void run(){
	    while(counter < x-1){
	        do{
	            counter++;
	        }
	        while( array.get(counter));
	        int tmp = counter;
	        for(int i = tmp; i < array.size(); i+=tmp){
	            if( i!= tmp)
	                array.put(i,true);
	        }
	        try{
	        Thread.sleep(0L, 1);
	        }
	        catch (Exception e){}
	    }
	}

	public void setArray(int x){
	    this.x = x;
	    for(int i = 2; i < x; i++)
	        array.put(i, false);
	}
	public void printArray(){

	    for(int i = 2; i < array.size();i++){
	        if( !array.get(i))
	        System.out.println(i);

	    }
	}
} 

Klasa główna

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
	public static void main(String[] args) throws InterruptedException{
		Sieve task = new Sieve();
		task.setArray(50000);
		Long beg = new Date().getTime();
		ExecutorService exec = Executors.newCachedThreadPool();
		for(int i = 0; i < 1;i++){
			exec.execute(task);
		}
		exec.shutdown();
		Long sleep = 300L;
		Thread.sleep(sleep);
		task.printArray();
		System.out.println("Time is "+(new Date().getTime()-beg-sleep));
	}
}

Problem polega na tym, że wraz ze wzrostem liczby wątków czasy wykonania nie zmniejszają się. Czasami wręcz zwiększają się... Czym może to być spowodowane?
Dodatkowo zauważyłem, że spory czas po zakończeniu obliczeń wątek DestroyJavaVM bardzo długo "pracuje". Czyżby to Garbage Collector sprzątał po HashMapie?

Ja chyba czegoś nie rozumiem. Przecież w u ciebie w tym kodzie odpalasz zwykłe seryjne sito, tylko że robisz to wiele razy. Ergo nie dziwota że działa wolniej -> procesor musi dzielić czas pomiędzy wątki. Spowolnienie zapewne jest liniowe i takie też być powinno...

Dodałem

System.out.print(tmp+" id="+Thread.currentThread().getId()+"\n");

do metody run()
Wyniki prezentują się następująco

2 id=8
3 id=9
5 id=10
7 id=8
11 id=11
13 id=10
17 id=9
19 id=8
23 id=11
29 id=10
31 id=9
37 id=8
41 id=11
43 id=10
47 id=9
53 id=8
59 id=9
61 id=8
67 id=11
71 id=10
73 id=9
79 id=8
83 id=11
89 id=10
97 id=9
Time is 5

Skoro kolejne liczby są "wykreślane" przez kolejne wątki, to dlaczego te wątki nie działają współbieżnie?

Dopiero teraz zobaczyłem co ty robisz. Słyszał kiedy o takich rzeczach jak synchronizacja albo race condition? To doczyta. Jakby to chciaż była ConcurrentHashMap...
Poza tym ile masz rdzeni procesora? Jak odpalisz ten kod to ile rdzeni zabiera JVM?

Słyszał i czytał, ale widocznie za mało, bo nie bardzo wiem jak to tutaj zastosować. Jeżeli dam dodam parametr synchronized do metody run, to nie będzie wtedy współbieżności. Bez bicia przyznam, że nie wiem jak inaczej mógłbym to usprawnić. Początkowo, program usuwał elementy z HashMapy( żeby nie sprawdzać 6 dwa razy itp), ale nie wiedziałem jak rozwiązać problem synchronizacji.
Rdzenie mam 4, ale nie wiem jak sprawdzić ile z nich jest wykorzystywanych przez JVM( JVisualVM ma takie możliwości?)

No ewidentnie za mało skoro "synchronizujesz" to sobie przez sleep() (którego być tu nie powinno!). Albo użyj ConcurrentHashMap albo synchronizuj samą operację na mapie tam gdzie ją wykonujesz, tzn

synchronized(array){
  operacja_na_mapie
}

A tego sleepa wywal. Powinno działać lepiej...

Synchronizacja synchronizacją, a tu się spory błąd wkradł i tak sobie czekał i czekał. Tym błędem było złe liczenie czasu. Zamiast liczyć czas, wykonywania wątków to mój program liczył czas wypisywania wyników.
Aby realnie mierzyć czas wykonywania programu dodałem do klasy Sieve zmienną threads. Którą później inkrementuję i dekrementuję w odpowiednich miejscach:

 public void run(){
    threads++;
    while(counter < x-1){
    ...
    }
    threads--;
}

W klasie Main dodałem coś takiego.

        Long time = 0L;
        while(true){
            if(task.threads == 0){
                time = new Date().getTime()-beg;
                break;
            }
            else 
                Thread.sleep(10);
           }

        task.printArray();
        System.out.println("Time is "+time);

Czyli sprawdzam w pętli czy wątki poboczne się zakończyły( a przynajmniej ja to tak rozumiem). Następnie wypisuję zawartość mapy i czas. Niestety nie działa to tak jak chciałem. Ostatnim poleceniem w metodzie run jest threads--; a mimo to wątek wydaje się liczyć dłużej. Czym może to być spowodowane?
Wyniki dla powyższego kodu i dla kodu po sleepie:

2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Time is 5



Po sleepie
2
3
5
7
11
13
17
19

Napisałem od początku metodę run. W ogólnym odczuciu program działa wolniej, ale przynajmniej współbieżnie. Niestety dla dużych zakresów liczb( 20 000). Wyniki są błędne, jakieś sugestie czemu?

public void run() {
		synchronized(this){
			threads++;
		}
		List<Integer> list = new ArrayList<Integer>();
		synchronized (list) {
			while (counter < Math.sqrt(x)){
				// 
				list.add(counter++);
				try {
					Thread.sleep(0,1);
				} catch (InterruptedException e) {}
			}
		}
		Collections.sort(list);
		for (Integer it : list) {
			int n = it.intValue();
			for (int i = n; i < x; i += n) {
				if (array.get(i) && i == n)
					break;
				if( i > n)
					array.put(i, true);
			}
		}
		synchronized(this){
			threads--;
		}
	}

Bo wykonujesz operacje na kolekcji poza sekcją krytyczną? o_O Robisz "sort" na liście poza blokiem synchronizacji i jak to niby ma działać? W ogóle zresztą modyfikujesz tam tą tablicę co jest jakimś WTFem. Nie lecą ci czasem ConcurrentModificationException? ;]

Shalom napisał(a):

W ogóle zresztą modyfikujesz tam tą tablicę co jest jakimś WTFem. Nie lecą ci czasem ConcurrentModificationException? ;]

Nie leci. Dlaczego WTF?

Ogólnie mój pomysł był następujący:
W pierwszej pętli while w każdym wątku wrzucam do listy kolejne liczby z podanego zakresu( np 2-200).
Po każdym addzie robię Thread.sleepa(* Thread.yield*() nie zmienia mi wątków¹) aby kolejne liczby były dodawane do list kolejnych wątków.
czyli:

Problemem jest fakt, że korzystanie ze sleepa znacząco spowalnia program. Niekorzystanie ze sleepa nie skraca czasu przy zwiększaniu liczby wątków.
Ze sleepem:
2-2000, 1 watek, czas: 2242
2-2000, 4 wątki, czas:589
Z yieldem
2-200000, 1 watek, czas: 496
2-200000, 4 wątki, czas: 513

¹ Czy może to wynikać z używania* CachedThreadPool*?

public void run() {
                synchronized(this){
                        threads++;
                }
                List<Integer> list = new ArrayList<Integer>();
                synchronized (list) {
                        while (counter < Math.sqrt(x)){
                                // 
                                list.add(counter++);
                                try {
                                        Thread.sleep(0,1);
                                } catch (InterruptedException e) {}
                        }
                }

To nie jest WTF, to jest ROTFL.

Robisz lokalną zmienną list i się na niej synchronizujesz? No człowieku, przecież ta lista jest lokalna dla każdego wątku. To na zmiennej counter powinieneś się synchronizować.

Z tym, że synchronizowanie się na prymitywach jest trochę bez sensu, bo od Javy 5 mamy takie bajery: http://docs.oracle.com/javase/1.5.0/docs/api/java/util/concurrent/atomic/package-summary.html

Collections.sort(list);
Czy przypadkiem nie sortujesz listy, która już jest posortowana? counter leci tylko w górę, a ty dodajesz kolejne wartości counter. No chyba, że masz sieczkę przez brak synchronizacji na counter.

                for (Integer it : list) {
                        int n = it.intValue();
                        for (int i = n; i < x; i += n) {
                                if (array.get(i) && i == n)
                                        break;
                                if( i > n)
                                        array.put(i, true);
                        }
                }

To już jest fraktalny WTF.

Po pierwsze zakładasz, że pola w mapie będą aktualizowane w określonym porządku, a przecież masz niesynchronizowane wątki.
Po drugie ta pętla nie ma nic wspólnego z sitem Erastotenesa. Sito działa mniej więcej tak:

1. ustaw wszystkie liczby na złożone (u ciebie to false)
2. począwszy od 2 rób następującą rzecz:

• wykreśl wszystkie wielokrotności aktualnej liczby (jako optymalizacja - jeśli aktualna liczba jest złożona, to można pominąć ten krok),
• przejdź do kolejnej liczby,
  A więc to ma być zagnieżdżenie typu: for -> if -> for, a nie for -> for -> if.

Zresztą na Wiki masz kod, który to potwierdza:

        boolean[] numbersTable = new boolean[n+1];
        for(int i = 2; i*i <= n; i++)
        {
            if (numbersTable[i] == true)
                continue;
            for (int j = 2 * i ; j <= n; j += i)
                numbersTable[j] = true;
 
        }

Wprowadziłem zmiany w metodzie run:

• Zamieniłem counter na AtomicInteger
• Wywaliłem sortowanie listy
• Przebudowałem pętle z sitem.

private AtomicInteger counter = new AtomicInteger(2);

	public void run() {
		List<Integer> list = new ArrayList<Integer>();	
			while (counter.get() < x) {
				list.add(counter.get());
				counter.incrementAndGet();
			//	 try {
				// Thread.sleep(1);
				// } catch (InterruptedException e) {}
				// Thread.yield();
		}
		for (Integer it : list) {
			  System.out.println(it.intValue());
			int n = it.intValue();
			if (array.get(n))
				continue;
			for (int i = 2 * n; i <= x; i += n) 
					array.put(i, true);
		}
	}

Zrobiłem też drugą wersję, która nie korzysta z list. Zamiast listy, używam metody synchronizowanej, która zwraca mi aktualną wartość.

private int counter = 2;

	public void run() {
		int n = 0;
		while(true){
			n = getCounter();
			if( n == -1)
				break;
			if (array.get(n))
				continue;
			for (int i = 2 * n; i <= x; i += n) {
					array.put(i, true);
			}
		}
	}
	private synchronized int getCounter(){
		if( counter < x)
			return counter++;
		else return -1;
	}

Niestety w obu przypadkach, zwiększenie liczby wątków nie zwiększa prędkości działania. Jakieś sugestie dlaczego?

Algorytm dalej jest niedeterministyczny, wystarczy na początku każdej iteracji wstawić sleep(random) i da złe wyniki. To że działa (o ile rzeczywiście działa) jest łutem szczęścia.

Niestety w obu przypadkach, zwiększenie liczby wątków nie zwiększa prędkości działania. Jakieś sugestie dlaczego?

Podaj cały aktualny kod.

Algorytm dalej jest niedeterministyczny

Dlaczego? Dodałem dwa sleepy a mimo to algorytm wydaje się działać dobrze. Co prawda nie sprawdziłem wszystkich liczb, ale dla ~340k liczb pierwsze 10k się zgadzało. Dodatkowo zgodnie z twierdzeniem o liczbach pierwszych ilość liczb w badanym przedziale też raczej mi się zgadzała. 324150 względem moich 348513.

Klasa główna

import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class ConcurrentTest {
    public static void main(String[] args) throws InterruptedException{
        Sieve task = new Sieve();
        int x = 5000000;
        int t = 4;
        task.setArray(x);
        Long beg = new Date().getTime();
        ExecutorService exec = Executors.newCachedThreadPool();
      for(int i = 0; i < t;i++){
            exec.execute(task);
        }
        exec.shutdown();
        Long time = 0L;
        while(true)
            if(exec.isTerminated()){
                time = new Date().getTime()-beg;
                break;
            }

        task.printArray();
        System.out.println("Time is "+time);
    }
}

Klasa Sieve

import java.util.concurrent.ConcurrentHashMap;

public class Sieve implements Runnable {
	private ConcurrentHashMap<Integer, Boolean> array = new ConcurrentHashMap<Integer, Boolean>();
	private int x;
	private int counter = 2;
//	private AtomicInteger counter = new AtomicInteger(2);
//	 
//    public void run() {
//            List<Integer> list = new ArrayList<Integer>();        
//                    while (counter.get() < x) {
//                            list.add(counter.get());
//                            counter.incrementAndGet();
//                    //         try {
//                            // Thread.sleep(1);
//                            // } catch (InterruptedException e) {}
//                            // Thread.yield();
//            }
//            for (Integer it : list) {
//                      System.out.println(it.intValue());
//                    int n = it.intValue();
//                    if (array.get(n))
//                            continue;
//                    for (int i = 2 * n; i <= x; i += n) 
//                                    array.put(i, true);
//            }
//    }
	public void run() {

		int n = 0;
		while(true){
			sleep((int)Math.random()*500);
			n = getCounter();
			if( n == -1)
				break;
			if (array.get(n))
				continue;
			for (int i = 2 * n; i <= x; i += n) {
				sleep((int)Math.random()*500);
					array.put(i, true);
			}
		}
	}
	private synchronized int getCounter(){
		if( counter < x)
			return counter++;
		else return -1;
	}
	public void setArray(int x) {
		this.x = x;
		for (int i = 2; i <= x; i++)
			array.put(i, false);
	}

	public void printArray() {

		for (int i = 2; i <= array.size(); i++) {
			if (!array.get(i))
				System.out.println(i);

		}
	}
	private void sleep(int time){
		try {
			Thread.sleep(time);
		} catch (InterruptedException e) {}
	}
}

W tym rozwiązaniu możliwe jest, że wątki będą niepotrzebnie przetwarzać wielokrotności liczb pierwszych, np.

Wątek 1 pobiera liczbę 2 i traci procesor
Wątek 2 pobiera liczbę 3 i traci procesor
Wątek 3 pobiera liczbę 4, sprawdza, że 4 jeszcze nie jest odsiane (wątek 1. jeszcze nie zdążył tego zrobić) i zaczyna odsiewać wielokrotności 4.

Wątek 3. wykona niepotrzebną pracę, gdyż wątek 1. też odsiałby te liczby (wielokrotności 4 są też wielokrotnościami 2).

Ano tak, @__krzysiek85 ma rację. Algo w sumie jednak zadziała dobrze, tylko wykona trochę niepotrzebnej pracy. Kod trzeba sprofilować, może to zrobię niedługo.

Spróbuj zamienić ConcurrentHashMap na zwykłą tablicę booleanów.

Na początku zakładałem, że nie będę tylko zmieniał flagi false na true, lecz będę faktycznie usuwał elementy z mapy. Fakt zmiany zmiany wartości true na false wyniknął po problemach z synchronizacją. Po zmianie mapy na tablicę booleanów problem braku przyspieszenia dalej nie został rozwiązany. Wróciłem więc do HashMapy i postanowiłem usuwać wartości co zmniejsza trochę liczbę iteracji. Dodatkowo przeprowadziłem kilka testów dla różnej liczby wątków.

Testy przy zmianie wartości na true.

Testy przy usuwaniu/wykreślaniu

Przy usuwaniu następuje nieznaczna poprawa szybkości, jednak problem z wątkami nadal występuje.

Twój procesor jest w stanie obsłużyć 20 wątków jednocześnie?

Nie, ale byłem ciekawy jakie będą wyniki. Spodziewałem się znacznego spowolnienia.

HashMapa do implementacji sita? To musi być okropnie powolne.
Takie rzeczy się robi na BitSetach. A jeszcze lepiej własna implementacja BitSet na AtomicIntegerArray, bez jakiejkolwiek synchronizacji, wtedy może wielowątkowa implementacja będzie mieć jakiś sens.