SEEDS Creators Blog

とにかく大量のメールを受信するだけサーバが欲しかったのでPerlで作ってみました。

Net::Server::Mail::SMTP + Parallel::Preforkを使ったプリフォーク型サーバにしました。
Net::Server::Mail::SMTPでSMTPコマンドそれぞれの処理を書き、
Parallel::Preforkでプリフォークの制御を行っています。

以下がサーバのプログラムになります。
プログラムの構成としては、Starletのコードを参考にしています。

package TestMailReceiver;
use strict;
use warnings;
use IO::Socket::INET;
use Parallel::Prefork;
use Net::Server::Mail::SMTP;
use Socket qw(IPPROTO_TCP TCP_NODELAY);
sub new {
my ($class, %args) = @_;
my $self = bless {
host        => $args{host} || 0,
port        => $args{port} || 25,
max_workers => $args{max_workers} || 10,
};
$self;
}
sub setup_listener {
my $self = shift;
$self->{listen_sock} ||= IO::Socket::INET->new(
Listen    => SOMAXCONN,
LocalPort => $self->{port},
LocalAddr => $self->{host},
Proto     => 'tcp',
ReuseAddr => 1,
) or die "failed to listen to port $self->{port}:$!";
if ($^O eq 'linux') {
setsockopt($self->{listen_sock}, IPPROTO_TCP, 9, 1)
and $self->{_using_defer_accept} = 1;
}
}
sub accept_loop {
my ($self, $max_reqs_per_child) = @_;
my $proc_req_count = 0;
while (! defined $max_reqs_per_child || $proc_req_count < $max_reqs_per_child) {
if (my $conn = $self->{listen_sock}->accept) {
$self->{_is_deferred_accept} = $self->{_using_defer_accept};
$conn->blocking(0)
or die "failed to set socket to nonblocking mode:$!";
$conn->setsockopt(IPPROTO_TCP, TCP_NODELAY, 1)
or die "setsockopt(TCP_NODELAY) failed:$!";
$proc_req_count++;
my $smtp = Net::Server::Mail::SMTP->new( 'socket' => $conn );
$smtp->set_callback( 'RCPT' => sub { return (1) } );
$smtp->set_callback( 'DATA' => sub { return (1, 250, 'message queued') });
$smtp->process;
$conn->close;
}
}
}
sub run {
my ($self) = @_;
$self->setup_listener();
if ($self->{max_workers} != 0) {
# use Parallel::Prefork
my %pm_args = (
max_workers => $self->{max_workers},
trap_signals => {
TERM => 'TERM',
HUP  => 'TERM',
},
);
my $pm = Parallel::Prefork->new(\%pm_args);
while ($pm->signal_received !~ /^(TERM|USR1)$/) {
$pm->start and next;
$self->accept_loop();
$pm->finish;
}
} else {
# run directly, mainly for debugging
local $SIG{TERM} = sub { exit 0; };
while (1) {
$self->accept_loop();
}
}
}
1;
package main;
my $server = TestMailReceiver->new(
host => '',
port => 25,
max_workers => 200,
);
$server->run;

メールを受信してファイル書き込まずに捨てるだけのサーバです。
メールをファイルに書き出す処理など、普通のサーバで行われる処理を全部無くしているので、
大量のメールを受信してもだいぶ軽い動作になりました。

TODOとしては、テストでメール受信統計などを取ることができるようになればいいなと思ってます。

以下、主に使っているモジュールについての解説です。

Net::Server::Mail::SMTP

Net::Server::Mail::SMTPはSMTPプロトコルを実装したモジュールです。
Net::Server::Mail::SMTPを使えば、自前のSMTPサーバを簡単に作ることができます。
HELO, MAIL, RCPT, DATAなどのコマンドに対する処理をコールバックの形で書きます。

use strict;
use warnings;
use utf8;
use IO::Socket::INET;
use Net::Server::Mail::SMTP;
my @local_domains = qw(example.com example.org localhost);
my $msgid = 1;
sub add_queue {
return $msgid++;
}
my $server = IO::Socket::INET->new(
Listen => 1,
LocalPort => 2500,
);
my $conn;
while ($conn = $server->accept) {
my $smtp = Net::Server::Mail::SMTP->new(
socket => $conn,
);
# HELOコマンドの処理
$smtp->set_callback(
HELO => sub {
my ($session, $hostname) = @_;
if ($hostname eq 'localhost') {
return (0, 553, q(I don't like this hostname, try again.));
}
return 1;
},
);
# RCPTコマンドの処理
$smtp->set_callback(
RCPT => sub {
my ($session, $recipient) = @_;
my $domain;
if ($recipient =~ /@(.*)\s*$/) {
$domain = $1;
}
if (not defined $domain) {
return (0, 513, 'Syntax error.');
}
elsif (not(grep $domain eq $_, @local_domains)) {
return (0, 554, "$recipient: Recipient address rejected: Relay access denied");
}
return (1);
}
);
# DATAコマンドの処理
$smtp->set_callback(
DATA => sub {
my ($session, $data) = @_;
my $sender = $session->get_sender();
my @recipients = $session->get_recipients();
return (0, 554, 'Error: no valid recipients') unless @recipients;
# キューに追加。add_queueは自前で実装する必要がある
my $msgid = add_queue($sender, \@recipients, $data) or return(0);
return (1, 250, "message queued $msgid");
}
);
$smtp->process();
$conn->close();
}

このコードを実行し、以下のようにtelnetで繋いで各コマンドの動作を確かめると、
意図したレスポンスがサーバから返ってくることが確認できます。

$ telnet localhost 2500
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
220 debian2 SMTP Net::Server::Mail (Perl) Service ready
HELO localhost
553 I don't like this hostname, try again.
HELO hogehoge
250 Requested mail action okay, completed
MAIL FROM: 
250 sender hogehoge@example.com OK
RCPT TO: 
554 hogehoge@example.net: Recipient address rejected: Relay access denied
RCPT TO: 
250 recipient hogehoge@example.com OK
DATA
354 Start mail input; end with .
test
.
250 message queued 1
quit
221 debian2 Service closing transmission channel
Connection closed by foreign host.

テストでSMTPサーバを作成したい場合などに便利ですね。

自動テストのためにメールサーバを一時的に起動する

Parallel::Prefork

Parallel::Preforkはプリフォークサーバを書くためのモジュールです。
似た名前で似た使い方をするParallel::ForkManagerというモジュールがありますが、
Parallel::Preforkはシグナル管理が可能になっています。

たとえば次のようなプリフォーク型Echoサーバを書くことができます(あんま有用ではないですが)。

use strict;
use warnings;
use utf8;
use IO::Socket::INET;
use Parallel::Prefork;
sub MaxRequestsPerChild() { 100 }
my $listen_sock = IO::Socket::INET->new(
Listen => 5,
LocalAddr => '0.0.0.0:5000',
Proto  => 'tcp',
) or die $!;
my $pm = Parallel::Prefork->new({
max_workers => 10,
trap_signals => {
TERM => 'TERM',
HUP  => 'TERM',
}
});
while ($pm->signal_received ne 'TERM') {
# ワーカープロセス生成処理
$pm->start and next;
#### ここからワーカープロセス処理
# 1ワーカーがリクエストを受け付ける数
my $reqs_before_exit = MaxRequestsPerChild;
$SIG{TERM} = sub { $reqs_before_exit = 0 };
while ($reqs_before_exit-- > 0) {
if (my $conn = $listen_sock->accept()) {
while (my $str = ) {
print $conn "$reqs_before_exit($$): ".$str;
}
$conn->close;
}
}
# ワーカープロセスの終了処理
$pm->finish;
}
# 子プロセス待ち受け
$pm->wait_all_children;

このサーバを実行して、psコマンドで確認してみると「max_workersで設定した数値 + 親プロセス」分のプロセスが存在していることが確認できます。

$ perl prefork_echo.pl &
$ ps a | grep prefork_echo.pl
3963 pts/3    S      0:00 perl prefork_echo.pl
3964 pts/3    S      0:00 perl prefork_echo.pl
3965 pts/3    S      0:00 perl prefork_echo.pl
3966 pts/3    S      0:00 perl prefork_echo.pl
3968 pts/3    S      0:00 perl prefork_echo.pl
3969 pts/3    S      0:00 perl prefork_echo.pl
3970 pts/3    S      0:00 perl prefork_echo.pl
3971 pts/3    S      0:00 perl prefork_echo.pl
3972 pts/3    S      0:00 perl prefork_echo.pl
3973 pts/3    S      0:00 perl prefork_echo.pl
3983 pts/3    S      0:00 perl prefork_echo.pl
4085 pts/3    S+     0:00 grep prefork_echo

また、telnetを用いて接続も確認できます。

$ telnet localhost 5000
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
test
99(3967): test
test1
99(3967): test1
test2
99(3967): test2```
何回か繋ぎ直すと、異なるプロセスが対応する様子が確認できます。
pstreeなどで親プロセスの番号を調べて、SIGTERMを送るとサーバを終了させることができます。

$ kill -TERM 3963

このように簡単にプリフォークサーバを書けるようになっています。
<h2><span style="color: #2196f3">まとめ</span></h2>
メール送信系のシステムをテストする場合は、Net::Server::Mail::SMTPで受信サーバを書くのがとても良いです。
また、プリフォークサーバを書きたい場合は、Parallel::Preforkがとても便利です。
このようなツールを書くのに便利なモジュールがあって、やっぱりCPANは素晴らしいですね。
<h2><span style="color: #2196f3">参考</span></h2>
・Net::SMTP vs. Email::Send(er)?
<a href="http://blog.azumakuniyuki.org/2011/08/netsmtp-vs-emailsender.html">http://blog.azumakuniyuki.org/2011/08/netsmtp-vs-emailsender.html</a>
・Parallel::Prefork - Perl でマルチプロセスなサーバを書く方法
<a href="http://labs.cybozu.co.jp/blog/kazuho/archives/2008/04/parallel-prefork.php">http://labs.cybozu.co.jp/blog/kazuho/archives/2008/04/parallel-prefork.php</a>