<commit_before>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog.shard

import com.precog.yggdrasil.YggConfigComponent

import com.precog.common.jobs._
import com.precog.common.security._
import com.precog.daze.QueryLogger

import blueeyes.util.Clock
import blueeyes.json._

import com.weiglewilczek.slf4s.Logging

import java.util.concurrent.atomic.AtomicBoolean

import org.joda.time.DateTime

import akka.dispatch.{ Future, ExecutionContext }
import akka.actor.{ ActorSystem, Cancellable }
import akka.util.Duration

import scalaz._

/**
 * A `QueryCancelledException` is thrown in a `Future` to indicate that the
 * query terminated abnormally because it was cancelled from the outside.
 */
case class QueryCancelledException(msg: String) extends Exception(msg)

/**
 * A `QueryExpiredException` is thrown in a `Future` to indicate that the
 * query terminated abnormally because its allotted time had expired.
 */
case class QueryExpiredException(msg: String) extends Exception(msg)

trait ManagedQueryModuleConfig {

  /** Determines how often the Jobs service is polled for status updates. */
  def jobPollFrequency: Duration

  def clock: Clock
}

/**
 * A managed query is a query that is run in the context of a job. This context
 * is modelled as a monad transformer. Basically, this gives the world outside
 * the monad the ability to affect the world inside of it. It is not pure, but
 * can be useful as it allows the query to be affected by cancellation requests,
 * for instance.
 *
 * The main idea is to run the query itself within a `ShardQuery[+_]`. The
 * monad instance for a `ShardQuery` is obtained through `createJob(...)`. This
 * is then stripped off by `completeJob(...)`.
 *
 * Note that if the job service is down, then a ShardQueryMonad will still be
 * returned, it just won't actually have a job associated with it and, thus,
 * cannot be cancelled (though timeouts still work fine).
 */
trait ManagedQueryModule extends YggConfigComponent with Logging {
  import scalaz.syntax.monad._
  import JobQueryState._

  type YggConfig <: ManagedQueryModuleConfig

  trait ShardQueryMonad extends QueryTMonad[JobQueryState, Future] with QueryTHoist[JobQueryState] {
    def jobId: Option[JobId]
    def Q: JobQueryStateMonad
  }

  def defaultTimeout: Duration

  /**
   * A mix-in for `QueryLogger`s that forcefully aborts a shard query on fatal
   * errors.
   */
  trait ShardQueryLogger[M[+_], P] extends QueryLogger[M, P] {
    def M: ShardQueryMonad

    abstract override def die(): M[Unit] = {
      M.Q.abort()
      super.die()
    }
  }

  implicit def jobActorSystem: ActorSystem

  def jobManager: JobManager[Future]

  /**
   * Creates a new job with the given apiKey, and uses that to construct a
   * ShardQueryMonad that is bound to this job. The behaviour of the
   * ShardQueryMonad depends on the job state and allows computations running
   * within the monad to be cancelled, expired, etc.
   *
   * Queries that are run within the ShardQueryMonad should be completed with
   * `completeJob` to ensure the job is put into a terminal state when the
   * query completes.
   */
  def createJob(apiKey: APIKey, data: Option[JValue], timeout: Option[Duration])(implicit asyncContext: ExecutionContext): Future[ShardQueryMonad] = {
    val start = System.currentTimeMillis
    val futureJob = jobManager.createJob(apiKey, "Quirrel Query", "shard-query", data, Some(yggConfig.clock.now())).map {
      j => logger.debug("Job created in %d ms".format(System.currentTimeMillis - start)); j
    }
    for {
      job <- futureJob map { job => Some(job) } recover { case _ => None }
      queryStateManager = job map { job =>
        val mgr = JobQueryStateManager(job.id, yggConfig.clock.now() plus timeout.getOrElse(defaultTimeout).toMillis)
        mgr.start()
        mgr
      } getOrElse FakeJobQueryStateManager(yggConfig.clock.now() plus timeout.getOrElse(defaultTimeout).toMillis)
    } yield {
      new ShardQueryMonad {
        val jobId = job map (_.id)
        val Q: JobQueryStateMonad = queryStateManager
        val M: Monad[Future] = new blueeyes.bkka.FutureMonad(asyncContext)
      }
    }
  }

  /**<commit_msg>Minor change based on kjn feedback<commit_after>/*
 *  ____    ____    _____    ____    ___     ____ 
 * |  _ \  |  _ \  | ____|  / ___|  / _/    / ___|        Precog (R)
 * | |_) | | |_) | |  _|   | |     | |  /| | |  _         Advanced Analytics Engine for NoSQL Data
 * |  __/  |  _ <  | |___  | |___  |/ _| | | |_| |        Copyright (C) 2010 - 2013 SlamData, Inc.
 * |_|     |_| \_\ |_____|  \____|   /__/   \____|        All Rights Reserved.
 *
 * This program is free software: you can redistribute it and/or modify it under the terms of the 
 * GNU Affero General Public License as published by the Free Software Foundation, either version 
 * 3 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; 
 * without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See 
 * the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License along with this 
 * program. If not, see <http://www.gnu.org/licenses/>.
 *
 */
package com.precog.shard

import com.precog.yggdrasil.YggConfigComponent

import com.precog.common.jobs._
import com.precog.common.security._
import com.precog.daze.QueryLogger

import blueeyes.util.Clock
import blueeyes.json._

import com.weiglewilczek.slf4s.Logging

import java.util.concurrent.atomic.AtomicBoolean

import org.joda.time.DateTime

import akka.dispatch.{ Future, ExecutionContext }
import akka.actor.{ ActorSystem, Cancellable }
import akka.util.Duration

import scalaz._

/**
 * A `QueryCancelledException` is thrown in a `Future` to indicate that the
 * query terminated abnormally because it was cancelled from the outside.
 */
case class QueryCancelledException(msg: String) extends Exception(msg)

/**
 * A `QueryExpiredException` is thrown in a `Future` to indicate that the
 * query terminated abnormally because its allotted time had expired.
 */
case class QueryExpiredException(msg: String) extends Exception(msg)

trait ManagedQueryModuleConfig {

  /** Determines how often the Jobs service is polled for status updates. */
  def jobPollFrequency: Duration

  def clock: Clock
}

/**
 * A managed query is a query that is run in the context of a job. This context
 * is modelled as a monad transformer. Basically, this gives the world outside
 * the monad the ability to affect the world inside of it. It is not pure, but
 * can be useful as it allows the query to be affected by cancellation requests,
 * for instance.
 *
 * The main idea is to run the query itself within a `ShardQuery[+_]`. The
 * monad instance for a `ShardQuery` is obtained through `createJob(...)`. This
 * is then stripped off by `completeJob(...)`.
 *
 * Note that if the job service is down, then a ShardQueryMonad will still be
 * returned, it just won't actually have a job associated with it and, thus,
 * cannot be cancelled (though timeouts still work fine).
 */
trait ManagedQueryModule extends YggConfigComponent with Logging {
  import scalaz.syntax.monad._
  import JobQueryState._

  type YggConfig <: ManagedQueryModuleConfig

  trait ShardQueryMonad extends QueryTMonad[JobQueryState, Future] with QueryTHoist[JobQueryState] {
    def jobId: Option[JobId]
    def Q: JobQueryStateMonad
  }

  def defaultTimeout: Duration

  /**
   * A mix-in for `QueryLogger`s that forcefully aborts a shard query on fatal
   * errors.
   */
  trait ShardQueryLogger[M[+_], P] extends QueryLogger[M, P] {
    def M: ShardQueryMonad

    abstract override def die(): M[Unit] = {
      M.Q.abort()
      super.die()
    }
  }

  implicit def jobActorSystem: ActorSystem

  def jobManager: JobManager[Future]

  /**
   * Creates a new job with the given apiKey, and uses that to construct a
   * ShardQueryMonad that is bound to this job. The behaviour of the
   * ShardQueryMonad depends on the job state and allows computations running
   * within the monad to be cancelled, expired, etc.
   *
   * Queries that are run within the ShardQueryMonad should be completed with
   * `completeJob` to ensure the job is put into a terminal state when the
   * query completes.
   */
  def createJob(apiKey: APIKey, data: Option[JValue], timeout: Option[Duration])(implicit asyncContext: ExecutionContext): Future[ShardQueryMonad] = {
    val start = System.currentTimeMillis
    val futureJob = jobManager.createJob(apiKey, "Quirrel Query", "shard-query", data, Some(yggConfig.clock.now())).onComplete {
      _ => logger.debug("Job created in %d ms".format(System.currentTimeMillis - start))
    }
    for {
      job <- futureJob map { job => Some(job) } recover { case _ => None }
      queryStateManager = job map { job =>
        val mgr = JobQueryStateManager(job.id, yggConfig.clock.now() plus timeout.getOrElse(defaultTimeout).toMillis)
        mgr.start()
        mgr
      } getOrElse FakeJobQueryStateManager(yggConfig.clock.now() plus timeout.getOrElse(defaultTimeout).toMillis)
    } yield {
      new ShardQueryMonad {
        val jobId = job map (_.id)
        val Q: JobQueryStateMonad = queryStateManager
        val M: Monad[Future] = new blueeyes.bkka.FutureMonad(asyncContext)
      }
    }
  }

  /**
{'fraction_non_alphanumeric': 0.08889101521385513, 'fraction_numerical': 0.001913692469620132, 'mean_word_length': 3.3207937164117403, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5031169', 'n_tokens_mistral': 3181, 'n_tokens_neox': 3077, 'n_words': 1113}