問題タブ [haskell-pipes]

私が大好きな Pipes を使用するプロジェクトのプログラムを作成しました。ただし、コードの単体テストに苦労しています。

Pipe In Out IO ()HSpec でテストしたいタイプ (たとえば) の一連の関数があります。これについてどうすればいいですか？

たとえば、次のドメインがあるとします。

そしてこのパイプ：

仕様を書きたいと思います：

私は Haskell とパイプ ライブラリの非常に初心者であり、関数でこのプログラムの高いメモリ使用量を引き起こしている原因を理解したいと言って前置きする必要がありtestます。

具体的には、r1値を生成するフォールドで、使用されtestない限り、最終結果が生成されるまで MyRecord 値の蓄積が見deepseqられます。~ 500000 行 / ~ 230 MB のサンプル データ セットでは、メモリ使用量が 1.5 GB を超えて増加します。

値を生成するフォールドr2は、定数メモリで実行されます。

私が理解したいのは：

1) 最初のフォールドで MyMemory 値がビルドされる原因は何ですか?また、使用deepseqするとそれが修正されるのはなぜですか? 一定のメモリ使用量を達成するために使用するまで、ランダムに物を投げていましたdeepseqが、なぜそれが機能するのかを理解したいと思います。deepseqMaybe Intの同じ結果タイプを生成しながら、使用せずに一定のメモリ使用量を達成できますか?

2)。同じ問題が発生しない原因となる 2 番目の折り目の違いは何ですか?

タプルの代わりに整数のみを使用する場合sum、Pipes.Prelude の組み込み関数を使用できることはわかっていますが、最終的には解析エラーを含む 2 番目の要素を処理する必要があります。

私は独自のモナドProducerを使用して、ランダム性に依存する値を作成するを持っています:Random

Randomまたはmwc-randomから実行できるラッパーです。STIO

プロデューサーはpolicies、単純な強化学習アルゴリズムから、ますます優れたポリシーを生成します。

にインデックスを付けることで、たとえば 5,000,000 回の反復後にポリシーを効率的にプロットできますpolicies。

ここで、500,000 ステップごとに中間ポリシーをプロットして、それらがどのように収束するかを確認したいと考えています。プロデューサーを使用して、たとえば 10 個のポリシー (500,000 回の反復ごとに 1 つ) のpoliciesリスト ( ) を抽出し、それらすべてをプロットする関数をいくつか作成しました。[Policy s a]

ただし、これらの関数は、学習反復の合計回数が同じ (つまり 5,000,000) であっても、上記のように最終ポリシーをプロットするよりもはるかに時間がかかり (10 倍)、多くのメモリ (4 倍) を使用します。これは、ガベージ コレクターを禁止するリストを抽出したためではないかと考えています。

慣用的なパイプ スタイルでは、すべての要素をメモリにロードするのではなく、要素が生成されるとすぐに要素を消費します。

Producerがランダムなモナド (つまりRandom) の上にあり、生成したい効果が にある場合、このようなパイプを消費する正しいアプローチは何IOですか?

別の言い方をすれば、 aProducer (Policy s a) Random xを aに接続したいのですConsumer (Policy s a) IO x。

これは、別の質問に対するガブリエルの回答と密接に関連しています。|||関数 fromのようなものをArrowChoice(Proxyライブラリからpipes) に対して提供する関数を作成しました。これはパターン マッチし、5 つの相互再帰関数を持ちます。Pipes.Coreパターン マッチングの代わりに関数 from を使用する代替実装を見つけたいと思います。

left私は最初に他の答えの実装を適応させようとしました。次のようになります。

ただし、この実装は上流と下流が同じであることに依存しており、それをさらに一般化する方法がわかりませんでした。私が得ることができた最も近いものは次のとおりです。

の変種のコード スニペット|||はかなり長いので、何らかの方法でこれを実行できるという証拠として提供しているだけです。関数の名前はdownstreamOr. 他のすべての関数は、相互に再帰的なヘルパー関数です。

これらのいずれかがパターン マッチングなしでどのように実装できるかを確認すると役に立ちます。ありがとうございます。何か明確にできることがあればお知らせください。

私はパイプ エコシステム、特にパイプ同時実行性を使用してストリーミング関数を作成しています。これは運用ライブラリに基づいており、ネットワーク経由でサーバーまたは stdin/out にコマンドを渡す小さなプログラム スニペットをすばやく作成できます。シェル コマンドを呼び出してから、応答を読み返します。この場合、それはアスタリスクですが、同様のものに一般化できます。

最初はパイプを念頭に置いてこれを書きましたが、うまくいきません。次のコードが機能しない理由は、astPipe が a を返すのPipe _ _ IO aに対して、 i と o の両方が pipes-concurrency から両方とも を返すためConsumer/Producer _ IO ()です。astPipeyieldを用意Maybe ByteStringしてから、出力をConsumerconsumerにすることを考えましMaybe ByteStringたが、それでもProducer返される問題は解決しません()。

解決にかなり近づいたような気がしますが、なかなか解決できません。このファイルでスタックを実行するだけで複製できるはずです。

エラー：

エンコードされた大きな(<bloblength><blob>)*バイナリ ファイルのデシリアライザーを作成しているときに、さまざまな Haskell の Produce-Transform-Consume ライブラリに行き詰まりました。これまでのところ、4 つのストリーミング ライブラリを認識しています。

• Data.Conduit : 広く使用されており、非常に慎重なリソース管理が行われています
• Pipes : 同様conduit( Haskell Cast #6は と の違いをうまく明らかにしていconduitますpipes)
• Data.Binary.Get : getWord32be などの便利な関数を提供しますが、ストリーミングの例は扱いにくいです
• System.IO.Streams : 一番使いやすいようです

Word32これは、ストリーミングをしようとしたときに問題が発生する場所の簡略化された例ですconduit。もう少し現実的な例では、最初Word32に blob の長さを決定する a を読み取り、次にその長さの lazyByteStringを生成します (その後、さらにデシリアライズされます)。しかし、ここでは、バイナリ ファイルからストリーミング形式で Word32 を抽出しようとしています。

プログラムの出力は、読み取られた Word32 の数だけです。最初のチャンク (約 32KiB) を読み取った後、ストリームが終了することがわかります。何らかの理由mbsでが neverであるため、チャンクが消費されたときにストリームを停止するものをNothing確認する必要があります。null bs明らかに、私のコンジットtransformは故障しています。ソリューションへの 2 つのルートが表示されます。

1. はのawait2 番目のチャンクに行きたくないByteStreamので、次のチャンクをプルする別の関数はありますか? 私が見た例 (例: Conduit 101 ) では、これはどのように行われたかではありません
2. これは、セットアップの方法が間違っているだけtransformです。

これはどのように適切に行われますか？これは正しい方法ですか？(パフォーマンスは重要です。)

更新:これを使用してそれを行う悪い方法は次のSystems.IO.Streamsとおりです。

「悪い」とは、時間と空間の要求が非常に高く、デコード例外を処理しないことを意味します。

私は次のタイプを持っています：

アイデアは、ネットワーク通信の純粋な表現として使用することです。つまり、次のようになります。

これは、不純な function によって評価されますeval。

さて、これは何ですか (あるとすれば?) 私が見る限り、これはモナドでも矢印でもありません。ストリーム/パイプ/オートマトン/fsm と継続モナドの間の何かのようです。これは、このタイプのものをより適切に表現することができるかもしれないと私に思わせますが、何ですか?

私 (Haskell の初心者) は、Web ページから受け取った ByteString に対してアンパック操作を実行しようとしています。基本的に、Webページからいくつかの単語を検索したいので、ストリームをトークン化し、単語から単語を検索しようとしています。

しかし、私はエラーを下回っています

ハックから、その署名が

デフォルトでは、パイプはプル ベースです。これは、彼のプル カテゴリのポイントフルオペレータである>->経由で実装されたオペレータによるものです。私の理解では、これは、のようなコードがある場合、コンシューマーの本体が最初に呼び出され、データを待機するとプロデューサーが呼び出されることを意味します。+>>bindproducer >-> consumer

ここpipesのドキュメントで、コードを使用してプル ベースのパイプをプッシュ ベースのパイプに変換できることを見てきました。つまり、上記のコードでは、プロデューサーが最初に実行され、値が生成され、次にコンシューマーが消費しようとします。それは本当に便利なようで、その方法を知りたいです。(reflect .)Pipes.Coreproducer >-> consumer

また、ここでの議論で>->、パイプを好転させるのは簡単であるため (リフレクトを使用すると思いますか?)、プッシュ ベースの対応するものがないことを見てきましたが、その方法や例を見つけることはできません。

ここに私が試みたいくつかのコードがあります:

pipes-csv ライブラリを使用して csv ファイルを読み込んでいます。最初の行を読んで、あとで続きを読みたい。残念ながら、 Pipes.Prelude.head 関数が戻った後。パイプは何とか閉じられています。最初に csv の先頭を読み取り、残りを後で読み取る方法はありますか。

最初にヘッダーを読み取らなければ、csv 全体を問題なく読み取ることができます。