ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/cvsroot/Coro/myhttpd/httpd.pl
(Generate patch)

Comparing cvsroot/Coro/myhttpd/httpd.pl (file contents):
Revision 1.83 by root, Sun Dec 3 17:49:22 2006 UTC vs.
Revision 1.93 by root, Sun Mar 21 00:21:14 2010 UTC

1use Coro; 1use Coro;
2use Coro::Semaphore; 2use Coro::Semaphore;
3use Coro::SemaphoreSet;
3use Coro::Event; 4use Coro::EV;
4use Coro::Socket; 5use Coro::Socket;
5use Coro::Signal; 6use Coro::Signal;
6use Coro::AIO (); 7use Coro::AIO ();
7 8
9use Fcntl;
8use HTTP::Date; 10use HTTP::Date;
9use POSIX (); 11use POSIX ();
10 12
11use Compress::Zlib (); 13use Compress::Zlib ();
12 14
13no utf8; 15use common::sense;
14use bytes;
15 16
16# at least on my machine, this thingy serves files 17# at least on my machine, this thingy serves files
17# quite a bit faster than apache, ;) 18# quite a bit faster than apache, ;)
18# and quite a bit slower than thttpd :( 19# and quite a bit slower than thttpd :(
19 20
20$SIG{PIPE} = 'IGNORE'; 21$SIG{PIPE} = 'IGNORE';
21 22
22our $accesslog; 23our $accesslog;
23our $errorlog; 24our $errorlog;
25our @listen_sockets;
24 26
25our $NOW; 27our $NOW;
26our $HTTP_NOW; 28our $HTTP_NOW;
27 29
28our $ERROR_LOG; 30our $ERROR_LOG;
29our $ACCESS_LOG; 31our $ACCESS_LOG;
32our $TRANSFER_LOCK = new Coro::SemaphoreSet; # lock to be acquired per ip
30 33
31Event->timer(interval => 1, hard => 1, cb => sub { 34our $update_time = EV::periodic 0, 1, undef, sub {
32 $NOW = time; 35 $NOW = time;
33 $HTTP_NOW = time2str $NOW; 36 $HTTP_NOW = time2str $NOW;
34})->now; 37};
38$update_time->invoke;
35 39
36if ($ERROR_LOG) { 40if ($ERROR_LOG) {
37 use IO::Handle; 41 use IO::Handle;
38 open $errorlog, ">>$ERROR_LOG" 42 open $errorlog, ">>$ERROR_LOG"
39 or die "$ERROR_LOG: $!"; 43 or die "$ERROR_LOG: $!";
… …
48} 52}
49 53
50sub slog { 54sub slog {
51 my $level = shift; 55 my $level = shift;
52 my $format = shift; 56 my $format = shift;
57
58 $format = sprintf $format, @_ if @_;
59
53 my $NOW = (POSIX::strftime "%Y-%m-%d %H:%M:%S", gmtime $::NOW); 60 my $NOW = (POSIX::strftime "%Y-%m-%d %H:%M:%S", gmtime $::NOW);
54 printf "$NOW: $format\n", @_; 61 print "$NOW: $format\n";
55 printf $errorlog "$NOW: $format\n", @_ if $errorlog; 62 print $errorlog "$NOW: $format\n", @_ if $errorlog;
56} 63}
57 64
58our $connections = new Coro::Semaphore $MAX_CONNECTS || 250; 65our $connections = new Coro::Semaphore $::MAX_CONNECTS || 250;
59our $httpevent = new Coro::Signal; 66our $httpevent = new Coro::Signal;
60 67
61our $queue_file = new transferqueue $MAX_TRANSFERS; 68our $queue_file = new transferqueue $::MAX_TRANSFERS;
62our $queue_index = new transferqueue 10; 69our $queue_index = new transferqueue 10;
63 70
64our $tbf_top = new tbf rate => $TBF_RATE || 100000; 71our $tbf_top = new tbf rate => $::TBF_RATE || 100000;
65 72
66my $unused_bytes = 0; 73my $unused_bytes = 0;
67my $unused_last = time; 74my $unused_last = time;
68 75
69sub unused_bandwidth { 76sub unused_bandwidth {
70 $unused_bytes += $_[0]; 77 $unused_bytes += $_[0];
71 if ($unused_last < $NOW - 30 && $unused_bytes / ($NOW - $unused_last) > 50000) { 78 if ($unused_last < $NOW - 30 && $unused_bytes / ($NOW - $unused_last) > 50000) {
72 $unused_last = $NOW; 79 $unused_last = $NOW;
73 $unused_bytes = 0; 80 $unused_bytes = 0;
74 $queue_file->force_wake_next; 81 $queue_file->force_wake_next
75 slog 1, "forced filetransfer due to unused bandwidth"; 82 and slog 1, "forced filetransfer due to unused bandwidth";
76 }
77}
78
79my @newcons;
80my @pool;
81
82# one "execution thread"
83sub handler {
84 while () {
85 if (@newcons) {
86 eval {
87 conn->new (@{pop @newcons})->handle;
88 };
89 slog 1, "$@" if $@ && !ref $@;
90
91 $httpevent->broadcast; # only for testing, but doesn't matter much
92
93 $connections->up;
94 } else {
95 last if @pool >= $MAX_POOL;
96 push @pool, $Coro::current;
97 schedule;
98 }
99 } 83 }
100} 84}
101 85
102sub listen_on { 86sub listen_on {
103 my $listen = $_[0]; 87 my $listen = $_[0];
… …
107 # the "main thread" 91 # the "main thread"
108 async { 92 async {
109 slog 1, "accepting connections"; 93 slog 1, "accepting connections";
110 while () { 94 while () {
111 $connections->down; 95 $connections->down;
112 push @newcons, [$listen->accept]; 96 my @conn = $listen->accept;
113 #slog 3, "accepted @$connections ".scalar(@pool); 97 #slog 3, "accepted @$connections ".scalar(@pool);
114 if (@pool) { 98
115 (pop @pool)->ready; 99 async_pool {
116 } else { 100 eval {
117 async \&handler; 101 conn->new (@conn)->handle;
102 };
103 slog 1, "$@" if $@ && !ref $@;
104
105 $httpevent->broadcast; # only for testing, but doesn't matter much
106
107 $connections->up;
118 } 108 }
119 } 109 }
120 }; 110 };
121} 111}
122 112
123my $http_port = new Coro::Socket 113my $http_port = new Coro::Socket
124 LocalAddr => $SERVER_HOST, 114 LocalAddr => $::SERVER_HOST,
125 LocalPort => $SERVER_PORT, 115 LocalPort => $::SERVER_PORT,
126 ReuseAddr => 1, 116 ReuseAddr => 1,
127 Listen => 50, 117 Listen => 50,
128 or die "unable to start server"; 118 or die "unable to start server";
129 119
130listen_on $http_port; 120listen_on $http_port;
131 121
132if ($SERVER_PORT2) { 122if ($::SERVER_PORT2) {
133 my $http_port = new Coro::Socket 123 my $http_port = new Coro::Socket
134 LocalAddr => $SERVER_HOST, 124 LocalAddr => $::SERVER_HOST,
135 LocalPort => $SERVER_PORT2, 125 LocalPort => $::SERVER_PORT2,
136 ReuseAddr => 1, 126 ReuseAddr => 1,
137 Listen => 50, 127 Listen => 50,
138 or die "unable to start server"; 128 or die "unable to start server";
139 129
140 listen_on $http_port; 130 listen_on $http_port;
141} 131}
142 132
143package conn; 133package conn;
144 134
145use strict; 135use common::sense;
146use bytes;
147 136
148use Socket; 137use Socket;
149use HTTP::Date; 138use HTTP::Date;
150use Convert::Scalar 'weaken'; 139use Convert::Scalar 'weaken';
151use IO::AIO; 140use IO::AIO;
152 141
153IO::AIO::min_parallel $::AIO_PARALLEL; 142IO::AIO::min_parallel $::AIO_PARALLEL;
154 143
155Event->io (fd => IO::AIO::poll_fileno, 144our $AIO_WATCHER = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
156 poll => 'r', async => 1,
157 cb => \&IO::AIO::poll_cb);
158 145
159our %conn; # $conn{ip}{self} => connobj 146our %conn; # $conn{ip}{self} => connobj
160our %uri; # $uri{ip}{uri}{self} 147our %uri; # $uri{ip}{uri}{self}
161our %blocked; 148our %blocked;
162our %mimetype; 149our %mimetype;
… …
222 for (keys %blocked) { 209 for (keys %blocked) {
223 delete $blocked{$_} unless $blocked{$_}[0] > $::NOW; 210 delete $blocked{$_} unless $blocked{$_}[0] > $::NOW;
224 } 211 }
225} 212}
226 213
227Event->timer (interval => 60, cb => \&prune_caches); 214our $PRUNE_WATCHER = EV::timer 60, 60, \&prune_caches;
228 215
229sub slog { 216sub slog {
230 my $self = shift; 217 my $self = shift;
231 main::slog($_[0], "$self->{remote_id}> $_[1]"); 218 main::slog($_[0], "$self->{remote_id}> $_[1]");
232} 219}
… …
425sub map_uri { 412sub map_uri {
426 my $self = shift; 413 my $self = shift;
427 my $host = $self->{server_name}; 414 my $host = $self->{server_name};
428 my $uri = $self->{uri}; 415 my $uri = $self->{uri};
429 416
417 $host =~ /[\/\\]/
418 and $self->err(400, "bad request");
419
430 # some massaging, also makes it more secure 420 # some massaging, also makes it more secure
431 $uri =~ s/%([0-9a-fA-F][0-9a-fA-F])/chr hex $1/ge; 421 $uri =~ s/%([0-9a-fA-F][0-9a-fA-F])/chr hex $1/ge;
432 $uri =~ s%//+%/%g; 422 $uri =~ s%//+%/%g;
433 $uri =~ s%/\.(?=/|$)%%g; 423 $uri =~ s%/\.(?=/|$)%%g;
434 1 while $uri =~ s%/[^/]+/\.\.(?=/|$)%%; 424 1 while $uri =~ s%/[^/]+/\.\.(?=/|$)%%;
… …
481 } else { 471 } else {
482 $self->err (404, "not found"); 472 $self->err (404, "not found");
483 } 473 }
484 } else { 474 } else {
485 475
486 stat $path 476 Coro::AIO::aio_stat $path
487 or $self->err (404, "not found"); 477 and $self->err (404, "not found");
488 478
489 $self->{stat} = [stat _]; 479 $self->{stat} = [stat _];
490 480
491 # idiotic netscape sends idiotic headers AGAIN 481 # idiotic netscape sends idiotic headers AGAIN
492 my $ims = $self->{h}{"if-modified-since"} =~ /^([^;]+)/ 482 my $ims = $self->{h}{"if-modified-since"} =~ /^([^;]+)/
… …
553 }; 543 };
554 544
555 my @code = (200, "ok"); 545 my @code = (200, "ok");
556 my ($l, $h); 546 my ($l, $h);
557 547
558 if ($self->{h}{range} =~ /^bytes=(.*)$/) { 548 if ($self->{h}{range} =~ /^bytes=(.*)$/i) {
559 for (split /,/, $1) { 549 for (split /,/, $1) {
560 if (/^-(\d+)$/) { 550 if (/^-(\d+)$/) {
561 ($l, $h) = ($length - $1, $length - 1); 551 ($l, $h) = ($length - $1, $length - 1);
562 } elsif (/^(\d+)-(\d*)$/) { 552 } elsif (/^(\d+)-(\d*)$/) {
563 ($l, $h) = ($1, ($2 ne "" || $2 >= $length) ? $2 : $length - 1); 553 ($l, $h) = ($1, ($2 ne "" || $2 >= $length) ? $2 : $length - 1);
… …
602 592
603 if ($self->{method} eq "GET") { 593 if ($self->{method} eq "GET") {
604 $self->{time} = $::NOW; 594 $self->{time} = $::NOW;
605 $self->{written} = 0; 595 $self->{written} = 0;
606 596
607 open my $fh, "<", $self->{path} 597 my $fh = Coro::AIO::aio_open $self->{path}, Fcntl::O_RDONLY, 0
608 or die "$self->{path}: late open failure ($!)"; 598 or die "$self->{path}: late open failure ($!)";
609 599
610 $h -= $l - 1; 600 $h -= $l - 1;
611 601
612 my $transfer = $queue->start_transfer ($h); 602 my $transfer = $queue->start_transfer ($h);
613 my $locked; 603 my $locked;
614 my $bufsize = $::WAIT_BUFSIZE; # initial buffer size 604 my $bufsize = $::WAIT_BUFSIZE; # initial buffer size
615 605
616 while ($h > 0) { 606 while ($h > 0) {
607 Coro::cede;
608 my $transfer_lock = $TRANSFER_LOCK->guard ($self->{remote_id});
609
617 unless ($locked) { 610 unless ($locked) {
618 if ($locked ||= $transfer->try ($::WAIT_INTERVAL)) { 611 if ($locked ||= $transfer->try ($::WAIT_INTERVAL)) {
619 $bufsize = $::BUFSIZE; 612 $bufsize = $::BUFSIZE;
620 $self->{time} = $::NOW; 613 $self->{time} = $::NOW;
621 $self->{written} = 0; 614 $self->{written} = 0;
… …
629 622
630 Coro::AIO::aio_read $fh, $l, ($h > $bufsize ? $bufsize : $h), my $buf, 0 623 Coro::AIO::aio_read $fh, $l, ($h > $bufsize ? $bufsize : $h), my $buf, 0
631 or last; 624 or last;
632 625
633 $tbf->request (length $buf); 626 $tbf->request (length $buf);
634 my $w = syswrite $self->{fh}, $buf 627 my $w = $self->{fh}->syswrite ($buf)
635 or last; 628 or last;
636 $::written += $w; 629 $::written += $w;
637 $self->{written} += $w; 630 $self->{written} += $w;
638 $l += $w; 631 $l += $w;
639 } 632 }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines