<commit_before>package org.clulab.processors.examples

import java.io.BufferedOutputStream
import java.io.File
import java.io.FileNotFoundException
import java.io.FileOutputStream
import java.io.FilenameFilter
import java.io.PrintWriter
import java.io.StringWriter
import java.nio.charset.StandardCharsets

import org.clulab.dynet.ConstEmbeddingsGlove
import org.clulab.dynet.Utils
import org.clulab.processors.Document
import org.clulab.processors.Processor
import org.clulab.processors.clu.CluProcessor
import org.clulab.processors.fastnlp.FastNLPProcessorWithSemanticRoles
import org.clulab.serialization.DocumentSerializer

import scala.collection.parallel.ForkJoinTaskSupport
import scala.collection.parallel.ForkJoinTasks
import scala.collection.parallel.ParSeq
import scala.io.Source

object ParallelProcessorExample {

  class Timer(val description: String) {
    var elapsedTime: Option[Long] = None
    var startTime: Option[Long] = None

    def time[R](block: => R): R = {
      val t0 = System.currentTimeMillis()
      val result: R = block // call-by-name
      val t1 = System.currentTimeMillis()

      elapsedTime = Some(t1 - t0)
      result
    }

    def start(): Unit = {
      val t0 = System.currentTimeMillis()

      startTime = Some(t0)
    }

    def stop(): Unit = {
      if (startTime.isDefined) {
        val t1 = System.currentTimeMillis()

        elapsedTime = Some(t1 - startTime.get)
      }
    }

    override def toString: String = {
      if (elapsedTime.isDefined)
        s"\tTime\t$description\t${diffToString(elapsedTime.get)}"
      else if (startTime.isDefined)
        s"\tStart\t$description\t${startTime.get}\tms"
      else
        s"\tTimer\t$description"
    }

    def diffToString(diff: Long): String = {
      val days = (diff / (1000 * 60 * 60 * 24)) / 1
      val hours = (diff % (1000 * 60 * 60 * 24)) / (1000 * 60 * 60)
      val mins = (diff % (1000 * 60 * 60)) / (1000 * 60)
      val secs = (diff % (1000 * 60)) / 1000
      val msecs = (diff % (1000 * 1)) / 1

      f"$days:$hours%02d:$mins%02d:$secs%02d.$msecs%03d"
    }
  }

  def mainWithCallback(args: Array[String])(callback: (File, String) => Unit): Unit = {
    val utf8: String = StandardCharsets.UTF_8.toString

    def findFiles(collectionDir: String, extension: String): Seq[File] = {
      val dir = new File(collectionDir)
      val filter = new FilenameFilter {
        def accept(dir: File, name: String): Boolean = name.endsWith(extension)
      }

      val result = Option(dir.listFiles(filter))
          .getOrElse( {
            val here = new File(".").getAbsolutePath
            println(s"Can't find $collectionDir from $here.")
            throw new FileNotFoundException(collectionDir)
          })
      result
    }

    def parallelize[T](seq: Seq[T], threads: Int): ParSeq[T] = {
      val forkJoinPoolConstructor = {
        // Get something of the right type.
        val defaultForkJoinPool = ForkJoinTasks.defaultForkJoinPool
        // Find the constructor.
        defaultForkJoinPool.getClass.getConstructor(classOf[Int])
      }

      def newForkJoinPool(threads: Int) = {
        // Invoke the constructor.
        forkJoinPoolConstructor.newInstance(threads.asInstanceOf[Integer])

        // For the record, this is the standard version
        //new ForkJoinPool(threads)
      }

      val forkJoinPool = newForkJoinPool(threads)
      val forkJoinTaskSupport = new ForkJoinTaskSupport(forkJoinPool)
      val parSeq = seq.par

      parSeq.tasksupport = forkJoinTaskSupport
      parSeq
    }

    def printDocument(document: Document, printWriter: PrintWriter): Unit = document.prettyPrint(printWriter)

    val inputDir = args(0)
    val outputDir = args(1)
    val extension = args(2)
    val threads = args(3).toInt

    val files = findFiles(inputDir, extension)
    val parFiles = parallelize(files, threads)

    Utils.initializeDyNet()

    val processor: Processor = new FastNLPProcessorWithSemanticRoles()
    val documentSerializer = new DocumentSerializer

    val untimed = processor.annotate("I am happy to join with you today in what will go down in history as the greatest demonstration for freedom in the history of our nation.")

    val timer = new Timer(s"$threads threads processing ${parFiles.size} files")
    timer.start()

    parFiles.foreach { file =>
      println(s"Processing ${file.getName}...")

      val text = {
        val source = Source.fromFile(file, utf8)
        val text = source.mkString

        source.close
        text
      }

      val outputFile = new File(outputDir + "/" + file.getName)
      val document = processor.annotate(text)
      val printedDocument = {
        val stringWriter = new StringWriter
        val printWriter = new PrintWriter(stringWriter)

        printDocument(document, printWriter)
        printWriter.close()

        val result = stringWriter.toString
        result
      }
      val savedDocument = documentSerializer.save(document)
      val outputDocument = printedDocument + savedDocument

      callback(outputFile, outputDocument)
    }

    timer.stop()<commit_msg>Use library to parallelize<commit_after>package org.clulab.processors.examples

import java.io.BufferedOutputStream
import java.io.File
import java.io.FileNotFoundException
import java.io.FileOutputStream
import java.io.FilenameFilter
import java.io.PrintWriter
import java.io.StringWriter
import java.nio.charset.StandardCharsets

import org.clulab.dynet.Utils
import org.clulab.processors.Document
import org.clulab.processors.Processor
import org.clulab.processors.fastnlp.FastNLPProcessorWithSemanticRoles
import org.clulab.serialization.DocumentSerializer
import org.clulab.utils.ThreadUtils

import scala.io.Source

object ParallelProcessorExample {

  class Timer(val description: String) {
    var elapsedTime: Option[Long] = None
    var startTime: Option[Long] = None

    def time[R](block: => R): R = {
      val t0 = System.currentTimeMillis()
      val result: R = block // call-by-name
      val t1 = System.currentTimeMillis()

      elapsedTime = Some(t1 - t0)
      result
    }

    def start(): Unit = {
      val t0 = System.currentTimeMillis()

      startTime = Some(t0)
    }

    def stop(): Unit = {
      if (startTime.isDefined) {
        val t1 = System.currentTimeMillis()

        elapsedTime = Some(t1 - startTime.get)
      }
    }

    override def toString: String = {
      if (elapsedTime.isDefined)
        s"\tTime\t$description\t${diffToString(elapsedTime.get)}"
      else if (startTime.isDefined)
        s"\tStart\t$description\t${startTime.get}\tms"
      else
        s"\tTimer\t$description"
    }

    def diffToString(diff: Long): String = {
      val days = (diff / (1000 * 60 * 60 * 24)) / 1
      val hours = (diff % (1000 * 60 * 60 * 24)) / (1000 * 60 * 60)
      val mins = (diff % (1000 * 60 * 60)) / (1000 * 60)
      val secs = (diff % (1000 * 60)) / 1000
      val msecs = (diff % (1000 * 1)) / 1

      f"$days:$hours%02d:$mins%02d:$secs%02d.$msecs%03d"
    }
  }

  def mainWithCallback(args: Array[String])(callback: (File, String) => Unit): Unit = {
    val utf8: String = StandardCharsets.UTF_8.toString

    def findFiles(collectionDir: String, extension: String): Seq[File] = {
      val dir = new File(collectionDir)
      val filter = new FilenameFilter {
        def accept(dir: File, name: String): Boolean = name.endsWith(extension)
      }

      val result = Option(dir.listFiles(filter))
          .getOrElse( {
            val here = new File(".").getAbsolutePath
            println(s"Can't find $collectionDir from $here.")
            throw new FileNotFoundException(collectionDir)
          })
      result
    }

    def printDocument(document: Document, printWriter: PrintWriter): Unit = document.prettyPrint(printWriter)

    val inputDir = args(0)
    val outputDir = args(1)
    val extension = args(2)
    val threads = args(3).toInt

    val files = findFiles(inputDir, extension)
    val parFiles = ThreadUtils.parallelize(files, threads)

    Utils.initializeDyNet()

    val processor: Processor = new FastNLPProcessorWithSemanticRoles()
    val documentSerializer = new DocumentSerializer

    val untimed = processor.annotate("I am happy to join with you today in what will go down in history as the greatest demonstration for freedom in the history of our nation.")

    val timer = new Timer(s"$threads threads processing ${parFiles.size} files")
    timer.start()
    
    parFiles.foreach { file =>
      println(s"Processing ${file.getName}...")

      val text = {
        val source = Source.fromFile(file, utf8)
        val text = source.mkString

        source.close
        text
      }

      val outputFile = new File(outputDir + "/" + file.getName)
      val document = processor.annotate(text)
      val printedDocument = {
        val stringWriter = new StringWriter
        val printWriter = new PrintWriter(stringWriter)

        printDocument(document, printWriter)
        printWriter.close()

        val result = stringWriter.toString
        result
      }
      val savedDocument = documentSerializer.save(document)
      val outputDocument = printedDocument + savedDocument

      callback(outputFile, outputDocument)
    }

    timer.stop()
{'fraction_non_alphanumeric': 0.10183387270765912, 'fraction_numerical': 0.017691477885652642, 'mean_word_length': 2.875836120401338, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '3232370', 'n_tokens_mistral': 2912, 'n_tokens_neox': 2713, 'n_words': 689}