Mojo::Reactor::Pollのソースを読んだときに調べたこと

reactorの意味

1. 化学反応器、反応装置
2. 原子炉

fileno関数

ファイルハンドルをfilenoに渡すと、
ファイルディスクリプタを取得できる。

Mojo::Reactor::Pollでは、
ioで渡されたファイルハンドルのファイルディスクリプタをキーとして$self->{io}に保存する。

example.pl

use strict;
use warnings;

open my $fh1, 'test.txt' or die "Can't open";
open my $fh2, 'test.txt' or die "Can't open";
warn fileno $fh1;
warn fileno $fh2;
close $fh1;
close $fh2;

open my $fh3, 'test.txt' or die "Can't open";
warn fileno $fh3;
close $fh3;

# 3 at fileno.pl line 6.
# 4 at fileno.pl line 7.
# 3 at fileno.pl line 12. -> closeしたらファイルディスクリプタが再利用される

IO::Poll

Mojo::Reactor::Pollでは、
ファイルハンドルのポーリングはIO::Pollを使用している。

以下のように使用している。

Mojo/Reactor/Poll.pm

# タイムアウト設定
$poll->poll($timeout);

# Read OKとなったハンドルに対してコールバックを実行
++$i and $self->_sandbox('Read', $self->{io}{fileno $_}{cb}, 0)
  for $poll->handles(POLLIN | POLLPRI | POLLHUP | POLLERR);

# Write OKとなったハンドルに対してコールバックを実行
++$i and $self->_sandbox('Write', $self->{io}{fileno $_}{cb}, 1)
   for $poll->handles(POLLOUT);

cometなチャットサーバ

以下のページでIO::Pollで書かれたコメットサーバをMojo::Reactor::Pollで書き直してみた。
http://d.hatena.ne.jp/jojo_a_gogogo/20070802/1186055332

comet.pl

use strict;
use warnings;
use IO::Socket;
use Mojo::Reactor::Poll;
use Mojo::Message::Request;

my $port = shift || 3000;

my $server = IO::Socket::INET->new(
    LocalPort => $port,
    Listen    => 10,
    Reuse     => 1,
) or die $@;

my $clients = {};
my @messages = ();

my $poll = Mojo::Reactor::Poll->new;
$poll->io($server => sub {
    my ($reactor, $writable) = @_;
    if (!$writable) {
        my $socket = $server->accept;
        $socket->blocking(0);
        $poll->io($socket => sub {
            my ($reactor, $writable) = @_;
            if (!$writable) {
                my $buf;
                while (defined $socket->sysread($buf, 8192)) {
                    my $req = Mojo::Message::Request->new;
                    $req->parse($buf);

                    unless ($buf) {
                        remove($poll, $socket);
                        last;
                    }

                    my $action = $req->param('a');

                    if ($action eq 'write') {
                        my $message = $req->param('message');
                        my $nickname = $req->param('nickname');

                        my $content = "{\"message\":\"$message\", \"nickname\":\"$nickname\"}";
                        push @messages, $content;

                        for my $client (values %{$clients}) {
                            $client->syswrite(
                                "HTTP/1.1 200 OK\r\n".
                                "Content-Type:text/html; charset=utf-8\r\n".
                                "Content-Length:".(length $content)."\r\n".
                                "Expires:-1\r\n".
                                "Pragma:no-cache\r\n".
                                "Cache-Control:must-revalidate, no-cache, no-store\r\n".
                                "\r\n".
                                $content
                            );
                            remove($poll, $socket);
                        }

                        $clients = {};
                    }
                    elsif ($action eq 'poll') {
                        #for debug
                        print "now polling...".fileno($socket);
                    }
                    elsif ($action eq 'init') {
                        #disconnect
                        my $message;
                        $message = join ',', @messages;
                        $socket->syswrite("{\"init\":[".$message."]}");
                        remove($poll, $socket);
                    }
                    else {
                        #disconnect
                        $socket->syswrite("oops bad request!!");
                        remove($poll, $socket);
                    }
                }
            }
        });
        $clients->{fileno $socket} = $socket;
    }
    else {
        warn "write";
    }
});

sub remove {
    my ($poll, $socket) = @_;
    delete $clients->{fileno $socket};
    $poll->remove($socket);
    $socket->close();
}

$poll->start unless $poll->is_running;

参考

https://metacpan.org/pod/Mojo::Reactor::Poll
https://metacpan.org/pod/IO::Poll