Scala中管道IO流的使用为何无限挂起？

我尝试实现李豪毅（Li Haoyi）博客中讨论的流接口，编写了一个简单测试实现，但运行时无限挂起，无法找到原因。

测试代码 #

import java.io.PipedOutputStream
import java.io.PipedInputStream
import java.io.InputStream
import java.io.OutputStream
import scala.concurrent.{Future, Promise, Await}
import scala.concurrent.ExecutionContext
import java.util.concurrent.Executors
import scala.concurrent.duration._
import scala.collection.immutable.ListMap

implicit val ec = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))

def copy(input: InputStream, output: OutputStream): Unit = {
  val buffer = new Array[Byte](1024)
  var bytesRead = input.read(buffer)
  while (bytesRead != -1) {
    output.write(buffer, 0, bytesRead)
    bytesRead = input.read(buffer)
  }
}

def curthread(label: String): Unit = println(label, Thread.currentThread().getName())

trait Writable{
  def writeBytesTo(out: OutputStream): Unit
}

trait Readable extends Writable{
  def readBytesThrough[T](f: InputStream => T): T
  def writeBytesTo(out: OutputStream): Unit = readBytesThrough(copy(_, out))
}

def convertToUpperCase(input: InputStream, output: OutputStream): Unit = {
  println("convertToUpperCase")
  curthread("convertToUpperCase")
  val buffer = new Array[Byte](1024) // buffer size
  println("convertToUpperCase: input.read")
  var bytesRead = input.read(buffer)

  while (bytesRead != -1) {
    println(s"convertToUpperCase read $bytesRead")
    val upperBuffer = buffer.map(_.toChar.toUpper.toByte)
    println("convertToUpperCase: output.write")
    output.write(upperBuffer, 0, bytesRead)
    println("convertToUpperCase: input.read")
    bytesRead = input.read(buffer)
  }

  input.close()
  output.close()
}

class Put extends Readable {
  private val output = new PipedOutputStream

  def readBytesThrough[T](f: InputStream => T): T = {
    println("Put: readBytesThrough")
    curthread("Put")
    f(new PipedInputStream(output))
  }
  
  def receive(x: Any): Unit = {
    println(s"Put: receive($x)")
    curthread("Put")
    output.write(x.toString.getBytes("utf-8"))
    println("done receiving")
  }

  def receivedLast(): Unit = {
    println("Put: receivedLast")
    curthread("Put")
    output.close()
  }
}

case class Parsed(value: String) extends Writable {
  def writeBytesTo(out: OutputStream): Unit = {
    println("Parsed: writeBytesTo")
    curthread("Parsed")
    out.write(value.getBytes("utf-8"))
  }
}

def uppercase(r: Readable): Readable = new Readable {
  def readBytesThrough[T](f: InputStream => T): T = {
    val output = new PipedOutputStream
    Future {
      r.readBytesThrough(input => {
        convertToUpperCase(input, output)
      })
    }
    f(new PipedInputStream(output))
  }
}

def parse(r: Readable): Writable = {
  curthread("parse")
  val x = r.readBytesThrough(scala.io.Source.fromInputStream(_).mkString)
  Parsed(x)
}

def display(w: Writable): Unit = {
  curthread("display")
  w.writeBytesTo(System.out)
}

val put = new Put

curthread("main")

Future {
    put.receive("foobarbaz")
    put.receivedLast()
}

display(parse(uppercase(put)))

运行输出 #

(main,run-main-0)
Put: receive(foobarbaz)
(Put,pool-11-thread-1)
(parse,run-main-0)
Put: readBytesThrough
(Put,pool-11-thread-2)
convertToUpperCase
(convertToUpperCase,pool-11-thread-2)
convertToUpperCase: input.read

问题原因 #

核心死锁源于管道流的连接时机错误：

1. Put类的output是无参构造的PipedOutputStream，此时尚未连接任何PipedInputStream。
2. 线程池调度导致put.receive先于Put.readBytesThrough执行，output.write因没有对应的读取端（PipedInputStream未创建）进入无限阻塞，无法继续执行put.receivedLast()关闭输出流。
3. 后续convertToUpperCase中的input.read因无数据可读阻塞，主线程在parse的mkString操作中等待管道流数据/关闭，最终形成多重阻塞死锁。

解决方法 #

修改Put类，通过同步机制确保PipedInputStream创建完成后再执行写入操作，避免无读取端时的写入阻塞：

class Put extends Readable {
  private val pipeReady = Promise[Unit]()
  private val output = new PipedOutputStream

  def readBytesThrough[T](f: InputStream => T): T = {
    println("Put: readBytesThrough")
    curthread("Put")
    // 创建管道输入流并完成连接
    val input = new PipedInputStream(output)
    // 通知写入线程管道已准备好
    pipeReady.success(())
    f(input)
  }
  
  def receive(x: Any): Unit = {
    println(s"Put: receive($x)")
    curthread("Put")
    // 等待管道连接建立后再写入
    Await.result(pipeReady.future, Duration.Inf)
    output.write(x.toString.getBytes("utf-8"))
    println("done receiving")
  }

  def receivedLast(): Unit = {
    println("Put: receivedLast")
    curthread("Put")
    output.close()
  }
}

修改后，receive会等待readBytesThrough创建好PipedInputStream并完成管道连接后再执行写入，确保数据能正常被convertToUpperCase读取，后续流程可正常执行直至输出结果。

内容的提问来源于stack exchange，提问作者alcorn