|
Size: 19190
Comment:
|
← Revision 29 as of 2009-05-04 15:51:42 ⇥
Size: 6181
Comment:
|
| Deletions are marked like this. | Additions are marked like this. |
| Line 3: | Line 3: |
| Quando viene creato un socket usando il wrapper {{{ntk.wrap.sock.Sock}}}, viene registrato nella mappa di {{{asyncore}}} (un modulo standard di python stackless) e viene avviato un microthread nel '''modulo microsock'''. | Lo scopo del '''modulo microsock''' è quello di permettere la gestione di operazioni di I/O sui socket in un ambiente multithread cooperativo. <<BR>> In un tale ambiente è necessario evitare l'uso di chiamate di sistema bloccanti per l'intero processo, come sono le chiamate di I/O sui socket. Quando viene creato un socket usando il wrapper {{{ntk.wrap.sock.Sock}}}, viene in effetti creato anche un oggetto {{{asyncore.dispatcher}}} che lo incapsula e questo viene registrato nella mappa di {{{asyncore}}} (un modulo standard di python stackless); inoltre viene avviato un microthread nel modulo microsock. |
| Line 19: | Line 23: |
| Vediamo quali passi segue la creazione e connessione di un socket, prima di analizzare quello che avviene quando si trasmette o riceve qualcosa. == Creazione di un socket == Quando si crea un socket all'interno del programma (con {{{socket.socket(...)}}}) avvengono queste operazioni: 1. viene effettivamente creato un vero socket (con {{{stdsocket.socket(...)}}}). 1. viene creato anche un {{{microsock.stacklesssocket}}}. Questo nel suo costruttore fa queste operazioni: a. memorizza il vero socket. a. si crea e memorizza una istanza dell'oggetto {{{microsock.dispatcher}}}, che è una derivata di {{{asyncore.dispatcher}}}. a. {{{microsock.dispatcher}}} nel suo costruttore fa queste operazioni: I. memorizza il vero socket. I. la classe ha i membri {{{connectChannel}}}, {{{acceptChannel}}} e {{{recvChannel}}}. I. istanzia per {{{recvChannel}}} un Channel con {{{micro_send = True}}}. I. prepara una stringa buffer per i send e una lista di buffers per i sendto ({{{sendBuffer}}} e {{{sendToBuffers}}}) I. prepara una stringa buffer per i recv ({{{readBufferString}}}) I. '''''TODO''': non ci fa nulla con il membro {{{readBufferList}}}, va rimosso?'' I. prepara una costante {{{maxreceivebuf}}} = 65536 I. memorizza il vero socket nella mappa di asyncore (questo lo fa {{{asyncore.dispatcher.__init__}}} => {{{asyncore.dispatcher.set_socket}}} => {{{asyncore.dispatcher.add_channel}}}) I. la sua rimozione dalla mappa avverrà con {{{microsock.stacklesssocket.__del__}}} => {{{microsock.dispatcher.close}}} => {{{asyncore.dispatcher.close}}} => {{{asyncore.dispatcher.del_channel}}} 1. viene restituito l'oggetto {{{microsock.stacklesssocket}}}. In seguito tutte le chiamate {{{__getattr__}}} al finto socket (cioè all'oggetto stacklesssocket ritornato) vengono inoltrate al dispatcher (esempi di queste chiamate sono {{{s.listen}}}, {{{s.bind}}}, ...) mentre le chiamate {{{__setattr__}}} vengono memorizzate localmente. ??? == Connessione di una coppia di socket == === con protocollo TCP === Se vogliamo implementare un server con il protocollo TCP, dobbiamo creare un socket di tipo {{{SOCK_STREAM}}}, associargli indirizzo e porta, richiamare il metodo listen e poi accept. Quando si richiama il metodo '''{{{listen}}}''' di un {{{microsock.dispatcher}}} (ereditato dal {{{asyncore.dispatcher}}}) il dispatcher si memorizza lo stato {{{accepting = True}}} e poi esegue il vero metodo listen sul socket. Il metodo listen non è bloccante. Quando si richiama il metodo '''{{{accept}}}''' di un {{{microsock.dispatcher}}} (overridato rispetto al {{{asyncore.dispatcher}}}) il dispatcher si istanzia un oggetto [[../ClasseChannel|Channel]] con {{{micro_send = True}}} e lo memorizza sul membro {{{acceptChannel}}}. Su questo oggetto richiama il metodo {{{recv}}}. <<BR>> Questo metodo a sua volta richiama il metodo {{{receive}}} di uno {{{stackless.channel}}}. Questo metodo blocca la tasklet, cioè il microthread attivo, e schedula gli altri microthread. Quello bloccato non sarà più ri-schedulato fino a che non ci saranno messaggi su quel canale. <<BR>> Quando il socket reale riceve una richiesta di connessione, anche se non è stato chiamato in realtà il suo metodo {{{accept}}}, la funzione {{{asyncore_poll}}} (periodicamente richiamata in un microthread apposito, come detto sopra) se ne accorge come se il socket fosse in attesa di lettura. Viene quindi richiamato il metodo {{{handle_accept}}}, da overridare. <<BR>> Il metodo {{{handle_accept}}} (overridato rispetto al {{{asyncore.dispatcher}}}) verifica che sul {{{channel}}} contenuto sul membro {{{acceptChannel}}} ci sia qualcuno effettivamente in ascolto, guardando l'attributo {{{balance}}}. Poi richiama il metodo {{{asyncore.dispatcher.accept}}} della classe base (non più quello overridato detto prima); questo richiama {{{accept}}} del socket, ma stavolta siamo sicuri che non si bloccherà. Restituirà un socket connesso e l'indirizzo del socket remoto. <<BR>> Il metodo {{{handle_accept}}} poi incapsula il socket ricevuto in un oggetto {{{stacklesssocket}}} e lo invia al {{{channel}}} contenuto sul membro {{{acceptChannel}}}. Siccome il Channel è con {{{micro_send = True}}}, questo invio viene effettuato in un nuovo microthread. '''''TODO''': Questo è davvero necessario? Ci siamo prima accertati che il balance fosse negativo...!'' <<BR>> Il microthread prima bloccato viene ora ri-schedulato e il metodo {{{microsock.dispatcher.accept}}} restituisce un {{{stacklesssocket}}} connesso. Se vogliamo implementare un client con il protocollo TCP, dobbiamo creare un socket di tipo {{{SOCK_STREAM}}} e poi richiamare il metodo connect. Quando si richiama il metodo '''{{{connect}}}''' di un {{{microsock.dispatcher}}} questo per prima cosa richiama il metodo base, cioè il metodo {{{asyncore.dispatcher.connect}}}. <<BR>> Il metodo {{{asyncore.dispatcher.connect}}} utilizza il metodo {{{socket.connect_ex}}} fornito dalla libreria di Python. Questo avvia un tentativo di connessione ma la chiamata non si blocca. Invece restituisce un numero (errno) con il quale indica se il tentativo è andato a buon fine o se è in corso o se c'è stato un altro errore. <<BR>> Se la connessione si completa, il metodo {{{asyncore.dispatcher.connect}}} richiama {{{handle_connect_event}}}, questo valorizza a True il suo membro {{{connected}}} e poi richiama il metodo {{{handle_connect}}}, da overridare. Il metodo {{{microsock.dispatcher.connect}}} in questo caso non ha bisogno di fare altro. <<BR>> Se, invece, la connessione non si completa, il metodo {{{asyncore.dispatcher.connect}}} ritorna. Il metodo {{{microsock.dispatcher.connect}}} se ne accorge dal fatto che il membro {{{connected}}} è ancora a False. Quindi si prepara nel suo membro {{{connectChannel}}} un Channel con {{{prefer_sender = True}}} (vedi la [[../ClasseChannel|classe Channel]]) e si mette in ascolto su quel canale. In questo modo il microthread si blocca in attesa della connessione, senza bloccare gli altri microthread. <<BR>> Quando il sistema completa la connessione sul socket reale, la funzione {{{asyncore_poll}}} (periodicamente richiamata in un microthread apposito, come detto sopra) se ne accorge come se il socket fosse in attesa di lettura. Viene quindi richiamato il metodo {{{handle_connect}}}. <<BR>> Il metodo {{{handle_connect}}} (overridato rispetto al {{{asyncore.dispatcher}}}) invia un messaggio al Channel memorizzato nel membro {{{connectChannel}}} così da ri-schedulare il microthread prima bloccato. === con protocollo UDP === Se vogliamo implementare un server con il protocollo UDP, dobbiamo creare un socket di tipo {{{SOCK_DGRAM}}}, associargli indirizzo e porta. <<BR>> '''''TODO''': Perché avviamo un UDPServer per ogni scheda se poi l'argomento {{{dev}}} non viene mai usato nella fase di attesa messaggi? Cioè, non viene fatta l'associazione del socket alla scheda con {{{sk_bindtodevice}}}.'' <<BR>> Dopo di questo non ci sono altre operazioni da fare, quindi nessuna bloccante, prima di passare alla trasmissione/ricezione dati. Se vogliamo implementare un client con il protocollo UDP, dobbiamo creare un socket di tipo {{{SOCK_DGRAM}}}, associare il socket ad una interfaccia di rete con {{{sk_bindtodevice}}} (supportato solo su linux) e poi richiamare il metodo {{{connect}}} dove al posto dell'indirizzo dell'host specifichiamo la stringa parola chiave {{{<broadcast>}}}. <<BR>> In questo modo otteniamo un socket pronto a comunicare in broadcast su tutta la LAN acceduta tramite quella interfaccia di rete. Per questo il programma Netsukuku si crea un socket UDP per ogni scheda di rete da gestire e li connette ed usa tutti per inviare messaggi ai suoi vicini (vedi la classe {{{BcastClient}}} nel [[../ModuloRPC|modulo RPC]]). <<BR>> Abbiamo già visto come il metodo {{{connect}}} di {{{microsock.dispatcher}}} finisce con il richiamare per prima cosa il metodo {{{socket.connect_ex}}} fornito dalla libreria di Python. Questo, quando si usa come indirizzo la parola chiave {{{<broadcast>}}} connette il socket e fa subito ritorno. <<BR>> '''''TODO''': Questa è una mia deduzione. La documentazione di Python non è così esplicita. E' corretta?'' == Trasmissione dati tra due socket == La trasmissione di dati tra due socket connessi con il protocollo TCP avviene in Netsukuku con l'uso dei metodi send, sendall e recv. Quando si richiama il metodo {{{send}}} di un {{{microsock.dispatcher}}} questo memorizza il messaggio da inviare in coda al buffer nel suo membro {{{sendBuffer}}} e fa subito ritorno al chiamante, segnalando di aver trasmesso tutti i bytes richiesti. <<BR>> La presenza di dati in questo buffer fa sì che il metodo {{{writable}}} restituisca {{{True}}}. Questo a sua volta, come detto sopra, fa sì che {{{asyncore.poll}}}, periodicamente e solo quando la trasmissione non risulterebbe bloccante, richiami il metodo {{{handle_write_event}}} che a sua volta richiama {{{handle_write}}} overridato nella classe {{{microsock.dispatcher}}}. <<BR>> Il metodo {{{microsock.dispatcher.handle_write}}} richiama il metodo {{{asyncore.dispatcher.send}}} sulla classe base, non più quello overridato. Questo effetua l'invio di un certo numero di bytes senza bloccare il processo e restituisce il numero di bytes trasmessi. Il metodo {{{microsock.dispatcher.handle_write}}} di seguito toglie dal buffer i bytes già inviati e fa ritorno. Abbiamo detto che il metodo {{{microsock.dispatcher.send}}} fa subito ritorno segnalando la trasmissione di tutti i bytes richiesti. In realtà la trasmissione avviene in seguito. <<BR>> Se si vuole essere sicuri che la trasmissione sia già avvenuta si chiama il metodo {{{microsock.dispatcher.sendall}}}. Questo metodo fa ritorno solo quando il buffer nel suo membro {{{sendBuffer}}} è completamente svuotato. <<BR>> Per attendere questo evento effettua un loop in cui intervalla il test alla schedulazione degli altri microthread con {{{micro_block}}}. '''''TODO''': Il loop in questione è in effetti un busy-wait. C'è un qualche motivo per cui qui non aggiungiamo un time.sleep(0.001) ?'' Per quanto riguarda la ricezione di dati, il metodo {{{readable}}} di un {{{microsock.dispatcher}}}, ereditato tal quale dalla {{{asyncore.dispatcher}}}, restituisce sempre {{{True}}}. Questo a sua volta, come detto sopra, fa sì che {{{asyncore.poll}}}, periodicamente e solo quando ci sono dei dati disponibili per la lettura, richiami il metodo {{{handle_read_event}}} che a sua volta richiama {{{handle_read}}} overridato nella classe {{{microsock.dispatcher}}}. <<BR>> Il metodo {{{microsock.dispatcher.handle_read}}} richiama il metodo {{{recv}}} di {{{asyncore.dispatcher}}}, non quello overridato da {{{microsock.dispatcher}}}. Questo richiama il metodo {{{recv}}} sul socket reale che leggerà un massimo di {{{maxreceivebuf}}} bytes senza bloccarsi. Questi dati vengono inviati come messaggio sul Channel memorizzato sul membro {{{recvChannel}}}. <<BR>> Siccome il Channel è stato istanziato con {{{micro_send = True}}} questa trasmissione sullo {{{stackless.channel}}} avviene in realtà su un altro microthread, lasciando così immediatamente libero di continuare il microthread che richiama periodicamente la {{{asyncore.poll}}}. <<BR>> In questo modo si ottiene che i dati sono in attesa, senza bloccare gli altri microthread, di essere letti dallo pseudo socket. <<BR>> '''Da notare''' che finora non abbiamo parlato del metodo {{{recv}}} overridato nella classe {{{microsock.dispatcher}}}. Infatti il meccanismo sopra descritto consente la lettura di un certo numero di bytes (quanto il valore corrente di maxreceivebuf) dal socket ogni qualvolta che siano presenti, anche se il microthread che sta "gestendo" l'oggetto {{{stacklesssocket}}} non sta al momento eseguendo una chiamata a {{{recv}}}. Da notare '''anche''' che il valore di maxreceivebuf cambia in base al valore della precedente chiamata al metodo {{{recv}}} della classe {{{microsock.dispatcher}}}, che non è detto che corrisponda alla attuale lettura -- apparentemente non era questo l'intento dello sviluppatore. <<BR>> '''''TODO''': che senso ha un tale comportamento? Mi pare possa essere fonte di diverse problematiche. Perché non leggiamo dal socket solo dopo che è stato chiamato recv e al massimo un numero di bytes tali da soddisfare la chiamata?'' Il metodo {{{recv}}} overridato nella classe {{{microsock.dispatcher}}} non fa altro che ricevere i dati dal Channel memorizzato sul membro {{{recvChannel}}}. Tutti i bytes che vi trova li mette nel buffer {{{readBufferString}}}. Poi ne restituisce al massimo quanti richiesti dal chiamante. <<BR>> Per la precisione, prima di tutto il metodo guarda se nel buffer {{{readBufferString}}} ci sono già un numero di bytes da soddisfare del tutto il chiamante. Altrimenti si possono verificare 2 casi: * Ci sono nel Channel alcuni messaggi in attesa di essere letti. In questo caso il metodo legge tutto quello che può dal Channel e accumula nel buffer. '''''TODO''': In realtà questo è il comportamento che si intenderebbe tenere, mi pare di capire dai commenti sul codice; ma non è implementato correttamente: ci vorrebbe un while invece di un if. Verificare e correggere.'' * Ci sono già alcuni bytes nel buffer (anche se non tanti quanti il massimo richiesto) e non ci sono altri messaggi da essere letti nel Channel. In questo caso il metodo si prepara a restituire immediatamente quanto finora disponibile. * Ci sono 0 bytes nel buffer e non ci sono altri messaggi da essere letti nel Channel. In questo caso il metodo deve bloccarsi in attesa di qualche byte. Lo fa mettendosi in ascolto sul Channel, quindi blocca il microthread ma non il resto dell'applicazione. Alla fine, da quello che ha a disposizione sul buffer, rimuove e restituisce al chiamante al massimo ''n'' bytes, quanti richiesti. ==== Appunti ==== TCP il client invia con sendall, il server legge con varie recv, il server risponde con send, il client legge con varie recv. <<BR>> UDP il client invia con sendto, il server legge con recvfrom, non ci sono risposte. |
Nel modulo microsock viene definita la classe {{{microsock.dispatcher}}} che è una classe derivata di {{{asyncore.dispatcher}}}. Grazie a questa classe possiamo definire come deve agire il programma quando un microthread richiama i classici metodi di I/O su un presunto socket. Per una analisi di dettaglio rimandiamo alle pagine sulla [[../ModuloMicrosockCreazione|creazione di un socket]], [[../ModuloMicrosockConnessione|connessione di due socket]] e [[../ModuloMicrosockTrasmissione|trasmissione dati tra due socket]]. |
| Line 139: | Line 37: |
| <<BR>> La funzione {{{select.select}}} è chiamata nella funzione {{{asyncore.poll}}}; noi ce ne avvaliamo facendo l'override nella classe {{{microsock.dispatcher}}} dei metodi di {{{asyncore.dispatcher}}} che vengono chiamati dalla funzione {{{asyncore.poll}}} stessa. Vedi le pagine citate sopra. |
Il modulo microsock
Lo scopo del modulo microsock è quello di permettere la gestione di operazioni di I/O sui socket in un ambiente multithread cooperativo.
In un tale ambiente è necessario evitare l'uso di chiamate di sistema bloccanti per l'intero processo, come sono le chiamate di I/O sui socket.
Quando viene creato un socket usando il wrapper ntk.wrap.sock.Sock, viene in effetti creato anche un oggetto asyncore.dispatcher che lo incapsula e questo viene registrato nella mappa di asyncore (un modulo standard di python stackless); inoltre viene avviato un microthread nel modulo microsock.
Il microthread avviato esegue la funzione ManageSockets. Questa richiama asyncore.poll(0.05) ciclicamente (intervallando con micro_block per passare lo scheduler agli altri) fin quando esiste un socket (nel caso di Netsukuku per sempre) nella mappa di asyncore.
La mappa di asyncore associa il fd (file descriptor, un numero intero) ad una istanza di asyncore.dispatcher che incapsula il vero oggetto sul quale vanno fatte le operazioni di I/O (che può essere un file, o un socket, ...).
Nota: La funzione asyncore.poll è bloccante. Il parametro passato è il numero di secondi dopo i quali cui si vuole cha la chiamata faccia ritorno.
Al suo interno la funzione richiama time.sleep oppure select.select, a seconda se ci sono o meno socket da controllare. In entrambi i casi queste funzioni usano chiamate di sistema che non impegnano la CPU, ma nemmeno permettono agli altri microthread di venire schedulati.
Per questo dobbiamo richiamarla con piccoli timeout intervallando con chiamate a micro_block. Vedi anche la trattazione dell'argomento nella pagina del modulo xtime.
Il metodo asyncore.poll, per prima cosa fa uso dei due metodi readable e writable di asyncore.dispatcher su tutti i dispatcher registrati nella sua mappa. Questi metodi possono essere overridati; nella implementazione base restituiscono True. Il loro obiettivo è quello di identificare i socket che hanno intenzione di inviare (writable) o di leggere (readable) qualcosa.
Di seguito, per i socket che sono readable o writable secondo i significati sopra esposti, tramite il metodo select.select scopre quali hanno dei dati da ricevere oppure sono subito in grado di inviare dati. Cioè per quali socket le chiamate recv e send non risulterebbero bloccanti.
Nel modulo microsock viene definita la classe microsock.dispatcher che è una classe derivata di asyncore.dispatcher. Grazie a questa classe possiamo definire come deve agire il programma quando un microthread richiama i classici metodi di I/O su un presunto socket. Per una analisi di dettaglio rimandiamo alle pagine sulla creazione di un socket, connessione di due socket e trasmissione dati tra due socket.
Il microthread che esegue ManageSockets
Abbiamo detto che fin dal primo momento in cui esiste un socket viene avviato un microthread che esegue la funzione ManageSockets presente in questo modulo.
La vita di questo microthread è un aspetto critico di tutto il programma Netsukuku. Se per un qualsiasi motivo questo microthread dovesse bloccarsi in attesa di un evento che tarda a verificarsi (o addirittura non si verificherà mai) l'intera struttura di comunicazione di Netsukuku si blocca.
Per questo motivo va usata estrema cautela in tutti i punti del codice (in particolare in questo modulo) in cui questo microthread passa e che fanno uso di chiamate bloccanti.
Le chiamate bloccanti usate in questo microthread sono di due tipi.
Chiamate che bloccano l'intero programma. Sono tutte quelle che riguardano l'I/O sul socket reale. Ci sono le chiamate accept per i server, le chiamate connect per i client, le chiamate send, sendto e le chiamate recv, recvfrom.
Chiamate che bloccano questo microthread ma lasciano spazio agli altri microthread. Sono le chiamate send e receive dei stackless.channel o della nostra classe Channel che li incapsula.
Per quanto riguarda le chiamate di I/O sul socket, ci si affida alla funzione select.select la quale ci dice con certezza (e senza bloccarsi indefinitamente) se una certa chiamata risulterebbe bloccante. Quindi se usata correttamente ci può guidare ad un uso non bloccante delle altre chiamate.
La funzione select.select è chiamata nella funzione asyncore.poll; noi ce ne avvaliamo facendo l'override nella classe microsock.dispatcher dei metodi di asyncore.dispatcher che vengono chiamati dalla funzione asyncore.poll stessa. Vedi le pagine citate sopra.
Per quanto riguarda la trasmissione di messaggi sui channel, bisogna:
Per le chiamate receive accertarsi che qualcuno stia inviando qualcosa; usare quindi l'attributo balance. In realtà al momento non ne vengono fatte di chiamate receive all'interno del microthread in questione.
Per le chiamate send si può agire in due modi:
Di norma non è essenziale sapere il momento esatto in cui sono state portate a compimento; si possono eseguire quindi in un separato microthread. Per questo motivo la classe Channel può venire istruita (usando il costruttore con micro_send=True) di eseguire tutte le chiamate send sul channel incapsulato in un nuovo microthread. TODO: Domanda: molteplici chiamate a send in microthread distinti verranno poi rilette sicuramente nell'ordine in cui sono state inviate?
Oppure vanno fatte solo se si è sicuri che una altra tasklet sia in attesa. In questo caso va preferito il costruttore della classe Channel con prefer_sender=True, perché così la chiamata send trasmette i dati, sblocca la tasklet ricevente, ma mantiene la schedulazione al microthread corrente.