<commit_before>package jigg.pipeline

/*
 Copyright 2013-2015 <NAME>

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
*/

import scala.xml.{Node, Elem}
import scala.concurrent.{Await, Future}
import scala.concurrent.duration.Duration
import scala.util.{Success, Failure}
import jigg.util.XMLUtil.RichNode

trait ExternalProcessSentencesAnnotator extends Annotator {

  // each local annotator runs in serial
  trait LocalAnnotator extends SentencesAnnotator {
    override final def nThreads = 1
  }

  def mkLocalAnnotator(): LocalAnnotator

  val localAnnotators = (0 until nThreads).map(_=>mkLocalAnnotator())

  override def annotate(annotation: Node): Node = {
    val documentSeq = annotation \\ "document"
    val sentencesSeq = documentSeq map { d => (d \ "sentences").head } // seq of "sentences" nodes
    val sentenceSeqs = sentencesSeq map (_ \ "sentence") // seq of "seq of sentences"

    val sentences = sentenceSeqs.flatten
    val offsets = sentenceSeqs.map(_.size).scanLeft(0)(_+_)

    val annotatedSentences = annotateInParallel(sentences)

    val newDocumentSeq: Seq[Node] = (0 until documentSeq.size) map { docidx =>
      val begin = offsets(docidx)
      val end = offsets(docidx + 1)
      val newSentenceSeqInDoc = (begin until end) map (annotatedSentences)
      val newSentences = sentencesSeq(docidx) replaceChild newSentenceSeqInDoc

      documentSeq(docidx) addOrOverwriteChild newSentences
    }
    annotation replaceChild newDocumentSeq
  }

  def annotateInParallel(sentences: Seq[Node]): Seq[Node] = {
    val dividedSentences = divideBy(sentences, nThreads)
    assert(dividedSentences.size == nThreads)
    assert(localAnnotators.size == nThreads)

    implicit val context = scala.concurrent.ExecutionContext.global

    val maybeAnnotatedSentences: Seq[Future[Seq[Node]]] =
      dividedSentences zip localAnnotators map {
        case (sentences, ann) => Future(sentences map ann.newSentenceAnnotation) recover {
          case e => annotateErrorToBatch(sentences, e)
        }
      }
    for (a <- maybeAnnotatedSentences) Await.ready(a, Duration.Inf)
    maybeAnnotatedSentences.map(_.value.get.get).flatten

    // (dividedSentences zip maybeAnnotatedSentences).map { case (d, m) =>
    //   m onComplete {
    //     case Success(batchResult) => batchResult
    //     case Failure(error) => annotateErrorToBatch(d, error)
    //   }
    // }.flatten
  }

  def divideBy(sentences: Seq[Node], n: Int): Seq[Seq[Node]] = {
    val divided: Seq[Seq[Node]] =
      if (sentences.size < n) sentences +: Array.fill(n-1)(Seq[Node]())
      else sentences.grouped(sentences.size / n).toIndexedSeq
    assert(divided.size == n || divided.size == n + 1)
    if (divided.size == n + 1) divided.take(n - 1) :+ (divided(n - 1) ++ divided(n))
    else divided
  }

  // TODO
  private def annotateErrorToBatch(original: Seq[Node], error: Throwable): Seq[Node] = {
    original
  }
}<commit_msg>Fix bugs; now mecab can run in parallel in server<commit_after>package jigg.pipeline

/*
 Copyright 2013-2015 <NAME>

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

     http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
*/

import scala.xml.{Node, Elem}
import scala.concurrent.{Await, Future}
import scala.concurrent.duration.Duration
import scala.util.{Success, Failure}
import jigg.util.XMLUtil.RichNode

trait ExternalProcessSentencesAnnotator extends Annotator { self=>

  // each local annotator runs in serial
  trait LocalAnnotator extends SentencesAnnotator {
    override def name = self.name
    override final def nThreads = 1
  }

  def mkLocalAnnotator(): LocalAnnotator

  // We don't create local annotator here, since each local annotator may depend on
  // some variables that are not yet instantitated; e.g., `command` in `LocalMecabAnnotator`.
  lazy val localAnnotators = (0 until nThreads).map(_=>mkLocalAnnotator())

  override def annotate(annotation: Node): Node = {
    val documentSeq = annotation \\ "document"
    val sentencesSeq = documentSeq map { d => (d \ "sentences").head } // seq of "sentences" nodes
    val sentenceSeqs = sentencesSeq map (_ \ "sentence") // seq of "seq of sentences"

    val sentences = sentenceSeqs.flatten
    val offsets = sentenceSeqs.map(_.size).scanLeft(0)(_+_)

    val annotatedSentences = annotateInParallel(sentences)

    val newDocumentSeq: Seq[Node] = (0 until documentSeq.size) map { docidx =>
      val begin = offsets(docidx)
      val end = offsets(docidx + 1)
      val newSentenceSeqInDoc = (begin until end) map (annotatedSentences)
      val newSentences = sentencesSeq(docidx) replaceChild newSentenceSeqInDoc

      documentSeq(docidx) addOrOverwriteChild newSentences
    }
    annotation replaceChild newDocumentSeq
  }

  def annotateInParallel(sentences: Seq[Node]): Seq[Node] = {
    val dividedSentences = divideBy(sentences, nThreads)
    assert(dividedSentences.size == nThreads)
    assert(localAnnotators.size == nThreads)

    implicit val context = scala.concurrent.ExecutionContext.global

    val maybeAnnotatedSentences: Seq[Future[Seq[Node]]] =
      dividedSentences zip localAnnotators map {
        case (sentences, ann) => Future(sentences map ann.newSentenceAnnotation) recover {
          case e => annotateErrorToBatch(sentences, e)
        }
      }
    for (a <- maybeAnnotatedSentences) Await.ready(a, Duration.Inf)
    maybeAnnotatedSentences.map(_.value.get.get).flatten
  }

  def divideBy(sentences: Seq[Node], n: Int): Seq[Seq[Node]] = {
    val divided: Seq[Seq[Node]] =
      if (sentences.size < n) sentences +: Array.fill(n-1)(Seq[Node]())
      else sentences.grouped(sentences.size / n).toIndexedSeq
    assert(divided.size == n || divided.size == n + 1)
    if (divided.size == n + 1) divided.take(n - 1) :+ (divided(n - 1) ++ divided(n))
    else divided
  }

  // TODO
  private def annotateErrorToBatch(original: Seq[Node], error: Throwable): Seq[Node] = {
    original
  }
}
{'fraction_non_alphanumeric': 0.10086244701067096, 'fraction_numerical': 0.00643180821517322, 'mean_word_length': 3.751388888888889, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, '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': '5832458', 'n_tokens_mistral': 2130, 'n_tokens_neox': 2039, 'n_words': 609}