<commit_before>package utils

import scala.concurrent.ExecutionContext
import play.api.libs.iteratee._
import scala.concurrent.{ExecutionContext, Future, Promise}
import play.api.libs.iteratee.{Cont, Error}
import scala.util.{Try, Failure, Success}
import akka.agent.Agent
import org.mapdb.DBMaker
import org.mapdb.Serializer
import java.io._
import scala.util.Random

object Buffering {
  def diskBuffer(e: Enumerator[Array[Byte]])(implicit ec: ExecutionContext): Enumerator[Array[Byte]] = {
    val buffer = new Buffer()
    (e run Iteratee.foreach(buffer.enqueue))
      .foreach(_ => buffer.finish)
    Enumerator.generateM(buffer.dequeue).onDoneEnumerating(buffer.close)
  }


  class Buffer {

    val db = DBMaker.newTempFileDB
      .asyncWriteEnable
      .transactionDisable
      .deleteFilesAfterClose
      .closeOnJvmShutdown
      .make
    val queue = db.createQueue("buffer", Serializer.BYTE_ARRAY, false)
    val pending = scala.collection.mutable.Queue.empty[Promise[Option[Array[Byte]]]]
    var isDone = false

    def enqueue(v: Array[Byte]) =
      this.synchronized {
        pending.dequeueFirst(_ => true) match {
          case Some(p) => p.success(Some(v))
          case None => addToQueue(v)
        }
      }

    def dequeue: Future[Option[Array[Byte]]] =
      this.synchronized {
        takeFromQueue match {
          case Some(v) => Future.successful(Some(v))
          case None if isDone => Future.successful(None)
          case None if pending.isEmpty =>
            val p = Promise[Option[Array[Byte]]]()
            pending.enqueue(p)
            p.future
          case e =>
            println(e)
            throw new RuntimeException("Unhandled state")
        }
      }

    def finish = this.synchronized {
      isDone = true
      pending.foreach(_.success(takeFromQueue))
    }

    def close = db.close

    private def addToQueue(v: Array[Byte]) { if (!db.isClosed) queue.add(v) }
    private def takeFromQueue: Option[Array[Byte]] = Option(queue.poll)
  }

}<commit_msg>Using unbounded thread pool for buffer<commit_after>package utils

import play.api.libs.iteratee._
import scala.concurrent.{ExecutionContext, Future, Promise}
import play.api.libs.iteratee.{Cont, Error}
import scala.util.{Try, Failure, Success}
import akka.agent.Agent
import org.mapdb.DBMaker
import org.mapdb.Serializer
import java.io._
import scala.util.Random
import java.util.concurrent.Executors

object Buffering {

  implicit val ec = ExecutionContext.fromExecutorService(
      Executors.newCachedThreadPool)

  def diskBuffer(e: Enumerator[Array[Byte]]): Enumerator[Array[Byte]] = {
    val buffer = new Buffer()
    (e run Iteratee.foreach(buffer.enqueue))
      .foreach(_ => buffer.finish)
    Enumerator.generateM(buffer.dequeue).onDoneEnumerating(buffer.close)
  }


  class Buffer {

    val db = DBMaker.newTempFileDB
      .asyncWriteEnable
      .transactionDisable
      .deleteFilesAfterClose
      .closeOnJvmShutdown
      .make
    val queue = db.createQueue("buffer", Serializer.BYTE_ARRAY, false)
    val pending = scala.collection.mutable.Queue.empty[Promise[Option[Array[Byte]]]]
    var isDone = false

    def enqueue(v: Array[Byte]) =
      this.synchronized {
        pending.dequeueFirst(_ => true) match {
          case Some(p) => p.success(Some(v))
          case None => addToQueue(v)
        }
      }

    def dequeue: Future[Option[Array[Byte]]] =
      this.synchronized {
        takeFromQueue match {
          case Some(v) => Future.successful(Some(v))
          case None if isDone => Future.successful(None)
          case None if pending.isEmpty =>
            val p = Promise[Option[Array[Byte]]]()
            pending.enqueue(p)
            p.future
          case e =>
            println(e)
            throw new RuntimeException("Unhandled state")
        }
      }

    def finish = this.synchronized {
      isDone = true
      pending.foreach(_.success(takeFromQueue))
    }

    def close = db.close

    private def addToQueue(v: Array[Byte]) { if (!db.isClosed) queue.add(v) }
    private def takeFromQueue: Option[Array[Byte]] = Option(queue.poll)
  }

}
{'fraction_non_alphanumeric': 0.11140647655872402, 'fraction_numerical': 0.0, 'mean_word_length': 2.8430826369545033, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '5552914', 'n_tokens_mistral': 1277, 'n_tokens_neox': 1255, 'n_words': 268}