問題タブ [iterate]

私は Haskell で次のコードを持っています。これはファイルの最初の行を読み取るには問題ありませんが、ディレクトリ内のファイルのすべてのコンテンツ (多くのファイルを再帰的に) を読み取る必要があります。firstLineE 関数を変更しようとしていますが、行を変更する方法がわかりません: EIO.enumFile 1024 filename $ joinI $ ((mapChunks B.pack) ><> EC.enumLinesBS) . これに関するドキュメントがありますか、それともいくつかの例を手伝ってもらえますか?

ドキュメントを確認していますが、Iteratee は私にとって非常に新しいものです。

http://www.mew.org/~kazu/proj/enumerator/

http://blog-mno2.csie.org/blog/2011/11/19/yet-another-guide-to-understand-iteratee-for-haskell-programmers/

私が見た Scala の Iteratee パターンの 3 つの説明には、すべて 3 つの入力ケースが含まれています。たとえば、ジェームズから：

詳細については、James、Runar、Joshによるブログを参照してください。

私の質問は単純です: なぜ正確に空の入力ケースが必要なのですか?

iteratee パターンは、値のストリームのプロデューサーとコンシューマーの間の関係を定義します。直感的には、入力が空の場合、イテラティーを「実行」するプロデューサーは、その空のアイテムを単純に折りたたんで、空でない入力が使用可能になるまでイテラーティーを呼び出さないように思われます。

iteratees の pull ベースの類似物である、はるかによく知られているイテレータは空のケースを定義しませんが、イテレータの「内部」で要素がフィルター処理されている可能性があります。

上記のすべてのブログでは、空の入力の必要性について言及していますが、完全に排除できない理由については明確に説明していません。示されているイテレータの例では、空の入力がノーオペレーションとして扱われていることに気付きました。

空の入力メッセージを解決する必要がある、もっともらしい「現実世界っぽい」問題の例をコードで示してください。

序章

私は多くのプロジェクトで Scalaz 7の iteratees を使用していますが、主に大規模なファイルを処理するためです。iteratee パッケージを置き換えるように設計された Scalazストリームへの切り替えを開始したいと思います (率直に言って、多くの部分が欠落しており、使用するのが面倒です)。

ストリームは、Haskell にも実装されているマシン(繰り返しのアイデアの別のバリエーション) に基づいています。私は Haskell マシン ライブラリを少し使用しましたが、マシンとストリームの関係は (少なくとも私には) 完全には明らかではなく、ストリーム ライブラリのドキュメントはまだ少しまばらです。

この質問は、反復の代わりにストリームを使用して実装したい単純な解析タスクに関するものです。誰も私を打ち負かすことができなければ、私は自分で質問に答えますが、この移行を行っている (または少なくとも検討している) のは私だけではないと確信しています。公の場でやったほうがいいと思った。

仕事

トークン化され、品詞でタグ付けされた文を含むファイルがあるとします。

1 行に 1 つのトークンがあり、単語と品詞は 1 つのスペースで区切られ、空白行は文の境界を表します。このファイルを解析して、文のリストを返したいと思います。これは、文字列のタプルのリストとして表すこともできます。

いつものように、無効な入力やファイル読み取りの例外が発生した場合に正常に失敗したり、手動でリソースを閉じることを心配したりしたくありません。

反復ソリューション

最初に、いくつかの一般的なファイル読み取り用のもの (実際には iteratee パッケージの一部である必要があります。現在、この高レベルのものはリモートで提供されていません):

そして、センテンス リーダー:

そして最後に、解析アクション:

それが機能することを実証できます。

これで完了です。

私が欲しいもの

iteratee の代わりに Scalaz ストリームを使用して実装されたほぼ同じプログラム。

私はscalaでscalaz 6.0を使用しています。入力ストリームから読み取るために iteratees を使用しています。

私が持っている simple.txt という単純なファイルを次に示します。

これ
は
テストです

私の iteratee は行を印刷するために io モナドを構築します

enumeratorM を使用する場合

正しい出力を取得します。

使おうとすると

「これ」だけがコンソールに返されます。getLines は標準入力ストリームを使用します。iteratee にデバッグ ステートメントを追加しましたが、getLines は最初の行の後に EOF() を送信しているようで、解決できませんでした。

IterateeJava 内からPlay の s を使用することは可能ですか? JavaIterateeでの使用に関する例やドキュメントを見つけることができませんでした。Scalaのみです。PLay API を使用して Java で Iteratees を動作させることは、コード的にはもう少し面倒だと思います (多くの anon s)...Funtion1<?,>

可能であれば、HTTPs チャンク転送エンコーディングを介してアップロードされたマルチパート ファイルのアップロードを受け入れ、これらのメッセージ チャンクをダウンストリームで S3 ストアに解析できるアプリ コントローラーを作成したいと考えています。Javaでこれにどのようにアプローチできるかについてのアイデアはありますか?

乾杯。

私は Scalaz 7 iteratee を使用して、一定のヒープ空間で大量の (つまり、制限のない) データ ストリームを処理してきました。

コードでは、次のようになります。

ここで、一度にP個のデータ チャンクを処理しながら、並列処理を実行したいと考えています。それでもヒープ領域を制限する必要がありますが、データのPチャンクと計算の累積結果を格納するのに十分なヒープがあると想定するのは合理的です。

私はTaskクラスを認識しており、列挙子をマッピングしてタスクのストリームを作成することを考えています。

しかし、非決定性を管理する方法はまだわかりません。ストリームを消費している間、可能な限りPタスクが実行されていることを確認するにはどうすればよいですか?

初挑戦：

私が最初に試した解決策は、ストリームを折りたたんで、Future各チャンクを処理する Scala を作成することでした。しかし、プログラムは GC オーバーヘッド エラーで爆発しました (おそらく、すべてのFutures を作成しようとしたときに、すべてのチャンクをメモリにプルしていたためです)。代わりに、反復対象は、既にP個のタスクが実行されているときに入力の消費を停止し、それらのタスクのいずれかが終了したときに再び再開する必要があります。

2 回目の試行:

私の次の試みは、ストリームをPサイズの部分にグループ化し、各部分を並行して処理し、結合してから次の部分に進むことでした。

これは利用可能なプロセッサーを完全には活用しませんが (特に、それぞれの処理に必要な時間Chunkは大きく異なる可能性があるため)、改善​​にはなります。ただし、enumerateegroupではメモリ リークが発生しているようです。ヒープの使用量が突然増加します。

scala と Play を使用して、ストリーミング Web ラジオ フレームワークを作成しています。私は実際のストリーミングを Iteratees に依存していますが、貪欲なクライアントがデータをあまりにも速くダウンロードするのを防ぎ、すべてのクライアントのストリームを消費するという問題に直面しています。そのために、列挙子がデータを生成する速度を調整する列挙子を作成しようとしました。これが私の列挙型の外観です

アイデアは、タイマー タスクを使用して throttlingIteratee を作成し、それを Enumeratee.grouped 関数とペアにするというものです。これはかなりうまくいっているようですが、チャンク サイズにどの値を使用すればよいかわかりません。これで、オーディオの再生とほぼ同じレートでチャンクを生成したいと考えています。私のオーディオ ファイルは 82kpbs でエンコードされており、それをバイト単位で計算しようとしましたが、得られた値が小さすぎるようで、オーディオはデータがストリーミングされるよりも速く再生されます。

私の質問は 2 つあります。私が実施している基本的なアプローチは適切ですか? そうである場合、オーディオ ファイルのビット レートに関してチャンク サイズを設定するにはどうすればよいでしょうか。

バックグラウンド

この質問で述べたように、私は Scalaz 7 iteratees を使用して、一定のヒープ空間で大量の (つまり、制限のない) データ ストリームを処理しています。

私のコードは次のようになります。

問題

メモリ リークに遭遇したようですが、バグが Scalaz にあるのか、それとも私のコードにあるのかを知るには、Scalaz/FP に精通していません。直観的には、このコードは-size スペースのP倍(のオーダー) しか必要としないと思います。Chunk

注: に遭遇した同様の質問を見つけましたがOutOfMemoryError、私のコードは を使用していませんconsume。

テスト

問題を特定するためにいくつかのテストを実行しました。要約すると、リークはzipWithIndexとの両方groupが使用された場合にのみ発生するようです。

テストのコード:

質問

• バグは私のコードにありますか?
• これを一定のヒープスペースで機能させるにはどうすればよいですか?