如何为我的 Task 实现实现相当于 Promise.all 的功能？答案

【问题标题】：How to implement the equivalent of Promise.all for my Task implementation?如何为我的 Task 实现实现相当于 Promise.all 的功能？
【发布时间】：2019-08-07 12:31:28
【问题描述】：

这是我的Task 实现（即一种Promise，但符合单子定律并且可以取消）。它的工作坚如磐石：

const Task = k =>
  ({runTask: (res, rej) => k(res, rej)});

const tAp = tf => tk =>
  Task((res, rej) => tf.runTask(f => tk.runTask(x => res(f(x)), rej), rej));

const tOf = x => Task((res, rej) => res(x));

const tMap = f => tk =>
  Task((res, rej) => tk.runTask(x => res(f(x)), rej));

const tChain = fm => mx =>
  Task((res, rej) => mx.runTask(x => fm(x).runTask(res, rej), rej));

const log = x => console.log(x);
const elog = e => console.error(e);

const fetchName = (id, cb) => {
  const r = setTimeout(id_ => {
    const m = new Map([[1, "Beau"], [2, "Dev"], [3, "Liz"]]);

    if (m.has(id_))
      return cb(null, m.get(id_));

    else
      return cb("unknown id", null);
  }, 0, id);

  return () => clearTimeout(r);
};

const fetchNameAsync = id =>
  Task((res, rej) =>
    fetchName(id, (err, data) =>
      err === null
        ? res(data)
        : rej(err)));

const a = tAp(tMap(x => y => x.length + y.length)
  (fetchNameAsync(1)))
    (fetchNameAsync(3));

const b = tAp(tMap(x => y => x.length + y.length)
  (fetchNameAsync(1)))
    (fetchNameAsync(5));

a.runTask(log, elog); // 7
b.runTask(log, elog); // Error: "unknown id"

但是，我不知道如何实现awaitAll，它应该具有以下特征：

它可以通过个人Tasks 的一组结果来解析
或者它立即拒绝第一个错误并取消所有其他Tasks
它以“并行”方式执行Tasks

const awaitAll = ms =>
  Task((res, rej) => ms.map(mx => mx.runTask(...?)));

感谢任何提示！

【问题讨论】：

“它以“并行”的方式执行任务” Promise.all 不是这样做的。 Promise.all 不执行任何操作。当您使用Promise.all 时，相关操作已经开始。 Promise.all 所做的只是等待他们完成，收集结果。
@T.J.Crowder 抱歉，这种行为只适用于单子任务，在我的例子中是一个扩展单子Either 类型。我的声明与Promises 有关。
@bob - 我认为你会在awaitAll 上挣扎，至少在当前的实现中是这样。据我所知，Task 不一定是异步的。一些是，但不是全部。所以如果awaitAll的工作是启动任务并并行运行，就会有一些任务是同步的问题。我看到的唯一方法是要求所有任务都是异步的，在这种情况下，实现相当简单：循环启动它们的任务并挂钩它们的完成，然后如果其中任何一个失败则失败，或者填写任务的如果有效，则插入结果数组中。
它们是否同步并不重要。如果它们是同步的，它们可能会来自 awaitAll 所在的同一堆栈中的 throw，在这种情况下，您可能必须实现冗余错误处理。
@T.J.Crowder 没错。所以你应该只创建异步Tasks，如果你需要非阻塞行为。

标签： javascript functional-programming continuations continuation-passing

【解决方案1】：

这是从此处的其他答案以及链接的民间故事/任务中汲取灵感的另一种方式。我们不会实现复杂的tAll 来处理迭代任务列表和组合任务，而是将关注点分离到单独的函数中。

这是一个简化的tAnd -

const tAnd = (t1, t2) =>
{ const acc = []

  const guard = (res, i) => x =>
    ( acc[i] = x
    , acc[0] !== undefined && acc[1] !== undefined
        ? res (acc)
        : void 0
    )

  return Task
    ( (res, rej) =>
        ( t1 .runTask (guard (res, 0), rej) // rej could be called twice!
        , t2 .runTask (guard (res, 1), rej) // we'll fix this below
        )
    )
}

它是这样工作的 -

tAnd
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

// ~2 seconds later
// [ 'a', 'b' ]

现在 tAll 很容易实现 -

const tAll = (t, ...ts) =>
  t === undefined
    ? tOf ([])
    : tAnd (t, tAll (...ts))

Wups，不要忘记沿途变平-

const tAll = (t, ...ts) =>
  t === undefined
    ? tOf ([])
    : tMap
        ( ([ x, xs ]) => [ x, ...xs ]
        , tAnd (t, tAll(...ts))
        )

它是这样工作的 -

tAll
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~2 seconds later
// [ 'a', 'b', 'c', 'd', 'e', 'f', 'g' ]

tAll 也能正确处理错误 -

tAll
  ( delay (100, 'test failed')
  , Task ((_, rej) => rej ('test passed'))
  )
  .runTask (console.log, console.error)

// test passed

与我们最初的tAll 相比，即使我们限制了我们程序的范围，要正确获得tAnd 也是非常困难的。合并的任务应该只解决一次，或拒绝一次 - 不能同时解决。这意味着还应避免双重解决/拒绝。执行这些约束需要更多代码 -

const tAnd = (t1, t2) =>
{ let resolved = false
  let rejected = false

  const result = []

  const pending = ([ a, b ] = result) =>
    a === undefined || b === undefined

  const guard = (res, rej, i) =>
    [ x =>
        ( result[i] = x
        , resolved || rejected || pending ()
            ? void 0
            : ( resolved = true
              , res (result)
              )
        )
    , e =>
        resolved || rejected
          ? void 0
          : ( rejected = true
            , rej (e)
            )
    ]

  return Task
    ( (res, rej) =>
        ( t1 .runTask (...guard (res, rej, 0))
        , t2 .runTask (...guard (res, rej, 1))
        )
    )
}

展开下面的sn-p，在自己的浏览器中验证结果-

const Task = k =>
  ({ runTask: (res, rej) => k (res, rej) })

const tOf = v =>
  Task ((res, _) => res (v))

const tMap = (f, t) =>
  Task
    ( (res, rej) =>
        t.runTask
          ( x => res (f (x)) 
          , rej
          )
    )

const tAnd = (t1, t2) =>
{ let resolved = false
  let rejected = false
  
  const result = []

  const pending = ([ a, b ] = result) =>
    a === undefined || b === undefined

  const guard = (res, rej, i) =>
    [ x =>
        ( result[i] = x
        , resolved || rejected || pending ()
            ? void 0
            : ( resolved = true
              , res (result)
              )
        )
    , e =>
        resolved || rejected
          ? void 0
          : ( rejected = true
            , rej (e)
            )
    ]

  return Task
    ( (res, rej) =>
        ( t1 .runTask (...guard (res, rej, 0))
        , t2 .runTask (...guard (res, rej, 1))
        )
    )
}

const tAll = (t, ...ts) =>
  t === undefined
    ? tOf ([])
    : tMap
        ( ([ x, xs ]) => [ x, ...xs ]
        , tAnd (t, tAll (...ts))
        )

const delay = (ms, x) =>
  Task (r => setTimeout (r, ms, x))

tAnd
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

tAll
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~2 seconds later
// [ 'a', 'b' ]
// [ 'a', 'b', 'c', 'd', 'e', 'f', 'g' ]

tAll
  ( delay (100, 'test failed')
  , Task ((_, rej) => rej ('test passed'))
  )
  .runTask (console.log, console.error)

// Error: test passed

串行处理

最棘手的一点是并行处理要求。如果需求要求 serial 行为，那么实现会非常容易 -

const tAnd = (t1, t2) =>
  Task
    ( (res, rej) =>
        t1 .runTask
          ( a =>
              t2 .runTask
                ( b =>
                    res ([ a, b ])
                , rej
                )
          , rej
          )
    )

当然，tAll 的实现保持不变。注意现在的延迟差异，因为现在任务是按顺序运行的 -

tAnd
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

// ~2.5 seconds later
// [ 'a', 'b' ]

还有tAll 的许多任务-

tAll
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~ 9 seconds later
// [ 'a', 'b', 'c', 'd', 'e', 'f', 'g' ]

展开下面的sn-p，在自己的浏览器中验证结果-

const Task = k =>
  ({ runTask: (res, rej) => k (res, rej) })

const tOf = v =>
  Task ((res, _) => res (v))

const tMap = (f, t) =>
  Task
    ( (res, rej) =>
        t.runTask
          ( x => res (f (x)) 
          , rej
          )
    )

const tAnd = (t1, t2) =>
  Task
    ( (res, rej) =>
        t1 .runTask
          ( a =>
              t2 .runTask
                ( b =>
                    res ([ a, b ])
                , rej
                )
          , rej
          )
    )

const tAll = (t, ...ts) =>
  t === undefined
    ? tOf ([])
    : tMap
        ( ([ x, xs ]) => [ x, ...xs ]
        , tAnd (t, tAll (...ts))
        )

const delay = (ms, x) =>
  Task (r => setTimeout (r, ms, x))

tAnd
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

// ~2.5 seconds later
// [ 'a', 'b' ]

tAll
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~ 9 seconds later
// [ 'a', 'b', 'c', 'd', 'e', 'f', 'g' ]

tAll
  ( delay (100, 'test failed')
  , Task ((_, rej) => rej ('test passed'))
  )
  .runTask (console.log, console.error)

// Error: test passed

如何实现tOr和tRace

为了完整起见，这里是tOr。注意tOr这里相当于folktale的Task.concat-

const tOr = (t1, t2) =>
{ let resolved = false
  let rejected = false

  const guard = (res, rej) =>
    [ x =>
        resolved || rejected
          ? void 0
          : ( resolved = true
            , res (x)
            )
    , e =>
        resolved || rejected
          ? void 0
          : ( rejected = true
            , rej (e)
            )
    ]

  return Task
    ( (res, rej) =>
        ( t1 .runTask (...guard (res, rej))
        , t2 .runTask (...guard (res, rej))
        )
    )
}

解决或拒绝两个任务中最先完成的任务 -

tOr
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

// ~500 ms later
// 'b'

还有tRace-

const tRace = (t = tOf (undefined), ...ts) =>
  ts .reduce (tOr, t)

解决或拒绝许多任务中最先完成的任务 -

tRace
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~300 ms later
// 'f'

展开下面的sn-p，在自己的浏览器中验证结果-

const Task = k =>
  ({ runTask: (a, b) => k (a, b) })

const tOr = (t1, t2) =>
{ let resolved = false
  let rejected = false

  const guard = (res, rej) =>
    [ x =>
        resolved || rejected
          ? void 0
          : ( resolved = true
            , res (x)
            )
    , e =>
        resolved || rejected
          ? void 0
          : ( rejected = true
            , rej (e)
            )
    ]

  return Task
    ( (res, rej) =>
        ( t1 .runTask (...guard (res, rej))
        , t2 .runTask (...guard (res, rej))
        )
    )
}

const tRace = (t = tOf (undefined), ...ts) =>
  ts. reduce (tOr, t)

const delay = (ms, x) =>
  Task (r => setTimeout (r, ms, x))

tOr
  ( delay (2000, 'a')
  , delay (500, 'b')
  )
  .runTask (console.log, console.error)

// ~500 ms later
// 'b' 

tRace
  ( delay (2000, 'a')
  , delay (500, 'b')
  , delay (900, 'c')
  , delay (1500, 'd')
  , delay (1800, 'e')
  , delay (300, 'f')
  , delay (2000, 'g')
  )
  .runTask (console.log, console.error)

// ~300 ms later
// note `f` appears in the output first because this tRace demo finishes before the tOr demo above
// 'f'

tRace
  ( delay (100, 'test failed')
  , Task ((_, rej) => rej ('test passed'))
  )
  .runTask (console.log, console.error)

// Error: test passed

如何实现tAp

在 cmets 中，我们谈论的是 applicative，tAp。我认为tAll 使实现变得相当容易-

const tAp = (f, ...ts) =>
  tMap
    ( ([ f, ...xs ]) => f (...xs)
    , tAll (f, ...ts)
    )

tAp 接受一个任务包装函数和任意数量的任务包装值，并返回一个新任务 -

const sum = (v, ...vs) =>
  vs.length === 0
    ? v
    : v + sum (...vs)

tAp
  ( delay (2000, sum)
  , delay (500, 1)
  , delay (900, 2)
  , delay (1500, 3)
  , delay (1800, 4)
  , delay (300, 5)
  )
  .runTask (console.log, console.error)

// ~2 seconds later
// 15

除非任务有副作用，否则我看不出tAp 的“并行”实现违反应用法则的原因。

展开下面的sn-p，在自己的浏览器中验证结果-

const Task = k =>
  ({ runTask: (res, rej) => k (res, rej) })

const tOf = v =>
  Task ((res, _) => res (v))

const tMap = (f, t) =>
  Task
    ( (res, rej) =>
        t.runTask
          ( x => res (f (x)) 
          , rej
          )
    )

const tAp = (f, ...ts) =>
  tMap
    ( ([ f, ...xs ]) => f (...xs)
    , tAll (f, ...ts)
    )

const tAnd = (t1, t2) =>
{ let resolved = false
  let rejected = false
  
  const result = []

  const pending = ([ a, b ] = result) =>
    a === undefined || b === undefined

  const guard = (res, rej, i) =>
    [ x =>
        ( result[i] = x
        , resolved || rejected || pending ()
            ? void 0
            : ( resolved = true
              , res (result)
              )
        )
    , e =>
        resolved || rejected
          ? void 0
          : ( rejected = true
            , rej (e)
            )
    ]

  return Task
    ( (res, rej) =>
        ( t1 .runTask (...guard (res, rej, 0))
        , t2 .runTask (...guard (res, rej, 1))
        )
    )
}

const tAll = (t, ...ts) =>
  t === undefined
    ? tOf ([])
    : tMap
        ( ([ x, xs ]) => [ x, ...xs ]
        , tAnd (t, tAll (...ts))
        )

const delay = (ms, x) =>
  Task (r => setTimeout (r, ms, x))

const sum = (v, ...vs) =>
  vs.length === 0
    ? v
    : v + sum (...vs)

tAp
  ( delay (2000, sum)
  , delay (500, 1)
  , delay (900, 2)
  , delay (1500, 3)
  , delay (1800, 4)
  , delay (300, 5)
  )
  .runTask (console.log, console.error)

// ~2 seconds later
// 15

【讨论】：

让我们continue this discussion in chat。
@bob 和 Bergi 感谢您了解这些细节。 tAnd 是一个更合适的名称。我已经更新了答案。
我还进行了一些额外的更新以仅强制执行单个解析或拒绝。 @Jonas，此更新包括我对 guard 的新想法，它会生成一对函数。
@user633183 也许我在Task的上下文中误解了ap。 - 如果你仍然对这个细节感兴趣，这里是explanation
我已经在实践中使用了这个实现有一段时间了，它的工作非常可预测和流畅。但是，我将并行应用程序移到了单独的类型 Parallel 中，它还实现了 monoid 类型类（append=tRace 和永不解决的 Parallel 作为空元素）。值得注意的是，只需要很少的代码就可以单元地处理异步内容。

【解决方案2】：

这是使用计数器和包裹在另一个任务中的循环的一种可能方法。使用计数器是因为任务可以按任何顺序完成，否则很难知道外部任务何时可以最终解决 -

const assign = (o = {}, [ k, v ]) =>
  Object .assign (o, { [k]: v })

const tAll = (ts = []) =>
{ let resolved = 0
  const acc = []
  const run = (res, rej) =>
  { for (const [ i, t ] of ts .entries ())
      t .runTask
        ( x =>
            ++resolved === ts.length
              ? res (assign (acc, [ i, x ]))
              : assign (acc, [ i, x ])
        , rej
        )
  }
  return Task (run)
}

我们编写了一个简单的delay 函数来测试它-

const delay = (ms, x) =>
  Task ((res, _) => setTimeout (res, ms, x))

const tasks =
  [ delay (200, 'a')
  , delay (300, 'b')
  , delay (100, 'c')
  ]

tAll (tasks) .runTask (console.log, console.error)
// ~300 ms later
// => [ 'a', 'b', 'c' ]

如果任何任务失败，外部任务将被拒绝 -

const tasks =
  [ delay (200, 'a')
  , delay (300, 'b')
  , Task ((_, rej) => rej (Error('bad')))
  ]

tAll (tasks) .runTask (console.log, console.error)
// => Error: bad

展开下面的sn-p，在自己的浏览器中验证结果-

const assign = (o = {}, [ k, v ]) =>
  Object .assign (o, { [k]: v })

const Task = k =>
  ({runTask: (res, rej) => k(res, rej)});

const tAll = (ts = []) =>
{ let resolved = 0
  const acc = []
  const run = (res, rej) =>
  { for (const [ i, t ] of ts .entries ())
      t .runTask
        ( x =>
            ++resolved === ts.length
              ? res (assign (acc, [ i, x ]))
              : assign (acc, [ i, x ])
        , rej
        )
  }
  return Task (run)
}

const delay = (ms, x) =>
  Task ((res, _) => setTimeout (res, ms, x))

const tasks =
  [ delay (200, 'a')
  , delay (300, 'b')
  , delay (100, 'c')
  ]

tAll (tasks) .runTask (console.log, console.error)
// ~300 ms later
// => [ 'a', 'b', 'c' ]

这是tAll 的另一种实现，它将for 换成forEach，并删除了一个命令式块{ ... } -

const tAll = (ts = []) =>
{ let resolved = 0
  const acc = []
  const run = (res, rej) => (t, i) =>
    t .runTask
      ( x =>
          ++resolved === ts.length
            ? res (assign (acc, [ i, x ]))
            : assign (acc, [ i, x ])
      , rej
      )
  return Task ((res, rej) => ts .forEach (run (res, rej)))
}

【讨论】：

@bob yvw。这是一个棘手的问题，不幸的是我看不到不涉及某种命令式风格的解决方案。我在答案的末尾添加了一个您可能更喜欢的变体。如果你想出其他的，请分享^_^
"某种命令式风格" ...好吧，变量可以变成参数，resolved可以省略，...
这不会像 Promise.all 那样用于空数组，但我的回答也不是

【解决方案3】：

另一种解决方案是使用带有 2 Task 基本情况的递归，然后只允许在两个变量中管理状态：

  const tAll = ([first, second, ...rest]) =>
   !second
     ? first
     : rest.length 
        ? tMap(
            results => results.flat()
          )(tAll([ tAll([first, second]), tAll(rest) ]))
        : Task((res, rej, a, b, done) => (
            first.runTask(
               value => !done && b ? (res([value, b.value]), done = true) : (a = { value }),
               err => !done && (rej(err), done = true)
            ),
            second.runTask(
               value => !done && a ? (res([a.value, value]), done = true) : (b = { value }),
              err => !done && (rej(err), done = true)
            ) 
         ));

【讨论】：

有趣！但是，当使用来自我的答案的相同输入时，这会产生 'a' 而不是预期的 [ 'a', 'b', 'c' ]。在尝试理解你在这里变出的魔法之前，我会等待修复。也许包括正常运行的 sn-p？
@user633183 是的，仍在进行中，here it is（偷了你的测试用例:)）
@JonasWilms 我自己实际上使用了tAll 作为名字，但为了这个问题改变了它，有趣！
@bob 我刚刚从 user633183 手中接过它（naomik 更容易写）;)
@JonasWilms 我在第二个回答中解开了更多的电线。这是一个有趣的项目^^