MokaByte Numero 17 - Marzo 1998
		Sotto la foresta di Java
	di Piergiuseppe  Spinelli

Una vasta libreria standard come quella del JDK ha il valore dell'oro puro quando si tratta di implementare velocemente dei prototipi o di consegnare nei tempi prestabiliti (vale a dire ieri) la prima versione funzionante di un modulo software.
Troppo spesso però il prototipo viene promosso sul campo a sistema finale o, al contrario, un componente di sistema è bocciato per problemi di performances senza valutare che la sostituzione di un paio di classi standard con altre ottimizzate ad hoc potrebbe ridare la carica al codice apparentemente pigro ed inaffidabile.
Vediamo come sia possibile avvalersi al massimo delle classi della libreria JDK superando eventuali ostacoli dovuti alla loro implementazione e traendo spunto proprio da quest'ultima per giungere ad una soluzione ottimale.
Java ci viene incontro in questo bisogno di conoscenza mettendo a nostra disposizione i sorgenti commentati di tutte le classi standard di sistema. Questo è senza dubbio un buon punto d'inizio.
Prenderemo come esempio un problema in cui mi sono imbattuto portando in Java un sistema d'acquisizione dati e sperimenteremo una tecnica ispirata ad un articolo apparso su JavaWorld, dal titolo: "How to decouple the Observer/Observable object model" .
 

Pattern Observable/Observer: l'implementazione Java
Java fornisce varie classi standard che implementano alcuni pattern d'uso frequente. Uno di questi è il pattern Observable/Observer che ha un vasto campo d'applicazione ogniqualvolta una sorgente di informazione (programma di calcolo, trigger di una porta di comunicazione, gestore di eventi di input, etc.) debba tenere aggiornati un certo numero di client sui propri mutamenti di stato.

Consideriamo un modulo d'acquisizione dati di processo da una linea di produzione industriale. Esso potrebbe, ad esempio, leggere ad intervalli regolari i contatori dei pezzi scartati divisi per causale di scarto e mantenuti nei controllori programmabili della macchina. Quando i valori letti differiscono da quelli precedenti, il modulo d'acquisizione notifica l'evento a tutti gli oggetti del sistema che abbiano dichiarato interesse per tale informazione.

Nell'esempio, ll modulo d'acquisizione è un oggetto Observable mentre tutti i fruitori dell'informazione sono degli Observer registrati su di esso. La potenza di questo pattern è data dal fatto che l'oggetto osservato non deve avere alcuna conoscenza del numero e del tipo dei propri osservatori, e nemmeno del loro scopo. Tipici fruitori dei dati sugli scarti macchina possono essere:

• Un oggetto che visualizza sul monitor i contatori in forma numerica (attivo solo se richiamato dall'utente)
• Un modulo che registra in un database i dati acquisiti (sempre attivo)

Altro punto di forza è l'estrema semplicità d'uso del modello. Il codice Java relativo all'esempio suona più o meno così:
 

public class AcquisisciDati extends Observable implements Runnable{
...

public class ClienteTipo implements Observer{
...

Si noti che Observer è un'interfaccia mentre Observable è una classe astratta. Ciò potrebbe imporre qualche limitazione a causa dell'ereditarietà semplice di Java (ad esempio AcquisisciDati deve implementare l'interfaccia Runnable non potendo ereditare direttamente dalla classe Thread) ma questo, di solito, è un problema minore.

Dalla teoria alla pratica, chi avesse rilasciato la versione finale del proprio sistema d'acquisizione dati seguendo l'esempio precedente andrebbe sicuramente incontro a spiacevoli sorprese. Facilmente, alla faccia della trasparenza degli osservatori, il nostro modulo d'acquisizione potrebbe andare in crash con l'introduzione di un nuovo client oppure funzionare a dovere con tutti ma non quando sono contemporaneamente attivi, e così via senza metter limite alla fantasia.
 

Il problema del coupling stretto tra Observable ed Observer
Prima di inoltrarsci nel cuore del problema, diamo uno sguardo ai sorgenti della classe Observable, ed in particolare al metodo notifyObservers():

public void notifyObservers(Object arg) {
        int size=0;
        synchronized (this) {
                if (!hasChanged()) return;
                size = obs.size();
                if (size > arr.length) arr = new Observer[size];
                obs.copyInto(arr);
                clearChanged();
        }
        for (int i = size -1; i>=0; i--) {
                if (arr[i] != null) arr[i].update(this, arg);
        }
}

• Le eccezioni non sono catturate durante l'esecuzione di notifyObservers(). In questo caso, visto che i metodi update() vengono eseguiti nel contesto del thread che gestisce l'oggetto Observable, esso va in crash.
• NotifyObservers() viene eseguito internamente ad un blocco try{}catch(). In questo caso, l'errore potrebbe ripetersi ad ogni esecuzione. Visto che l'ordine di notifica ai vari client registrati ci è sconosciuto, ci troveremmo nella situazione in cui solo alcuni Observer ricevono notifica (quelli chiamati prima dell'errore) mentre i rimanenti, pur innocenti, ne restano esclusi.

Questo tipo di problemi di accoppiamento (coupling) tra parti di codice era già conosciuto ben prima dell'affermazione dei sistemi ad oggetti. Il listato di notifyObserver() ci conferma come i metodi update() dei vari osservatori vengano eseguiti a carico del tempo e delle risorse del thread che esegue l'oggetto osservato.

1. Vogliamo mantenere la stessa interfaccia applicativa verso gli oggetti Observer.
2. Vogliamo modificare il meno possibile le classi già implementate come sottoclassi di Observable.
3. Vogliamo che i tempi di scansione di AcquisisciDati non siano influenzati da quelli di esecuzione dei vari client.
4. Non vogliamo che anomalie in un Observer si riflettano sull'oggetto ossevato e sui restanti osservatori.
5. Non vogliamo che le risorse (p.e. lo stack) del thread in cui gira AcquisisciDati siano impegante dai client registrati.
6. Non vogliamo che eventi troppo frequenti saturino il sistema.
7. Non vogliamo che metodi update() dall'esecuzione troppo lunga monopolizzino la CPU.

• Istanzia un oggetto di servizio, la nested class Broker (composta in gran parte da codice copiato da Observable) e con un task dedicato alle notifiche ai vari lettori. Tale classe è a sua volta un Observer ed è l'unico effettivamente registrato per il relativo ThObservable.
• Accoda le notifiche da inoltrare impaccandole in un oggetto Notification in modo che il thread Broker possa reinviarle asincronamente senza perdere l'informazione sull'esatto momento in cui l'evento avvenuto.
• Modifica il metodo notifyObservers del Broker in modo che catturi gli eventuali errori di ogni singola chiamata ad update() ed elmini imediatamente la registrazione dell'osservatore che ha presentato il problema.
• Implementa la coda delle notifiche, la nested class Queue, per controllare se una notifica per il medesimo oggetto è già in coda. In questo caso si limita ad incrementare il contatore del numero di eventi all'interno della Notification già accodata.

Implementazione di ThObservable
Il codice di ThObservable, quindi, è fondamentalmente una modifica della classe standard Observable con l'aggiunta di una coda di eventi ed un task apposito per smistarli. Il metodo addObserver() standard della classe Observable viene sostituito in modo da registrare i vari oggetti sul Broker piuttosto che direttamente sull'istanza di ThObservable. Quando quest'ultima chiama il proprio metodo notifyObservers(), esso esegue solo il metodo update() del proprio Broker piuttosto che quelli di tutti gli Observer registrati.

L'introduzione di un secondo metodo per la registrazione degli osservatori, addSyncObserver(), consente di dare una certa priorità a quegli oggetti che sono riconosciuti come sicuri in termini di correttezza e risorse impegnate. Ciò viene realizzato chiamando il metodo addObserver() originale. Nel nostro esempio, il modulo di registrazione su database, che è sempre eseguito ed è presumibilmente parte del sistema base, potrebbe essere ritenuto ben testato e sotto controllo e, quindi, essere registrato con addSyncObserver() invece che con addObserver().

package ThObservable;
import java.util.*;
public abstract class ThObservable extends Observable{
        public class Notification {...}
        private class Broker implements Observer, Runnable {...}

        private Broker b=new Broker();
        private Thread bTh=null;
        private long TMT=200;
        private int priority=5;

        public ThObservable(int priority, long timeout){
                super();
                TMT=timeout;
                this.priority=priority;
                super.addObserver(b);
                bTh = new Thread(b);
                bTh.setDaemon(true);
                bTh.setPriority(priority);
                bTh.start();
        }

        public synchronized void addObserver(Observer o){
                super.deleteObserver(o);
                b.addObserver(o);
        }
        public synchronized void addSyncObserver(Observer o){
                b.deleteObserver(o);
                super.addObserver(o);
        }
        public synchronized void deleteObserver(Observer o){
                super.deleteObserver(o);
                b.deleteObserver(o);
        }
        public synchronized int countObservers(){
                return obs.size() + super.countObservers() - 1;
        }

        public Notification getNotification(){return b.currentNotification;}

        public void finalize(){
                if(bTh!=null){
                        Thread t=bTh;
                        bTh=null;
                        t.notify();
                }
        }
}

Il codice di Broker, che svolge la maggior parte del lavoro, è in buona parte copiato da quello originale di Observable. Il loop infinito del thread rimane in sospeso su una wait() quando la coda è vuota e non impegna inutilmente la CPU. Il metodo update(), richiamato da ThObservable, si limita ad accodare l'oggetto e risvegliare il thread con notify(). Gli oggetti estratti dalla coda sono passati come parametro ai metodi update() di tutti gli osservatori registrati. Tali metodi sono eseguiti all'interno di un blocco try/catch e, nel caso venga intercettata un'eccezione, la registrazione dell'Observer colpevole viene imediatamente rimossa.

private class Broker implements Observer, Runnable {

        Hashtable pars=new Hashtable();
        private class Queue{...}
        private Queue q=new Queue(null);
        private Vector obs;
        private Observer[] arr = new Observer[2];
        Notification currentNotification=null;

        Broker() {obs = new Vector();}
        synchronized void addObserver(Observer o) {
                if (!obs.contains(o)) { obs.addElement(o);}
        }
        synchronized void deleteObserver(Observer o){
                bs.removeElement(o);
        }
        public void run(){
                while(Thread.currentThread()==bTh){
                        synchronized(this){
                                currentNotification=q.get();
                                if(currentNotification==null){
                                        try{ wait(); }catch(Exception e){}
                                        continue;
                                }
                        }
                        int size=0;
                        synchronized (this) {
                                size = obs.size();
                                if(size>arr.length) {arr = new Observer[size];}
                                obs.copyInto(arr);
                        }
                        for(int i = size -1; i>=0; i--){
                                if(arr[i] != null){
                                        try{
                                        arr[i].update(ThObservable.this,
                                                                        currentNotification.obj);
                                        }catch(Exception e){
                                                deleteObserver(arr[i]);
                                                e.printStackTrace();
                                        }
                                }
                        }
                }
        }
        public synchronized void update(Observable obs,Object par){
                q.put(par);
                notify();
        }
}
 

public class Notification {
        public long first=System.currentTimeMillis();
        public long last;
        public long num=1;
        public Object obj;
        Notification(Object obj){
                this.obj=obj;
                last=first;
        }
}

private class Queue{
        private Queue next=null;
        private Queue tail=null;
        Notification not;
        Queue(Notification not){this.not=not;}
        synchronized void put(Object par){
                if(par==null) par=this;
                Notification n=(Notification)pars.get(par);
                if(n!=null){
                        n.last=System.currentTimeMillis();
                        n.num++;
                        return;
                }
                n=new Notification(par);
                pars.put(par, n);
                Queue q=new Queue(n);
                if(next==null)  next=q;
                else tail.next=q;
                tail=q;
        }
        synchronized Notification get(){
                Queue q=next;
                if(next!=null) {
                        next=next.next;
                        if(next==null) tail=null;
                        pars.remove(q.not.obj);
                        if(q.not.obj==this) q.not.obj=null;
                        return q.not;
                }
                return null;
        }
        synchronized boolean hasMoreElements(){return (q.next!=null);}
}

    public class AcquisisciDati extends ThObservable implements Runnable{

La buona notizia è che gli osservatori non hanno bisogno di alcuna modifica. Sostituendo alla classe Observable della libreria standard la nuova versione asincrona si otterranno performance migliori nella maggior parte dei casi e, soprattutto, si manterrà un controllo più accurato sul comportamento dei vari oggetti osservatori.
Allocazione del tempo e blocco delle risorse
Resta in sospeso il settimo e famigerato punto della specifica: non consentire "...che metodi update() dall'esecuzione troppo lunga monopolizzino la CPU". Mi vengono in mente diverse soluzioni, ma nessuna di applicazione così generale da essere inserita nella classe astratta ThObservable:

• assegnare un tempo massimo d'esecuzione consentita per ogni servizio al momento della registrazione e, controllandolo in fase di chiamata, eliminare le registrazioni degli osservatori troppo ingombranti. Questa non è una strada di semplice realizzazione in Java perchè non è possibile sollevare eccezioni dall'esterno di un thread ed il metodo Thread.Interrupt() non è implementato sulla maggior parte delle JVM. Sarebbe quindi necessario distruggere e ricreare il Broker ogni volta che esso si imbatte in un client prepotente.
• creare dinamicamente un thread Broker per ogni osservatore registrato: questa sembra una soluzione generalmente troppo pesante ma potrebbe ben adattarsi ad alcune situazioni e, comunque, presenta il grande vantaggio di disaccoppiare tutti i client tra loro. In coda al presente articolo è riportato il codice di RtObservable che, al costo di un maggior numero di thread creati, rende indipendenti le esecuzioni dei metodi update() dei vari Observer.
• miscelare le due ipotesi precedenti in modo che la priorità dei thread Broker abbinati agli osservatori più onerosi venga abbassata proporzionalmente al loro peso. I lettori interessati potranno sollazzarsi con l'implementazione di queste ipotesi: chi ne trovasse di migliori è pregato di tenermi aggiornato.

public class Virus implements Observer{
        public Virus() {super();}
        public void update(Observable obs, Object param){
                if(obs instanceof ThObservable.Notification) {
                        Object obj=((ThObservable.Notification)obs).obj;
                        new BadBoy(obj);
                }
        }
        private class BadBoy implements Runnable{
                private Object victim;
                private Thread me;
                badBoy(Object victim){
                        this.victim=victim;
                                me=new Thread(this);
                                me.start();
                        }
                        public void run(){
                                synchronized(victim){while(true);}
                        }
                }
        }
}

In merito è necessario precisare che un oggetto Java non è di per sè un monitor anche se tutti i suoi metodi sono dichiarati synchronized: la caratteristica principale che deve possedere un monitor è quella di essere l'unico responsabile per il proprio lock. Se chiunque in possesso di un riferimento ad un oggetto può bloccare a tempo indeterminato tutti i sui metodi sincronizzati allora il guscio offerto dall'incapsulazione del codice dentro una classe appare assai fragile!

Per le classi pubbliche è preferibile rinunziare ai metodi sincronizzati ed utilizzare, in loro vece, una sincronizzazione esplicita su un oggeto privato e, quindi, inaccessibile all'esterno. Ecco come implementare una classe che possa essere passata in modo sicuro come parametro agli Observer:

public class MailBox implements RtSource{
...
        private Object mailBox=null;
        private Object cond=new Object();
        public Object read(){
                synchronized(cond) {
                        if(mailBox==null)
                        try{cond.wait();}catch(Exception e) {}
                        return mailBox;
                }
        }
        public void write(Object obj){
                synchronized(cond) {
                        mailBox=obj;
                        cond.notify();
                }
        }
}

RtObservable: una versione multithreaded di ThObservable
Segue il codice di RtObservable che risolve il problema dell'accoppiamento residuo tra i vari client collegati.
Quì viene creato un thread con coda per ogni Observer.

Inoltre gli oggetti di tipo Notification sono presi da un pool invece che essere continuamente creati con new: in questo modo si cerca di diminuire il lavoro del garbage collector ottimizzando le perfomance genarli del sistema. Questa tecnica può funzionare solo se il riferimento al parametro di update() non viene conservato dall'osservatore per un uso successivo. Se non si ritiene che tale contratto possa essere rispettato, bisognerà eliminare dal codice la variabile pool ed i metodi obtainNotification() e returnNotification() rimpiazzandoli col consueto uso di new.
 

package RtObservable;
import java.util.*;
public abstract class RtObservable extends Observable{
        private Hashtable brokers=new Hashtable();
        private Notification pool=null;

        public class Notification {
                public long first=System.currentTimeMillis();
                public long last;
                public long num=1;
                public Object obj;
                Notification(Object obj){
                        this.obj=obj;
                        last=first;
                }
        }
        synchronized Notification obtainNotification(Object par){
                Notification n;
                if(pool!=null){
                        n = pool;
                        pool=(Notification)pool.obj;
                        n.first=System.currentTimeMillis();
                        n.last=n.first;
                        n.obj=par;
                }else{
                        n=new Notification(par);
                }
                return n;
        }
        synchronized void returnNotification(Notification n){
                n.obj=pool;
                pool=n;
        }
 

        public class Broker implements Observer, Runnable {
                Observer obs;
                int priority=5;
                Thread bTh=null;
                private Hashtable pars=new Hashtable();
                private Queue q=new Queue(null);
                Notification currentNotification=null;

                private class Queue{
                        private Queue next=null;
                        private Queue tail=null;
                        Notification not;
                        Queue(Notification not){this.not=not;}
                        synchronized void put(Object par){
                                if(par==null) par=this;
                                Notification n=(Notification)pars.get(par);
                                if(n!=null){
                                        n.last=System.currentTimeMillis();
                                        n.num++;
                                        return;
                                }
                                n=obtainNotification(par);
                                pars.put(par, n);
                                Queue q=new Queue(n);
                                if(next==null) next=q;
                                else tail.next=q;
                                tail=q;
                        }
                        synchronized Notification get(){
                                Queue q=next;
                                if(next!=null) {
                                        next=next.next;
                                        if(next==null) tail=null;
                                        pars.remove(q.not.obj);
                                        if(q.not.obj==this) q.not.obj=null;
                                        return q.not;
                                }
                                return null;
                        }
                        synchronized boolean hasMoreElements(){return (q.next != null);}
                }
 

 

                Broker(Observer obs, int priority){
                        this.obs=obs;
                        this.priority=priority;
                        bTh=new Thread(this);
                        bTh.setDaemon(true);
                        bTh.setPriority(priority);
                        bTh.start();
                }

                public void run(){
                        while(Thread.currentThread()==bTh){
                                synchronized(this){
                                        currentNotification=q.get();
                                        if(currentNotification==null){
                                                try{ wait(); }catch(Exception e){}
                                                continue;
                                        }
                                }
                                try{
                                        obs.update(RtObservable.this, currentNotification.obj);
                                }catch(Exception e){
                                        deleteObserver(obs);
                                        e.printStackTrace();
                                }
                                synchronized(this){
                                        returnNotification(currentNotification);
                                        currentNotification=null;
                                }
                        }
                }
                public synchronized void update(Observable obs, Object par){
                        q.put(par);
                        notify();
                }
        }
 

        public RtObservable(){super();}
        public synchronized void addObserver(Observer o, int priority){
                if(brokers.get(o)==null){
                        Broker b=new Broker(o, priority);
                        brokers.put(o, b);
                        super.addObserver(b);
                }
        }
        public synchronized void addObserver(Observer o){
                addObserver(o, Thread.currentThread().getPriority());
        }
        public synchronized void addSyncObserver(Observer o){super.addObserver(o);}
        public synchronized void deleteObserver(Observer o){
                Broker b=(Broker)brokers.get(o);
                if(b==null){
                        super.deleteObserver(o);
                }else{
                        synchronized(b){
                                super.deleteObserver(b);
                                brokers.remove(b);
                                b.bTh=null;
                                b.notify();
                        }
                }
        }
        public synchronized void deleteObservers(){
                super.deleteObservers();
                for(Enumeration e=brokers.elements();e.hasMoreElements();){
                        Broker b=(Broker)e.nextElement();
                        synchronized(b){
                                brokers.remove(b);
                                b.bTh=null;
                                b.notify();
                        }
                }
        }

        public Notification getNotification(Observer o){
                return ((Broker)brokers.get(o)).currentNotification;
        }
}

Bibliografia

[1] Albert Lopez - "How to decouple the Observer/Observable object model"
[2] http://java.sun.com/products/jdk/1.1/docs/api/Package-java.util.html
[3] Meilire Page-Jones - "The practical guide to structured design", Yourdon Press, New York, 1980.

	MokaByte rivista web su Java MokaByte ricerca nuovi collaboratori Chi volesse mettersi in contatto con noi può farlo scrivendo a mokainfo@mokabyte.it