Named pipe in Java e multithreading

Question

Sono corretto suppongo che entro i limiti dello stesso processo che ha 2 thread di lettura/scrittura su una pipe denominata, non blocca affatto il lettore/scrittore? Quindi con orari errati è possibile perdere alcuni dati?

E in caso di più processi - il lettore aspetterà che alcuni dati siano disponibili e lo scrittore sarà bloccato fino a quando il lettore leggerà tutti i dati forniti dal lettore?

Sto pianificando di utilizzare named pipe per passare diversi (decine, centinaia) di file da un processo esterno e consumarne uno nella mia applicazione Java. Scrivere semplici test unitari per utilizzare un thread per scrivere nella pipe e un altro - per la lettura dalla pipe, ha provocato sporadici errori di test a causa della mancanza di blocchi di dati.

Penso che sia dovuto alla filettatura e allo stesso processo, quindi il mio test non è corretto in generale. Questa supposizione è corretta?

Ecco qualche esempio che illustra il caso:

import java.io.{FileOutputStream, FileInputStream, File} 
import java.util.concurrent.Executors 
import org.apache.commons.io.IOUtils 
import org.junit.runner.RunWith 
import org.scalatest.FlatSpec 
import org.scalatest.junit.JUnitRunner 

@RunWith(classOf[JUnitRunner]) 
class PipeTest extends FlatSpec { 

    def md5sum(data: Array[Byte]) = { 
    import java.security.MessageDigest 
    MessageDigest.getInstance("MD5").digest(data).map("%02x".format(_)).mkString 
    } 

    "Pipe" should "block here" in { 
    val pipe = new File("/tmp/mypipe") 
    val srcData = new File("/tmp/random.10m") 
    val md5 = "8e0a24d1d47264919f9d47f5223c913e" 
    val executor = Executors.newSingleThreadExecutor() 
    executor.execute(new Runnable { 
     def run() { 
     (1 to 10).foreach { 
      id => 
      val fis = new FileInputStream(pipe) 
      assert(md5 === md5sum(IOUtils.toByteArray(fis))) 
      fis.close() 
     } 
     } 
    }) 
    (1 to 10).foreach { 
     id => 
     val is = new FileInputStream(srcData) 
     val os = new FileOutputStream(pipe) 
     IOUtils.copyLarge(is, os) 
     os.flush() 
     os.close() 
     is.close() 
     Thread.sleep(200) 
    } 
    } 

} 

senza Thread.sleep (200) il test non riesce a passare per ragioni

• eccezione tubo rotto
• somma MD5 errata

con questo set di ritardo: funziona alla grande. Sto usando un file con 10 megabyte di dati casuali.

Answer 1

Questa è una condizione di gara molto semplice nel codice: si scrivono messaggi a dimensione fissa nella pipe e si presuppone che sia possibile leggere gli stessi messaggi. Tuttavia, non hai idea di quanti dati siano disponibili nella pipe per ogni lettura data.

Se si antepongono le scritture con il numero di byte scritti e si assicura che ogni lettura legge solo quel numero di byte, vedrete che le pipe funzionano esattamente come pubblicizzate.

Se si dispone di una situazione con più scrittori e/o più lettori, si consiglia di utilizzare una coda di messaggi effettiva. In realtà, raccomando comunque di usare una coda di messaggi, in quanto risolve il problema della demarcazione dei confini dei messaggi; non ha molto senso reinventare quella particolare ruota.

Answer 2

Sono corretto Suppongo che entro i limiti dello stesso processo di 2 thread di lettura/scrittura su una named pipe non blocchi affatto il lettore/scrittore?

A meno che non si utilizzi l'I/O non bloccante, che non si è.

Quindi con orari errati è possibile perdere alcuni dati?

A meno che non si utilizzi l'I/O non bloccante, che non si è.