Studio di una Architettura per un Sistema Distributivo ad Alta Affidabilità

Studio di unaArchitettura per un Sistema Distributivo

ad Alta Affidabilita

Autore: Dott. Roberto PeruzzoTutor: Prof. Alessandro Roncato

Mestre, 24 ottobre 2012

Indice

1 Introduzione 3

2 Rendere affidabile un sistema 72.1 Reliable Server Pooling . . . . . . . . . . . . . . . . . . . . 82.2 Cloud Computing . . . . . . . . . . . . . . . . . . . . . . . 8

2.2.1 Google AppEngine . . . . . . . . . . . . . . . . . . 92.2.2 Amazon AWS . . . . . . . . . . . . . . . . . . . . . 11

2.3 Scelta dell’infrastruttura . . . . . . . . . . . . . . . . . . . 11

3 Architettura proposta 153.1 Utilizzo di Java NIO . . . . . . . . . . . . . . . . . . . . . 173.2 Limiti nel prototipo di Pavan . . . . . . . . . . . . . . . . 213.3 Thread in un mondo Multiplexed . . . . . . . . . . . . . . 23

4 Implementazione 254.1 Come attivare i servizi Amazon AWS . . . . . . . . . . . . 254.2 Amazon SimpleDB . . . . . . . . . . . . . . . . . . . . . . 274.3 Amazon DynamoDB . . . . . . . . . . . . . . . . . . . . . 324.4 Il server . . . . . . . . . . . . . . . . . . . . . . . . . . . . 33

5 Prove Sperimentali 415.1 Preparazione dei test . . . . . . . . . . . . . . . . . . . . . 41

5.1.1 Hardware utilizzato . . . . . . . . . . . . . . . . . . 435.1.2 Modalita di esecuzione dei test . . . . . . . . . . . 43

5.2 Risultati test senza persistenza . . . . . . . . . . . . . . . 455.3 Risultati test con Amazon SimpleDB . . . . . . . . . . . . 47

iii

iv INDICE

5.4 Risultati test con Amazon DynamoDB . . . . . . . . . . . 49

6 Conclusioni 55

A Codice sorgente 59A.1 Server . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 59

A.1.1 Server.java . . . . . . . . . . . . . . . . . . . . . . . 59A.1.2 AsdaaServer.java . . . . . . . . . . . . . . . . . . . 61A.1.3 SelectorThread.java . . . . . . . . . . . . . . . . . . 63A.1.4 DynamoDBManager.java . . . . . . . . . . . . . . . 73

A.2 Client . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 77A.2.1 MultiplexingClient.java . . . . . . . . . . . . . . . . 77A.2.2 MultithreadClient.java . . . . . . . . . . . . . . . . 86

B Glossario 95

Elenco delle figure

1.1 Diagramma dell’attuale architettura da ottimizzare. . . . . 4

2.1 Google AppEngine vs. Amazon AWS. . . . . . . . . . . . . 12

2.2 Architettura con SimpleDB. . . . . . . . . . . . . . . . . . 13

2.3 Architettura con DynamoDB. . . . . . . . . . . . . . . . . 14

3.1 all’aumentare dei thread creati aumenta il tempo di crea-zione degli stessi. . . . . . . . . . . . . . . . . . . . . . . . 16

3.2 un Thread utilizza l’API Selector per gestire 3 canali dicomunicazione. . . . . . . . . . . . . . . . . . . . . . . . . 18

3.3 diagramma degli stati del server implementato. . . . . . . 19

3.4 diagramma degli stati nella soluzione proposta da Pavan. . 22

4.1 Amazon AWS console. . . . . . . . . . . . . . . . . . . . . 26

4.2 Amazon EC2 console. . . . . . . . . . . . . . . . . . . . . . 28

4.3 elenco delle Amazon Machine Image da scegliere. . . . . . 29

4.4 finsetra per creare una nuova tabella. . . . . . . . . . . . . 30

4.5 finsetra per creare una nuova tabella. . . . . . . . . . . . . 31

4.6 diagramma della classe SimpleDBManager.java. . . . . . . 32

4.7 diagramma della classe DynamoDBManager.java. . . . . . 33

4.8 diagramma della classe SelectorThread.java. . . . . . . . . 35

4.9 diagramma della classe Server.java. . . . . . . . . . . . . . 37

5.1 diagramma della classe MultithreadClient.java. . . . . . . . 42

5.2 diagramma della classe MultiplexingClient.java. . . . . . . 44

v

vi ELENCO DELLE FIGURE

5.3 rapporto tra quantita di dati inviati e tempi di rispostadel server. Totale messaggi inviati 200, Tempo di rispostamedio 108,71 ms, minimo 38 ms, massimo 282 ms. . . . . . 45

5.4 rapporto tra quantita di dati inviati e tempi di risposta delserver. Totale messaggi inviati 10481, Tempo di rispostamedio 261,51 ms, minimo 36 ms, massimo 57300 ms. . . . 46

5.5 rapporto tra quantita di dati inviati e tempi di risposta delserver. SimpleDB come persistenza dati. Totale messaggiinviati 200, Tempo di risposta medio 3599,295 ms, minimo203 ms, massimo 22638 ms. . . . . . . . . . . . . . . . . . 47

5.6 rapporto tra quantita di dati inviati e tempi di risposta delserver. SimpleDB come persistenza dati. Totale messaggiinviati 200, Tempo di risposta medio 1494,07 ms, minimo203 ms, massimo 4087 ms. . . . . . . . . . . . . . . . . . . 48

5.7 rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 5writes/s. Totale messaggi inviati 200, Tempo di rispostamedio 1701,3 ms, minimo 193 ms, massimo 4426 ms. . . . 50

5.8 rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 320writes/s. Totale messaggi inviati 200, Tempo di rispostamedio 904,06 ms, minimo 151 ms, massimo 2650 ms. . . . 51

5.9 rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 320writes/s. Totale messaggi inviati 800, Tempo di rispostamedio 223,38 ms, minimo 137 ms, massimo 746 ms. . . . . 52

5.10 rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 320writes/s, 10 Thread come client. Totale messaggi inviati658, Tempo di risposta medio 315,77 ms, minimo 144 ms,massimo 1529 ms. . . . . . . . . . . . . . . . . . . . . . . . 53

ELENCO DELLE FIGURE vii

5.11 rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 320writes/s, 100 thread come client. Totale messaggi inviati1707, Tempo di risposta medio 8191,05 ms, minimo 280ms, massimo 177857 ms. . . . . . . . . . . . . . . . . . . . 54

viii ELENCO DELLE FIGURE

Elenco delle tabelle

ix

x ELENCO DELLE TABELLE

Sommario

1

2 ELENCO DELLE TABELLE

Capitolo 1

Introduzione

Questo documento rappresenta il risultato di uno studio di algoritmie architetture utili per implementare un sistema ad alta affidabilita ealte prestazioni. Non si tratta semplicemente di un sistema capace digarantire un’elevata continuita del servizio ()uptime guarantee), ma deveessere in grado anche di non perdere informazioni in caso di guasti (faulttollerance) e di soddisfare un numero crescente di richieste (scalable). Unsistema per la gestione delle comunicazioni tra le centrali di smistamentoSMS e applicazioni esterne (per avere news sul meteo o sulla viabilita)puo esserne un esempio. Per meglio comprendere la problematica darisolvere partiamo con l’analizzare l’architettura descritta nella Figura1.1. Un insieme di client, il cui numero ee destinato ad aumentare neltempo, comunicato con un server di controllo con il compito di effettuarealcuni calcoli sui dati ricevuti e restituire un messaggio “ok ricevuto” alclient. Nel caso il server non sia raggiungibile a causa di un guasto allarete, i client invieranno i propri messaggi ad un indirizzo IP alternativodove un dispositivo inoltrera questi messaggi al server di controllo cheeseguira le stesse operazioni sopra descritte. La quantita medi di datiche ogni singolo client invia puo essere stimata in 100B/h. Nel caso ilserver principale, a causa di un guasto, interrompe la propria esecuzione,tutti i client rinviano dopo cinque minuti i messaggi che non hanno avutorisposta sperando che il guasto si risolva nel piu breve tempo possibile.

Il canale tra client e server utilizza una comunicazione GPRS a pa-

3

4 CAPITOLO 1. INTRODUZIONE

Un elevato numero di client (centraline)inviano moltissimi messaggi di pochi datial server per essere elaborati.

Nel caso l'IP PRINCIPALE non sia raggiungibile, i clients inviano i loro messaggi ad un IP SECONDARIO.

Database

IP SECONDARIO

IP PRINCIPALE

Server che elabora le richieste dei client

Dispositivo che esegue unsemplice forward dei messaggial server.

Figura 1.1: Diagramma dell’attuale architettura da ottimizzare.

5

gamento; il costo viene calcolato in base al numero di Byte inviate ericevuti, quindi varia in modo proporzionale al numero di messaggi cheattraversano il canale: piu messaggi sono spediti, piu alti saranno i costi.Quindi in caso di guasti del sistema, ogni rinvio di un messaggio da partedei client comporta un aumento del traffico nel canale e di conseguenzauna lievitazione dei costi proporzionalmente al tempo di “blocco”.

L’obbiettivo che si vuole raggiungere e di progettare una nuova ar-chitettura che sia in grado di garantire un’alta affidabilita in modo daminimizzare il piu possibile situazioni di guasto.

6 CAPITOLO 1. INTRODUZIONE

Capitolo 2

Rendere affidabile un sistema

Rendere un sistema affidabile significa garantire l’operabilita (uptimeguarantee) del sistema anche in presenza di guasti (fault tolerant). Lamaggior parte delle tecniche di tolleranza ai guasti fa uso di tecnichedi ridondanza per replicare le sorgenti che contengono le informazioni(Server Pooling). Esistono due tipi di ridondanza:

• spaziale, si utilizzano componenti hardware o software replicati ingrado di sostituire l’elemento guasto;

• temporale, basata sulla ripetizione di operazioni o sequenze di ope-razioni.

Un sistema in cui le informazioni vengono replicate non offre solo unamaggiore affidabilita, ma e in grado di fornire:

• migliori prestazioni, grazie ad una distribuzione del carico tra levarie repliche; questo permette inoltre di diminuire la latenza dellacomunicazione permettendo ai client di contattare “la replica” piuvicina;

• una migliore disponibilita del servizio, se ogni replica e isolata fi-sicamente ed elettricamente dalle altre, il guasto di una unita noninfluenza le altre;

7

8 CAPITOLO 2. RENDERE AFFIDABILE UN SISTEMA

Il rovescio della medaglia e che la ridondanza aumenta la complessitadi tutto il sistema poiche bisogna gestire lo smistamento dei messaggi ver-so le varie repliche, mantenere una consistenza dei dati replicati ed infinemantenere il determinismo dello stato delle repliche, in pratica riceverele stesse richieste da parte dei un client nello stesso ordine temporale.

2.1 Reliable Server Pooling

Vista la complessita nel gestire un insieme di server replicati, nel 5 dicem-bre 2000 prese vita un gruppo di lavoro IETF RSerPool [1] con l’obbiet-tivo di sviluppare l’architettura e i protocolli necessari per la gestione e ilfunzionamento di gruppi di server che supportano applicazioni altamenteaffidabili [15].

Il framework realizzato permette ad un insieme di server con informa-zioni replicate di essere visti come un singolo punto di accesso ai dati; incaso di fallimenti le applicazioni che lo utilizzano possono essere deviateverso un altro server in modo del tutto trasparente e senza il riavvio dellamacchina.

Nonostante il framework aiuti gli sviluppatori ad implementare appli-cazioni con un alto grado di affidabilita, presenta alcune problematicheda risolvere:

• dal punto di vista infrastrutturale, l’acquisto e la gestione di tuttala parte hardware e sistemistica ha un costo molto alto;

• il gruppo di lavoro e stato chiuso il 27 aprile 2009, quindi non c’ela possibilita di avere un supporto tecnico in caso di problemi.

La soluzione a questi problemi sta nel Cloud Computing descritto nelprossimo paragrafo.

2.2 Cloud Computing

Con la frase cloud computing si identifica quel servizio che fornisce ca-pacita di calcolo e di archiviazione ad una comunita eterogenea di utiliz-zatori finali in modo tale che quest’ultimi non si debbano minimamente

2.2. CLOUD COMPUTING 9

preoccupare di come gestire e garantire l’operabilita del sistema anche incaso di guasti, poiche tali sistemi utilizzano una ridondanza spaziale inmodo del tutto trasparente all’utilizzatore finale. Esistono tre tipologiedi cloud computing:

• Infrastructure as a Service (IaaS),

• Platform as a Service (PaaS),

• Software as a Service (SaaS).

Nel caso di studio sappiamo che il numero dei client e alto e si presumesia destinato ad aumentare con il passare del tempo, inoltre le comuni-cazioni tra client e server devono utilizzare il minor numero di messaggipossibili anche in caso di guasti in modo da consumare meno banda possi-bile. In altre parole il sistema deve essere scalabile e affidabile. I servizi dicloud computing hanno entrambi queste caratteristiche: in base al caricodi lavoro e alla quantita di dati da archiviare e possibile adattare (aumen-tando o diminuendo) la capacita di calcolo e di archiviazione in temporeale; in caso di guasti hardware i dati sono replicati in modo trasparen-te e la capacita di calcolo e garantita da un sistema di virtualizzazione.Tutto questo viene fornito a basso costo poiche, come l’energia elettrica,si pagano solo le risorse consumate (modello pay-as-you-go) senza doversipreoccupare dell’amministrazione di grossi datacenter.

Di seguito si analizzeranno alcuni tra i principali fornitori di servizicloud nel mercato per identificarne il piu adatto al caso, tenendo presentedi questi due vincoli:

1. la nostra applicazione sara scritta in Java visto che gran parte dellefunzionalita sono gia state scritte in questo linguaggio;

2. la nostra applicazione dovra implementare una comunicazione clien-t/server tramite un socket TPC.

2.2.1 Google AppEngine

Google AppEngine [10] offre un sistema per poter sviluppare applicazioniweb a traffico elevato senza dover preoccuparsi di gestire l’infrastruttura.


Grazie a tale strumento e possibile realizzare applicazioni web utilizzandotecnologie Java Standard (Java 6 JVM) e farle girare sull’infrastrutturascalabile di Google. Questo tipo di servizio rientra nella tipologia PaaS.

Le istanze di calcolo, chiamate App Engine Backends, eseguono l’ap-plicazione con una capacita di memoria che va da 128MB fino a 1GB euna CPU con una frequenza configurabile da 600MHz a 4,8GHz. Altracaratteristica interessante e la possibilita di definire il numero massimodi istanze attive; in base al volume di dati elaborato il sistema attivanuove istanze automaticamente in modo da gestire l’aumento del caricodi lavoro senza degradare le prestazioni dell’applicazione. Google Ap-pEngine permette inoltre di servire piu richieste in parallelo garantendouna corretta sincronizzazione tra i thread (thread safe).

Google AppEngine e dotato inoltre di un servizio per la persistenza deidati (App Engine datastore [9]) con due diverse tipologie di data storagein base alla disponibilita e alla consistenza che si vuole garantire:

• il Master/Slave datastore utilizza un sistema di replicazione master-slave che replica in modo asincrono il dato quando viene scrittofisicamente nel data center. Questa soluzione offre un’elevata con-sistenza in lettura e nelle interrogazioni, a fronte di temporaneiperiodi di indisponibilita del servizio in caso di problemi del datacenter o downtime pianificati;

• nel High Replication datastore, i dati sono replicati (in modo sin-crono) in piu data center tramite un meccanismo basato sul Paxosalgorithm (7). Questo sistema offre un alto grado di disponibilita,ma il prezzo da pagare e un elevata latenza nella scrittura dei datidovuta proprio alla replicazione distribuita.

Ogni applicazione viene eseguita in un ambiente sandboxed sicuro edisolato in modo da prevenire che un’applicazione interferisca con le altre.In questo ambiente le applicazioni non possono aprire socket TCP o ac-cedere ad un altro host direttamente, ma fare solamente richieste HTTPo HTTPS. Questa caratteristica rappresenta una restrizione critica perlo sviluppo di questo progetto, poiche le classi gia sviluppate per la co-municazione client/server prevedono l’utilizzo di un canale TCP ed un

2.3. SCELTA DELL’INFRASTRUTTURA 11

protocollo ad-hoc. Utilizzare una comunicazione HTTP o HTTPS il vo-lume di ogni singolo pacchetto aumenterebbe a causa degli http headersda aggiungere al messaggio con l’inevitabile conseguenza di un aumentodella quantita di dati scambiati che si traduce con un aumento dei costidel canale. Alla luce di questa limitazione si e deciso di analizzare le ca-ratteristiche tecniche di un altro tra i principali fornitori di servizi cloud,descritto nel paragrafo seguente.

2.2.2 Amazon AWS

Amazon AWS [4], a differenza di Google, offre un’infrastruttura modu-lare che permette di scegliere i componenti che piu si adattano al tuoproblema creando un’architettura ad-hoc. In questo caso infatti si parladi un’infrastruttura come servizio IaaS. Oltre a servizi per il em data-storage (es. Amazon SimpleDB [3], Amazon DynamoDB [2]) mette adisposizione ambienti di calcolo virtuali personalizzabili dove e possibilecaricare differenti sistemi operativi, gestire i permessi di accesso alla re-te e personalizzare il proprio ambiente applicativo. Altra caratteristicaimportante e che il servizio offerto e “elastico” nel senso che e in gra-do di scalare automaticamente in base al carico di lavoro, configurandoopportunamente servizi di bilanciamento del carico e monitoraggio dellerisorse garantendo un uptime di 99,95% durante un’intero anno. Grazie aqueste caratteristiche e possibile costruire un’architettura personalizzatacombinando i vari servizi offerti secondo le proprie esigenze, senza pre-occuparsi della manutenzione dell’infrastruttura (es. upgrade, softwarepatch, ecc.) con un notevole risparmio monetario.

2.3 Scelta dell’infrastruttura

Dopo aver analizzato le caratteristiche di questi due sistemi (vedi Figura2.1) la scelta e ricaduta su Amazon.

Per il salvataggio dei dati abbiamo scelto inizialmente per AmazonSimpleDB poiche offriva sia un alta disponibilita del servizio che un’e-levata propensione alla scalabilita trattandosi di una base di dati nonrelazionale (NoSQL). Un vincolo imposto nel nostro progetto e quello di


Amazon AWS Google AppEngine

Java Yes Yes

Socket TCP Yes No

Latency in data persistence

Good Bad

System Customization Yes No

martedì 23 ottobre 2012

Figura 2.1: Google AppEngine vs. Amazon AWS.

2.3. SCELTA DELL’INFRASTRUTTURA 13

Client Units

Server(using Java NIO)

AmazonEC2

AmazonSimpleDB

Figura 2.2: Architettura con SimpleDB.

avere un’alta consistenza in modo da leggere sempre l’ultimo dato scritto;SimpleDB e in grado di garantire questo ma a discapito della latenza.

In Figura 2.2 e descritta l’architettura di partenza da cui siamo par-titi per implementare il server Java NIO [12] e l’interfacciamento conSimpleDB. Dai test fatti (vedi paragrafi successivi) le performace non sisono dimostrate soddisfacenti, quindi bisognava sostituire SimpleDB conun servizio che permettesse di diminuire il tempo di latenza dell’interosistema.

Amazon AWS ha lanciato il 19 gennaio 2012 un nuovo servizio perla persistenza dei dati chiamato Amazon DynamoDB. Questo nuovo ser-vizio oltre ai vantaggi gia visti con SimpleDB, ci consente di avere unaforte consistenza nei dati che garantisce la lettura dell’ultimo dato scritto,mantenendo una bassa latenza a qualsiasi livello di scalabilita. L’inter-vento quindi e stato indolore poiche abbiamo introdotto una nuova classejava in grado di gestire la comunicazione con il nuovo database (Figura2.3).


Client Units

Server(using Java NIO)

AmazonEC2

AmazonDynamoDB

Figura 2.3: Architettura con DynamoDB.

Capitolo 3

Architettura proposta

Una delle caratteristiche fondamentali dell’architettura che sara svilup-pata e la scalabilita ovvero, la capacita del server di gestire una quantitacrescente di connessioni simultanee.

Un’approccio tradizionale, che si basa sull’utilizzo del thread-poolingper gestire piu connessioni, trova il suo principale limite nel costante au-mento del tempo dedicato al context switching tra i thread proporzionaleall’aumento del numero di thread stessi. Inoltre, nel caso specifico di que-sta ricerca, ogni thread esegue operazioni I/O in modo concorrente checomporta un’ulteriore complessita al sistema peggiorandone le prestazio-ni. Il problema che nasce da un accesso concorrente sta nel fatto che laJava VM non sa quale thread e pronto ad eseguire un’operazione I/O,pertanto potrebbe mettere in esecuzione un thread non ancora pronto perun operazione I/O, mentre lasciare in attesa un thread gia pronto. Nelpeggior dei casi potrebbe accadere che un thread rimanga in attesa cosıa lungo da andare in timeout. Inoltre all’aumentare del numero di threadcreati aumenta il carico per la CPU e la quantita di memoria che verrariservata allo stack di ogni thread degradando le prestazioni dell’interosistema. Quest’ultima affermazione trova fondamento nei test fatti daLawery [11] in cui la creazione di nuovi thread diventa sempre piu lentaall’aumentare dei thread creati (vedi Figura 3.1).

In questi test i thread creati non eseguono alcuna operazione di cal-colo; in un caso reale dove ogni thread esegue operazioni di calcolo o di

15

16 CAPITOLO 3. ARCHITETTURA PROPOSTA

Figura 3.1: all’aumentare dei thread creati aumenta il tempo di creazionedegli stessi.

3.1. UTILIZZO DI JAVA NIO 17

I/O il degrado delle prestazioni sara di sicuro maggiore; questo si traducein un aumento del tempo di esecuzione e anche un aumento dei tempi dirisposta ai client.

Per evitare che un canale di comunicazione quindi rimanga in attesacosı a lungo da rischiare il timeout e quindi che il client non riceva larisposta attesa, bisognerebbe procedere con l’invio del messaggio nonappena il canale e pronto a scrivere e allo stesso tempo fare in modo cheun singolo thread sia in grado di gestire piu connessioni simultaneamente(multiplexing). Questo tipo di operazioni possono essere gestita graziealla programmazione ad eventi offerta dalla libreira Java NIO che verradescritta nel prossimo paragrafo.

3.1 Utilizzo di Java NIO

A partire dalla versione J2SE 1.4 Java introduce il concetto di socket non-bloccanti grazie alla libreria NIO (New I/O) che permette di gestire piuconnessioni TCP con un unico thread (vedi Figura 3.2) senza bloccarnel’esecuzione.

Ogni canale deve gestire i seguenti quattro eventi: connect, accept,read e write. Quando un client invia un messaggio al server quest’ultimoattiva un canale che solleva l’evento accept; non appena il canale accettala connessione viene sollevato l’evento connect. A questo punto il canale epronto per poter leggere i dati inviati dal client, quindi l’evento sollevatoe quello read; una volta letti tutti i dati ed elaborato la risposta il canalee pronto per inviare la risposta quindi viene sollevato l’evento write.

Questi eventi vengono gestiti dalla classe SelectorThread.java laquale si occupa di monitorare il Selector restando in attesa di eventi I/Oe redistribuirli al gestore (handler) appropriato in base alla tipologia dievento generato. Nella nostra architettura avremo bisogno di un gestoreper aprire connessioni in uscita, uno per accettare le connessioni in en-trata e uno per leggere e scrivere nel canale; questi gestori dovrannoimplementare rispettivamente le interfacce ConnectSelectorHandler,AcceptSelectorHandler e ReadWriteSelectorHandler.


Thread

Selector

Channel Channel Channel

Figura 3.2: un Thread utilizza l’API Selector per gestire 3 canali dicomunicazione.

3.1. UTILIZZO DI JAVA NIO 19

Waiting for Data(read interest)

Reading(no interest)

Processing Request

(no interest)

Writing(no interest)

Waiting to write(write interest)

Request partially read(enable read interest)

Ready to read

Requestread

Send reply(enable write interest)

Ready to write

Replay partially sent(enable write interest)

Replay sent(enable read interest)

Start

Figura 3.3: diagramma degli stati del server implementato.


Aspetto di fondamentale importanza in una architettura dove le ope-razioni di scrittura e lettura non sono bloccanti e la gestione dei datiparzialmente letti o scritti. Quando viene generato un evento di letturasignifica che nel read-buffer del socket sono disponibili alcuni byte, manon e precisato se questi byte sono una parte di un pacchetto, un pac-chetto intero o piu pacchetti. Lo stesso vale in fase di scrittura, i bytevengono scritti nel write-buffer del socket in base allo spazio disponibilenel buffer senza preoccuparsi che i dati del pacchetto siano stati scritticompletamente.

Nello schema proposto da Nuno Santos nel suo articolo [14] sono evi-denziati cinque stati per descrivere il ciclo di vita di un handler (vediFigura 3.3):

1. Waiting for data: l’handler e in attesa di ricevere i dati provenientidal canale di comunicazione, quindi per prima cosa attiva l’interes-se nel ricevere un evento di tipo lettura e attende che esso vengagenerato;

2. Reading: dopo aver ricevuto un evento di lettura, l’handler ricevei dati provenienti dal socket e inizzia a riassemblare il pacchetto.Durante questo stato non c’e l’interesse a ricevere nessun altro tipodi evento. Se il pacchetto viene riassemblato completamente sipassa allo stato successivo (stato 3), altrimenti vengono salvati idati parziali e viene riattivato l’interesse in lettura (stato 1).

3. Processing request: non appena il pacchetto viene completamenteriassemblato, l’handler inizia a processarlo. Nel nostro caso prov-vedera a controllare la validita del CRC e poi procedera a salvareil messaggio nel database. Finche l’handler rimane in questo stato,tutti gli interessi in eventi di tipo I/O sono disabilitati.

4. Writing: finito di processare il pacchetto e non appena la risposta epronta l’handler la spedisce immediatamente utilizzando una scrit-tura non-bloccante. Se non c’e spazio sufficiente nel write-bufferdel socket per contenere tutto il pacchetto, sara necessario spedireil resto dei dati successivamente passando allo stato 5, altrimenti

3.2. LIMITI NEL PROTOTIPO DI PAVAN 21

l’handler si preparera a ricevere il pacchetto successivo tornandoallo stato 1.

5. Waiting to write: se una scrittura non-bloccante ritorna senza averscritto tutti i dati che compongono il pacchetto di risposta, l’hand-ler attiva l’interesse nel ricevere un evento di scrittura. Quando ilbuffer di scrittura avra nuovamente dello spazio disponibile, verrasollevato l’evento per scrivere il resto del pacchetto.

Il ciclo di vita dell’handler permette di comprendere cosa accade quan-do riceve o spedisce dati attraverso il canale di comunicazione. Moltointeressante risulta la gestione dell’interesse in lettura/scrittura descrittanel diagramma di stato in Figura 3.3. Grazie a questo meccanismo epossibile capire:

• in fase di lettura quando un’intero pacchetto di dati e stato ricevutoe quindi procedere con l’elaborazione dei dati;

• in fase di scrittura quando il pacchetto di risposta e stato inviatocompletamente e quindi verificare se ci sono altri dati da leggere ose chiudere la comunicazione.

Questo rende l’architettura riutilizzabile per risolvere svariate proble-matiche con protocolli di comunicazione differenti tra loro implementandol’interfaccia ProtocolDecoder.java a seconda dei casi.

3.2 Limiti nel prototipo di Pavan

Il prototipo sviluppato da Pavan e descritto nella sua tesi [13] presen-ta l’implemetazione di un server ECHO dove ogni pacchetto inviato alserver viene immediatamente restituito per intero al client senza alcunaelaborazione. In questo prototipo non esistono dei gestori (handlers) chevengono invocati a seconda della tipologia di evento sollevato; il serverServerSelect.java continua ad iterare l’insieme dei canali attivi e appe-na ne trova uno pronto ad effettuare un’operazione di lettura, catturatutti i dati presenti nel buffer per poi immediatamente inviarli indietro


Waiting to Data

Reading

Writing

Replay sent

Start

Send replay

Ready to read

Figura 3.4: diagramma degli stati nella soluzione proposta da Pavan.

al client cosı come sono arrivati. Il buffer del canale ha una dimensionedi 16KB ed i pacchetti possono avere una dimensione massima di 1200Byte, quindi il buffer ha una dimensione tale per contenere l’intero mes-saggio. Il primo limite di questa implementazione quindi e la dimensionedei pacchetti che possono essere scambiati tra client e server.

Nello scenario presentato nell’introduzione (capitolo 1) dove ogni mes-saggio e formato da pochi Byte il problema non si pone, ma con una vi-sione aperta al futuro e alla possibilita di applicare una tale architetturaad altri servizi (es. trasferimento di video da client a server o viceversa)la soluzione di Pavan e piuttosto limitativa. Inoltre impostare un bufferdi 16KB puo raggiungere facilmente run out of memory all’aumentare delnumero di connessioni e quindi non essere scalabile. Altro limite nell’im-plementazione di Pavan deriva dalla mancanza di uno stato che disabilitimomentaneamente la possibilita di sollevare altri eventi I/O durante l’e-laborazione dei dati contenuti nel pacchetto (nel nostro caso controllo del

3.3. THREAD IN UN MONDO MULTIPLEXED 23

CRC e salvataggio dei dati). Operazioni come il salvataggio dei dati inun database sono di tipo bloccante ed hanno un tempo di latenza piu omeno lungo in base alla tipologia di tecnologia utilizzata; in questa faseil server deve mettere in attesa il canale finche la risposta da inviare alclient non sara pronta. Il non rispetto di una tale attesa potrebbe de-terminare situazione di perdita di dati dovuta all’interruzione prematuradelle operazioni di salvataggio.

In base a queste considerazioni il prototipo di Pavan, presentatonel diagramma di stato illustrato nella Figura 3.4, dev’essere rivisto ereimplementato sequendo quanto descritto nel Paragrafo 3.1.

3.3 Thread in un mondo Multiplexed

Utilizzare un singolo thread che svolga tutto il lavoro (dispatch and pro-cess request) ha il grosso limite di non poter “nascondere” la latenzanecessaria per operazioni I/O su disco; di conseguenza lettura e scrittu-ra su database influenzano negativamente le prestazioni di calcolo (JavaNIO non supporta operazioni non-blocking su filesystem).

Per prima cosa vediamo in modo generico quali sono le tipologie diarchitettura che e possibile realizzare per risolvere il problema:

• M dispatchers/no workers: parecchi thread di event-dispatcher cheeseguono tutto il lavoro;

• 1 dispatcher/N workers: un singolo event-dispatcher thread e unitaelaborative multiple;

• M dispatchers/N workers: thread multipli per lo smistamento deglieventi e unita elaborative multiple;

Solitamente per risolvere un problema complesso si cerca di suddivi-derlo in due o piu sotto problemi di facile risoluzione, ossia divide et im-pera. L’idea iniziale quindi era quella di suddividere gli event-dispatchere i worker (unita elaborative) in piu thread; in questo modo il dispatcherdelega ai worker la parte di elaborazione e salvataggio sul database e si


occupa solamente di gestire le connessioni con i client. Basandosi sull’e-sperienza fatta da Santos questa non e la soluzione migliore poiche un’ar-chitettura del genere presenta un grande numero di punti di interazionetra dispatcher e worker rendendo altamente problematica la sincronizza-zione nell’accedere al database, essendo una risorsa condivisa. Come estato descritto all’inizio di questo Capitolo quando piu thread devono ac-cedere in maniera concorrente una risorsa condivisa, sorge il problema diimplementare un codice thread-safe. Questo si scontra con la natura delSelector e i suoi SelectionKey in quanto non sono “sicuri” per accessimulti-thread. Se un SelectionKey viene modificato da un thread mentreun altro thread sta chiamando il metodo select() dell’oggetto Selectora cui appartiene lo stesso SelectionKey, il risultato e che select() termi-na brutalmente con un’eccezione. Questo solitamente accade quando unworker chiude i canali di comunicazione e indirettamente cancella anche iSelectionKey corrispondenti, quindi se select() viene chiamato, troverail SelectionKey cancellato inaspettatamente. Santos dopo un tentativoper rafforzare il sistema con regole piu severe nel controllo dei Selector, leproprie SelectionKeys e i RegisteredChannels, si e accorto che l’architet-tura iniziava a diventare sempre piu complicata, inefficiente e propensaagli errori.

Alla luce di queste considerazioni e stato scelto di partire con l’archi-tettura piu semplice (M dispatchers/no workers). Come abbiamo vistoall’inizio di questo capitolo il numero di thread va limitato altrimentile prestazioni rischiano di degenerare. Secondo Santos, l’ideale sarebbeda mantenere costante ad un valore di 4 o 5 il rapporto tra numero didispatcher e CPU:

dispatchers

CPUs= 5

Prima di inoltrarci in problematiche relative al numero di thread inesecuzione ed il loro bilanciamento del carico, e opportuno capire se ilcodice proposto da Santos, opportunamente modificato, e idoneo a questoprogetto. Fondamentale capire quindi se i tempi di risposta del serversono soddisfacenti o meno.

Capitolo 4

Implementazione

In questo capitolo verra illustrato per prima cosa come configurare gliambienti di calcolo e di persistenza dei dati con Amazon AWS e il co-dice Java utilizzato per interfacciarsi con tali servizi. Come ultima cosavedremo il funzionamento del Server e le modifiche apportate al codiceproposto da Santos.

4.1 Come attivare i servizi Amazon AWS

Per utilizzare l’infrastruttura offerta da Amazon bisogna creare un ac-count utente registrandosi al sito http://aws.amazon.com. Amazon dala possibilita di testare i propri servizi gratuitamente grazie al AWS FreeUsage Tier; l’infrastruttura offerta in questo caso ha alcune limitazioni(vedi dettaglio [5]), che abbiamo dovuto superare durante i test in mo-do da raggiungere buone performance. Come vedremo nel paragrafo deitest, abbiamo dovuto superare questi limiti gratuiti.

Creato l’account sara possibile configurare i servizi di cui si ha biso-gno grazie alla console (vedi Figura 4.1) che Amazon mette a disposizioneall’utente. Per questo progetto e stata utilizzata un’istanza EC2 dove ab-biamo installato il server. L’attivazione e molto semplice, basta cliccaresul link “EC2 Virtual Services in the Cloud” per aprire la console di ge-stione delle istanze virtuali di EC2 (vedi Figura 4.2), cliccare sul bottone

25

26 CAPITOLO 4. IMPLEMENTAZIONE

Figura 4.1: Amazon AWS console.

4.2. AMAZON SIMPLEDB 27

“Launch Instance e scegliere la macchina virtuale che si vuole utilizzare(vedi Figura 4.3).

Per attivare il database si deve cliccare sul link “DynamoDB” presen-te sempre nella console (Figura 4.1) alla sezione “Database”. Una voltaentrati nella console di DymanoDB si procede con la creazione delle ta-belle (Figura 4.4) indicando il nome della tabella e il nome della chiaveprimaria. Successivamente bisogna indicare la capacita di lettura e scrit-tura1 con la quale si vuole lavorare; il valore di default per il serviziogratuito e di 10 unita/sec per la lettura e 5 unita/sec per la scrittura(Figura 4.5).

4.2 Amazon SimpleDB

Amazon mette a disposizione una libreria Java [6] offrendo un insieme diAPI in grado di interagire con l’infrastruttura dei servizi AWS. Per glisviluppatori che utilizzano Eclipse Java IDE, e possibile installare il plug-in open source AWS Toolkit for Eclipse [7], che permette di agevolarli infase di sviluppo, debug e deploy dell’applicazione. Nel caso specifico diSimpleDB, Amazon fornisce una serie di API che permettono di crearedomini (domains, astrazione che puo essere paragonata alle tabelle per undatabase relazionale) e elementi (items, astrazione che puo essere parago-nata ai record di una tabella), interrogare i dati, aggiornarli e rimuoverlidalla base di dati. Qui di seguito metto in evidenza i frammenti di codiceutilizzati nel nostro prototipo.

Per interagire con la base di dati viene utilizzata la classe AmazonSimpleDBClientche mette a disposizione i metodi per accedere e manipolare i dati in essacontenuti.

// Get PropertiesCredentialsPropertiesCredentials pCredentials =

1una unita di scrittura permette di effettuare una scrittura al secondo per elementidi dimensione non superiore a 1KB. Allo stesso modo una unita di lettura permettedi effettuare una lettura consistente al secondo di un elemento di dimensioni nonsuperiori a 1KB. Se l’elemento ha una dimensione superiore allora richiedera unamaggior capacita in scrittura o lettura.


Figura 4.2: Amazon EC2 console.


Figura 4.3: elenco delle Amazon Machine Image da scegliere.

new PropertiesCredentials(ClassLoader.getSystemResourceAsStream('AwsCredentials.properties'));

// Activate the Amazon SimpleDB Clientthis.sdb = new AmazonSimpleDBClient(new BasicAWSCredentials(

pCredentials.getAWSAccessKeyId(),pCredentials.getAWSSecretKey()));

this.sdb.createDomain(new CreateDomainRequest(domain));

Ogni item (oggetto di tipo ReplaceableItem) contenuto nel dominio erappresentato da un nome identificativo (rappresenta la primary key) e daun numero arbitrario di attributi (oggetto di tipo ReplaceableAttribute)a seconda delle nostre esigenze.

List<ReplaceableItem> data = new ArrayList<ReplaceableItem>();

data.add(new ReplaceableItem().withName(String.valueOf(recordId)).withAttributes(new ReplaceableAttribute()

.withName(”clientId”).withValue(clientId),new ReplaceableAttribute().withName(”body”)

.withValue(String.valueOf(byteRead)),new ReplaceableAttribute().withName(”created”).withValue(

String.valueOf(System.currentTimeMillis()))));


Figura 4.4: finsetra per creare una nuova tabella.


Figura 4.5: finsetra per creare una nuova tabella.


Figura 4.6: diagramma della classe SimpleDBManager.java.

4.3 Amazon DynamoDB

Con l’uscita di DymanoDB Amazon ha aggiornato la propria libreria Java[8] inserendo le nuove funzionalita per interagire con il nuovo servizio.La modalita di operare e la stessa vista precedentemente con SimpleDB,cambiano ovviamente i comandi utilizzati come e possibile vedere qui diseguito.

// Get PropertiesCredentialsInputStream credentialsAsStream = Thread.currentThread()

.getContextClassLoader()

.getResourceAsStream(”AwsCredentials.properties”);

AWSCredentials credentials = new PropertiesCredentials(credentialsAsStream);

// Activate the Amazon DynamoDB Clientthis.dynamoDB = new AmazonDynamoDBClient(credentials);

// Create an itemMap<String, AttributeValue> item =

new HashMap<String, AttributeValue>();item.put(”messageId”, new AttributeValue(messageId));

4.4. IL SERVER 33

Figura 4.7: diagramma della classe DynamoDBManager.java.

item.put(”clientId”, new AttributeValue(clientId));item.put(”body”, new AttributeValue(body));item.put(”created”, new AttributeValue().withN(

Long.toString(created)));

4.4 Il server

Il cuore del server e rappresentato dal ciclo while, riportato qui di segui-to, responsabile della gestione degli eventi che si trova all’interno dellaclasse io.SelectorThread descritta dall’illustrazione Figura 4.8. Questaclasse e stata implementata ispirandosi al modello di gestione degli eventiutilizzato dal toolkit SWING di Java: gli eventi generati dall’interfaccia


utente vengono salvati in una coda che viene controllata da un thread(istanza della classe java.awt.EventQueue) che si occupa di instradaregli eventi in entrata verso il listener interessato. La classe EventQueuedeve assicurare che tutte le operazioni eseguite dagli oggetti AWT sianoeffettuate in un unico thread. Allo stesso modo la classe SelectorThreadha incarico simile, in particolare:

• solo il thread creato dall’istanza di questa classe ha accesso al Se-lector e a tutti i socket da esso gestiti. Se un thread esterno vuoleavere accesso agli oggetti gestiti da questo Selector, allora dovrautilizzare i metodi invockeLater() o invokeAndWait().

• il thread non dev’essere utilizzato per eseguire lunghe operazioni.

Come gia detto nel Capitolo 3, questo tipo di architettura e stata scel-ta principalmente per mantenere basso il livello di complessita evitandol’onere di mantenere la sicronizzazione tra i thread.

Vediamo in dettaglio ora il funzionamento del ciclo principale.Per prima cosa vengono eseguiti eventuali task la cui esecuzione era

stata programmata da thread esterni (tramite le chiamate invockeLater()o invokeAndWait()).

public void run() {// Here's where everything happens. The select method will// return when any operations registered above have occurred, the// thread has been interrupted, etc.while (true) {

// Execute all the pending tasks.doInvocations();

...

...

...

}}

A questo punto si controlla che non ci sia una richiesta di chiusuradel Selector, in tal caso vengono portate a termine tutte le attivita incorso e poi il Selector viene chiuso. Se non ci sono richieste di chiusura,si procede con il verificare se qualcuno e pronto ad eseguire operazioni di

4.4. IL SERVER 35

Figura 4.8: diagramma della classe SelectorThread.java.


I/O; tramite la chiamata select() e possibile capire quante operazioni(identificate da una SelectedKey) tra i vari canali registrati sono pronteper essere eseguite.

// Time to terminate?if (closeRequested) {

return;}

Iterando questo insieme di selection key le operazioni vengono distri-buite al gestore appropriato, una volta che tutte le operazioni sono stateeseguite si ritorna all’inizio del loop principale e si ripete tutto il processodi nuovo.

// Someone is ready for IO, get the ready keysIterator it = selector.selectedKeys().iterator();// Walk through the collection of ready keys and dispatch// any active event.while (it.hasNext()) {

SelectionKey sk = (SelectionKey)it.next();

...

...

...}

Prendiamo ora in analisi la classe AsdaaServer.java (vedi Figura4.9)che avvia il server. Questa classe e un’estensione di Server.java

implementata da Santos a cui abbiamo aggiunto alcune modifiche peradattarla al nostro problema. Due sono le fasi principali quando vieneinstanziato l’oggetto di tipo AsdaaServer:

1. aprire il socket e abilitare il Selector a ricevere connessioni (chia-mata super(listenPort));

2. aprire la connessione con la base di dati (vedi il Paragrafo 4.3 percome creare la connessione e salvare i dati).

Vediamo cosa accade all’interno della chiamata super(listenPort).Per prima cosa si istanzia un oggetto di tipo SelectorThread e lo si abilitaad accettare connessioni alla porta specificata dal parametro listenPort.

4.4. IL SERVER 37

Figura 4.9: diagramma della classe Server.java.


public Server(int listenPort) throws Exception {st = new SelectorThread();Acceptor acceptor = new Acceptor(listenPort, st, this);acceptor.openServerSocket();logger.info(”Listening on port: ” + listenPort);System.out.println(”Listening on port: ” + listenPort);

}

Per abilitare il Selector viene chiamato il metodo openServerSocket()che si preoccupa di creare un socket e di registrare il nuovo canale di co-municazione con l’istruzione registerChannelLater() abilitandolo adaccettare streaming di dati passandogli come parametro la tipologia dievento SelectionKey.OP ACCEPT.

public void openServerSocket() throws IOException {ssc = ServerSocketChannel.open();InetSocketAddress isa = new InetSocketAddress(listenPort);ssc.socket().bind(isa, 100);

// This method might be called from any thread. We must use// the xxxLater methods so that the actual register operation// is done by the selector's thread. No other thread should access// the selector directly.ioThread.registerChannelLater(ssc,

SelectionKey.OP_ACCEPT,this,new CallbackErrorHandler() {

public void handleError(Exception ex) {listener.socketError(Acceptor.this, ex);}});}

Non appena un client si connette al server viene creato un PacketChannel

per gestire la connessione e tutti i dati che verranno inviati nel canale.Non dobbiamo trascurare il fatto che tra i parametri passati a questachiamata c’e l’oggetto new SimpleProtocolDecoder(); esso identificail protocollo utilizzato per lo scambio di messaggi, quindi premette distabilire quando il server ha ricevuto per intero il messaggio inviato dalclient.

public void socketConnected(Acceptor acceptor, SocketChannel sc) {logger.info(”[” + acceptor + ”] Socket connected: ”

+ sc.socket().getInetAddress());try {

4.4. IL SERVER 39

// We should reduce the size of the TCP buffers or else we will// easily run out of memory when accepting several thousands of// connctionssc.socket().setReceiveBufferSize(2 ∗ 1024);sc.socket().setSendBufferSize(2 ∗ 1024);// The contructor enables reading automatically.PacketChannel pc = new PacketChannel(sc, st,

new SimpleProtocolDecoder(), this);pc.resumeReading();

} catch (IOException e) {e.printStackTrace();

logger.error(”[” + sc.socket().getInetAddress().getHostAddress()+ ”:” + sc.socket().getPort()+ ”] Error on socket connection: ” + e.getMessage());

}}

Una volta che il messaggio e stato ricevuto per intero dal server verrachiamato il metodo packetArrived(PacketChannel pc, ByteBuffer pckt).Per prima cosa si converte il ByteBuffer in un byte array, si salvano idati nella base di dati e per finire invia indietro al client il messaggio “okricevuto”. Prima di spedire il pacchetto indietro, viene eseguita l’istru-zione pckt.flip() reimpostanto a zero la posizione corrente del buffer.Questa metodo dev’essere invocato ogni volta che, dopo una lettura dalcanale, si vuole preparare una sequenza di scrittura.

All’interno di questo metodo dovra essere inserita anche la verificaCRC del messaggio che per ora non e stata ancora implementata.

public void packetArrived(PacketChannel pc, ByteBuffer pckt) {logger.info(”[” + pc.toString() + ”] Packet received. Size: ” + pckt.remaining());

// Convert ByteBuffer to bytearraybyte[] bytearray = new byte[pckt.remaining()];pckt.get(bytearray);

// Save to the repositorydynamoDB.save(pc.toString(), new String(bytearray));

// Send packet back to the clientlogger.debug(”Send packet back”);pckt.flip();pc.sendPacket(pckt);logger.debug(”Packet sent.”);

}

Capitolo 5

Prove Sperimentali

In questo paragrafo metteremo in luce le problematiche che abbiamo in-contrato nello sviluppo dell’integrazione tra il server Java e i servizi offertida Amazon AWS. Grazie ai test fatti abbiamo potuto individuare i limitidei servizi Amazon e trovare soluzioni alternative per poter oltrepassarli.

5.1 Preparazione dei test

Per ogni test abbiamo creato un archivio JAR per il server e uno per ilclient in modo da agevolarne l’esecuzione. Nel corso del progetto abbiamosviluppato due tipologie di client che propongono due differenti approcci:

• MultithreadClient, (classe descritta in Figura 5.1) implementa ilmetodo tradizionale un thread per connessione TCP. Questo cipermette di simulare uno scenario che si avvicina maggiormentealla realta: ogni connessione attiva invia i messaggi ad intervalli ditempo casuali.

• MultiplexingClient, (classe descritta in Figura 5.2) utilizza la libre-ria Java NIO che permette di gestire le connessioni multiple conun unico thread. Questo client per prima cosa apre tutte le con-nessioni con il server e poi abilita l’invio dei messaggi al server. Imessaggi vengono inviati in modo sequenziale, in altre parole ogni

41

42 CAPITOLO 5. PROVE SPERIMENTALI

Figura 5.1: diagramma della classe MultithreadClient.java.

5.1. PREPARAZIONE DEI TEST 43

connessione aperta attende di ricevere risposta dal server prima diinviare un nuovo messaggio. Questo approccio permette di indi-viduare piu facilmente quante connessioni il server e in grado digestire simultaneamente.

Per avviare questi client vengono utilizzati due script bash ai qualie possibile passare il numero di porta del server e il numero di threadda lanciare (per multithreadClient.sh) o il numero di connessioni daaprire (per multiplexingClient.sh).

5.1.1 Hardware utilizzato

A lato server e stato utilizzata un’istanza Amazon EC2 Linux Micro con613MB di RAM, 8GB di spazio su disco e Linux 3.2. Per la persistenzadei dati e stato utilizzato inizialmente Amazon SimpleDB con a dispo-sizione 25 ore macchina e 1GB di spazio al mese; nella seconda fase delprogetto e stato utilizzato Amazon DynamoDB con 100MB di spazio,una capacita di scrittura di 5 write/sec e una capacita di lettura di 10read/sec. Per migliorare le performance dei test la capacita di scritturae di lettura sono state incrementate fino a 320 writeread/sec.

Il client —————– (recuperare dati client).La rete utilizzata nei test e quella del dipartimento di Informatica

dell’universita Ca’ Foscari che offre una banda di 100Mbps nell’intranet(LAN) e una banda di 10GBps per la connessione verso l’esterno (WAN).

Server e Client sono installati su macchine che non appartengono allastessa LAN.

5.1.2 Modalita di esecuzione dei test

I test fatti li possiamo classificare in tre momenti:

1. un semplice test preliminare sul server realizzato senza persistenzadei dati, lo scopo era quello di verificare la corretta comunicazionetra client e server;

2. test per verificare le prestazioni dell’architettura persistendo i daticon Amazon SimpleDB;


Figura 5.2: diagramma della classe MultiplexingClient.java.

5.2. RISULTATI TEST SENZA PERSISTENZA 45

Figura 5.3: rapporto tra quantita di dati inviati e tempi di risposta delserver. Totale messaggi inviati 200, Tempo di risposta medio 108,71 ms,minimo 38 ms, massimo 282 ms.

3. test per verificare le prestazioni dell’architettura persistendo i daticon Amazon DynamoDB;

5.2 Risultati test senza persistenza

L’andamento a “scalini” visibile nella Figura 5.3 indica come i tempi di ri-sposta tendono ad aumentare non in maniera proporzionale all’aumentaredella dimensione del messaggio, a quanto pare esiste un intervallo entrocui la latenza sembra stabilizzarsi. Aumentando di 50 volte il numerodi connessioni (Figura 5.4) i tempi di latenza hanno dei picchi piuttostoalti, ma con una frequenza cosı bassa che la latenza media mantiene unbuon valore.


Figura 5.4: rapporto tra quantita di dati inviati e tempi di risposta delserver. Totale messaggi inviati 10481, Tempo di risposta medio 261,51ms, minimo 36 ms, massimo 57300 ms.

5.3. RISULTATI TEST CON AMAZON SIMPLEDB 47

Figura 5.5: rapporto tra quantita di dati inviati e tempi di risposta delserver. SimpleDB come persistenza dati. Totale messaggi inviati 200,Tempo di risposta medio 3599,295 ms, minimo 203 ms, massimo 22638ms.

5.3 Risultati test con Amazon SimpleDB

Osservando questi due test si puo notare come la latenza abbia un valoremolto alto gia con poche connessioni attive. Questi test sono stati fattia distanza di qualche ora e nonostante abbiamo gli stessi parametri d’in-gresso, hanno un andamento e delle prestazioni completamente diverse;questo e da attribuirsi alla diversa congestione della linea internet utiliz-zata in facolta. Non sono stati fatti ulteriori test con SimpleDB percheritenuti gia troppo poco performanti questi appena fatti.


Figura 5.6: rapporto tra quantita di dati inviati e tempi di risposta delserver. SimpleDB come persistenza dati. Totale messaggi inviati 200,Tempo di risposta medio 1494,07 ms, minimo 203 ms, massimo 4087 ms.

5.4. RISULTATI TEST CON AMAZON DYNAMODB 49

5.4 Risultati test con Amazon DynamoDB

Dai risultati ottenuti nel primo test con DynamoDB (Figura 5.7) sem-bra non essere cambiato nulla rispetto alle performance di SimpleDB. Ilmotivo sta nella capacita di scrittura nella base di dati; la configurazio-ne base del servizio prevede al massimo 5 write/sec. Quando arrivanomessaggi con una frequenza piu alta, questi vengono accodati e elabo-rati appena possibile (paradigma FIFO) determinando un rallentamentodel tempo di risposta. Una unita di scrittura permette di scrivere nellabase di dati un elemento di al massimo 1KB al secondo. Se l”elementosupera questo limite e necessario calcolare la capacita di cui si ha bi-sogno; ad esempio se ogni elemento ha dimensioni di 1,5KB e si vuoleottenere 100 scritture al secondo dovremmo approvvigionarci 100 (wri-te/sec) x 2 (1,5KB arrotondato all’intero superiore piu vicino) = 200unita di capacita. La capacita di scrittura ha un costo che varia in basealla regione in cui si trova DynamoDB che si utilizza; in questa paginahttp://aws.amazon.com/dynamodb/pricing/ ci sono i prezzi che varianoda un minimo di $ 0,01 ad un massimo di $ 0,012 all’ora per ogni 10unita di scrittura. Visto il limitato budget a disposizione per effettuarei test abbiamo scelto una capacita di scrittura pari a 320 writes/sec cheabbiamo preventivamente stimato come una spesa media mensile di circa$ 275,96.

L’aumento della capacita di scrittura ha diminuito i tempi di laten-za (Figura 14) migliorando discretamente le prestazioni. Il sospetto diquesto risultato si fonda sull’implementazione del MultiplexingClient chegestisce connessioni multiple sullo stesso thread e ha un degrado delleprestazioni all’aumentare delle connessioni dovute all’overhead determi-nato dal context-switching nel passare da una connessione all’altra. Aconfermare questo sospetto e stato il test descritto in Figura 15 doveabbiamo utilizzato un’unica connessione. Alla luce di questo abbiamoimplementato un client multi-thread per abbattere l’overhead di gestionee avere un singolo thread per ogni connessione.

Il risultato ottenuto in Figura 16 dimostra come i tempi di rispostasiano migliorati di quasi 2/3 rispetto a prima, confermando come la ge-stione di connessioni multiple con un singolo thread sia un’operazione


Figura 5.7: rapporto tra quantita di dati inviati e tempi di risposta delserver. Utilizzato DynamoDB per la persistenza, 5 writes/s. Totale mes-saggi inviati 200, Tempo di risposta medio 1701,3 ms, minimo 193 ms,massimo 4426 ms.


Figura 5.8: rapporto tra quantita di dati inviati e tempi di risposta delserver. Utilizzato DynamoDB per la persistenza, 320 writes/s. Totalemessaggi inviati 200, Tempo di risposta medio 904,06 ms, minimo 151ms, massimo 2650 ms.


Figura 5.9: rapporto tra quantita di dati inviati e tempi di risposta delserver. Utilizzato DynamoDB per la persistenza, 320 writes/s. Totalemessaggi inviati 800, Tempo di risposta medio 223,38 ms, minimo 137ms, massimo 746 ms.


Figura 5.10: rapporto tra quantita di dati inviati e tempi di risposta delserver. Utilizzato DynamoDB per la persistenza, 320 writes/s, 10 Threadcome client. Totale messaggi inviati 658, Tempo di risposta medio 315,77ms, minimo 144 ms, massimo 1529 ms.

costosa. La stessa cosa si verifica anche a lato server, all’aumentare delleconnessioni simultanee da gestire i tempi di risposta di dilatano (vediFigura 17).


Figura 5.11: rapporto tra quantita di dati inviati e tempi di rispostadel server. Utilizzato DynamoDB per la persistenza, 320 writes/s, 100thread come client. Totale messaggi inviati 1707, Tempo di rispostamedio 8191,05 ms, minimo 280 ms, massimo 177857 ms.

Capitolo 6

Conclusioni

In questo documento e stato illustrato come un server ad alte prestazioniimplementato grazie all’utilizo della libreria Java NIO, sia in grado di in-terfacciarsi ad una base di dati dove poter salvare i dati inviati dai client.Lo studio dell’architettura per questo progetto ha preso spunto dall’im-plementazione fatta da Cristian Pavan e descritta nella testi “Server adAlte Prestazioni e ad Alta Affidabilita [13]. Una modifica sostanziale alserver di Pavan riguarda il differente comportamento (vedi il diagrammadi stato in Figura 3.3) nel gestire il flusso di dati in ricezione ed invio;questo permette di avere una maggiore flessibilita di dimensione dei pac-chetti scambiati tra client e server, un vantaggio non indifferente vistoche la tendenza per il futuro e quella di scambiare moli di dati sempremaggiori. Altra cosa che differenzia questo progetto da quanto fatto nel-la tesi di Pavan e l’utilizzo di un servizio di Cloud Computing sia per ilcalcolo che per lo storage dei dati definendo un’architettura nuova chenon si basa sui soliti schemi dove il centro di elaborazione viene gestitointernamente.

I risultati ottenuti nei primi test senza la persistenza dei dati sonomolto buoni

55

56 CAPITOLO 6. CONCLUSIONI

Ringraziamenti

57

58 CAPITOLO 6. CONCLUSIONI

Appendice A

Codice sorgente

A.1 Server

A.1.1 Server.java

/∗(c) 2004, Nuno Santos, [email protected] under terms of the GNU public licensehttp://www.gnu.org/licenses/licenses.html#TOCGPL

∗/package handlers;

import io.SelectorThread;

import java.io.IOException;import java.net.URL;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;

import org.apache.log4j.Logger;import org.apache.log4j.xml.DOMConfigurator;

/∗∗∗ A simple server for demonstrating the IO Multiplexing framework in action.∗ After accepting a connection, it will read packets as defined by the∗ SimpleProtocolDecoder class and echo them back.∗∗ This server can accept and manage large numbers of incoming connections. For∗ added fun remove the System.out statements and try it with several thousand∗ (>10.000) clients. You might have to increase the maximum number of sockets∗ allowed by the operating system.

59

60 APPENDICE A. CODICE SORGENTE

∗∗ @author Nuno Santos, modified by Roberto Peruzzo∗/

public class Server implements AcceptorListener, PacketChannelListener {

// Log4jstatic Logger logger = Logger.getLogger(Server.class);

private final SelectorThread st;

/∗∗∗ Starts the server.∗∗ @param listenPort∗ The port where to listen for incoming connections.∗ @throws Exception∗/

public Server(int listenPort) throws Exception {st = new SelectorThread();Acceptor acceptor = new Acceptor(listenPort, st, this);acceptor.openServerSocket();logger.info(”Listening on port: ” + listenPort);System.out.println(”Listening on port: ” + listenPort);

}

public static void main(String[] args) throws Exception {

URL log4jURL = ClassLoader.getSystemResource(”log4jserver.xml”);logger.debug(”Log4j config file URL: ” + log4jURL.toString());DOMConfigurator.configure(log4jURL);logger.info(”Log4j config loded: ” + log4jURL.toString());

int listenPort = Integer.parseInt(args[0]);new Server(listenPort);

}

// ////////////////////////////////////////// Implementation of the callbacks from the// Acceptor and PacketChannel classes// /////////////////////////////////////////∗∗∗ A new client connected. Creates a PacketChannel to handle it.∗/

public void socketConnected(Acceptor acceptor, SocketChannel sc) {logger.info(”[” + acceptor + ”] Socket connected: ”

+ sc.socket().getInetAddress());try {

// We should reduce the size of the TCP buffers or else we will// easily run out of memory when accepting several thousands of// connctionssc.socket().setReceiveBufferSize(2 ∗ 1024);sc.socket().setSendBufferSize(2 ∗ 1024);// The contructor enables reading automatically.

A.1. SERVER 61

PacketChannel pc = new PacketChannel(sc, st,new SimpleProtocolDecoder(), this);

pc.resumeReading();} catch (IOException e) {

e.printStackTrace();logger.error(”[” + sc.socket().getInetAddress().getHostAddress()

+ ”:” + sc.socket().getPort()+ ”] Error on socket connection: ” + e.getMessage());

}}

public void socketError(Acceptor acceptor, Exception ex) {logger.error(”[” + acceptor + ”] Error: ” + ex.getMessage());

}

public void packetArrived(PacketChannel pc, ByteBuffer pckt) {logger.debug(”[” + pc.toString() + ”] Packet received. Size: ”

+ pckt.remaining());pc.sendPacket(pckt);

}

public void socketException(PacketChannel pc, Exception ex) {logger.error(”[” + pc.toString() + ”] Error: ” + ex.getMessage());

}

public void socketDisconnected(PacketChannel pc) {logger.info(”[” + pc.toString() + ”] Disconnected.”);

}

/∗∗∗ The answer to a request was sent. Prepare to read the next request.∗/

public void packetSent(PacketChannel pc, ByteBuffer pckt) {try {

pc.resumeReading();} catch (Exception e) {

e.printStackTrace();logger.error(”[” + pc.toString() + ”] Error on resume reading: ”

+ e.getMessage());}

}}

A.1.2 AsdaaServer.java

/∗∗∗∗/

package it.studioaqua.unive.asdaa.handlers;

import handlers.PacketChannel;


import handlers.Server;import it.studioaqua.unive.asdaa.repository.DynamoDBManager;

import java.net.URL;import java.nio.ByteBuffer;


/∗∗∗ @author roberto∗∗/

public class AsdaaServer extends Server {

// Log4jstatic Logger logger = Logger.getLogger(AsdaaServer.class);

private DynamoDBManager dynamoDB;

/∗∗∗ @throws Exception∗∗/

public AsdaaServer(int listenPort) throws Exception {super(listenPort);

// Open the connection with Repository for saving messagesdynamoDB = new DynamoDBManager(”Messages”);

}


URL log4jURL = ClassLoader.getSystemResource(”log4jserver.xml”);logger.debug(”Log4j config file URL: ”+log4jURL.toString());DOMConfigurator.configure(log4jURL);logger.info(”Log4j config loded: ”+log4jURL.toString());

int listenPort = Integer.parseInt(args[0]);new AsdaaServer(listenPort);

}

/∗∗∗ The full packet was received, so resend it to the client (like echo service)∗/

public void packetArrived(PacketChannel pc, ByteBuffer pckt) {logger.info(”[” + pc.toString() + ”] Packet received. Size: ” + pckt.remaining()

);

// Convert ByteBuffer to bytearraybyte[] bytearray = new byte[pckt.remaining()];

A.1. SERVER 63

pckt.get(bytearray);

// Save to the repositorydynamoDB.save(pc.toString(), new String(bytearray));

// Send packet back to the clientlogger.debug(”Send packet back”);pckt.flip();pc.sendPacket(pckt);logger.debug(”Packet sent.”);

}

}

A.1.3 SelectorThread.java

/∗(c) 2004, Nuno Santos, [email protected] under terms of the GNU public licensehttp://www.gnu.org/licenses/licenses.html#TOCGPL

∗/package io;

import java.io.IOException;import java.nio.channels.∗;import java.util.∗;

/∗∗∗ Event queue for I/O events raised by a selector. This class receives the∗ lower level events raised by a Selector and dispatches them to the∗ appropriate handler. It also manages all other operations on the selector,∗ like registering and unregistering channels, or updating the events of∗ interest for each monitored socket.∗∗ This class is inspired on the java.awt.EventQueue and follows a similar∗ model. The EventQueue class is responsible for making sure that all∗ operations on AWT objects are performed on a single thread, the one managed∗ internally by EventQueue. The SelectorThread class performs a similar∗ task. In particular:∗∗ − Only the thread created by instances of this class should be allowed∗ to access the selector and all sockets managed by it. This means that∗ all I/O operations on the sockets should be peformed on the corresponding∗ selector's thread. If some other thread wants to access objects managed∗ by this selector, then it should use <code>invokeLater()</code> or the∗ <code>invokeAndWait()</code> to dispatch a runnable to this thread.∗∗ − This thread should not be used to perform lenghty operations. In∗ particular, it should never be used to perform blocking I/O operations.∗ To perform a time consuming task use a worker thread.∗


∗∗ This architecture is required for two main reasons:∗∗ The first, is to make synchronization in the objects of a connection∗ unnecessary. This is good for performance and essential for keeping∗ the complexity low. Getting synchronization right within the objects∗ of a connection would be extremely tricky.∗∗ The second is to make sure that all actions over the selector, its∗ keys and related sockets are carried in the same thread. My personal∗ experience with selectors is that they don't work well when being∗ accessed concurrently by several threads. This is mostly the result∗ of bugs in some of the version of Sun's Java SDK (these problems were∗ found with version 1.4.2 02). Some of the bugs have already been∗ identified and fixed by Sun. But it is better to work around them∗ by avoiding multithreaded access to the selector.∗∗ @author Nuno Santos∗/

final public class SelectorThread implements Runnable {/∗∗ Selector used for I/O multiplexing ∗/private Selector selector;

/∗∗ The thread associated with this selector ∗/private final Thread selectorThread;

/∗∗∗ Flag telling if this object should terminate, that is,∗ if it should close the selector and kill the associated∗ thread. Used for graceful termination.∗/

private boolean closeRequested = false;

/∗∗∗ List of tasks to be executed in the selector thread.∗ Submitted using invokeLater() and executed in the main∗ select loop.∗/@SuppressWarnings(”rawtypes”)

private final List pendingInvocations = new ArrayList(32);

/∗∗∗ Creates a new selector and the associated thread. The thread∗ is started by this constructor, thereby making this object∗ ready to be used.∗∗ @throws IOException∗/

public SelectorThread() throws IOException {// Selector for incoming time requestsselector = Selector.open();selectorThread = new Thread(this);selectorThread.start();

A.1. SERVER 65

}

/∗∗∗ Raises an internal flag that will result on this thread dying∗ the next time it goes through the dispatch loop. The thread∗ executes all pending tasks before dying.∗/

public void requestClose() {closeRequested = true;// Nudges the selector.selector.wakeup();}

/∗∗∗ Adds a new interest to the list of events where a channel is∗ registered. This means that the associated event handler will∗ start receiving events for the specified interest.∗∗ This method should only be called on the selector thread. Otherwise∗ an exception is thrown. Use the addChannelInterestLater() when calling∗ from another thread.∗∗ @param channel The channel to be updated. Must be registered.∗ @param interest The interest to add. Should be one of the∗ constants defined on SelectionKey.∗/

public void addChannelInterestNow(SelectableChannel channel,int interest) throws IOException {

if (Thread.currentThread() != selectorThread) {throw new IOException(”Method can only be called from selector thread”);}SelectionKey sk = channel.keyFor(selector);changeKeyInterest(sk, sk.interestOps() | interest);}

/∗∗∗ Like addChannelInterestNow(), but executed asynchronouly on the∗ selector thread. It returns after scheduling the task, without∗ waiting for it to be executed.∗∗ @param channel The channel to be updated. Must be registered.∗ @param interest The new interest to add. Should be one of the∗ constants defined on SelectionKey.∗ @param errorHandler Callback used if an exception is raised when executing the task.∗/

public void addChannelInterestLater(final SelectableChannel channel,final

int

interest

,


final

CallbackErrorHandler

errorHandler

)

{

// Add a new runnable to the list of tasks to be executed in the selector threadinvokeLater(new Runnable() {

public void run() {try {addChannelInterestNow(channel, interest);} catch (IOException e) {errorHandler.handleError(e);}}});}

/∗∗∗ Removes an interest from the list of events where a channel is∗ registered. The associated event handler will stop receiving events∗ for the specified interest.∗∗ This method should only be called on the selector thread. Otherwise∗ an exception is thrown. Use the removeChannelInterestLater() when calling∗ from another thread.∗∗ @param channel The channel to be updated. Must be registered.∗ @param interest The interest to be removed. Should be one of the∗ constants defined on SelectionKey.∗/

public void removeChannelInterestNow(SelectableChannel channel,int interest) throws IOException {

if (Thread.currentThread() != selectorThread) {throw new IOException(”Method can only be called from selector thread”);}SelectionKey sk = channel.keyFor(selector);changeKeyInterest(sk, sk.interestOps() & ˜interest);}

/∗∗∗ Like removeChannelInterestNow(), but executed asynchronouly on∗ the selector thread. This method returns after scheduling the task,∗ without waiting for it to be executed.∗∗ @param channel The channel to be updated. Must be registered.∗ @param interest The interest to remove. Should be one of the∗ constants defined on SelectionKey.∗ @param errorHandler Callback used if an exception is raised when∗ executing the task.

A.1. SERVER 67

∗/public void removeChannelInterestLater(final SelectableChannel channel,

final

int

interest

,

final

CallbackErrorHandler

errorHandler

)

{

invokeLater(new Runnable() {public void run() {

try {removeChannelInterestNow(channel, interest);} catch (IOException e) {errorHandler.handleError(e);}}});}

/∗∗∗ Updates the interest set associated with a selection key. The∗ old interest is discarded, being replaced by the new one.∗∗ @param sk The key to be updated.∗ @param newInterest∗ @throws IOException∗/

private void changeKeyInterest(SelectionKey sk,int newInterest) throws IOException {

/∗ This method might throw two unchecked exceptions:∗ 1. IllegalArgumentException − Should never happen. It is a bug if it happens∗ 2. CancelledKeyException − Might happen if the channel is closed while∗ a packet is being dispatched.∗/

try {sk.interestOps(newInterest);} catch (CancelledKeyException cke) {IOException ioe = new IOException(”Failed to change channel interest.”);ioe.initCause(cke);throw ioe;}}


/∗∗∗ Like registerChannelLater(), but executed asynchronouly on the∗ selector thread. It returns after scheduling the task, without∗ waiting for it to be executed.∗∗ @param channel The channel to be monitored.∗ @param selectionKeys The interest set. Should be a combination of∗ SelectionKey constants.∗ @param handler The handler for events raised on the registered channel.∗ @param errorHandler Used for asynchronous error handling.∗∗ @throws IOException∗/

public void registerChannelLater(final SelectableChannel channel,final int selectionKeys,

final

SelectorHandler

handlerInfo

,

final CallbackErrorHandler errorHandler) {invokeLater(new Runnable() {

public void run() {try {registerChannelNow(channel, selectionKeys, handlerInfo);} catch (IOException e) {errorHandler.handleError(e);}}});}

/∗∗∗ Registers a SelectableChannel with this selector. This channel will∗ start to be monitored by the selector for the set of events associated∗ with it. When an event is raised, the corresponding handler is∗ called.∗∗ This method can be called multiple times with the same channel∗ and selector. Subsequent calls update the associated interest set∗ and selector handler to the ones given as arguments.∗∗ This method should only be called on the selector thread. Otherwise∗ an exception is thrown. Use the registerChannelLater() when calling∗ from another thread.∗∗ @param channel The channel to be monitored.∗ @param selectionKeys The interest set. Should be a combination of∗ SelectionKey constants.∗ @param handler The handler for events raised on the registered channel.

A.1. SERVER 69

∗/public void registerChannelNow(SelectableChannel channel,

int selectionKeys,SelectorHandler handlerInfo) throws IOException {

if (Thread.currentThread() != selectorThread) {throw new IOException(”Method can only be called from selector thread”);}

if (!channel.isOpen()) {throw new IOException(”Channel is not open.”);}

try {if (channel.isRegistered()) {SelectionKey sk = channel.keyFor(selector);assert sk != null : ”Channel is already registered with other selector”;sk.interestOps(selectionKeys);Object previousAttach = sk.attach(handlerInfo);assert previousAttach != null;} else {channel.configureBlocking(false);channel.register(selector, selectionKeys, handlerInfo);}} catch (Exception e) {IOException ioe = new IOException(”Error registering channel.”);ioe.initCause(e);throw ioe;}}

/∗∗∗ Executes the given task in the selector thread. This method returns∗ as soon as the task is scheduled, without waiting for it to be∗ executed.∗∗ @param run The task to be executed.∗/@SuppressWarnings(”unchecked”)

public void invokeLater(Runnable run) {synchronized (pendingInvocations) {pendingInvocations.add(run);}selector.wakeup();}

/∗∗∗ Executes the given task synchronously in the selector thread. This∗ method schedules the task, waits for its execution and only then∗ returns.∗∗ @param run The task to be executed on the selector's thread.∗/

public void invokeAndWait(final Runnable task)


throws InterruptedException

{if (Thread.currentThread() == selectorThread) {

// We are in the selector's thread. No need to schedule// executiontask.run();} else {

// Used to deliver the notification that the task is executedfinal Object latch = new Object();synchronized (latch) {

// Uses the invokeLater method with a newly created taskthis.invokeLater(new Runnable() {

public void run() {task.run();// Notifieslatch.notify();}});// Wait for the task to complete.latch.wait();}// Ok, we are done, the task was executed. Proceed.}}

/∗∗∗ Executes all tasks queued for execution on the selector's thread.∗∗ Should be called holding the lock to <code>pendingInvocations</code>.∗∗/

private void doInvocations() {synchronized (pendingInvocations) {

for (int i = 0; i < pendingInvocations.size(); i++) {Runnable task = (Runnable) pendingInvocations.get(i);task.run();}pendingInvocations.clear();}}

/∗∗∗ Main cycle. This is where event processing and∗ dispatching happens.∗/@SuppressWarnings(”rawtypes”)public void run() {

// Here's where everything happens. The select method will// return when any operations registered above have occurred, the// thread has been interrupted, etc.while (true) {

// Execute all the pending tasks.doInvocations();

A.1. SERVER 71

// Time to terminate?if (closeRequested) {

return;}

int selectedKeys = 0;try {selectedKeys = selector.select();} catch (IOException ioe) {

// Select should never throw an exception under normal// operation. If this happens, print the error and try to// continue working.ioe.printStackTrace();continue;}

if (selectedKeys == 0) {// Go back to the beginning of the loopcontinue;}

// Someone is ready for IO, get the ready keysIterator it = selector.selectedKeys().iterator();// Walk through the collection of ready keys and dispatch// any active event.while (it.hasNext()) {SelectionKey sk = (SelectionKey)it.next();it.remove();try {

// Obtain the interest of the keyint readyOps = sk.readyOps();// Disable the interest for the operation that is ready.// This prevents the same event from being raised multiple// times.sk.interestOps(sk.interestOps() & ˜readyOps);SelectorHandler handler =

(SelectorHandler) sk.attachment();

// Some of the operations set in the selection key// might no longer be valid when the handler is executed.// So handlers should take precautions against this// possibility.

// Check what are the interests that are active and// dispatch the event to the appropriate method.if (sk.isAcceptable()) {

// A connection is ready to be completed((AcceptSelectorHandler)handler).handleAccept();

} else if (sk.isConnectable()) {// A connection is ready to be accepted((ConnectorSelectorHandler)handler).handleConnect();


} else {ReadWriteSelectorHandler rwHandler =

(ReadWriteSelectorHandler)handler;// Readable or writableif (sk.isReadable()) {

// It is possible to readrwHandler.handleRead();}

// Check if the key is still valid, since it might// have been invalidated in the read handler// (for instance, the socket might have been closed)if (sk.isValid() && sk.isWritable()) {

// It is read to writerwHandler.handleWrite();}}} catch (Throwable t) {

// No exceptions should be thrown in the previous block!// So kill everything if one is detected.// Makes debugging easier.closeSelectorAndChannels();t.printStackTrace();return;}}}}

/∗∗∗ Closes all channels registered with the selector. Used to∗ clean up when the selector dies and cannot be recovered.∗/

private void closeSelectorAndChannels() {@SuppressWarnings(”rawtypes”)

Set keys = selector.keys();for (@SuppressWarnings(”rawtypes”)

Iterator iter = keys.iterator(); iter.hasNext();) {SelectionKey key = (SelectionKey)iter.next();try {key.channel().close();} catch (IOException e) {

// Ignore}}try {selector.close();} catch (IOException e) {

// Ignore}}}

A.1. SERVER 73

A.1.4 DynamoDBManager.java

/∗∗∗ (c) 2011, Roberto Peruzzo, [email protected]∗ relased under terms of the GNU public license∗ http://www.gnu.org/licenses/licenses.html#TOCGPL∗/

package it.studioaqua.unive.asdaa.repository;

import java.io.IOException;import java.io.InputStream;import java.nio.charset.CharacterCodingException;import java.util.HashMap;import java.util.Map;

import org.apache.log4j.Logger;

import com.amazonaws.AmazonClientException;import com.amazonaws.AmazonServiceException;import com.amazonaws.auth.AWSCredentials;import com.amazonaws.auth.PropertiesCredentials;import com.amazonaws.services.dynamodb.AmazonDynamoDBClient;import com.amazonaws.services.dynamodb.model.AttributeValue;import com.amazonaws.services.dynamodb.model.Condition;import com.amazonaws.services.dynamodb.model.CreateTableRequest;import com.amazonaws.services.dynamodb.model.DescribeTableRequest;import com.amazonaws.services.dynamodb.model.DescribeTableResult;import com.amazonaws.services.dynamodb.model.KeySchema;import com.amazonaws.services.dynamodb.model.KeySchemaElement;import com.amazonaws.services.dynamodb.model.ProvisionedThroughput;import com.amazonaws.services.dynamodb.model.PutItemRequest;import com.amazonaws.services.dynamodb.model.PutItemResult;import com.amazonaws.services.dynamodb.model.ScanRequest;import com.amazonaws.services.dynamodb.model.ScanResult;import com.amazonaws.services.dynamodb.model.TableDescription;import com.amazonaws.services.dynamodb.model.TableStatus;

/∗∗∗ This class manage the interaction between Amazon SimpleDB and SelectedServer.∗ It persists every message received by SelectedServer.∗∗ @author Studio AQuA∗∗/

public class DynamoDBManager {

// Log4jstatic Logger logger = Logger.getLogger(DynamoDBManager.class);

private AmazonDynamoDBClient dynamoDB;private String table;


/∗∗∗ Constructor∗ @param tableName∗/

public DynamoDBManager(String tableName) {try {

logger.debug(”Get AWS Credentials.”);// Get PropertiesCredentialsInputStream credentialsAsStream = Thread.currentThread().

getContextClassLoader().getResourceAsStream(”AwsCredentials.properties”);

AWSCredentials credentials = new PropertiesCredentials(credentialsAsStream);

logger.debug(credentials.toString());

// Activate the Amazon DynamoDB Clientthis.dynamoDB = new AmazonDynamoDBClient(credentials);

// Describe our new tableDescribeTableRequest describeTableRequest = new DescribeTableRequest().

withTableName(tableName);TableDescription tableDescription = this.dynamoDB.describeTable(

describeTableRequest).getTable();logger.info(”Table Description: ” + tableDescription);

String tableStatus = tableDescription.getTableStatus();if (tableStatus.equals(TableStatus.ACTIVE.toString())) {

logger.info(”Table ”+tableName+ ” already exists.”);} else {

// Create a table with a primary key named 'name', which holds a stringCreateTableRequest createTableRequest = new CreateTableRequest().

withTableName(tableName).withKeySchema(new KeySchema(new KeySchemaElement().

withAttributeName(”messageId”).withAttributeType(”S”))).withProvisionedThroughput(new ProvisionedThroughput().

withReadCapacityUnits(10L).withWriteCapacityUnits(320L));tableDescription = this.dynamoDB.createTable(createTableRequest).

getTableDescription();logger.info(”Created Table: ” + tableDescription);

}

this.table = tableName;} catch (AmazonServiceException ase) {

logger.error(”Caught an AmazonServiceException, which means your requestmade it ”

+ ”to AWS, but was rejected with an error response for some reason.”);logger.error(”Error Message: ” + ase.getMessage());logger.error(”HTTP Status Code: ” + ase.getStatusCode());logger.error(”AWS Error Code: ” + ase.getErrorCode());logger.error(”Error Type: ” + ase.getErrorType());logger.error(”Request ID: ” + ase.getRequestId());

A.1. SERVER 75

} catch (AmazonClientException ace) {logger.error(”Caught an AmazonClientException, which means the client

encountered ”+ ”a serious internal problem while trying to communicate with AWS, ”+ ”such as not being able to access the network.”);

logger.error(”Error Message: ” + ace.getMessage());} catch (IOException e) {

logger.error(”RepositoryManager creation: IO exception. ”+e.getMessage());

e.printStackTrace();}

}

/∗∗∗ Persists the client message into SimpleDB.∗ @param clientId − client identifier∗ @param body − message∗ @return∗ @throws CharacterCodingException∗/

public boolean save(String clientId, String body) {

String recordId = clientId+” ”+String.valueOf(System.currentTimeMillis());

// Add an itemtry {

logger.info(”Save message id ”+recordId+” on table ”+ this.table);logger.debug(”MESSAGE BODY: ”+body);

Map<String, AttributeValue> item = newItem(recordId, clientId, body, System.currentTimeMillis());

PutItemRequest putItemRequest = new PutItemRequest(this.table, item);PutItemResult putItemResult = dynamoDB.putItem(putItemRequest);logger.info(”Result: ” + putItemResult.toString());logger.info(”Message from client ”+clientId+” saved.”);} catch (AmazonServiceException ase) {logger.error(”[SAVE] Caught an AmazonServiceException, which means your

request made it ”+ ”to AWS, but was rejected with an error response for some reason.”);

logger.error(”[SAVE] Error Message: ” + ase.getMessage());logger.error(”[SAVE] HTTP Status Code: ” + ase.getStatusCode());logger.error(”[SAVE] AWS Error Code: ” + ase.getErrorCode());logger.error(”[SAVE] Error Type: ” + ase.getErrorType());logger.error(”[SAVE] Request ID: ” + ase.getRequestId());return false;

} catch (AmazonClientException ace) {logger.error(”[SAVE] Caught an AmazonClientException, which means the client


logger.error(”[SAVE] Error Message: ” + ace.getMessage());


return false;}

return true;}

/∗∗∗ Get table content∗ @param tableName∗ @return∗/

public ScanResult scan(String tableName, HashMap<String, Condition> scanFilter) {

ScanResult scanResult = null;

try {ScanRequest scanRequest = new ScanRequest(tableName);if(!scanFilter.isEmpty()) {

scanRequest.withScanFilter(scanFilter);}scanResult = dynamoDB.scan(scanRequest);logger.info(”Result: ” + scanResult);

} catch (AmazonServiceException ase) {logger.error(”[SAVE] Caught an AmazonServiceException, which means your

request made it ”+ ”to AWS, but was rejected with an error response for some reason.”);

logger.error(”[SAVE] Error Message: ” + ase.getMessage());logger.error(”[SAVE] HTTP Status Code: ” + ase.getStatusCode());logger.error(”[SAVE] AWS Error Code: ” + ase.getErrorCode());logger.error(”[SAVE] Error Type: ” + ase.getErrorType());logger.error(”[SAVE] Request ID: ” + ase.getRequestId());

} catch (AmazonClientException ace) {logger.error(”[SAVE] Caught an AmazonClientException, which means the client


logger.error(”[SAVE] Error Message: ” + ace.getMessage());}

return scanResult;}

/∗∗∗∗ @param tableName∗ @return∗/

public Long countTableItems(String tableName) {

// Describe our new tableDescribeTableRequest describeTableRequest = new DescribeTableRequest().

withTableName(tableName);DescribeTableResult tableDescription = this.dynamoDB.describeTable(

A.2. CLIENT 77

describeTableRequest);return tableDescription.getTable().getItemCount();}

/∗∗∗∗ @param name∗ @param clientId∗ @param body∗ @param created∗ @return∗/

private Map<String, AttributeValue> newItem(String messageId, String clientId,String body, long created) {

Map<String, AttributeValue> item = new HashMap<String, AttributeValue>();item.put(”messageId”, new AttributeValue(messageId));item.put(”clientId”, new AttributeValue(clientId));item.put(”body”, new AttributeValue(body));item.put(”created”, new AttributeValue().withN(Long.toString(created)));

return item;}

}

A.2 Client

A.2.1 MultiplexingClient.java

package it.studioaqua.unive.asdaa.client;

import handlers.∗;import io.SelectorThread;

import java.io.FileOutputStream;import java.io.IOException;import java.io.PrintStream;import java.net.InetAddress;import java.net.InetSocketAddress;import java.net.URL;import java.net.UnknownHostException;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.util.∗;


/∗∗


∗ A simple test client for the I/O Multiplexing based server. This class simulates several clientsusing an imaginary request− reply protocol to connect to a server. Several connections canbe established at the same time. Each one, generates a random packet to simulate a request,sends it to the server and waits for the answer before sending the next packet. This

implementation is based on I/O Multiplexing, so it should be able to handle severalthousand of connections. Using Redhat Linux 9.0 I was able to establish about 10.000connections before running out of memory (the system had only 256Mb of RAM).

∗ @author Nuno Santos∗/

public class MultiplexingClient implements ConnectorListener, PacketChannelListener {

// Log4jstatic Logger logger = Logger.getLogger(MultiplexingClient.class);

/∗∗∗ A single selector for all clients∗ @uml.property name=”st”∗ @uml.associationEnd∗/

private static SelectorThread st;/∗∗ Maximum size of the packets sent ∗/private static final int MAX_SIZE = 1200; //10 ∗ 1024;/∗∗ Minimum size of the packets sent ∗/private static final int MIN_SIZE = 50; //128;/∗∗ How many packets each client should send ∗/private static final int PACKETS_TO_SEND = 10;/∗∗ For generating random packet sizes. ∗/private static final Random r = new Random();/∗∗ How many connections to created ∗/private static int connectionCount;/∗∗ How many connections were opened so far ∗/private static int connectionsEstablished = 0;private static int connectionsFailed = 0;/∗∗ How many connections were disconnected so far ∗/private static int connectionsClosed = 0;/∗∗∗ Keeps a list of connections that have been established but not yet∗ started.∗/@SuppressWarnings(”rawtypes”)private static List establishedConnections = new ArrayList(512);

/∗∗ How many packets each instance sent so far. ∗/private int packetsSent = 0;

/∗∗∗ Studio AQuA∗ @uml.property name=”rigaStat”∗ @uml.associationEnd∗/

// Statisticsprivate Stats rigaStat = null;private final static ArrayList<Stats> statistiche = new ArrayList<Stats>();

A.2. CLIENT 79

@SuppressWarnings(”unused”)private int clientId;

private static long globalStart = 0;

/∗∗∗ Initiates a non−blocking connection attempt to the given address.∗∗ @param remotePoint∗ Where to try to connect.∗ @throws Exception∗/

public MultiplexingClient(InetSocketAddress remotePoint, int clientId) throwsException {

Connector connector = new Connector(st, remotePoint, this);connector.connect();logger.info(”[” + connector + ”] Connecting...”);this.clientId = clientId;

}

/∗∗∗ Creates a new packet with a size chosen randomly between MIN SIZE and∗ MAX SIZE.∗/

private ByteBuffer generateNextPacket() {// Generate a random size betweenint size = MIN_SIZE + r.nextInt(MAX_SIZE − MIN_SIZE);ByteBuffer buffer = ByteBuffer.allocate(size);buffer.put(SimpleProtocolDecoder.STX);for (int i = 0; i < size − 2; i++) {

buffer.put((byte) 'a');}buffer.put(SimpleProtocolDecoder.ETX);buffer.limit(buffer.position());buffer.flip();return buffer;

}

// ////////////////////////////////////////// Implementation of the callbacks from the// Acceptor and PacketChannel classes// /////////////////////////////////////////∗∗∗ A new client connected. Creates a PacketChannel to handle it.∗/@SuppressWarnings(”unchecked”)public void connectionEstablished(Connector connector, SocketChannel sc) {

try {// We should reduce the size of the TCP buffers or else we will// easily run out of memory when accepting several thousands of// connctionssc.socket().setReceiveBufferSize(2 ∗ 1024);sc.socket().setSendBufferSize(2 ∗ 1024);


// The contructor enables reading automatically.PacketChannel pc = new PacketChannel(sc, st,

new SimpleProtocolDecoder(), this);

// Do not start sending packets right away. Waits for all sockets// to connect. Otherwise, the load created by sending and receiving// packets will increase dramatically the time taken for all// connections to be established. It is better to establish all// connections and only then to start sending packets.establishedConnections.add(pc);connectionsEstablished++;logger.info(”[” + connector + ”] Connected: ”

+ sc.socket().getInetAddress() + ” (”+ connectionsEstablished + ”/” + connectionCount +

”)”);// If if all connections are established.checkAllConnected();

} catch (IOException e) {e.printStackTrace();

}}

public void connectionFailed(Connector connector, Exception cause) {logger.error(”[” + connector + ”] Error: ” + cause.getMessage());connectionsFailed++;checkAllConnected();

}

public void packetArrived(PacketChannel pc, ByteBuffer pckt) {logger.debug(”[”+ pc.toString() + ”] Packet arrived”);

rigaStat.setFine(System.currentTimeMillis());rigaStat.setRicevuti(pckt.capacity());

//synchronized (statistiche) {statistiche.add(rigaStat);

//}

if (packetsSent >= PACKETS_TO_SEND) {// This connection sent all packets that it was supposed to send.// Close.logger.info(”[”

+ pc.getSocketChannel().socket().getLocalPort()+ ”] Closed. Packets sent ” + PACKETS_TO_SEND

+ ”. Connection: ” + (connectionsClosed + 1) + ”/”+ connectionsEstablished);

pc.close();connectionClosed();

} else {// Still more packets to send.sendPacket(pc);

}

A.2. CLIENT 81

}

public void socketException(PacketChannel pc, Exception ex) {logger.error(”[” + pc.toString() + ”] Error: ” + ex.getMessage());connectionClosed();

}

public void socketDisconnected(PacketChannel pc) {logger.info(”[” + pc.toString() + ”] Disconnected.”);connectionClosed();

}

/∗∗∗ The request was sent. Prepare to read the answer.∗/

public void packetSent(PacketChannel pc, ByteBuffer pckt) {logger.debug(”[” + pc.toString() + ”] Packet sent.”);

rigaStat = new Stats();String host = pc.getSocketChannel().socket().getInetAddress().getHostAddress

()+”:”+ String.valueOf(pc.getSocketChannel().socket().

getLocalPort());rigaStat.setClientHost(host);rigaStat.setMsgId(packetsSent);rigaStat.setInviati(pckt.capacity());rigaStat.setInizio(System.currentTimeMillis());

try {pc.resumeReading();

} catch (Exception e) {e.printStackTrace();logger.error(”Packet sent error: ”+e.getMessage());

}}

// //////////////////////////// Helper methods// ///////////////////////////∗∗∗ Called when a connection is closed. Checks if all connections have been∗ closed and if so exits the virtual machine.∗/

private void connectionClosed() {connectionsClosed++;if (connectionsClosed >= connectionsEstablished) {

st.requestClose();System.exit(1);

}}

/∗∗∗ Sends a newly generated packet using the given PacketChannel


∗∗ @param pc∗/

private void sendPacket(PacketChannel pc) {// System.out.println(”[” + pc.toString() + ”] Sending packet.”);ByteBuffer packet = generateNextPacket();packetsSent++;pc.sendPacket(packet);

}

/∗∗∗ Checks if all connections have been established. If so, starts them all∗ by sending an initial packet to all of them.∗/

private void checkAllConnected() {// Starts sending packets only after all connections are established.if ((connectionsEstablished + connectionsFailed) == connectionCount) {

for (int i = 0; i < establishedConnections.size(); i++) {PacketChannel pc = (PacketChannel) establishedConnections

.get(i);sendPacket(pc);

}establishedConnections.clear();

}}


URL log4jURL = ClassLoader.getSystemResource(”log4jclient.xml”);logger.debug(”Log4j config file URL: ”+log4jURL.toString());DOMConfigurator.configure(log4jURL);logger.info(”Log4j config loded: ”+log4jURL.toString());

InetSocketAddress remotePoint = new InetSocketAddress(args[0],Integer.parseInt(args[1]));

connectionCount = 1;if (args.length > 2) {

connectionCount = Integer.parseInt(args[2]);}st = new SelectorThread();

globalStart = System.currentTimeMillis();

for (int i = 0; i < connectionCount; i++) {new MultiplexingClient(remotePoint,i);// Must sleep for a while between opening connections in order// to give the remote host enough time to handle them. Otherwise,// the remote host backlog will get full and the connection// attemps will start to be refused.Thread.sleep(10);

}

Runtime.getRuntime().addShutdownHook(new PrintStats());

A.2. CLIENT 83

}

private static class PrintStats extends Thread {

public void run() {

logger.info(”Execution TIME: ”+String.valueOf(System.currentTimeMillis()−globalStart)+”ms”);

String locale = ”qualcosa”;try {

locale = InetAddress.getLocalHost().getHostName();try {

String filename = ”logs/”+locale + ” PPM ”+ Calendar.getInstance().get(Calendar.

HOUR_OF_DAY)+ ”−” + Calendar.getInstance().get(

Calendar.MINUTE)+ ”−” + Calendar.getInstance().get(

Calendar.SECOND)+ ” ” + currentThread().getId()+ ”.log”;

logger.info(”\n\nSalvate statistiche su file: ” +filename);

FileOutputStream stats = new FileOutputStream(filename,

true);PrintStream out = new PrintStream(stats);out.println(”Host\tMsgID\tByteSent\tByteReceived\

tStart\tEnd\tTime”);

for (Iterator<Stats> it = statistiche.iterator(); it.hasNext();) {

Stats riga = it.next();out.println(riga.getClientHost() + ”\t” +

riga.getMsgId()+ ”\t” + riga.getInviati() + ”

\t”+ riga.getRicevuti() + ”\t” +

riga.getInizio()+ ”\t” + riga.getFine() + ”\t”

+ riga.getTempo());}out.close();

} catch (IOException ex) {logger.error(”Errore salvataggio dati: ” + ex.getMessage

());}

} catch (UnknownHostException ex) {logger.error(”Impossibile ricavare IP locale: ”

+ ex.getMessage());}


}}

/∗∗∗ @author roberto∗/

private static class Stats {

/∗∗∗ @uml.property name=”inviati”∗/

private long inviati;/∗∗∗ @uml.property name=”ricevuti”∗/

private long ricevuti;/∗∗∗ @uml.property name=”inizio”∗/

private long inizio;/∗∗∗ @uml.property name=”fine”∗/

private long fine;/∗∗∗ @uml.property name=”msgId”∗/

private int msgId;/∗∗∗ @uml.property name=”clientHost”∗/

private String clientHost;

/∗∗∗ @return∗ @uml.property name=”inizio”∗/

public long getInizio() {return inizio;

}

/∗∗∗ @param inizio∗ @uml.property name=”inizio”∗/

public void setInizio(long inizio) {this.inizio = inizio;

}

/∗∗∗ @return∗ @uml.property name=”fine”∗/

A.2. CLIENT 85

public long getFine() {return fine;

}

/∗∗∗ @param fine∗ @uml.property name=”fine”∗/

public void setFine(long fine) {this.fine = fine;

}

public Stats() {}

/∗∗ public Stats(Thread thread, int msgId, long inviati, long ricevuti,∗ long tempo) { this.inviati = inviati; this.ricevuti = ricevuti;∗ this.tempo = tempo; this.msgId = msgId; this.threadId =∗ thread.getId(); }∗/

/∗∗∗ @return the inviati∗ @uml.property name=”inviati”∗/

public long getInviati() {return inviati;

}

/∗∗∗ @param inviati the inviati to set∗ @uml.property name=”inviati”∗/

public void setInviati(long inviati) {this.inviati = inviati;

}

/∗∗∗ @return the ricevuti∗ @uml.property name=”ricevuti”∗/

public long getRicevuti() {return ricevuti;

}

/∗∗∗ @param ricevuti the ricevuti to set∗ @uml.property name=”ricevuti”∗/

public void setRicevuti(long ricevuti) {this.ricevuti = ricevuti;

}


/∗∗∗ @return the tempo∗/

public long getTempo() {return this.fine−this.inizio;

}

/∗∗∗ @return the msgId∗ @uml.property name=”msgId”∗/

public int getMsgId() {return msgId;

}

/∗∗∗ @param msgId the msgId to set∗ @uml.property name=”msgId”∗/

public void setMsgId(int msgId) {this.msgId = msgId;

}

/∗∗∗ @return∗ @uml.property name=”clientHost”∗/

public String getClientHost() {return clientHost;

}

/∗∗∗ @param clientHost∗ @uml.property name=”clientHost”∗/

public void setClientHost(String clientHost) {this.clientHost = clientHost;

}

}

}

A.2.2 MultithreadClient.java

package it.studioaqua.unive.asdaa.client;

A.2. CLIENT 87

import handlers.SimpleProtocolDecoder;import io.ProtocolDecoder;

import java.io.BufferedReader;import java.io.FileOutputStream;import java.io.IOException;import java.io.InputStream;import java.io.InputStreamReader;import java.io.OutputStream;import java.io.PrintStream;import java.net.InetAddress;import java.net.Socket;import java.net.URL;import java.net.UnknownHostException;import java.nio.ByteBuffer;import java.util.ArrayList;import java.util.Calendar;import java.util.Iterator;import java.util.Random;


/∗∗∗ @author Studio AQuA∗/

public class MultithreadClient implements Runnable {

// Log4jstatic Logger logger = Logger.getLogger(MultithreadClient.class);

private String host;private int porta;private static final int minWriteSize = 50;private static final int maxWriteSize = 1200;private static int minPause = (int) (150);private static int maxPause = (int) (1500);private Random rand = new Random();private final static ArrayList<Stats> statistiche = new ArrayList<Stats>();private static boolean gira = true;// private final Object contatore=new Object();// private int cont=0;

private static long startExecution = 0;

/∗∗∗ Used to convert raw bytes into packets. (Strategy design pattern)∗ @uml.property name=”protocolDecoder”∗ @uml.associationEnd∗/

private ProtocolDecoder protocolDecoder;

public MultithreadClient(String host, int port, int numThreads) {


this.host = host;this.porta = port;

this.protocolDecoder = new SimpleProtocolDecoder();

for (int i = 0; i < numThreads; ++i) {new Thread(this).start();

}}

/∗∗∗ Main del programma, accetta come parametri l'host a cui connettersi e la∗ porta∗∗ @param args∗ Array di paramentri da riga di comando∗ @throws Exception∗ Diverse eccezioni non gestite (chiusura del server ad∗ esempio)∗/

public static void main(String args[]) {

URL log4jURL = ClassLoader.getSystemResource(”log4jclient.xml”);logger.debug(”Log4j config file URL: ” + log4jURL.toString());DOMConfigurator.configure(log4jURL);logger.info(”Log4j config loded: ” + log4jURL.toString());

logger.info(”Inizio esecuzione ” + System.currentTimeMillis());

String host = args[0];int port = Integer.parseInt(args[1]);int numThreads = Integer.parseInt(args[2]);

if (args.length > 3) {maxPause = minPause = Integer.parseInt(args[3]);

}if (args.length == 5) {

maxPause = Integer.parseInt(args[4]);}

startExecution = System.currentTimeMillis();

new MultithreadClient(host, port, numThreads);Runtime.getRuntime().addShutdownHook(new PrintStats());

System.out.println(”\n\nPremere q (e invio) per chiudere il programma \n\

n”);

InputStreamReader reader = new InputStreamReader(System.in);BufferedReader myInput = new BufferedReader(reader);String str = new String();

A.2. CLIENT 89

try {str = myInput.readLine();if (str.equals(”q”))

gira = false;} catch (IOException e) {

logger.error(”errore lettura carattere: ” + e.getMessage());}

}

/∗∗∗ Metodo Run del thread∗/

public void run() {

byte buffer[] = new byte[maxWriteSize];long inizio;long fine;

Socket s = null;InputStream in = null;OutputStream out = null;try {

s = new Socket(host, porta);

in = s.getInputStream();out = s.getOutputStream();

} catch (UnknownHostException e) {logger.error(”Socket creation: unknown host ” + host + ” − ”

+ e.getMessage());e.printStackTrace();

} catch (IOException e) {logger.error(”Socket creation: IO exception − ” + e.getMessage());e.printStackTrace();

}

/∗∗ synchronized (contatore){ cont++; System.out.println(”Porta locale: ”∗ + s.getLocalPort()+” n:”+cont); }∗/

logger.debug(”Inizio a inviare dati...” + s.getLocalPort());int messageId = 1;while (gira) {

Stats rigaStat = new Stats();rigaStat.setThreadId(Thread.currentThread().getId());rigaStat.setMsgId(messageId);

buffer = this.generateNextPacket();int numToWrite = buffer.length;logger.debug(”Byte to write: ” + numToWrite);

rigaStat.setInviati(numToWrite);


try {out.write(buffer, 0, numToWrite);

} catch (IOException e) {logger.error(”Write on socket: ” + e.getMessage());e.printStackTrace();

}inizio = System.currentTimeMillis();int sofar = 0;

logger.debug(”byte scritti: ” + numToWrite);try {

while (sofar < numToWrite) {if (in.available() > 0) {

sofar += in.read(buffer, sofar, numToWrite −sofar);

}}

ByteBuffer packet = protocolDecoder.decode(ByteBuffer.wrap(buffer));

// A packet may or may not have been fully assembled,depending

// on the data available in the bufferif (packet != null) {

// A packet was reassembled.logger.debug(”Packet received.”);

}

} catch (IOException e) {logger.error(”Read on socket: ” + e.getMessage());e.printStackTrace();

}

fine = System.currentTimeMillis();

rigaStat.setRicevuti(sofar);rigaStat.setTempo(fine − inizio);

logger.debug(Thread.currentThread() + ” scritto ” + numToWrite

+ ” ricevuto ” + sofar + ” in ” + (fine − inizio) + ”ms”);

synchronized (statistiche) {statistiche.add(rigaStat);

}

int pause = minPause

+ (int) (rand.nextDouble() ∗ (maxPause − minPause));try {

Thread.sleep(pause);} catch (InterruptedException ex) {

logger.error(”” + ex.getMessage());synchronized (statistiche) {

A.2. CLIENT 91

statistiche.add(rigaStat);}

}

messageId++;}

try {out.close();in.close();s.close();

} catch (IOException e) {logger.error(”Closing socket: ” + e.getMessage());e.printStackTrace();

}

}

/∗∗∗ Creates a new packet with a size chosen randomly between MIN SIZE and∗ MAX SIZE.∗/

private byte[] generateNextPacket() {// Generate a random size betweenint size = minWriteSize + rand.nextInt(maxWriteSize − minWriteSize);ByteBuffer buffer = ByteBuffer.allocate(size);buffer.put(SimpleProtocolDecoder.STX);for (int i = 0; i < size − 2; i++) {

buffer.put((byte) 'a');}buffer.put(SimpleProtocolDecoder.ETX);buffer.limit(buffer.position());buffer.flip();

// Retrieve bytes between the position and limit// (see Putting Bytes into a ByteBuffer)byte[] bytes = new byte[buffer.remaining()];buffer.get(bytes, 0, bytes.length);

return bytes;}

private static class PrintStats extends Thread {

public void run() {

logger.info(”Execution time: ”+String.valueOf(System.currentTimeMillis()−MultithreadClient.startExecution));

String locale = ”qualcosa”;try {

locale = InetAddress.getLocalHost().getHostName();try {


String filename = ”logs/” + locale + ” PPM ”+ Calendar.getInstance().get(Calendar.

HOUR_OF_DAY)+ ”−” + Calendar.getInstance().get(

Calendar.MINUTE)+ ”−” + Calendar.getInstance().get(

Calendar.SECOND)+ ”.log”;

logger.info(”Salvate statistiche su file: ” + filename);FileOutputStream stats = new FileOutputStream(

filename,true);

PrintStream out = new PrintStream(stats);out.println(”Thread\tMsgID\tByteSent\tByteReceived\

tTime”);for (Iterator<Stats> it = statistiche.iterator(); it

.hasNext();) {Stats riga = it.next();

out.println(riga.getThreadId() + ”\t” + riga.getMsgId()

+ ”\t” + riga.getInviati() + ”\t”

+ riga.getRicevuti() + ”\t” +riga.getTempo());

}out.close();

} catch (IOException ex) {logger.error(”Errore salvataggio dati: ” + ex.getMessage

());}

} catch (UnknownHostException ex) {logger.error(”Impossibile ricavare IP locale: ”

+ ex.getMessage());}

}}

/∗∗∗ @author roberto∗/

private static class Stats {

/∗∗∗ @uml.property name=”inviati”∗/

private long inviati;/∗∗∗ @uml.property name=”ricevuti”∗/

private long ricevuti;/∗∗

A.2. CLIENT 93

∗ @uml.property name=”tempo”∗/

private long tempo;/∗∗∗ @uml.property name=”threadId”∗/

private long threadId;/∗∗∗ @uml.property name=”msgId”∗/

private int msgId;

public Stats() {}

/∗∗∗ @return the inviati∗ @uml.property name=”inviati”∗/

public long getInviati() {return inviati;

}

/∗∗∗ @param inviati the inviati to set∗ @uml.property name=”inviati”∗/

public void setInviati(long inviati) {this.inviati = inviati;

}

/∗∗∗ @return the ricevuti∗ @uml.property name=”ricevuti”∗/

public long getRicevuti() {return ricevuti;

}

/∗∗∗ @param ricevuti the ricevuti to set∗ @uml.property name=”ricevuti”∗/

public void setRicevuti(long ricevuti) {this.ricevuti = ricevuti;

}

/∗∗∗ @return the tempo∗ @uml.property name=”tempo”∗/

public long getTempo() {return tempo;


}

/∗∗∗ @param tempo the tempo to set∗ @uml.property name=”tempo”∗/

public void setTempo(long tempo) {this.tempo = tempo;

}

/∗∗∗ @return the threadId∗ @uml.property name=”threadId”∗/

public long getThreadId() {return threadId;

}

/∗∗∗ @param threadId the threadId to set∗ @uml.property name=”threadId”∗/

public void setThreadId(long threadId) {this.threadId = threadId;

}

/∗∗∗ @return the msgId∗ @uml.property name=”msgId”∗/

public int getMsgId() {return msgId;

}

/∗∗∗ @param msgId the msgId to set∗ @uml.property name=”msgId”∗/

public void setMsgId(int msgId) {this.msgId = msgId;

}

}}

Appendice B

Glossario

client In informatica, con client (in italiano detto anche cliente) si indicauna componente che accede ai servizi o alle risorse di un’altra com-ponente, detta server. In questo contesto si puo quindi parlare diclient riferendosi all’hardware o al software. Un computer collegatoad un server tramite una rete informatica (locale o geografica) edal quale richiede uno o piu servizi, utilizzando uno o piu protocollidi rete e un esempio di client hardware. Un programma di postaelettronica e un esempio di client software.

server In informatica il termine server (in inglese letteralmente servi-tore) indica genericamente un componente informatico che fornisceun qualunque tipo di servizi ad altre componenti (tipicamente chia-mate client, cioe cliente) attraverso una rete di computer. Si notiche il termine server, cosı come pure il termine client, possono essereriferiti sia alla componente hardware che alla componente softwareche forniscono le funzionalita o servizi di cui sopra. A seconda delcontesto, il termine server puo indicare quindi:

• un computer utilizzato per fornire servizi ad altri computer, aprescindere dalle sue caratteristiche hardware;

• un computer specifico appartenente alla fascia di mercato de-dicata all’uso come server, caratterizzato da alta affidabilita emaggiori prestazioni;

95

96 APPENDICE B. GLOSSARIO

• un processo (ovvero un programma software in esecuzione) chefornisca servizi ad altri processi (es. Server Web).

framework una struttura di supporto su cui un software puo essereorganizzato e progettato.

database

affidabilita Definendo come qualita di un assieme o di un sistema co-munque complesso la sua rispondenza ai criteri di specifica di fun-zionamento, si definisce affidabilita la capacita di rispettare le spe-cifiche di funzionamento nel tempo.

efficienza L’efficienza e la capacita di azione o di produzione con ilminimo di scarto, di spesa, di risorse e di tempo impiegati.

pay-as-you-go Paghi solamente per le risorse che stai utilizzando in unpreciso momento.

data center

repository ambiente di un sistema informativo, in cui vengono gestiti imetadati, attraverso tabelle relazionali; l’insieme di tabelle, regolee motori di calcolo tramite cui si gestiscono i metadati prende ilnome di metabase.

data store e un archivio di dati di un insieme di oggetti integrati. Que-sti oggetti sono modellati utilizzando le classi definite in schemi didatabase. Comprende non solo i repository di dati come i database,e un concetto piu generale che include i file anche piatti che possonomemorizzare i dati.

latenza In informatica si definisce latenza di un modulo di elaborazione,il tempo impiegato per la completa elaborazione di una operazioneesterna.

transazione In informatica, una transazione e una sequenza di opera-zioni, che puo concludersi con un successo o un insuccesso; in caso

97

di successo, il risultato delle operazioni deve essere permanente,mentre in caso di insuccesso si deve tornare allo stato precedenteall’inizio della transazione.

hardware indica la parte fisica di un computer, ovvero tutte quelle partielettroniche, meccaniche, magnetiche, ottiche che ne consentono ilfunzionamento.

cloud computing si indica un insieme di tecnologie che permettono, ti-picamente sotto forma di un servizio offerto da un provider al clien-te, di memorizzare/archiviare e/o elaborare dati (tramite CPU osoftware) grazie all’utilizzo di risorse hardware/software distribuitee virtualizzate in Rete.

infrastructure as a service (IaaS) e una piattaforma attraverso la qua-le le imprese possono avvalersi, sotto la forma di hardware, server,spazio disco utilizzando la forma di pagamento a pay-as-you-go.

platform as a service (PaaS) servizio che offre un’architettura hard-ware ed un framework software con un set di funzionalita che per-mettono di creare un prodotto o servizio completo.

software as a service (SaaS) e un modello di distribuzione del soft-ware applicativo dove un produttore di software sviluppa, opera(direttamente o tramite terze parti) e gestisce un’applicazione webche mette a disposizione dei propri clienti via internet.

noSQL

LAN una Local Area Network (LAN) (in italiano rete in area locale, orete locale) e una tipologia di rete informatica contraddistinta daun’estensione territoriale non superiore a qualche chilometro.

WAN Wide Area Network, rete geografica. Rete di computer che ricopreun’area geografica molto estesa, e che interconnette sia singoli hostche LAN, dislocati anche in continenti diversi.

98 APPENDICE B. GLOSSARIO

thread pool e un modello nell’implementazione software che prevede lacreazione di un certo numero di thread per eseguire un insieme dioperazioni che solitamente sono organizzate in code.

thread safety e un concetto utilizzato nell’implementazione softwareapplicabile nel contesto di applicazioni multi-thread. Un frammen-to di codice si dice essere thread-safe solamente se manipola strut-ture dati condivise in modo da garantire un’esecuzione sicura (senzaperdere dati) da piu thread contemporaneamente.

Bibliografia

[1] Reliable server pooling workgroup.

[2] Amazon, Amazon dynamodb. http://aws.amazon.com/dynamodb/.

[3] , Amazon simple db. http://aws.amazon.com/simpledb/.

[4] , Amazon web services. http://aws.amazon.com.

[5] , Aws free usage tier.

[6] , Aws sdk for java.

[7] , Aws toolkit for eclipse.

[8] , Release: Aws sdk for java 1.3.0.http://aws.amazon.com/releasenotes/5581335213116164.

[9] Google, App engine datastore. http://developers.google.com/appengine/docs/python/datastore/.

[10] , Google app engine.

[11] P. Lawrey, Java: What is the limit to the number of threads youcan create?

[12] O’Reilly, Java nio.

[13] C. Pavan, Server ad Alte Prestazioni e ad Alta Affidabilita, PhDthesis, Universita Ca’Foscari, 2011.

[14] N. Santos, Building highly scalable servers with java nio.

99

100 BIBLIOGRAFIA

[15] M. T. Thomas Dreibholz, High availability using reliable serverpooling, 2002.

Studio di una Architettura per un Sistema Distributivo ad Alta Affidabilità

Software

Transcript of Studio di una Architettura per un Sistema Distributivo ad Alta Affidabilità