PipedInputStream.java 디자인에 버그가 있습니까? 아니면 그 디자인을 오해 했습니까?

Question

나는 PipedInputStream과 PipedOutputStream에 관련된 문제가 있고, 나는 이러한 클래스의 디자인을 잘못 이해 한 경우 모르거나 내가 PipedInputStream을 이해 지금까지 PipedInputStream.java

에서 자바 코드에 버그가있는 경우와없는 PipedOutputStream은 두 개의 서로 다른 스레드간에 스트림을 생성하는 데 사용할 수있는 메커니즘을 구현합니다. 제작자 스레드는 PipedOutputStream에 뭔가를 쓰고 소비자 스레드는 연결된 PipedInputStream에서 스레드를 읽습니다. 버퍼링 된 통신을 허용하는 내부 버퍼가 있습니다. 기본적으로이 버퍼의 크기는 1024 바이트입니다.

소비자 스레드가 PipedInputStream을 읽고 버퍼가 비어 있으면 스레드가 대기합니다. 제작자 스레드가 PipedOutputStream을 쓰고 버퍼가 가득차면 스레드도 대기합니다.

PipedInputStream은 내부 버퍼를 유지합니다. PipedOutputStream은 PipedInputStream에 선언 된 함수 만 사용합니다.

모든 PipedInputStream의 내부 (원) 버퍼와 관련된 필드합니다 (byte [] buffer 상기 int in 및 int out가 PipedInputStream.java 볼 수 -as)는 모든 protected 선언된다. PipedInputStream은 2 개의 다른 PipedInputStream.receive 기능을 사용하여 데이터를 주입합니다.

모든 입력 스트림에는 read() 및 read(byte [], int, int)의 두 가지 읽기 버전이 있습니다. 모든 OutputStream에는 2 개의 기입 버젼 write(byte b)과 write(byte [], int, int)가 있습니다. 모두 단일 바이트 버전과 다중 바이트 버전이 있습니다. PipedInputStream 및 PipedOutputStream에는 이러한 기능이 있습니다.

PipedOutputStream.write(byte b)은 PipedInputStream에 연결된 바이트를 삽입하는 기능을 사용합니다 (PipedInputStream.receive(int b)). 이 수신 기능은 protected으로 선언되어 있으므로이 기능을 오버로드하여 PipedOutputStream에서 PipedInputStream으로 바이트 주입을 차단할 수 있습니다.

PipedOutputStream.write(byte b[], int offset, int len)은 에 연결된 바이트 배열을 삽입하기 위해 PipedInputStream.receive(byte [] b, int offset, int len)을 사용합니다.

그리고 내 문제는 다음과 같습니다 receive(int)의 PipedInputStream.receive(byte [], int, int), 멀티 바이트 대응 부분은 receive(int)으로 보호되지 않았다고 선언되지 않습니다. 기본 가시성 (패키지 가시성)이 있습니다. 따라서이 기능에 과부하가 걸리지 않고 PipedOutputStream에서 PipedInputStream으로 멀티 바이트 주입을 차단할 수 없습니다.

PipedInputStream.write(byte b[], int offset, int len)은 PipedInputStream.write(int b)을 호출하지 않습니다. 따라서 을 사용할 때 receive(int)을 오버로드해도 아무런 영향이 없습니다.

내가 이해하는 한 PipedInputStream.receive(byte[], int, int)은 protected이어야하며 PipedInputStream.receive(int)이어야합니다.선언은 :

protected synchronized void receive(byte [] b, int off, int len) throws IOException {

PipeReader와 PipeWriter (PipedInputStream 및 PipedOutputStream의 문자 버전) 버퍼 필드를 선언하고는 보호되지 않는 (패키지 가시성 방법 나타날

synchronized void receive(byte [] b, int off, int len) throws IOException {

같아야 !). Java의 리더/라이터 (JDK1. 1 이후)는, InputStream/OutputStream (JDK1.

그것은 PipedInputStream의 디자인에 실제 버그인가?, 이전 자바 버전에서 상속 PipedInputStream 디자인 사고의 protected visibity은?, 나 아니면 완전히 손실 있어요?

미리 감사드립니다.

PD : 다음은이 문제가 나타나는 예입니다. 이 프로그램은 컴파일되지 않습니다 (언급 된 가시성 문제로). 이 예제에서는 필요한 경우 버퍼의 자동 확장을 허용하는 PipedInputStream 하위 클래스를 만들려고합니다. 따라서 버퍼가 비어 있고 누군가가 스레드를 읽으려고하면 대기합니다. 그러나 버퍼가 꽉 차서 누군가가 (연결된 PipedOutputStream을 사용하여) 쓰기를 시도하면 스레드는 대기하지 않지만 버퍼가 더 많은 바이트를 저장하도록 확장됩니다. 소비자는 기다리지 만 생산자는 기다리지 않습니다.

이 예제의 기능 구현은 있지만 PipedInputStream 하위 클래스로 구현할 수 없는지 알고 싶습니다.

import java.io.IOException; 
import java.io.PipedInputStream; 
import java.io.PipedOutputStream; 

public class ExtensiblePipedInputStream extends PipedInputStream { 

/** 
* Default extensions' size 
*/ 
private static final int DEFAULT_EXTENSION = 1024; 

/** 
* The current extensions' size 
*/ 
protected int extension = DEFAULT_EXTENSION; 


// the same constructors than the super class (PipedInputStream)... 

public ExtensiblePipedInputStream() { 
    super(); 
} 

public ExtensiblePipedInputStream(PipedOutputStream src) throws IOException { 
    super(src); 
} 

public ExtensiblePipedInputStream(int pipeSize) { 
    super(pipeSize); 
} 

public ExtensiblePipedInputStream(PipedOutputStream src, int pipeSize) throws IOException { 
    super(src, pipeSize); 
} 

/** 
* This function ensures the specified capacity in the internal buffer. If 
* the specified capacity is less or equals than the current internal buffer 
* capacity it does nothing. If the specified capacity is greater than the 
* current one, then the buffer is extended to: at least allocate the new 
* capacity. This function extends the buffer using multiple factors of 
* extension size. 
* 
* @param capacity The capacity 
* @throws IOException if an IO error occurs 
* @throws IllegalArgumentException if capacity is negative 
*/ 
public synchronized void ensureCapacity(int capacity) throws IOException, IllegalArgumentException { 
    if (capacity < 0) { 
     throw new IllegalArgumentException("capacity < 0"); 
    } 

    if (capacity > buffer.length) { 
     int additionalSpace = capacity - buffer.length; 
     final int modExtension = additionalSpace % extension; 
     additionalSpace += (modExtension == 0) ? 0 : extension - modExtension; 

     setCapacity(buffer.length + additionalSpace); 
    } 
} 

/** 
* Returns capacity of the internal buffer (the buffer's size). 
* 
* @return The capacity or the internal buffer 
*/ 
public synchronized int getCapacity() { 
    return buffer.length; 
} 

/** 
* Returns the size of the next buffer's extensions. 
* 
* @return The size of the next buffer's extensions. 
*/ 
public synchronized int getExtension() { 
    return extension; 
} 

/** 
* This function extends and invokes PipedInputStream.receive. It only avoid 
* writers block by extending the internal buffer when needed. 
* 
* @param b The byte to be received 
* @throws IOException if an IO error occurs 
*/ 
@Override 
protected synchronized void receive(int b) throws IOException { 
    ensureCapacity(available() + 1); 
    super.receive(b); 
} 

/** 
* MY PROBLEM!!!! 
* 
* this function is not posible! 
* 
* PipedInputStream.receive(byte[], int, int) 
* has not protected visibility, it has package visibility!!!!! 
* 
* Why? 
* 
* @param b The array of bytes to be received 
* @param off The offset in the array of bytes. 
* @param len The number of bytes to be received. 
* @throws IOException If an IO error occurs 
*/ 
@Override 
protected synchronized void receive(byte b[], int off, int len) throws IOException { 
    ensureCapacity(available() + len); 
    super.receive(b, off, len); 
} 

/** 
* Changes the size of the internal buffer. The new size must be greater or 
* equals than the number of bytes stored in the internal buffer 
* (available()) 
* 
* @param capacity The new size of the internal buffer. 
* @throws IOException If an IO error occurs. 
* @throws IllegalArgumentException If capacity < available() 
*/ 
public synchronized void setCapacity(int capacity) throws IOException, IllegalArgumentException { 
    final int available = available(); 

    if (capacity < available) { 
     throw new IllegalArgumentException("capacity < available"); 
    } 

    final byte[] nbuf = new byte[capacity]; 
    if (available > 0) { 
     final int firstTransferAmount = Math.min(available, buffer.length - out); 
     System.arraycopy(buffer, out, nbuf, 0, firstTransferAmount); 
     if (in > 0) { 
      System.arraycopy(buffer, 0, nbuf, firstTransferAmount, in); 
     } 
     out = 0; 
     in = (available == capacity) ? 0 : available; 
    } 

    buffer = nbuf; 
} 

/** 
* Set the size of future extensions. It must be a value greater than 0. 
* 
* @param extension The size of future extensions. 
* @throws IllegalArgumentException If extension <= 0 
*/ 
public synchronized void setExtension(int extension) throws IllegalArgumentException { 
    if (extension <= 0) { 
     throw new IllegalArgumentException("extension <= 0"); 
    } 

    this.extension = extension; 
} 

} 

Answer 1

나는 파이프의 목적을 잘못 이해했다고 생각합니다. 파이프는 입니다. 두 스레드 간의 통신이 차단되었습니다. 작성자의 속도는 파이프가 메모리 사용에 효율적이지만 처리 속도가 가장 느린 구성 요소의 속도로 제한된다는 독자의 속도로 제한 될 것으로 예상됩니다.

비동기 쓰기를 원할 경우, 큐를 사용해야합니다. java.util.concurrent 패키지의 버전 중 하나가 적합해야합니다.

Answer 2

특정 유스 케이스를 지원하지 않는 API 디자인 (특히 서브 클래 싱을 포함하는 API 디자인)은 "제한된"또는 "제한된"것으로 설명 될 수 있지만 버그라고 부르는 경우가 있습니다.

당신이 구현하려는 것이 가치 있다고 생각한다면, 당신의 분석은 정확하다고 생각합니다. 확장 버퍼를 사용하여 파이프 된 입출력 스트림의 버전을 생성하는 것은 수정자가 다른 경우보다 훨씬 효과적입니다.

하지만 ... 불가능하지 않습니다. 보다 큰 규모의 메소드 오버라이드 (예 : read 및 write 메소드)를 통해 원하는대로 달성 할 수도 있고 처음부터 간단히 시작할 수도 있습니다.

---

다른 점은이 (StackOverflow)가 Java "버그"를보고하거나 향상 요청을하는 적절한 장소가 아니라는 것입니다. 이 점에 대해 정말로 생각한다면 OpenJDK 팀에 제안 된 변경 사항을 적용하여 패치를 개발하여 제출하십시오. 그러나 호환성 (기존 고객 서브 클래스 인 PipedInputStream 및 PipedOutputStream과의 호환성 포함)을 위반하는 항목은 거부 될 수 있습니다.이것은 이것이 그들이 아직 "고정"되지 않은 이유를 설명 할 수 있습니다!

---

^{1 - 배리가 지적 하듯이, 파이프로 연결된 스트림은 디자인에 의해 차단된다. 비 차단 버전은 잠재적으로 위험합니다.}