问题标签 [haskell-pipes]

我正在使用Haskell 管道包。

我正在尝试使用管道并发将生产者列表合并在一起。

我想要达到的是：

所以给定一个生产者 s1 和另一个生产者 s2: r = merge [s1, s2] 这将给出以下行为：

按照教程页面中的代码，我想出了：

按预期工作。

但是，我很难概括事物。

我的尝试：

不幸的是，这可以编译，但并没有做太多其他事情。我猜我弄得一团糟runEffectIO。我目前的其他方法runEffectIO没有产生更好的结果。

该程序：

命中val <- await但没有到达，liftIO $ putStrLn因此它不会产生任何输出。但是它没有挂起就可以正常退出。

当我代替然后mergeIO程序merge运行时，我希望输出 20 行。

在下面的代码中我怎么能

• 更改stdoutCharConsumer以便在打印输入流中的所有数据后打印换行符

• 实施mixin和mixin'

无需进入 Pipes.Internal？可能吗？我需要类似nextProducers 的功能。

我使用管道 4.1.0

UPD：现在，在我阅读了基于拉/推的流之后，我认为即使使用 Pipes.Internal 也是不可能的。这是真的吗？

这感觉有点远，但我写了一个管道来连接数据库，获取服务器上的数据库列表，连接到每个数据库，然后对每个数据库执行查询（用户计数），然后打印那些计数。不幸的是，这与我可以从我的真实示例中简化的程度差不多。我正在使用管道 4.1.0、管道安全 2.0.2 和 mysql-simple-0.2.2.4。这是代码：

这工作正常。但是，假设在其中一个数据库中，用户表被命名为 users 而不是 user，因此 mysql-simple 将在运行该查询时抛出异常。我想内联捕获该错误，并为这些查询返回 0 个用户，但继续前进。我尝试过的事情：

这行不通。有时它会打印失败并产生一个 0，只是在异常时立即终止。我想可能是因为我抛出了 queryProducer 异常，我应该再次调用它，所以我尝试了这个递归版本：

但这也失败了。然而，有时它实际上会执行几个查询，打印出几次失败并产生几个 0，然后再次以异常终止。我真的很困惑为什么会这样。

根据异步库，异常应该被发送到管道正在运行的线程中，所以它看起来不像是线程问题。

如果我的 queryProducer 的实现很重要，它是在管道 postgresql 查询函数之后建模的，推广到 Producer'，所以我可以将它嵌入到其他组合器中。在 mysql-simple 下面，在 mysql 库中有一个 throw，如果您的 sql 没有意义，它会抛出 ConnectionError，它会一直渗透到这个函数中。

我还尝试使用 EitherT 来尝试捕获异常，因为这似乎是过去在管道中完成的方式。但是管道教程中的文档在 3 到 4 之间消失了，这让我想知道是否仍然推荐该技术。不幸的是，我无法让它工作，因为我使用 queryProducer 而不是单一的 await/yields 的方式，我不确定如何构建它。

有时需要执行一些复杂的例程来检索或保存正在处理的数据。在这种情况下，需要将数据生成和数据处理逻辑分开。常见的方法是使用类似迭代的功能。有很多不错的库：管道、导管等。在大多数情况下，它们会做这些事情。但是据我所知，它们（可能除了管道）受到处理顺序的限制。

但是考虑一个日志查看器的例子：人类可能希望随机地来回走动。他也可以放大和缩小。我担心迭代者在这里无能为力。

一个简单的解决方案可能如下所示：

然后可以将其包装到 StateT 中，依此类推。所以，问题：

0）我是否完全错过了迭代的要点，它们在这里适用？

1）我只是重新发明了一个众所周知的轮子吗？

2) 很明显，增长和收缩操作非常无效，因为它们的复杂性与帧大小成正比。有没有更好的方法来延长这样的拉链？

我见过有人为各种惰性 IO 相关任务推荐管道/导管库。这些库究竟解决了什么问题？

此外，当我尝试使用一些与 hackage 相关的库时，很可能存在三个不同的版本。例子：

• 阿托秒差距
• 管道-attoparsec
• attoparsec-conduit

这让我很困惑。对于我的解析任务，我应该使用 attoparsec 还是 pipe-attoparsec/attoparsec-conduit？与普通的 attoparsec 相比，管道/导管版本给我带来了什么好处？

旧pipes 教程有以下示例。这段代码在 4.1.1 版中会是什么样子？

管道教程在他们的文档中提到了这一点：

如果你牺牲 Effects，你会得到 Haskell 的纯和惰性列表，你可以在恒定空间中使用可组合函数对其进行转换，但没有交错效果。

如果你牺牲了 Streaming，你会得到 mapM、forM 和“ListT done wrong”，它们是可组合且有效的，但在整个列表首先被处理并加载到内存之前不会返回单个结果。

但并不Streaming意味着交错效应？当你是时Streaming，你实际上是一个块一个块地消耗。那么，在逐块消费的同时，效果不是交错的吗？

Pipes.Aeson 库公开了以下函数：

如果我将此解析器与文件句柄一起使用 evalStateT 作为参数，则会从文件中读取单个 JSON 对象并进行解析。

问题是该文件包含多个对象（所有相同类型），我想在读取它们时折叠或缩小它们。

Pipes.Parse 提供：

但是正如你所看到的，这会返回一个新的解析器——我想不出一种方法来提供第一个解析器作为参数。

看起来 Parser 实际上是 StateT monad 转换器中的 Producer。我想知道是否有一种方法可以从 StateT 中提取 Producer，以便可以将 evalStateT 应用于 foldAll Parser，以及从 decode Parser 中提取 Producer。

不过，这可能是完全错误的方法。

简而言之，我的问题是：
使用 Pipes.Aeson 解析文件时，折叠文件中所有对象的最佳方法是什么？

假设我有一个这样的模块：

'produceString' 函数是一个字节串生产者，我关心的是折叠解析以产生某种结果。

以下两个程序显示了通过将字节字符串解析为一系列 JSON 整数来解决在字节字符串中查找最大值问题的不同方法。

方案一：

方案二：

不幸的是，当我分析它们时，这两个程序总共占用了大约 200MB 的内存，我希望使用流解析器可以解决这个问题。第一个程序将大部分时间和内存 (> 70%) 花费在(^.)Lens.Family 中，而第二个程序将其花费在fmapLens.Family.State.Strictzoom中。使用图如下。这两个程序都将大约 70% 的时间用于垃圾收集。

难道我做错了什么？Prelude 功能是max不是不够严格？我不知道库函数是否不好，或者我是否使用错误的库！（应该是后者。）

为了完整起见，这里有一个 git repocabal install ，如果你想亲眼看看我在说什么，你可以克隆并运行它，这里是两个程序的内存使用情况：

我有以下带有管道的代码，没有第二个管道（>-> P.mapM ( fillMD5)）就可以了。fillMD5是一个操作a -> IO a。

错误是：

我知道类型mapM是

但我看不出这个怎么解除Safe.SafeT？