<commit_before>package isabelle.control

import java.io.{BufferedInputStream, BufferedOutputStream, BufferedReader, BufferedWriter, FileInputStream, FileOutputStream, IOException, InputStream, InputStreamReader, OutputStreamWriter}
import java.lang
import java.lang.ProcessBuilder.Redirect
import java.lang.ref.Cleaner
import java.net.URL
import java.nio.file.{FileSystemNotFoundException, Files, Path, Paths}
import java.util.concurrent.{ArrayBlockingQueue, BlockingQueue, ConcurrentHashMap, ConcurrentLinkedQueue}

import isabelle.control.Isabelle.Setup
import org.apache.commons.io.FileUtils
import org.log4s
import org.log4s.{Debug, LogLevel, Logger, Warn, getLogger}

import scala.annotation.tailrec
import scala.collection.JavaConverters.{asScalaIteratorConverter, collectionAsScalaIterableConverter, enumerationAsScalaIteratorConverter}
import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.io.Source
import scala.sys.process.{Process, ProcessLogger}
import scala.util.{Failure, Success, Try}

class Isabelle(val setup: Setup, build: Boolean = false) {
  import Isabelle._

  private val sendQueue : BlockingQueue[(String, Try[String] => Unit)] = new ArrayBlockingQueue(1000)
  private val callbacks : ConcurrentHashMap[Int, Try[String] => Unit] = new ConcurrentHashMap()
  private val cleaner = Cleaner.create()

  // Must be Integer, not Int, because ConcurrentLinkedList uses null to indicate that it is empty
  private val garbageQueue = new ConcurrentLinkedQueue[Integer]()

  private def garbageCollect() : Option[String] = {
//    println("Checking for garbage")
    @tailrec def drain(objs: List[Int]) : List[Int] = garbageQueue.poll() match {
      case null => objs
      case obj =>
        if (objs.length > 1000) // Since Poly/ML has only a 64KB string size limit, we avoid too long lists of IDs in one go
          obj::objs
        else
          drain(obj::objs)
    }
    val objs = drain(Nil)
    if (objs.nonEmpty) {
      logger.debug(s"Sending GC command to Isabelle, ${objs.size} freed objects")
      Some(s"g${objs.mkString(" ")}\n")
    } else
      None
  }

  private def processQueue(inFifo: Path) : Unit = {
    logger.debug("Process queue thread started")
    val stream = new FileOutputStream(inFifo.toFile)
    val writer = new BufferedWriter(new OutputStreamWriter(stream, "ascii"))
    var count = 0

    @tailrec @inline
    def drainQueue() : Unit = {
      val elem = sendQueue.poll()
      if (elem!=null) {
        val (line,callback) = elem
        assert(line.endsWith("\n"), line)
//        logger.debug(s"Writing ${line.trim}")
        if (callback != null)
          callbacks.put(count, callback)
        writer.write(line)
        count += 1
        drainQueue()
      }
    }

    while (true) {
      val (line,callback) = sendQueue.take()
      assert(line.endsWith("\n"), line)
//      logger.debug(s"Writing! ${line.trim}")
      if (callback != null)
        callbacks.put(count, callback)
      writer.write(line)
      count += 1
      drainQueue()
      for (cmd <- garbageCollect()) {
        writer.write(cmd)
        count += 1
      }
//      println("Flushing.")
      writer.flush()
      Thread.sleep(100)
    }
  }

  private def parseIsabelle(outFifo: Path) : Unit = {
    val output = new FileInputStream(outFifo.toFile)
    Source.fromInputStream(output, "ascii").getLines.foreach { line =>
//      println(s"Received: [$line]")
      val spaceIdx = line.indexOf(' ')
      val (seq,content) = if (spaceIdx == -1) (line,"") else line.splitAt(spaceIdx+1)
      callbacks.remove(seq.trim.toInt) match {
        case null => println(s"No callback $seq")
        case callback =>
          if (content.nonEmpty && content(0) == '!')
            callback(Failure(IsabelleException(this, intStringToID(content.substring(1)))))
          else
            callback(Success(content))
      }
//      println(s"#callbacks = ${callbacks.size}")
    }
  }

  //noinspection SameParameterValue
  private def filePathFromResource(name: String, tmpDir: Path): Path = {
    val url = getClass.getResource(name)
    assert(url != null, name)
    try
      Path.of(url.toURI)
    catch {
      case _ : FileSystemNotFoundException =>
        val tmpPath = tmpDir.resolve(name.split('/').last)
        val tmpFile = tmpPath.toFile
        tmpFile.deleteOnExit()
        FileUtils.copyURLToFile(url, tmpFile)
        tmpPath
    }
  }

  private def startProcess() : java.lang.Process = {
    def wd = setup.workingDirectory
    /** Path to absolute string, interpreted relative to wd */
    def str(path: Path) = wd.resolve(path).toAbsolutePath.toString

    val tempDir = Files.createTempDirectory("isabellecontrol").toAbsolutePath
    tempDir.toFile.deleteOnExit()
    logger.debug(s"Temp directory: $tempDir")

    val isabelleBinary = setup.isabelleHome.resolve("bin").resolve("isabelle")
    val mlFile = filePathFromResource("control_isabelle.ml", tempDir)

    assert(setup.userDir.forall(_.endsWith(".isabelle")))


    val inputPipe = tempDir.resolve("in-fifo").toAbsolutePath
    inputPipe.toFile.deleteOnExit()
    val outputPipe = tempDir.resolve("out-fifo").toAbsolutePath
    outputPipe.toFile.deleteOnExit()
    if (Process(List("mkfifo", inputPipe.toString)).! != 0)
      throw new IOException(s"Cannot create fifo $inputPipe")
    if (Process(List("mkfifo", outputPipe.toString)).! != 0)
      throw new IOException(s"Cannot create fifo $outputPipe")


    val cmd = ListBuffer[String]()

    cmd += str(isabelleBinary) += "process"
    cmd += "-l" += setup.logic

    // TODO: escape pipe name for ML
    cmd += "-e" += s"""val (inputPipeName,outputPipeName) = ("$inputPipe","$outputPipe")"""

    cmd += "-f" += mlFile.toAbsolutePath.toString

    cmd += "-e" += "Control_Isabelle.handleLines()"

    for (root <- setup.sessionRoots)
      cmd += "-d" += str(root)

    logger.debug(s"Cmd line: ${cmd.mkString(" ")}")

    val processBuilder = new java.lang.ProcessBuilder(cmd.toSeq :_*)
    processBuilder.directory(wd.toAbsolutePath.toFile)
    for (userDir <- setup.userDir)
      processBuilder.environment.put("USER_HOME", str(userDir.getParent))

    val processQueueThread = new Thread("Send to Isabelle") {
      override def run(): Unit = processQueue(inputPipe) }
    processQueueThread.setDaemon(true)
    processQueueThread.start()

    val parseIsabelleThread = new Thread("Read from Isabelle") {
      override def run(): Unit = parseIsabelle(outputPipe) }
    parseIsabelleThread.setDaemon(true)
    parseIsabelleThread.start()

    val process = processBuilder.start()

    logStream(process.getErrorStream, Warn) // stderr
    logStream(process.getInputStream, Debug) // stdout

    process
  }

  if (build) buildSession(setup)
  private val process: lang.Process = startProcess()

  def isDestroyed = destroyed

  @volatile private var destroyed = false
  def destroy(): Unit = {
    destroyed = true
    garbageQueue.clear()
    process.destroy()

    def callCallback(cb: Try[String] => Unit): Unit =
      cb(Failure(IsabelleDestroyedException("Isabelle process has been destroyed")))

    for ((_,cb) <- sendQueue.asScala)
      callCallback(cb)
    sendQueue.clear()

    for (cb <- callbacks.values.asScala)
      callCallback(cb)
  }

  private def send(str: String, callback: Try[String] => Unit) : Unit = {
    if (destroyed)
      throw new IllegalStateException("Isabelle instance has been destroyed")<commit_msg>Fix: removed 100ms sleep in Isabelle-communication loop<commit_after>package isabelle.control

import java.io.{BufferedInputStream, BufferedOutputStream, BufferedReader, BufferedWriter, FileInputStream, FileOutputStream, IOException, InputStream, InputStreamReader, OutputStreamWriter}
import java.lang
import java.lang.ProcessBuilder.Redirect
import java.lang.ref.Cleaner
import java.net.URL
import java.nio.file.{FileSystemNotFoundException, Files, Path, Paths}
import java.util.concurrent.{ArrayBlockingQueue, BlockingQueue, ConcurrentHashMap, ConcurrentLinkedQueue}

import isabelle.control.Isabelle.Setup
import org.apache.commons.io.FileUtils
import org.log4s
import org.log4s.{Debug, LogLevel, Logger, Warn, getLogger}

import scala.annotation.tailrec
import scala.collection.JavaConverters.{asScalaIteratorConverter, collectionAsScalaIterableConverter, enumerationAsScalaIteratorConverter}
import scala.collection.mutable.ListBuffer
import scala.concurrent.duration.Duration
import scala.concurrent.{Await, ExecutionContext, Future, Promise}
import scala.io.Source
import scala.sys.process.{Process, ProcessLogger}
import scala.util.{Failure, Success, Try}

class Isabelle(val setup: Setup, build: Boolean = false) {
  import Isabelle._

  private val sendQueue : BlockingQueue[(String, Try[String] => Unit)] = new ArrayBlockingQueue(1000)
  private val callbacks : ConcurrentHashMap[Int, Try[String] => Unit] = new ConcurrentHashMap()
  private val cleaner = Cleaner.create()

  // Must be Integer, not Int, because ConcurrentLinkedList uses null to indicate that it is empty
  private val garbageQueue = new ConcurrentLinkedQueue[Integer]()

  private def garbageCollect() : Option[String] = {
//    println("Checking for garbage")
    @tailrec def drain(objs: List[Int]) : List[Int] = garbageQueue.poll() match {
      case null => objs
      case obj =>
        if (objs.length > 1000) // Since Poly/ML has only a 64KB string size limit, we avoid too long lists of IDs in one go
          obj::objs
        else
          drain(obj::objs)
    }
    val objs = drain(Nil)
    if (objs.nonEmpty) {
      logger.debug(s"Sending GC command to Isabelle, ${objs.size} freed objects")
      Some(s"g${objs.mkString(" ")}\n")
    } else
      None
  }

  private def processQueue(inFifo: Path) : Unit = {
    logger.debug("Process queue thread started")
    val stream = new FileOutputStream(inFifo.toFile)
    val writer = new BufferedWriter(new OutputStreamWriter(stream, "ascii"))
    var count = 0

    @tailrec @inline
    def drainQueue() : Unit = {
      val elem = sendQueue.poll()
      if (elem!=null) {
        val (line,callback) = elem
        assert(line.endsWith("\n"), line)
//        logger.debug(s"Writing ${line.trim}")
        if (callback != null)
          callbacks.put(count, callback)
        writer.write(line)
        count += 1
        drainQueue()
      }
    }

    while (true) {
      val (line,callback) = sendQueue.take()
      assert(line.endsWith("\n"), line)
//      logger.debug(s"Writing! ${line.trim}")
      if (callback != null)
        callbacks.put(count, callback)
      writer.write(line)
      count += 1
      drainQueue()
      for (cmd <- garbageCollect()) {
        writer.write(cmd)
        count += 1
      }
//      println("Flushing.")
      writer.flush()
//      Thread.sleep(100)
    }
  }

  private def parseIsabelle(outFifo: Path) : Unit = {
    val output = new FileInputStream(outFifo.toFile)
    Source.fromInputStream(output, "ascii").getLines.foreach { line =>
//      println(s"Received: [$line]")
      val spaceIdx = line.indexOf(' ')
      val (seq,content) = if (spaceIdx == -1) (line,"") else line.splitAt(spaceIdx+1)
      callbacks.remove(seq.trim.toInt) match {
        case null => println(s"No callback $seq")
        case callback =>
          if (content.nonEmpty && content(0) == '!')
            callback(Failure(IsabelleException(this, intStringToID(content.substring(1)))))
          else
            callback(Success(content))
      }
//      println(s"#callbacks = ${callbacks.size}")
    }
  }

  //noinspection SameParameterValue
  private def filePathFromResource(name: String, tmpDir: Path): Path = {
    val url = getClass.getResource(name)
    assert(url != null, name)
    try
      Path.of(url.toURI)
    catch {
      case _ : FileSystemNotFoundException =>
        val tmpPath = tmpDir.resolve(name.split('/').last)
        val tmpFile = tmpPath.toFile
        tmpFile.deleteOnExit()
        FileUtils.copyURLToFile(url, tmpFile)
        tmpPath
    }
  }

  private def startProcess() : java.lang.Process = {
    def wd = setup.workingDirectory
    /** Path to absolute string, interpreted relative to wd */
    def str(path: Path) = wd.resolve(path).toAbsolutePath.toString

    val tempDir = Files.createTempDirectory("isabellecontrol").toAbsolutePath
    tempDir.toFile.deleteOnExit()
    logger.debug(s"Temp directory: $tempDir")

    val isabelleBinary = setup.isabelleHome.resolve("bin").resolve("isabelle")
    val mlFile = filePathFromResource("control_isabelle.ml", tempDir)

    assert(setup.userDir.forall(_.endsWith(".isabelle")))


    val inputPipe = tempDir.resolve("in-fifo").toAbsolutePath
    inputPipe.toFile.deleteOnExit()
    val outputPipe = tempDir.resolve("out-fifo").toAbsolutePath
    outputPipe.toFile.deleteOnExit()
    if (Process(List("mkfifo", inputPipe.toString)).! != 0)
      throw new IOException(s"Cannot create fifo $inputPipe")
    if (Process(List("mkfifo", outputPipe.toString)).! != 0)
      throw new IOException(s"Cannot create fifo $outputPipe")


    val cmd = ListBuffer[String]()

    cmd += str(isabelleBinary) += "process"
    cmd += "-l" += setup.logic

    // TODO: escape pipe name for ML
    cmd += "-e" += s"""val (inputPipeName,outputPipeName) = ("$inputPipe","$outputPipe")"""

    cmd += "-f" += mlFile.toAbsolutePath.toString

    cmd += "-e" += "Control_Isabelle.handleLines()"

    for (root <- setup.sessionRoots)
      cmd += "-d" += str(root)

    logger.debug(s"Cmd line: ${cmd.mkString(" ")}")

    val processBuilder = new java.lang.ProcessBuilder(cmd.toSeq :_*)
    processBuilder.directory(wd.toAbsolutePath.toFile)
    for (userDir <- setup.userDir)
      processBuilder.environment.put("USER_HOME", str(userDir.getParent))

    val processQueueThread = new Thread("Send to Isabelle") {
      override def run(): Unit = processQueue(inputPipe) }
    processQueueThread.setDaemon(true)
    processQueueThread.start()

    val parseIsabelleThread = new Thread("Read from Isabelle") {
      override def run(): Unit = parseIsabelle(outputPipe) }
    parseIsabelleThread.setDaemon(true)
    parseIsabelleThread.start()

    val process = processBuilder.start()

    logStream(process.getErrorStream, Warn) // stderr
    logStream(process.getInputStream, Debug) // stdout

    process
  }

  if (build) buildSession(setup)
  private val process: lang.Process = startProcess()

  def isDestroyed: Boolean = destroyed

  @volatile private var destroyed = false
  def destroy(): Unit = {
    destroyed = true
    garbageQueue.clear()
    process.destroy()

    def callCallback(cb: Try[String] => Unit): Unit =
      cb(Failure(IsabelleDestroyedException("Isabelle process has been destroyed")))

    for ((_,cb) <- sendQueue.asScala)
      callCallback(cb)
    sendQueue.clear()

    for (cb <- callbacks.values.asScala)
      callCallback(cb)
  }

  private def send(str: String, callback: Try[String] => Unit) : Unit = {
    if (destroyed)
      throw new IllegalStateException("Isabelle instance has been destroyed")
{'fraction_non_alphanumeric': 0.12370110530147595, 'fraction_numerical': 0.003507843007478986, 'mean_word_length': 3.469091984619935, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6711671', 'n_tokens_mistral': 4669, 'n_tokens_neox': 4534, 'n_words': 1060}