【问题标题】:Trouble with lazy convolution fn in ClojureClojure 中惰性卷积 fn 的问题
【发布时间】:2010-07-15 20:36:30
【问题描述】:

我正在编写一些信号处理软件,我从写一个 discrete convolution function 开始。

这适用于前一万左右的值列表,但随着它们变大(比如 100k),我当然会开始遇到 StackOverflow 错误。

不幸的是,我在将命令式卷积算法转换为 递归和惰性版本 时遇到了很多问题,该算法实际上足够快(至少有一点优雅也会很好)。

我也不是 100% 确定我的这个功能完全正确,但是 - 如果我遗漏了什么/做错了什么,请告诉我。我认为这是正确的。

(defn convolve
  "
    Convolves xs with is.

    This is a discrete convolution.

    'xs  :: list of numbers
    'is  :: list of numbers
  "
  [xs is]
  (loop [xs xs finalacc () acc ()]
    (if (empty? xs)
      (concat finalacc acc)
      (recur (rest xs)
             (if (empty? acc)
               ()
               (concat finalacc [(first acc)]))
             (if (empty? acc)
               (map #(* (first xs) %) is)
               (vec-add
                (map #(* (first xs) %) is)
                (rest acc)))))))

我将非常感谢任何形式的帮助:我仍在了解 Clojure 的方位,并且使这种优雅、懒惰和/或递归变得非常棒。

我有点惊讶,用 Lisp 的命令式语言表达一个相当容易表达的算法是多么困难。但也许我做错了!

编辑: 只是为了展示用命令式语言表达是多么容易,并为人们提供运行良好且易于阅读的算法,这里是 Python 版本。除了更短、更简洁、更容易推理之外,它的执行速度比 Clojure 代码快几个数量级:甚至是我使用 Java 数组的命令式 Clojure 代码。

from itertools import repeat

def convolve(ns, ms):
    y = [i for i in repeat(0, len(ns)+len(ms)-1)]
    for n in range(len(ns)):
        for m in range(len(ms)):
            y[n+m] = y[n+m] + ns[n]*ms[m]
    return y

另一方面,这里是命令式 Clojure 代码。它还从卷积中丢弃最后一个非完全沉浸的值。因此,除了缓慢和丑陋之外,它并不是 100% 的功能。也不实用。

(defn imp-convolve-1
  [xs is]
  (let [ys (into-array Double (repeat (dec (+ (count xs) (count is))) 0.0))
        xs (vec xs)
        is (vec is)]
     (map #(first %)
          (for [i (range (count xs))]
            (for [j (range (count is))]
              (aset ys (+ i j)
                    (+ (* (nth xs i) (nth is j))
                       (nth ys (+ i j)))))))))

这太令人沮丧了。拜托,有人告诉我我刚刚错过了一些明显的东西。

编辑 3:

这是我昨天想到的另一个版本,展示了我希望如何表达它(尽管其他解决方案非常优雅;我只是放另一个版本!)

(defn convolve-2
  [xs is]
  (reduce #(vec-add %1 (pad-l %2 (inc (count %1))))
         (for [x xs]
           (for [i is]
             (* x i)))))

它使用了这个实用函数vec-add:

(defn vec-add
  ([xs] (vec-add xs []))
  ([xs ys]
     (let [lxs (count xs)
           lys (count ys)
           xs (pad-r xs lys)
           ys (pad-r ys lxs)]
       (vec (map #(+ %1 %2) xs ys))))
  ([xs ys & more]
     (vec (reduce vec-add (vec-add xs ys) more))))
     (vec (reduce vec-add (vec-add xs ys) more))))

【问题讨论】:

  • 这根本不是您问题的答案,但是...通过使用快速傅里叶变换实现卷积可大大减少操作数量(例如,请参阅 stackoverflow.com/questions/3084987/…)。
  • 我对函数式语言了解不多,不知道这是否有帮助,但您可能对stackoverflow.com/questions/803055/how-do-i-do-convolution-in-f 感兴趣。
  • @mtrw 在生产中,这肯定会使用 FFT 完成;不过,就目前而言,我希望有一个不错的、实用的解决方案。但我不确定这有多大可能(在 Clojure 中)。
  • 你有原始算法的链接吗?虽然尝试制作一个惰性函数版本是一个有趣的练习,但使用 Java 数组进行相当直译可能会更有效。
  • dspguide.com/ch6/3.htm 有原始算法(其他版本在其他页面上-基本上相同。所以没有办法以功能方式有效地做到这一点?

标签: functional-programming clojure convolution signal-processing


【解决方案1】:
(defn ^{:static true} convolve ^doubles [^doubles xs ^doubles is]
  (let [xlen (count xs)
        ilen (count is)
        ys   (double-array (dec (+ xlen ilen)))]
    (dotimes [p xlen]
      (dotimes [q ilen]
        (let [n (+ p q), x (aget xs p), i (aget is q), y (aget ys n)]
          (aset ys n (+ (* x i) y)))))
    ys))

如果我在 Clojure equiv 分支中执行此操作,我会参考 j-g-faustus 的版本。为我工作。在 i7 Mackbook Pro 上,1,000,000 点约为 400 毫秒,100,000 点约为 25 毫秒。

【讨论】:

  • 在作为参数传递之前调用(double-array xs)的数字序列
【解决方案2】:

堆栈溢出错误的可能原因是惰性 thunk 变得太深。 (concatmap 是懒惰的)。尝试将这些调用包装在 doall 中以强制评估它们的返回值。

至于功能更强大的解决方案,请尝试以下方法:

(defn circular-convolve
  "Perform a circular convolution of vectors f and g"
  [f g]
  (letfn [(point-mul [m n]
        (* (f m) (g (mod (- n m) (count g)))))
      (value-at [n]
        (reduce + (map #(point-mul % n) (range (count g)))))]
    (map value-at (range (count g)))))

使用reduce可以很方便的进行求和,由于map产生了一个惰性序列,所以这个函数也是惰性的。

【讨论】:

  • 不幸的是,将我的 mapsconcats 包裹在 doalls 中会花费非常长的时间(我不确定它是否会完成!)来计算一个包含 100,000 个数组的卷积点和另一个有 2... 你的循环卷积没有返回正确长度或正确值的列表(我不相信!)谢谢你的帮助!
【解决方案3】:

无法提供高性能的函数式版本,但您可以通过消除惰性和添加类型提示来获得命令式版本的 100 倍加速:

(defn imp-convolve-2 [xs is]
  (let [^doubles xs (into-array Double/TYPE xs)
        ^doubles is (into-array Double/TYPE is)
        ys (double-array (dec (+ (count xs) (count is)))) ]
    (dotimes [i (alength xs)]
      (dotimes [j (alength is)]
        (aset ys (+ i j)
          (+ (* (aget xs i) (aget is j))
            (aget ys (+ i j))))))
    ys))

xs 大小为 100k 和 is 大小为 2,你的 imp-convolve-1 在我的机器上包裹在一个 doall 中大约需要 6,000 毫秒,而这个需要大约 35 毫秒。

更新

这是一个惰性函数版本:

(defn convolve 
  ([xs is] (convolve xs is []))
  ([xs is parts]
    (cond
      (and (empty? xs) (empty? parts)) nil
      (empty? xs) (cons
                    (reduce + (map first parts))
                    (convolve xs is
                      (remove empty? (map rest parts))))
      :else (cons
              (+ (* (first xs) (first is))
                (reduce + (map first parts)))
              (lazy-seq
                (convolve (rest xs) is
                  (cons 
                    (map (partial * (first xs)) (rest is))
                    (remove empty? (map rest parts)))))))))

在大小 100k 和 2 上,它的时钟频率约为 600 毫秒(450-750 毫秒不等),而 imp-convolve-1 约为 6,000 毫秒,imp-convolve-2 约为 35 毫秒。

所以它是功能性的、懒惰的并且具有可容忍的性能。不过,它的代码量是命令式版本的两倍,而且我花了 1-2 个小时才找到,所以我不确定我是否明白这一点。

我完全支持纯函数,因为它们使代码更短或更简单,或者比命令式版本有其他好处。如果他们不这样做,我不反对切换到命令模式。

这是我认为 Clojure 很棒的原因之一,因为您可以根据需要使用任何一种方法。

更新 2:

我将通过说我喜欢 David Cabana 的 this functional implementation(第二个,在页面下方)来修改我的“在功能上这样做有什么意义”。

在与上述相同的输入大小(100,000 和 2)的情况下,它简短、易读且时间约为 140 毫秒,使其成为迄今为止我尝试过的性能最佳的功能实现。

考虑到它是功能性的(但不是惰性的),不使用类型提示并且适用于所有数字类型(不仅仅是双精度数),这令人印象深刻。

【讨论】:

  • 我很感激——我几乎没有使用命令式 Clojure 的经验!也许我应该学习...谢谢! (不接受,因为仍在寻找功能版本,但有我的 +1)
  • 虽然奇​​怪的是,当我用 100 万分尝试它时,我得到一个“进程过滤器错误:参数数量错误:nil,0”。虽然我看不到任何 Java 错误......不过,对于 100k 来说效果很好。我的 Python 版本适用于 100 万 - 还没有尝试过更多。
  • 它应该适用于最多 Integer/MAX_VALUE(约 20 亿)个点,它是 Java 数组的最大大小。 100万点我看不出有什么不同。确定不会在其他地方发生? “进程过滤器中的错误”听起来不像来自卷积函数。
  • 我会再调查一下:我还遇到了一个 Java 错误:OutOfMemory 或类似的东西。我想 20 亿就足够了,但如果能够使用 long 并且几乎没有限制,那就太好了:有办法做到这一点吗?
  • 有 20 亿个双精度数据,每个双精度数据 8 字节,您需要 16GB 的 RAM 才能保存双数据数据。在那个规模上,您可能想要研究分块(一次处理来自 xs 的 N 个数字,然后将结果在边界处粘合在一起)或懒惰。我不知道一种简单、高性能和实用的方法,我想我得把它留给专家了。就我自己而言,我倾向于认为高性能数字运算是最适合可变性的事情之一,如果没有别的,因为您不需要重新发明文献中找到的每个算法。
【解决方案4】:
(defn convolve [xs is]
  (if (> (count xs) (count is))
    (convolve is xs)
    (let [os (dec (+ (count xs) (count is)))
          lxs (count xs)
          lis (count is)]
      (for [i (range os)]
        (let [[start-x end-x] [(- lxs (min lxs (- os i))) (min i (dec lxs))]
              [start-i end-i] [(- lis (min lis (- os i))) (min i (dec lis))]]
          (reduce + (map *
                         (rseq (subvec xs start-x (inc end-x)))
                         (subvec is start-i (inc end-i)))))))))

可以用简洁的术语表达一个惰性的、功能性的解决方案。唉,> 2k 的性能是不切实际的。我有兴趣看看是否有办法在不牺牲可读性的情况下加快速度。

编辑: 在阅读了 drcabana 关于该主题的内容丰富的帖子 (http://erl.nfshost.com/2010/07/17/discrete-convolution-of-finite-vectors/) 后,我更新了我的代码以接受不同大小的向量。他的实现表现更好: xs size 3, is size 1000000, ~2s vs ~3s

编辑 2: 以 drcabana 的简单反转 xs 和 padding 的想法,我得出:

(defn convolve [xs is]
  (if (> (count xs) (count is))
    (convolve is xs)
    (let [is (concat (repeat (dec (count xs)) 0) is)]
      (for [s (take-while not-empty (iterate rest is))]
         (reduce + (map * (rseq xs) s))))))

这可能会尽可能简洁,但总体上仍然较慢,可能是由于需要一段时间。感谢博客作者的深思熟虑的方法。这里唯一的好处是上面的内容真的很懒,如果我问 (nth res 10000),它只需要前 10k 次计算即可得出结果。

【讨论】:

  • 对于您的第二个解决方案,填充不应该是:(dec (/ (count xs) 2))
【解决方案5】:

对于您提出的许多问题都没有真正的答案,但我有几个关于您没有问的问题。

  1. 您可能不应该在向量上使用nth。是的,它是 O(1),但是因为 nth 在 O(n) 中适用于其他序列,所以 (a) 并不清楚您希望输入是一个向量,并且 (b) 意味着如果您犯了一个错误,你的代码会莫名其妙地变慢,而不是立即失败。

  2. formap 都是惰性的,aset 只是副作用。这种组合是灾难的根源:对于类似 for 的副作用行为,请使用 doseq

  3. fordoseq 允许多个绑定,因此您不需要像(显然)在 Python 中那样堆积它们。

    (doseq [i (range (count cs))
            j (range (count is))] 
      ...)
    

    会做你想做的。

  4. #(first %) 更简洁地写成first;同样#(+ %1 %2)+

  5. 在一堆不需要成为向量的中间结果上调用vec 会减慢您的速度。特别是在vec-add 中,仅在生成最终返回值时调用vec 就足够了:在(vec (reduce foo bar)) 中,如果foo 从不将其中间结果用于随机访问,则没有理由将其转换为向量。

【讨论】:

    猜你喜欢
    • 1970-01-01
    • 2015-03-21
    • 2018-05-30
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2023-03-21
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多