Curs 8
Fire de executie

• Ce este un fir de executie ?
• Crearea unui fir de executie
  • Extinderea clasei Thread
  • Implementarea interfetei Runnable
• Ciclul de viata al unui fir de executie
• Stabilirea prioritatilor de executie
• Sincronizarea mai multor fire de executie
  • Scenariul producator / consumator
  • Blocarea unui obiect (cuvāntul cheie synchronized)
  • Metodele wait, notify si notifyAll
• Gruparea firelor de executie
• Comunicarea prin fluxuri de tip "pipe"

Definitie

Un fir de executie este o succesiune sceventiala de instructiuni care se executa īn cadrul unui proces.

	public class SimpleThread extends Thread {

		public SimpleThread(String nume) {
			super(nume);
			//apelez constructorul superclasei Thread
		}
		public void run() {
			//codul executat de firul de executie
		}
	}

		SimpleThread t = new SimpleThread("Java")
		//creeaza un fir de executie cu numele Java

		SimpleThread t = new SimpleThread("Java")		
		t.start()
		//creeaza si lanseaza un fir de executie

class Counter extends Thread {	//clasa care defineste firul de executie
	private int from, to, step;

	public Counter(int from, int to, int step) {
		this.from = from;
		this.to = to;
		this.step = step;
	}

	public void run() {
		for(int i = from; i <= to; i += step)
			System.out.print(i + " " );
	}
}

public class TestCounter {	//clasa principala
	public static void main(String args[]) {
		Counter cnt1, cnt2;     
		
		cnt1 = new Counter(0, 10, 2);
		//numara de la 0 la 100 cu pasul 5

		cnt2 = new Counter(100, 200, 10); 
		//numara de la 100 la 200 cu pasul 10

		cnt1.start();
		cnt2.start();
		//pornim firele de executie
		//ele vor fi distruse automat la terminarea lor
	}
}

	class FirExecutie extends Parinte, Thread // ilegal !
 

		public interface Runnable {
			pulic abstract void run( );
		}

	public class SimpleThread implements Runnable {

		private Thread simpleThread = null;

		public SimpleThread() {
			if (simpleThread == null) {
				simpleThread = new Thread(this);
				simpleThread.start();
		}

		public void run() {
			//codul executat de firul de executie
		}
	}

		private Thread simpleThread = null;

		simpleThread = new Thread( this );
		simpleThread.start();

Varianta 1 (standard)

class Counter implements Runnable {
	private Thread counterThread = null;
	private int from, to, step;
	public Counter(int from, int to, int step) {
		this.from = from;
		this.to = to;
		this.step = step;
		if (counterThread == null) {
			counterThread = new Thread(this);
			counterThread.start();
		}
	}
	public void run() {
		for(int i = from; i <= to; i += step)
			System.out.print(i + " " );
	}
}

public class TestThread2 {
	public static void main(String args[]) {
		Counter cnt1, cnt2; 
		//lansez primul fir de executie (prin constructor)
		cnt1 = new Counter(0, 100, 5);
		//lansez al doilea fir de executie (prin constructor)
		cnt2 = new Counter(100, 200, 10); 
	}
}

Varianta 2

class Counter implements Runnable {
	private int from, to, step;
	public Counter(int from, int to, int step) {
		this.from = from;
		this.to = to;
		this.step = step;
	}
	public void run() {
		for(int i = from; i <= to; i += step)
			System.out.print(i + " " );
	}
}

public class TestThread2 {
	public static void main(String args[]) {
		Counter cnt1, cnt2; 
		cnt1 = new Counter(0, 100, 5);
		cnt2 = new Counter(100, 200, 10); 
		new Thread( cnt1 ).start();
		//lansez primul fir de executie
		new Thread( cnt2 ).start();
		//lansez al doilea fir de executie
	}
}

1. New Thread
2. Runnable
3. Not Runnable
4. Dead

Starea "New Thread"

		Thread counterThread = new Thread ( this );
		//counterThread se gaseste in starea New Thread

Starea "Runnable"

		counterThread.start();
		//counterThread se gaseste in starea Runnable

• aloca resursele sistem necesare
• planifica firul de executie la CPU pentru a fi lansat
• apeleaza metoda run a obiectului reprezentat de firul de executie

Starea "Not Runnable"

• este "adormit" prin apelul metodei sleep.
• a apelat metoda wait, asteaptānd ca o anumita conditie sa fie satisfacuta
• este blocat īntr-o operatie de intrare/iesire

"Adormirea" unui fir de executie

	public static void sleep( long millis )
		throws InterruptedException
	public static void sleep( long millis, int nanos )
		throws InterruptedException

		try {
			Thread.sleep(1000);
			//face pauza de o secunda
		} catch (InterruptedException e) { 
			. . . 
		}

• Daca un fir de executie a fost "adormit", atunci el devine Runnable doar dupa scurgerea intervalului de timp specificat de instructiunea sleep.
• Daca un fir de executie asteapta o anumita conditie, atunci un alt obiect trebuie sa īl informeze daca acea conditie este īndeplinita sau nu; acest lucru se realizeaza prin instructiunile notify sau notifyAll
  (vezi "Sincronizarea firelor de executie").
• Daca un fir de executie este blocat īntr-o operatiune de intrare/iesire atunci el redevine Runnable atunci cānd acea operatiune s-a terminat.

Starea "Dead"

Terminarea unui fir de executie

• Prin scrierea unor metode run care sa-si termine executia īn mod natural; la terminarea metodei run se va termina automat si firul de executie, acesta intrānd īn starea Dead. Acesta este cazul din exemplul considerat anterior:

  	public void run() {
  		for(int i = from; i <= to; i += step)
  			System.out.print(i + " " );
  	}
  

  Dupa afisarea numerelor din intervalul specificat metoda se termina si odata cu ea si firul de executie repsectiv.
• Prin folosirea unei variabile de terminare. In cazul cānd metoda run trebuie sa execute o bucla infinita atunci aceasta trebuie controlata si printr-o variabila care sa opreasca aceasta bucla atunci cānd dorim ca firul de executie sa se termine. Uzual, aceasta este o variabila membra a clasei care descrie firul de executie care fie este publica, fie este asociata cu o metoda care īi schimba valoarea.

Terminarea unui fir de executie folosind metoda "īnvechita" stop

import java.io.*;
public class TestThread {
	public static void main(String args[]) throws IOException {
		WaitKey thread = new WaitKey();            
		thread.start();
		System.in.read();	//astept apasarea tastei Enter
		thread.stop();	//opresc firul de executie
		System.out.println("S-au scurs " + thread.sec + " secunde");
	}
}
class WaitKey extends Thread {
	public int sec = 0;
	public void run() {
		while (true) {
			try {
				Thread.sleep(1000);	//pauze de o secunda
				sec ++;			//s-a mai scurs o secunda
			} catch(InterruptedException e){}
		}
	}
}

Terminarea unui fir de executie folosind o variabila de terminare

import java.io.*;
public class TestThread {
	public static void main(String args[]) throws IOException {
		WaitKey thread = new WaitKey();            
		thread.start();
		System.in.read();	//astept apasarea tastei Enter
		thread.running = false;
		System.out.println("S-au scurs " + thread.sec + " secunde");
	}
}

class WaitKey extends Thread {
	public int sec = 0;
	public boolean running = true; //variabila de terminare
	public void run() {
		while ( running ) {
			try {
				Thread.sleep(1000);
				sec ++;
			} catch(InterruptedException e){}
		}
	}
}

Metoda isAlive

• true - daca firul este īn una din starile Runnable sau Not Runnable
• false - daca firul este īn una din starile New Thread sau Dead

		WaitKey thread = new WaitKey();
		// isAlive retuneaza false (starea este New Thread)

		thread.start();
		// isAlive retuneaza true (starea este Runnable)

		System.in.read();

		thread.running = false;
		// isAlive retuneaza false (starea este Dead)

	public static final int MIN_PRIORITY	- prioritatea minima
	public static final int NORM_PRIORITY	- prioritatea implicita
	public static final int MAX_PRIORITY	- prioritatea maxima

• un fir de executie cu o prioritate mai mare solicita procesorul
• metoda sa run se termina
• vrea sa faca explicit acest lucru apelānd metoda yield
• timpul alocat pentru executia sa a expirat (pe SO cu cuante de timp)

Exemplu de fir de executie "egoist"

//un fir de executie care numara pana la 100.000 din 100 īn 100
class Selfish extends Thread {
	public Selfish(String name) {
		super(name);
	}
	public void run() {
		int i = 0;
		while (i < 100000) {
			//bucla stransa care acapareaza procesorul
			i ++;
			if (i % 100 == 0) 
				System.out.println(getName()+" a ajuns la "+i);
		}
	}
}
//clasa principala
public class TestSelfishThread {
	public static void main(String args[]) {
		Selfish s1, s2;     
		s1 = new Selfish("Firul 1");
		s1.setPriority (Thread.MAX_PRIORITY);
		s2 = new Selfish("Firul 2");
		s2.setPriority (Thread.MAX_PRIORITY);
		s1.start();
		s2.start();
	}
}

		Firul 1 a ajuns la 100
		Firul 1 a ajuns la 200
		Firul 1 a ajuns la 300
		. . . 
		Firul 1 a ajuns la 99900
		Firul 1 a ajuns la 100000
		Firul 2 a ajuns la 100
		Firul 2 a ajuns la 200
		. . .
		Firul 2 a ajuns la 99900
		Firul 2 a ajuns la 100000

	public void run() {
		int i = 0;
		while (i < 100000) {
			i ++;
			if (i % 100 == 0) 
				System.out.println(getName()+" a ajuns la "+i);
			yield(); //cedez temporar procesorul
		}

• Producstorul genereaza numerele īntregi de la 1 la 10, fiecare la un interval neregulat cuprins īntre 0 si 100 de milisecunde. Pe masura ce le genereaza īncearca sa le plaseze īntr-o zona de memorie (o variabila īntreaga) de unde sa fie citite de catre consumator.
• Consumatorul va prelua, pe rānd, numerele generate de catre producator si va afisa valoarea lor pe ecran.

	
class Buffer {
	private int number = -1;
		
	public int get() {
        return number;
	}
			
	public void put(int number) {
		this.number = number;
	}
}

class Producator extends Thread {
	private Buffer buffer;
	public Producator(Buffer b) {
		buffer = b;
	}
	public void run() {
		for (int i = 0; i < 10; i++) {
			buffer.put(i);
			System.out.println("Producatorul a pus:\t" + i);
			try {
				sleep((int)(Math.random() * 100));
			} catch (InterruptedException e) { }
		}
	}
}

class Consumator extends Thread {
	private Buffer buffer;
	public Consumator(Buffer b) {
		buffer = b;
	}
	public void run() {
		int value = 0;
		for (int i = 0; i < 10; i++) {
			value = buffer.get();
			System.out.println("Consumatorul a primit:\t" + value);
		}
	}
}
//Clasa principala
public class TestSincronizare1 {
	public static void main(String[] args) {
		Buffer b = new Buffer();
		Producator p1 = new Producator(b);
		Consumator c1 = new Consumator(b);
		p1.start();
		c1.start();
	}
}

		Consumatorul a primit:	-1
		Consumatorul a primit:	-1
		Producatorul a pus:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Consumatorul a primit:	0
		Producatorul a pus:	1
		Producatorul a pus:	2
		Producatorul a pus:	3
		Producatorul a pus:	4
		Producatorul a pus:	5
		Producatorul a pus:	6
		Producatorul a pus:	7
		Producatorul a pus:	8
		Producatorul a pus:	9

• consumatorul nu asteapta īnainte de a citi ca producatorul sa genereze un numar si va prelua de mai multe ori acelasi numar.
• producatorul nu asteapta consumatorul sa preia numarul generat īnainte de a produce un altul, īn felul acesta consumatorul va "rata" cu siguranta unele numere (īn cazul nostru aproape pe toate).

1. Cele doua fire de executie nu trebuie sa acceseze simultan buffer-ul ; acest lucru se realizeaza prin blocarea obiectului Buffer atunci cānd este accesat de un fir de executie, astfel īncāt nici nu alt fir de executie sa nu-l mai poate accesa. (vezi "Blocarea unui obiect").
2. Cele doua fire de executie trebuie sa se coordoneze, adica producatorul trebuie sa gaseasca o modalitate de a "spune" consumatorului ca a plasat o valoare īn buffer, iar consumatorul trebuie sa comunice producatorului ca a preluat aceasta valoare, pentru ca acesta sa poata genera o alta. Pentru a realiza aceasta comunicare, clasa Thread pune la dispozitie metodele wait, notify, notifyAll. (vezi "Metodele wait, notify, notifyAll").

class Buffer {
	private int number = -1;
	private boolean available = false;
	public synchronized int get() {
		while (!available) {
			try {
				wait();
				//asteapta producatorul sa puna o valoare
			} catch (InterruptedException e) { }
  		}
		available = false;
		notifyAll();
		return number;
	}
	public synchronized void put(int number) {
		while (available) {
			try {
				wait();
				//asteapta consumatorul sa preia valoarea
			} catch (InterruptedException e) { }
     		}
		this.number = number;
		available = true;
		notifyAll();
	}
}

		Producatorul a pus:	0
		Consumatorul a primit:	0
		Producatorul a pus:	1
		Consumatorul a primit:	1
		. . .
		Producatorul a pus:	9
		Consumatorul a primit:	9

Definitie

Un segment de cod ce gestioneaza o resursa comuna mai multor de fire de executie separate si concurente se numeste sectiune critica. In Java o sectiune critica poate fi un bloc de instructiuni sau o metoda.

	
	public synchronized int get() {
		...
	}
	public synchronized void put(int number) {
		...
	}

	public synchronized void put(int number) {
		// buffer blocat de producator
		...
		// buffer deblocat de producator
	}
	public synchronized int get() {
		// buffer blocat de consumator
		...
		// buffer deblocat de consumator
	}

	public synchronized int get() {
		if (available) {
			available = false;
			return number;
		}
	}
	public synchronized int put(int number) {
		if (!available) {
	            available = true;
      	      this.number = number;
		}
	}

	public synchronized int get() {
		while (!available) {
			//nimic - astept ca variabila sa devina true
		}
		available = false;
		return number;
	}
	public synchronized int put(int number) {
		while (available) {
			//nimic - astept ca variabila sa devina false
		}
            available = true;
            this.number = number;
	}

		void wait( )
		void wait( long timeout )
		void wait( long timeout, long nanos )

• un alt fir de executie informeaza pe cei care "asteapta" la un anumit monitor sa se trezeasca; acest lucru se realizeaza printr-un apel al metodei notifyAll sau notify.
• perioada de astepatare specificata a expirat.

		public synchronized int get() {
			while (!available) {
				try {
					wait();
					//asteapta producatorul sa puna o valoare
				} catch (InterruptedException e) { }
     	   		}
			available = false;
			notifyAll(); 
			return number;
		}
	public synchronized void put(int number) {
			while (available) {
				try {
					wait();
					//asteapta consumatorul sa preia valoarea
				} catch (InterruptedException e) { }
     	   		}
			this.number = number;
			available = true;
			notifyAll();
		}
	}

	public Thread(ThreadGroup group, Runnable target)
	public Thread(ThreadGroup group, String name)
	public Thread(ThreadGroup group, Runnable target, String name)

	ThreadGroup grup1 = new ThreadGroup("Producatori");
	Thread p1 = new Thread(grup, "Producator 1");
	Thread p2 = new Thread(grup, "Producator 2");

	ThreadGroup grup2 = new ThreadGroup("Consumatori");
	Thread c1 = new Thread(grup, "Consumator 1");
	Thread c2 = new Thread(grup, "Consumator 2");
	Thread c3 = new Thread(grup, "Consumator 3");

public class EnumerateTest {
	public void listCurrentThreads() {
		ThreadGroup currentGroup = Thread.currentThread().getThreadGroup();

		//aflu numarul firelor de executie active
		int numThreads = currentGroup.activeCount();

		//pun intr-un vector referinte la firele de exec. active
		Thread[] listOfThreads = new Thread[numThreads];
		currentGroup.enumerate(listOfThreads);

		//le afisez pe ecran
		for (int i = 0; i < numThreads; i++)
			System.out.println("Thread #" + i + " = " + 
				listOfThreads[i].getName());
	}
}

	PipedReader, PipedWriter - pentru caractere, respectiv 
	PipedOutputStream, PipedInputStream - pentru octeti

	public PipedReader( )
	public PipedReader( PipedWriter pw ) throws IOException
	public PipedWriter( )
	public PipedWriter( PipedReader pr ) throws IOException

	public void connect( PipedWriter pw ) throws IOException
	public void connect( PipedReader pr ) throws IOException, 

		PipedWriter pw1 = new PipedWriter();
		PipedReader pr1 = new PipedReader(pw1);
	sau
		PipedReader pr2 = new PipedReader();
		PipedWriter pw2 = new PipedWriter(pr2);
	sau
		PipedReader pr = new PipedReader();
		PipedWriter pw = new PipedWirter();
		pr.connect(pw) //echivalent cu
		pw.connect(pr);

import java.io.*;

//clasa principala
public class TestPipes {
	public static void main(String[] args) throws IOException {

		PipedOutputStream pipeOut = new PipedOutputStream();
		PipedInputStream pipeIn = new PipedInputStream(pipeOut);

		DataOutputStream out = new DataOutputStream( pipeOut);
		DataInputStream in = new DataInputStream( pipeIn );
		
		Producator p1 = new Producator(out);
		Consumator c1 = new Consumator(in);

		p1.start();
		c1.start();
	}
}

class Producator extends Thread {
	private DataOutputStream out;

	public Producator(DataOutputStream out) {
		this.out = out;
	}

	public void run() {
		for (int i = 0; i < 10; i++) {
			try { out.writeInt(i); }
			catch (IOException e) {}
			System.out.println("Producatorul a pus:\t" + i);
			try {
				sleep((int)(Math.random() * 100));
			} catch (InterruptedException e) { }
		}
	}
}

class Consumator extends Thread {
	private DataInputStream in;

	public Consumator(DataInputStream in) {
		this.in = in;
	}
	public void run() {
		int value = 0;
		for (int i = 0; i < 10; i++) {
			try { value = in.readInt(); }
			catch (IOException e) {}			
			System.out.println("Consumatorul a primit:\t" + value);
		}
	}
}