经验估计大时间效率

algorithm java big-o scala

2022-09-01 02:40:35

背景

我想通过基准测试来估计库中某些方法的big oh性能。我不需要精确度 - 它足以证明某些东西是O（1），O（logn），O（n），O（nlogn），O（n^2）或比这更差。由于 big-oh 表示上限，因此估计 O（logn）的 O（logn）不是问题。

现在，我正在考虑找到最适合每个big-oh的数据的常数乘数k（但将高于所有结果），然后选择最合适的big-oh。

问题

有没有比我所追求的更好的方法来做到这一点？如果是这样，它们是什么？
否则，任何人都可以向我指出算法来估计k以获得最佳拟合，并比较每条曲线与数据的拟合程度？

注释和约束

鉴于到目前为止的评论，我需要明确一些事情：

这需要自动化。我不能“看”数据并做出判断。
我将对具有多种大小的方法进行基准测试。对于每种尺寸，我将使用一个经过验证的基准框架来提供可靠的统计结果。nn
实际上，我事先知道将要测试的大多数方法中的大杂烩。我的主要目的是为他们提供性能回归测试。
代码将用Scala编写，任何免费的Java库都可以使用。

例

这里有一个我想要衡量的东西的例子。我有一个带有此签名的方法：

def apply(n: Int): A

给定一个，它将返回序列的第 n 个元素。给定现有实现，此方法可以具有 O（1）、O（logn）或 O（n），并且稍加更改可能会使它错误地使用次优实现。或者，更容易的是，可以获得一些依赖于它的其他方法来使用它的次优版本。n

答案 1

为了开始，你必须做出一些假设。

n与任何常量项相比，它很大。
您可以有效地随机化输入数据
您可以以足够的密度进行采样，以便很好地处理运行时的分布

特别是，（3）很难与（1）一致实现。因此，您可能会遇到指数级最坏情况，但永远不会遇到最坏的情况，因此认为您的算法比平均水平要好得多。

话虽如此，您所需要的只是任何标准曲线拟合库。Apache Commons Math有一个完全足够的。然后，您可以使用要测试的所有常用项（例如常量，log n，n，n log n，nn，nn*n，e^n）创建一个函数，或者获取数据的对数并拟合指数，然后，如果您得到一个不接近整数的指数，请查看抛入对数n是否能提供更好的拟合。

（更详细地说，如果你适合和，或者更容易，你可以得到指数;在一次所有通用项的方案中，你将获得每个项的权重，所以如果你有大的地方，你也会选择这个项。C*x^aCalog C + a log xan*n + C*n*log(n)C

您需要将大小变化足够大，以便您可以区分不同的情况（如果您关心这些情况，则使用日志项可能很难），并且比参数更安全地具有不同的大小（可能3x过量会开始正常，只要您至少运行十几次左右）。

编辑：这是Scala代码，为您完成所有这些工作。与其解释每一小块，我会把它留给你去调查;它使用 C*x^a 拟合实现上述方案，并返回（（a，C），（a 的下限，a 的上限））。边界是相当保守的，正如你从运行这个东西几次可以看出的那样。的单位是秒（是无单位的），但不要太相信这一点，因为有一些循环开销（还有一些噪音）。Ca

class TimeLord[A: ClassManifest,B: ClassManifest](setup: Int => A, static: Boolean = true)(run: A => B) {
  @annotation.tailrec final def exceed(time: Double, size: Int, step: Int => Int = _*2, first: Int = 1): (Int,Double) = {
    var i = 0
    val elapsed = 1e-9 * {
      if (static) {
        val a = setup(size)
        var b: B = null.asInstanceOf[B]
        val t0 = System.nanoTime
        var i = 0
        while (i < first) {
          b = run(a)
          i += 1
        }
        System.nanoTime - t0
      }
      else {
        val starts = if (static) { val a = setup(size); Array.fill(first)(a) } else Array.fill(first)(setup(size))
        val answers = new Array[B](first)
        val t0 = System.nanoTime
        var i = 0
        while (i < first) {
          answers(i) = run(starts(i))
          i += 1
        }
        System.nanoTime - t0
      }
    }
    if (time > elapsed) {
      val second = step(first)
      if (second <= first) throw new IllegalArgumentException("Iteration size increase failed: %d to %d".format(first,second))
      else exceed(time, size, step, second)
    }
    else (first, elapsed)
  }

  def multibench(smallest: Int, largest: Int, time: Double, n: Int, m: Int = 1) = {
    if (m < 1 || n < 1 || largest < smallest || (n>1 && largest==smallest)) throw new IllegalArgumentException("Poor choice of sizes")
    val frac = (largest.toDouble)/smallest
    (0 until n).map(x => (smallest*math.pow(frac,x/((n-1).toDouble))).toInt).map{ i => 
      val (k,dt) = exceed(time,i)
      if (m==1) i -> Array(dt/k) else {
        i -> ( (dt/k) +: (1 until m).map(_ => exceed(time,i,first=k)).map{ case (j,dt2) => dt2/j }.toArray )
      }
    }.foldLeft(Vector[(Int,Array[Double])]()){ (acc,x) =>
      if (acc.length==0 || acc.last._1 != x._1) acc :+ x
      else acc.dropRight(1) :+ (x._1, acc.last._2 ++ x._2)
    }
  }

  def alpha(data: Seq[(Int,Array[Double])]) = {
    // Use Theil-Sen estimator for calculation of straight-line fit for exponent
    // Assume timing relationship is t(n) = A*n^alpha
    val dat = data.map{ case (i,ad) => math.log(i) -> ad.map(x => math.log(i) -> math.log(x)) }
    val slopes = (for {
      i <- dat.indices
      j <- ((i+1) until dat.length)
      (pi,px) <- dat(i)._2
      (qi,qx) <- dat(j)._2
    } yield (qx - px)/(qi - pi)).sorted
    val mbest = slopes(slopes.length/2)
    val mp05 = slopes(slopes.length/20)
    val mp95 = slopes(slopes.length-(1+slopes.length/20))
    val intercepts = dat.flatMap{ case (i,a) => a.map{ case (li,lx) => lx - li*mbest } }.sorted
    val bbest = intercepts(intercepts.length/2)
    ((mbest,math.exp(bbest)),(mp05,mp95))
  }
}

请注意，该方法预计需要大约 sqrt（2）nm*时间才能运行，假设使用了静态初始化数据，并且与您正在运行的任何内容相比相对便宜。以下是一些示例，其中的参数选择为运行 ~15 秒：multibench

val tl1 = new TimeLord(x => List.range(0,x))(_.sum)  // Should be linear
// Try list sizes 100 to 10000, with each run taking at least 0.1s;
// use 10 different sizes and 10 repeats of each size
scala> tl1.alpha( tl1.multibench(100,10000,0.1,10,10) )
res0: ((Double, Double), (Double, Double)) = ((1.0075537890632216,7.061397125245351E-9),(0.8763463348353099,1.102663784225697))

val longList = List.range(0,100000)
val tl2 = new TimeLord(x=>x)(longList.apply)    // Again, should be linear
scala> tl2.alpha( tl2.multibench(100,10000,0.1,10,10) )
res1: ((Double, Double), (Double, Double)) = ((1.4534378213477026,1.1325696181862922E-10),(0.969955396265306,1.8294175293676322))

// 1.45?!  That's not linear.  Maybe the short ones are cached?
scala> tl2.alpha( tl2.multibench(9000,90000,0.1,100,1) )
res2: ((Double, Double), (Double, Double)) = ((0.9973235607566956,1.9214696731124573E-9),(0.9486294398193154,1.0365312207345019))

// Let's try some sorting
val tl3 = new TimeLord(x=>Vector.fill(x)(util.Random.nextInt))(_.sorted)
scala> tl3.alpha( tl3.multibench(100,10000,0.1,10,10) )
res3: ((Double, Double), (Double, Double)) = ((1.1713142886974603,3.882658025586512E-8),(1.0521099621639414,1.3392622111121666))
// Note the log(n) term comes out as a fractional power
// (which will decrease as the sizes increase)

// Maybe sort some arrays?
// This may take longer to run because we have to recreate the (mutable) array each time
val tl4 = new TimeLord(x=>Array.fill(x)(util.Random.nextInt), false)(java.util.Arrays.sort)
scala> tl4.alpha( tl4.multibench(100,10000,0.1,10,10) )
res4: ((Double, Double), (Double, Double)) = ((1.1216172965292541,2.2206198821180513E-8),(1.0929414090177318,1.1543697719880128))

// Let's time something slow
def kube(n: Int) = (for (i <- 1 to n; j <- 1 to n; k <- 1 to n) yield 1).sum
val tl5 = new TimeLord(x=>x)(kube)
scala> tl5.alpha( tl5.multibench(10,100,0.1,10,10) )
res5: ((Double, Double), (Double, Double)) = ((2.8456382116915484,1.0433534274508799E-7),(2.6416659356198617,2.999094292838751))
// Okay, we're a little short of 3; there's constant overhead on the small sizes

无论如何，对于所述用例 - 您正在检查以确保顺序不会更改 - 这可能就足够了，因为在设置测试时，您可以稍微考虑一下这些值，以确保它们提供一些合理的东西。人们也可以创建寻求稳定性的启发式方法，但这可能有点过分了。

（顺便说一句，这里没有明确的预热步骤;Theil-Sen估计器的稳健拟合应该使得对于合理的大型基准测试来说没有必要。这也是为什么我不使用任何其他板凳框架的原因。它所做的任何统计数据都会从此测试中失去功率。

再次编辑：如果将方法替换为以下内容：alpha

  // We'll need this math
  @inline private[this] def sq(x: Double) = x*x
  final private[this] val inv_log_of_2 = 1/math.log(2)
  @inline private[this] def log2(x: Double) = math.log(x)*inv_log_of_2
  import math.{log,exp,pow}

  // All the info you need to calculate a y value, e.g. y = x*m+b
  case class Yp(x: Double, m: Double, b: Double) {}

  // Estimators for data order
  //   fx = transformation to apply to x-data before linear fitting
  //   fy = transformation to apply to y-data before linear fitting
  //   model = given x, slope, and intercept, calculate predicted y
  case class Estimator(fx: Double => Double, invfx: Double=> Double, fy: (Double,Double) => Double, model: Yp => Double) {}
  // C*n^alpha
  val alpha = Estimator(log, exp, (x,y) => log(y), p => p.b*pow(p.x,p.m))
  // C*log(n)*n^alpha
  val logalpha = Estimator(log, exp, (x,y) =>log(y/log2(x)), p => p.b*log2(p.x)*pow(p.x,p.m))

  // Use Theil-Sen estimator for calculation of straight-line fit
  case class Fit(slope: Double, const: Double, bounds: (Double,Double), fracrms: Double) {}
  def theilsen(data: Seq[(Int,Array[Double])], est: Estimator = alpha) = {
    // Use Theil-Sen estimator for calculation of straight-line fit for exponent
    // Assume timing relationship is t(n) = A*n^alpha
    val dat = data.map{ case (i,ad) => ad.map(x => est.fx(i) -> est.fy(i,x)) }
    val slopes = (for {
      i <- dat.indices
      j <- ((i+1) until dat.length)
      (pi,px) <- dat(i)
      (qi,qx) <- dat(j)
    } yield (qx - px)/(qi - pi)).sorted
    val mbest = slopes(slopes.length/2)
    val mp05 = slopes(slopes.length/20)
    val mp95 = slopes(slopes.length-(1+slopes.length/20))
    val intercepts = dat.flatMap{ _.map{ case (li,lx) => lx - li*mbest } }.sorted
    val bbest = est.invfx(intercepts(intercepts.length/2))
    val fracrms = math.sqrt(data.map{ case (x,ys) => ys.map(y => sq(1 - y/est.model(Yp(x,mbest,bbest)))).sum }.sum / data.map(_._2.length).sum)
    Fit(mbest, bbest, (mp05,mp95), fracrms)
  }

然后，当存在对数项时，您可以获得指数的估计值 - 存在错误估计值以选择对数项是否是正确的方法，但由您进行调用（即，我假设您将首先对此进行监督并读取脱落的数字）：

val tl3 = new TimeLord(x=>Vector.fill(x)(util.Random.nextInt))(_.sorted)
val timings = tl3.multibench(100,10000,0.1,10,10)

// Regular n^alpha fit
scala> tl3.theilsen( timings )
res20: tl3.Fit = Fit(1.1811648421030059,3.353753446942075E-8,(1.1100382697696545,1.3204652930525234),0.05927994882343982)

// log(n)*n^alpha fit--note first value is closer to an integer
//   and last value (error) is smaller
scala> tl3.theilsen( timings, tl3.logalpha )
res21: tl3.Fit = Fit(1.0369167329732445,9.211366397621766E-9,(0.9722967182484441,1.129869067913768),0.04026308919615681)

（编辑：修复了RMS计算，所以它实际上是平均值，再加上证明你只需要做一次计时，然后可以尝试两个拟合。

答案 2

我不认为你的方法总体上是行不通的。

问题在于，“大O”复杂性是基于一个极限，因为一些缩放变量趋向于无穷大。对于该变量的较小值，性能行为可能看起来完全适合不同的曲线。

问题在于，使用经验方法，您永远无法知道缩放变量是否足够大，以使极限在结果中显而易见。

另一个问题是，如果你在Java / Scala中实现它，你必须花费相当大的时间来消除由于JVM预热（例如.class加载，JIT编译，堆大小调整）和垃圾回收等因素而导致的时序失真和“噪音”。

最后，没有人会相信对复杂性的经验估计。或者至少，如果他们理解复杂性分析的数学，他们就不会这样做。

随访

针对这一评论：

您使用的样本越多、越大，估算值的重要性将显著提高。

这是真的，尽管我的观点是你（丹尼尔）没有考虑到这一点。

此外，运行时函数通常具有可以利用的特殊特征;例如，算法往往不会在某个巨大的n上改变它们的行为。

对于简单的情况，是的。

对于复杂的案例和现实世界的案例，这是一个可疑的假设。例如：

假设某个算法使用具有大而固定大小的主哈希数组的哈希表，并使用外部列表来处理冲突。对于小于主哈希数组大小的 N（== 条目数），大多数操作的行为将显示为。只有当 N 大于该值时，才能通过曲线拟合来检测真实行为。O(1)O(N)
假设该算法使用大量内存或网络带宽。通常，它会很好地工作，直到你达到资源限制，然后性能会严重下降。您如何解释这一点？如果它是“经验复杂性”的一部分，你如何确保你到达过渡点？如果你想排除它，你该怎么做？