<commit_before>/*
 * Copyright 2014-2017 Netflix, Inc.
 *
 * 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.
 */
package com.netflix.atlas.lwcapi

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit

import com.github.benmanes.caffeine.cache.CacheLoader
import com.github.benmanes.caffeine.cache.Caffeine
import com.netflix.atlas.core.model.DataExpr
import com.netflix.atlas.core.model.ModelExtractors
import com.netflix.atlas.core.model.Query
import com.netflix.atlas.core.model.Query.KeyQuery
import com.netflix.atlas.core.model.StyleVocabulary
import com.netflix.atlas.core.stacklang.Interpreter

import scala.util.Failure
import scala.util.Success
import scala.util.Try

/**
  * Splits a complete graph expression (StyleExpr) string into a set of subscriptions. Each
  * subscription is based on the underlying data expressions (DataExpr) that get pushed back
  * to the systems supplying data to LWCAPI.
  */
class ExpressionSplitter {

  import ExpressionSplitter._

  private val keepKeys = Set("nf.app", "nf.stack", "nf.cluster")

  private val interpreter = Interpreter(StyleVocabulary.allWords)

  /**
    * Processing the expressions can be quite expensive. In particular compiling regular
    * expressions to ensure they are valid. Generally the set of expressions should not
    * vary much over time and the evaluator library will regularly submit the full list
    * to sync with. This cache prevents the reprocessing for expressions that have already
    * been seen recently.
    */
  private val exprCache = {
    val loader = new CacheLoader[String, Try[List[DataExprMeta]]] {
      def load(expr: String): Try[List[DataExprMeta]] = parse(expr)
    }
    Caffeine
      .newBuilder()
      .expireAfterAccess(10, TimeUnit.MINUTES)
      .build(loader)
  }

  // TODO: https://github.com/Netflix/atlas/issues/729
  private val interner = new ConcurrentHashMap[Query, Query]()

  private[lwcapi] def intern(query: Query): Query = {
    query match {
      case Query.True =>
        query
      case Query.False =>
        query
      case q: Query.Equal =>
        interner.computeIfAbsent(q, _ => Query.Equal(q.k.intern(), q.v.intern()))
      case q: Query.LessThan =>
        interner.computeIfAbsent(q, _ => Query.LessThan(q.k.intern(), q.v.intern()))
      case q: Query.LessThanEqual =>
        interner.computeIfAbsent(q, _ => Query.LessThanEqual(q.k.intern(), q.v.intern()))
      case q: Query.GreaterThan =>
        interner.computeIfAbsent(q, _ => Query.GreaterThan(q.k.intern(), q.v.intern()))
      case q: Query.GreaterThanEqual =>
        interner.computeIfAbsent(q, _ => Query.GreaterThanEqual(q.k.intern(), q.v.intern()))
      case q: Query.Regex =>
        interner.computeIfAbsent(q, _ => Query.Regex(q.k.intern(), q.v.intern()))
      case q: Query.RegexIgnoreCase =>
        interner.computeIfAbsent(q, _ => Query.RegexIgnoreCase(q.k.intern(), q.v.intern()))
      case q: Query.In =>
        interner.computeIfAbsent(q, _ => Query.In(q.k.intern(), q.vs.map(_.intern())))
      case q: Query.HasKey =>
        interner.computeIfAbsent(q, _ => Query.HasKey(q.k.intern()))
      case q: Query.And =>
        interner.computeIfAbsent(q, _ => Query.And(intern(q.q1), intern(q.q2)))
      case q: Query.Or =>
        interner.computeIfAbsent(q, _ => Query.Or(intern(q.q1), intern(q.q2)))
      case q: Query.Not =>
        interner.computeIfAbsent(q, _ => Query.Not(intern(q.q)))
    }
  }

  private def parse(expression: String): Try[List[DataExprMeta]] = Try {
    val context = interpreter.execute(expression)
    val dataExprs = context.stack.flatMap {
      case ModelExtractors.PresentationType(t) => t.expr.dataExprs
      case _                                   => throw new IllegalArgumentException("expression is invalid")
    }

    // Offsets are not supported
    dataExprs.foreach { dataExpr =>
      if (!dataExpr.offset.isZero) {
        throw new IllegalArgumentException(
          s":offset not supported for streaming evaluation [[$dataExpr]]"
        )
      }
    }

    dataExprs.distinct.map { e =>
      val q = intern(compress(e.query))
      DataExprMeta(e, e.toString, q)
    }
  }

  def split(expression: String, frequency: Long): List[Subscription] = {
    exprCache.get(expression) match {
      case Success(dataExprs: List[DataExpr]) => dataExprs.map(e => toSubscription(e, frequency))
      case Failure(t)                         => throw t
    }
  }

  private def toSubscription(meta: DataExprMeta, frequency: Long): Subscription = {<commit_msg>reduce thread contention for splitter cache (#732)<commit_after>/*
 * Copyright 2014-2017 Netflix, Inc.
 *
 * 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.
 */
package com.netflix.atlas.lwcapi

import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.TimeUnit

import com.github.benmanes.caffeine.cache.Caffeine
import com.netflix.atlas.core.model.DataExpr
import com.netflix.atlas.core.model.ModelExtractors
import com.netflix.atlas.core.model.Query
import com.netflix.atlas.core.model.Query.KeyQuery
import com.netflix.atlas.core.model.StyleVocabulary
import com.netflix.atlas.core.stacklang.Interpreter
import com.netflix.spectator.api.Utils

import scala.util.Failure
import scala.util.Success
import scala.util.Try

/**
  * Splits a complete graph expression (StyleExpr) string into a set of subscriptions. Each
  * subscription is based on the underlying data expressions (DataExpr) that get pushed back
  * to the systems supplying data to LWCAPI.
  */
class ExpressionSplitter {

  import ExpressionSplitter._

  private val keepKeys = Set("nf.app", "nf.stack", "nf.cluster")

  private val interpreter = Interpreter(StyleVocabulary.allWords)

  /**
    * Processing the expressions can be quite expensive. In particular compiling regular
    * expressions to ensure they are valid. Generally the set of expressions should not
    * vary much over time and the evaluator library will regularly submit the full list
    * to sync with. This cache prevents the reprocessing for expressions that have already
    * been seen recently.
    *
    * Note: do not use LoadingCache, see `getFromCache`.
    */
  private val exprCache = Caffeine
    .newBuilder()
    .expireAfterAccess(10, TimeUnit.MINUTES)
    .build[String, Try[List[DataExprMeta]]]()

  // TODO: https://github.com/Netflix/atlas/issues/729
  private val interner = new ConcurrentHashMap[Query, Query]()

  /**
    * Avoid using `ConcurrentHashMap.computeIfAbsent` here, on some of the instance types
    * with more cores it causes a heavy thread contention. It is better for this use-case
    * to spend some additional CPU cycles computing a value that will not get used.
    */
  private def internQuery(q: Query, newQuery: => Query): Query = {
    Utils.computeIfAbsent[Query, Query](interner, q, _ => newQuery)
  }

  private[lwcapi] def intern(query: Query): Query = {
    query match {
      case Query.True =>
        query
      case Query.False =>
        query
      case q: Query.Equal =>
        internQuery(q, Query.Equal(q.k.intern(), q.v.intern()))
      case q: Query.LessThan =>
        internQuery(q, Query.LessThan(q.k.intern(), q.v.intern()))
      case q: Query.LessThanEqual =>
        internQuery(q, Query.LessThanEqual(q.k.intern(), q.v.intern()))
      case q: Query.GreaterThan =>
        internQuery(q, Query.GreaterThan(q.k.intern(), q.v.intern()))
      case q: Query.GreaterThanEqual =>
        internQuery(q, Query.GreaterThanEqual(q.k.intern(), q.v.intern()))
      case q: Query.Regex =>
        internQuery(q, Query.Regex(q.k.intern(), q.v.intern()))
      case q: Query.RegexIgnoreCase =>
        internQuery(q, Query.RegexIgnoreCase(q.k.intern(), q.v.intern()))
      case q: Query.In =>
        internQuery(q, Query.In(q.k.intern(), q.vs.map(_.intern())))
      case q: Query.HasKey =>
        internQuery(q, Query.HasKey(q.k.intern()))
      case q: Query.And =>
        internQuery(q, Query.And(intern(q.q1), intern(q.q2)))
      case q: Query.Or =>
        internQuery(q, Query.Or(intern(q.q1), intern(q.q2)))
      case q: Query.Not =>
        internQuery(q, Query.Not(intern(q.q)))
    }
  }

  private def parse(expression: String): Try[List[DataExprMeta]] = Try {
    val context = interpreter.execute(expression)
    val dataExprs = context.stack.flatMap {
      case ModelExtractors.PresentationType(t) => t.expr.dataExprs
      case _                                   => throw new IllegalArgumentException("expression is invalid")
    }

    // Offsets are not supported
    dataExprs.foreach { dataExpr =>
      if (!dataExpr.offset.isZero) {
        throw new IllegalArgumentException(
          s":offset not supported for streaming evaluation [[$dataExpr]]"
        )
      }
    }

    dataExprs.distinct.map { e =>
      val q = intern(compress(e.query))
      DataExprMeta(e, e.toString, q)
    }
  }

  /**
    * On instance types with a lot of cores, the loading cache causes a lot of thread
    * contention and most threads are blocked. This just does and get/put which potentially
    * recomputes some values, but for this case that is preferable.
    */
  private def getFromCache(k: String): Try[List[DataExprMeta]] = {
    val value = exprCache.getIfPresent(k)
    if (value == null) {
      val tmp = parse(k)
      exprCache.put(k, tmp)
      tmp
    } else {
      value
    }
  }

  def split(expression: String, frequency: Long): List[Subscription] = {
    getFromCache(expression) match {
      case Success(exprs: List[_]) =>
        val dataExprs = exprs.asInstanceOf[List[DataExpr]]
        exprs.map(e => toSubscription(e, frequency))
      case Failure(t) => throw t
    }
  }

  private def toSubscription(meta: DataExprMeta, frequency: Long): Subscription = {
{'fraction_non_alphanumeric': 0.11301687372519933, 'fraction_numerical': 0.004172074911922863, 'mean_word_length': 3.160046278441959, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 59, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6990139', 'n_tokens_mistral': 3256, 'n_tokens_neox': 3199, 'n_words': 901}