Invia dati a più socket utilizzando pipe, tee() e splice()

Question

Sto duplicando una pipe "master" con tee() per scrivere su più socket utilizzando splice(). Naturalmente questi tubi si svuotano a velocità diverse a seconda di quanto posso giuntare() alle prese di destinazione. Quindi quando passo ad aggiungere dati alla pipe "master" e poi tee() di nuovo, potrei avere una situazione in cui posso scrivere 64 KB nella pipe ma solo tee 4KB in uno dei pipe "slave". Suppongo quindi che se spago() tutta la pipe "master" alla presa, non sarò mai in grado di tee() i restanti 60KB a quella pipe slave. È vero? Immagino di poter tenere traccia di un tee_offset (a partire da 0) che ho impostato all'inizio dei dati "non sottoposti a" e quindi non lo ho corretto. Quindi in questo caso imposterei tee_offset su 4096 e non unire più di quello finché non sarò in grado di collegarlo a tutti gli altri pipe. Sono sulla strada giusta qui? Qualche suggerimento/avvertimento per me?

Answer 1

Se ho capito bene, hai una fonte di dati in tempo reale che vuoi collegare a più socket. Hai una sola pipe "sorgente" collegata a qualsiasi cosa stia producendo i tuoi dati, e hai una "destinazione" pipe per ogni socket su cui desideri inviare i dati. Quello che stai facendo è utilizzare tee() per copiare i dati dalla pipe di origine a ciascuna delle pipe di destinazione e splice() per copiarlo dalle pipe di destinazione ai socket stessi.

Il problema fondamentale che stai per colpire qui è se uno dei socket semplicemente non può tenere il passo - se stai producendo dati più velocemente di quanto tu possa inviarlo, allora avrai un problema. Questo non è legato al tuo uso delle pipe, è solo una questione fondamentale. Quindi, ti consigliamo di scegliere una strategia per far fronte in questo caso: ti suggerisco di gestirlo anche se non ti aspetti che sia comune, perché spesso queste cose ti vengono in mente. Le tue scelte di base consistono nel chiudere il socket incriminato o saltare i dati fino a quando non viene cancellato il suo buffer di output: quest'ultima opzione potrebbe essere più adatta per lo streaming audio/video, ad esempio.

Il problema che è correlato all'uso di pipe, tuttavia, è che su Linux la dimensione del buffer di una pipe è alquanto inflessibile. Il valore predefinito è 64 KB da Linux 2.6.11 (la chiamata tee() è stata aggiunta in 2.6.17) - vedere lo pipe manpage. Dal 2.6.35 questo valore può essere modificato tramite l'opzione F_SETPIPE_SZ su fcntl() (vedere fcntl manpage) fino al limite specificato da /proc/sys/fs/pipe-size-max, ma il buffering è ancora più difficile da modificare su richiesta rispetto a uno schema allocato dinamicamente nello spazio utente. essere. Ciò significa che la tua capacità di gestire socket lenti sarà alquanto limitata, indipendentemente dal fatto che sia accettabile in base alla frequenza con cui prevedi di ricevere ed essere in grado di inviare dati.

Supponendo che questa strategia di buffering sia accettabile, sei corretto nel presupposto che dovrai tenere traccia della quantità di dati che ogni pipe di destinazione ha consumato dalla sorgente ed è sicuro scartare solo i dati che tutte le pipe di destinazione hanno consumato . Ciò è alquanto complicato dal fatto che lo tee() non ha il concetto di offset: è possibile copiare solo dall'inizio della pipe. La conseguenza di ciò è che è possibile copiare solo alla velocità del socket più lento, dal momento che non è possibile utilizzare tee() per copiare in una pipe di destinazione finché alcuni dati non sono stati consumati dalla fonte e non è possibile eseguire questo fino a quando tutti i socket hanno i dati che stai per consumare.

Come gestirlo dipende dall'importanza dei dati.Se hai davvero bisogno della velocità di tee() e splice(), e sei sicuro che un socket lento sarà un evento estremamente raro, potresti fare qualcosa del genere (ho pensato che tu stia usando IO non bloccante e un singolo thread , ma qualcosa di simile potrebbe funzionare anche con più thread):

1. assicurarsi che tutti i tubi siano non-blocking (utilizzare fcntl(d, F_SETFL, O_NONBLOCK) per rendere ogni descrittore di file non bloccante).
2. Inizializza una variabile read_counter per ogni pipe di destinazione su zero.
3. Utilizzare qualcosa come epoll() per attendere finché non c'è qualcosa nella pipe di origine.
4. Loop su tutti i tubi di destinazione in cui read_counter è zero, chiamando tee() per trasferire i dati a ciascuno. Assicurati di passare SPLICE_F_NONBLOCK nelle bandiere.
5. Incrementa read_counter per ciascun tubo di destinazione dell'importo trasferito da tee(). Tieni traccia del valore risultante più basso.
6. Trovare il valore risultante più basso di read_counter - se non è zero, eliminare la quantità di dati dalla pipe di origine (utilizzando una chiamata splice() con una destinazione aperta su /dev/null, ad esempio). Dopo aver scartato i dati, sottrarre la quantità scartata da read_counter su a tutti i tubi (poiché questo era il valore più basso, non è possibile che ne risultasse negativo).
7. Ripetere dal punto .

Nota: una cosa che mi ha scattato in passato è che SPLICE_F_NONBLOCK colpisce se i tee() e splice() operazioni sui tubi sono non-blocking, e il O_NONBLOCK si imposta con fnctl() colpisce se le interazioni con altre chiamate (es. read() e write()) non sono bloccanti. Se vuoi che tutto sia non-bloccante, imposta entrambi. Ricorda inoltre di rendere i tuoi socket non bloccanti o le chiamate splice() per il trasferimento dei dati potrebbero bloccarsi (a meno che non sia quello che vuoi, se stai usando un approccio con thread).

Come potete vedere, questa strategia ha un grosso problema: non appena un socket si blocca, tutto si ferma: il tubo di destinazione per quel socket si riempirà, e quindi il tubo sorgente diventerà stagnante. Quindi, se raggiungi la fase in cui tee() restituisce EAGAIN nel passaggio , allora dovrai chiudere il socket, o almeno "disconnetterlo" (cioè toglierlo dal ciclo) in modo da non scrivere qualsiasi altra cosa fino a quando il suo buffer di output è vuoto. La scelta dipende dal fatto che il flusso di dati possa essere ripristinato dal fatto che alcuni di essi sono saltati.

Se si vuole far fronte con latenza di rete con più grazia, allora si sta andando ad avere bisogno di fare di più buffering, e questo sta andando a coinvolgere sia i buffer user-space (che nega invece i vantaggi di tee() e splice()) o forse buffer basato su disco. Il buffering basato su disco sarà quasi certamente molto più lento del buffering spazio utente, e quindi non appropriato visto che presumibilmente si desidera molta velocità da quando si è scelto tee() e splice() in primo luogo, ma lo cito per completezza.

Una cosa che vale la pena notare se si finisce per inserire dati da spazio utente in qualsiasi punto è il vmsplice() chiamata che può eseguire "gather output" from user-spazio in un tubo, in modo analogo al richiamo writev().Questo potrebbe essere utile se stai facendo abbastanza buffer che hai diviso i tuoi dati tra più buffer allocati diversi (ad esempio se stai usando un approccio di allocatore di pool).

Infine, si potrebbe immaginare scambiando prese tra il regime di "veloce" di usare tee() e splice() e, se non riescono a tenere il passo, li passare a un più lento user-space buffering. Questo complicherà la tua implementazione, ma se gestisci un numero elevato di connessioni e solo una piccola parte di esse è lenta, stai comunque riducendo la quantità di copia allo spazio utente che è in qualche modo coinvolta. Tuttavia, questo sarebbe solo sempre una misura a breve termine per far fronte a problemi di rete transienti - come ho detto in origine, hai un problema fondamentale se i tuoi socket sono più lenti della tua sorgente. Alla fine avresti raggiunto qualche limite di buffering e hai bisogno di saltare i dati o chiudere le connessioni.

Nel complesso, vorrei prendere in considerazione con attenzione perché è necessario la velocità di tee() e splice() e se, per il vostro caso d'uso, semplicemente buffer in user-space in memoria o su disco sarebbe più appropriato. Se sei sicuro che le velocità saranno sempre alte, e il buffering limitato è accettabile, l'approccio descritto sopra dovrebbe funzionare.

Inoltre, una cosa che dovrei menzionare è che questo renderà il tuo codice estremamente specifico per Linux - non sono a conoscenza di queste chiamate supportate in altre varianti di Unix. La chiamata sendfile() è più limitata di splice(), ma potrebbe essere piuttosto più portatile. Se vuoi davvero che le cose siano portabili, attenersi al buffering dello spazio utente.

Fammi sapere se c'è qualcosa di cui ho parlato su cui desideri maggiori dettagli.