関数プログラミングと無限論理多項式(21)

C# の例(7)

C# で Channel というものを使うことができるので Go への移植の参考になるかもしれないので調べてみます。C#によるマルチコアのための非同期/並列処理プログラミング を少し参考にしていますが、けっこう古い本なので Channel や IAsyncEnumerable は載っていないようです。

以下のコードで Generator は仮想サーバー、Receiver は仮想ブラウザーとなります。ch_get は仮想サーバーからの get を表します。まず get だけの場合を考えます。

        // 仮想サーバー
        public async Task Generator(Channel<int> ch_get)
        {
            for (int i = 0; i < 10; i++)
            {
                int n = i;
                if (await ch_get.Writer.WaitToWriteAsync())
                {
                    if (!ch_get.Writer.TryWrite(n))
                    {
                        break;
                    }
                }
            }
            ch_get.Writer.Complete();
        }

        // 仮想ブラウザー
        public async IAsyncEnumerable<int> Receiver()
        {
            Channel<int> ch_get = Channel.CreateBounded<int>(1);
            Task task = Task.Run(() => Generator(ch_get));
            while (await ch_get.Reader.WaitToReadAsync())
            {
                if (!ch_get.Reader.TryRead(out int n))
                {
                    break;
                }
                yield return n;
            }
            await task;
        }

        private async void button1_Click(object sender, EventArgs e)
        {
            IteratorClass iterator = new IteratorClass();
            string res = "";
            await foreach (int n in iterator.Receiver())
            {
                res += n.ToString() + " ";
            }
            textBox1.Text = res;
        }

実行結果: 0 1 2 3 4 5 6 7 8 9

以下のコードで ch_set は仮想サーバーへの set を表します。get と set の両方がある場合を考えます。

        // 仮想サーバー
        public async Task GeneratorIO(Channel<int> ch_get, Channel<int> ch_set)
        {
            int n = 1;
            for (int i = 0; i < 10; i++)
            {
                if (await ch_get.Writer.WaitToWriteAsync())
                {
                    if (!ch_get.Writer.TryWrite(n))
                    {
                        break;
                    }
                }
                if (await ch_set.Reader.WaitToReadAsync())
                {
                    if (!ch_set.Reader.TryRead(out n))
                    {
                        break;
                    }
                }
            }
            ch_get.Writer.Complete();
        }

        // 仮想ブラウザー
        public async IAsyncEnumerable<int> ReceiverIO()
        {
            Channel<int> ch_get = Channel.CreateBounded<int>(1);
            Channel<int> ch_set = Channel.CreateBounded<int>(1);
            Task task = Task.Run(() => GeneratorIO(ch_get, ch_set));
            while (await ch_get.Reader.WaitToReadAsync())
            {
                if (!ch_get.Reader.TryRead(out int n))
                {
                    break;
                }
                yield return n;
                if (await ch_set.Writer.WaitToWriteAsync())
                {
                    if (!ch_set.Writer.TryWrite(n * 2))
                    {
                        break;
                    }
                }
            }
            await task;
        }

        private async void button1_Click(object sender, EventArgs e)
        {
            IteratorClass iterator = new IteratorClass();
            string res = "";
            await foreach (int n in iterator.ReceiverIO())
            {
                res += n.ToString() + " ";
            }
            textBox1.Text = res;
        }

実行結果: 1 2 4 8 16 32 64 128 256 512

C#によるマルチコアのための非同期/並列処理プログラミング