生成器 – 比使用`Task / produce / consume`更好的方式表示懒惰集合表示为协同程序

2023年11月21日 304次阅读

使用Tasks非常方便

表达懒惰的集合/生成器.

例如：

function fib()
    Task() do
        prev_prev = 0
        prev = 1
        produce(prev)
        while true
            cur = prev_prev + prev
            produce(cur)
            prev_prev = prev
            prev = cur
        end
    end
end

collect(take(fib(), 10))

输出：

10-element Array{Int64,1}:
  1
  1
  2
  3
  5
  8
 13
 21
 34

但是,它们根本不遵循良好的迭代器约定.
他们尽可能地表现得很糟糕

它们不使用返回的状态

start(fib()) == nothing #It has no state

所以他们改变迭代器对象本身.
一个正确的迭代器使用它的状态,而不是自己改变它,所以它们多个调用者可以立即迭代它.
用start创建该状态,然后在下一个阶段推进它.

辩论说,该状态应该是不可改变的,接下来将返回一个新状态,这样可以简单地发挥作用. (另一方面,分配新内存 – 虽然在堆栈上)

更进一步,隐藏状态,它不会在下一个进展.
以下不起作用：

@show ff = fib()
@show state = start(ff)
@show next(ff, state)

输出：

ff = fib() = Task (runnable) @0x00007fa544c12230
state = start(ff) = nothing
next(ff,state) = (nothing,nothing)

相反,隐藏状态在完成期间被提升：
以下作品：

@show ff = fib()
@show state = start(ff)
@show done(ff,state)     
@show next(ff, state)

输出：

ff = fib() = Task (runnable) @0x00007fa544c12230
state = start(ff) = nothing
done(ff,state) = false
next(ff,state) = (1,nothing)

在完成期间推进状态并不是世界上最糟糕的事情.
毕竟,通常情况下很难知道你何时完成,而不是试图找到下一个状态.人们希望在接下来之前总能调用.
仍然不是很好,因为发生以下情况：

ff = fib()
state = start(ff)
done(ff,state)
done(ff,state)
done(ff,state)
done(ff,state)
done(ff,state)
done(ff,state)
@show next(ff, state)

输出：

next(ff,state) = (8,nothing)

这真的是你现在的期望.可以合理地假设完成多次调用是安全的.

基本上任务会使得迭代器变差.在许多情况下,它们与其他需要迭代器的代码不兼容. (他们很多,但很难分辨哪个).
这是因为在这些“生成器”函数中,任务并不真正用作迭代器.它们用于低级控制流程.
并且这样优化.

那么更好的方法是什么？
为fib编写迭代器也不错：

immutable Fib end
immutable FibState
    prev::Int
    prevprev::Int
end

Base.start(::Fib) = FibState(0,1)
Base.done(::Fib, ::FibState) = false
function Base.next(::Fib, s::FibState)
    cur = s.prev + s.prevprev
    ns = FibState(cur, s.prev)
    cur, ns
end

Base.iteratoreltype(::Type{Fib}) = Base.HasEltype()
Base.eltype(::Type{Fib}) = Int
Base.iteratorsize(::Type{Fib}) = Base.IsInfinite()

但是有点不那么直观.
对于更复杂的功能,它不太好.

所以我的问题是：
什么是更好的方式来获得像Task一样工作的东西,作为从单个函数构建迭代器的一种方法,但是表现得很好？

如果有人已经用宏来编写包来解决这个问题,我不会感到惊讶.

最佳答案 Tasks的当前迭代器接口非常简单：

# in share/julia/base/task.jl
275 start(t::Task) = nothing
276 function done(t::Task, val)
277     t.result = consume(t)
278     istaskdone(t)
279 end
280 next(t::Task, val) = (t.result, nothing)

不知道为什么开发人员选择将消耗步骤放在完成函数而不是下一个函数中.这就是产生你奇怪的副作用的原因.对我来说,实现这样的界面听起来更直接：

import Base.start; function Base.start(t::Task) return t end
import Base.next;  function Base.next(t::Task, s::Task) return consume(s), s end
import Base.done;  function Base.done(t::Task, s::Task) istaskdone(s) end

因此,这就是我提出的问题答案.

我认为这个更简单的实现更有意义,满足您的上述标准,甚至具有输出有意义状态的预期结果：任务本身！ (如果你真的想要,你可以“检查”,只要这不涉及消费：p).

但是,有一些警告：

>警告1：任务必须具有返回值,表示迭代中的最后一个元素,否则可能发生“意外”行为.

我假设开发人员选择了第一种方法来避免这种“无意识的”输出;但是我相信这应该是预期的行为！期望用作迭代器的任务应该通过设计定义适当的迭代端点(通过明确的返回值)！

示例1：错误的方法

julia> t = Task() do; for i in 1:10; produce(i); end; end;
julia> collect(t) |> show
Any[1,2,3,4,5,6,7,8,9,10,nothing] # last item is a return value of nothing
                                  # correponding to the "return value" of the
                                  # for loop statement, which is 'nothing'.
                                  # Presumably not the intended output!

示例2：另一种错误的方法

julia> t = Task() do; produce(1); produce(2); produce(3); produce(4); end;
julia> collect(t) |> show
Any[1,2,3,4,()] # last item is the return value of the produce statement,
                # which returns any items passed to it by the last
                # 'consume' call; in this case an empty tuple.
                # Presumably not the intended output!

例3：(以我的拙见)正确的方式去做！

julia> t = Task() do; produce(1); produce(2); produce(3); return 4; end;
julia> collect(t) |> show
[1,2,3,4] # An appropriate return value ending the Task function ensures an
          # appropriate final value for the iteration, as intended.

>警告2：在迭代中不应该进一步修改/消耗任务(通常需要使用迭代器),除非理解这有意在迭代中导致’跳过'(这将是一个hack充其量,并且可能不可取).

例：

julia> t = Task() do; produce(1); produce(2); produce(3); return 4; end;
julia> for i in t; show(consume(t)); end
24

更微妙的例子：

julia> t = Task() do; produce(1); produce(2); produce(3); return 4; end;
julia> for i in t   # collecting i is a consumption event
        for j in t  # collecting j is *also* a consumption event
          show(j)
        end
       end # at the end of this loop, i = 1, and j = 4
234

>警告3：通过这种方案,您可以“继续离开的地方”行为.例如

julia> t = Task() do; produce(1); produce(2); produce(3); return 4; end;
julia> take(t, 2) |> collect |> show
[1,2]
julia> take(t, 2) |> collect |> show
[3,4]

但是,如果希望迭代器始终从任务的预消耗状态开始,则可以修改start函数以实现此目的：

import Base.start; function Base.start(t::Task) return Task(t.code) end;
import Base.next;  function Base.next(t::Task, s::Task) consume(s), s end;
import Base.done;  function Base.done(t::Task, s::Task) istaskdone(s) end;

julia> for i in t
         for j in t
           show(j)
         end
       end # at the end of this loop, i = 4, and j = 4 independently
1234123412341234

有趣的是,请注意这个变体如何影响’警告2’中的“内部消费”情景：

julia> t = Task() do; produce(1); produce(2); produce(3); return 4; end;
julia> for i in t; show(consume(t)); end
1234
julia> for i in t; show(consume(t)); end
4444

看看你能否发现为什么这是有意义的！ 🙂

说完这一切之后,有一个哲学观点,关于一个任务在开始,下一个和完成命令的行为方式是否重要,甚至重要的是,这些函数被认为是“an informal interface”：即它们应该是是“引擎盖下”的功能,不打算手动调用.

因此,只要他们完成工作并返回预期的迭代值,你就不应该太在意他们是如何做到这一点的,即使技术上他们并没有完全按照’规范’这样做,因为你永远不应该首先手动调用它们.

点赞