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.89 by root, Sun Jan 24 20:44:34 2010 UTC vs.
Revision 1.95 by root, Wed Apr 24 23:49:34 2013 UTC

1use AnyEvent ();
2
1use Coro; 3use Coro;
2use Coro::Semaphore; 4use Coro::Semaphore;
5use Coro::SemaphoreSet;
3use Coro::EV; 6use Coro::EV;
4use Coro::Socket; 7use Coro::Socket;
5use Coro::Signal; 8use Coro::Signal;
6use Coro::AIO (); 9use Coro::AIO ();
7 10
11use Fcntl;
8use HTTP::Date; 12use HTTP::Date;
9use POSIX (); 13use POSIX ();
10 14
11use Compress::Zlib (); 15use Compress::Zlib ();
12 16
13no utf8; 17use common::sense;
14use bytes;
15 18
16# at least on my machine, this thingy serves files 19# at least on my machine, this thingy serves files
17# quite a bit faster than apache, ;) 20# quite a bit faster than apache, ;)
18# and quite a bit slower than thttpd :( 21# and quite a bit slower than thttpd :(
19 22
20$SIG{PIPE} = 'IGNORE'; 23$SIG{PIPE} = 'IGNORE';
21 24
22our $accesslog; 25our $accesslog;
23our $errorlog; 26our $errorlog;
27our @listen_sockets;
24 28
25our $NOW; 29our $NOW;
26our $HTTP_NOW; 30our $HTTP_NOW;
27 31
28our $ERROR_LOG; 32our $ERROR_LOG;
29our $ACCESS_LOG; 33our $ACCESS_LOG;
34our $TRANSFER_LOCK = new Coro::SemaphoreSet; # lock to be acquired per ip
30 35
31our $update_time = EV::periodic 0, 1, undef, sub { 36our $update_time = EV::periodic 0, 1, undef, sub {
32 $NOW = time; 37 $NOW = time;
33 $HTTP_NOW = time2str $NOW; 38 $HTTP_NOW = time2str $NOW;
34}; 39};
… …
49} 54}
50 55
51sub slog { 56sub slog {
52 my $level = shift; 57 my $level = shift;
53 my $format = shift; 58 my $format = shift;
59
60 $format = sprintf $format, @_ if @_;
61
54 my $NOW = (POSIX::strftime "%Y-%m-%d %H:%M:%S", gmtime $::NOW); 62 my $NOW = (POSIX::strftime "%Y-%m-%d %H:%M:%S", gmtime $::NOW);
55 printf "$NOW: $format\n", @_; 63 print "$NOW: $format\n";
56 printf $errorlog "$NOW: $format\n", @_ if $errorlog; 64 print $errorlog "$NOW: $format\n", @_ if $errorlog;
57} 65}
58 66
59our $connections = new Coro::Semaphore $MAX_CONNECTS || 250; 67our $connections = new Coro::Semaphore $::MAX_CONNECTS || 250;
60our $httpevent = new Coro::Signal; 68our $httpevent = new Coro::Signal;
61 69
62our $queue_file = new transferqueue $MAX_TRANSFERS; 70our $queue_file = new transferqueue $::MAX_TRANSFERS;
63our $queue_index = new transferqueue 10; 71our $queue_index = new transferqueue 10;
64 72
65our $tbf_top = new tbf rate => $TBF_RATE || 100000; 73our $tbf_top = new tbf rate => $::TBF_RATE || 100000;
66 74
67my $unused_bytes = 0; 75my $unused_bytes = 0;
68my $unused_last = time; 76my $unused_last = time;
69 77
70sub unused_bandwidth { 78sub unused_bandwidth {
71 $unused_bytes += $_[0]; 79 $unused_bytes += $_[0];
72 if ($unused_last < $NOW - 30 && $unused_bytes / ($NOW - $unused_last) > 50000) { 80 if ($unused_last < $NOW - 30 && $unused_bytes / ($NOW - $unused_last) > 50000) {
73 $unused_last = $NOW; 81 $unused_last = $NOW;
74 $unused_bytes = 0; 82 $unused_bytes = 0;
75 $queue_file->force_wake_next; 83 $queue_file->force_wake_next
76 slog 1, "forced filetransfer due to unused bandwidth"; 84 and slog 1, "forced filetransfer due to unused bandwidth";
77 } 85 }
78} 86}
79 87
80sub listen_on { 88sub listen_on {
81 my $listen = $_[0]; 89 my $listen = $_[0];
… …
103 } 111 }
104 }; 112 };
105} 113}
106 114
107my $http_port = new Coro::Socket 115my $http_port = new Coro::Socket
108 LocalAddr => $SERVER_HOST, 116 LocalAddr => $::SERVER_HOST,
109 LocalPort => $SERVER_PORT, 117 LocalPort => $::SERVER_PORT,
110 ReuseAddr => 1, 118 ReuseAddr => 1,
111 Listen => 50, 119 Listen => 50,
112 or die "unable to start server"; 120 or die "unable to start server";
113 121
114listen_on $http_port; 122listen_on $http_port;
115 123
116if ($SERVER_PORT2) { 124if ($::SERVER_PORT2) {
117 my $http_port = new Coro::Socket 125 my $http_port = new Coro::Socket
118 LocalAddr => $SERVER_HOST, 126 LocalAddr => $::SERVER_HOST,
119 LocalPort => $SERVER_PORT2, 127 LocalPort => $::SERVER_PORT2,
120 ReuseAddr => 1, 128 ReuseAddr => 1,
121 Listen => 50, 129 Listen => 50,
122 or die "unable to start server"; 130 or die "unable to start server";
123 131
124 listen_on $http_port; 132 listen_on $http_port;
125} 133}
126 134
127package conn; 135package conn;
128 136
129use strict; 137use common::sense;
130use bytes;
131 138
132use Socket; 139use Socket;
133use HTTP::Date; 140use HTTP::Date;
134use Convert::Scalar 'weaken'; 141use Convert::Scalar 'weaken';
135use IO::AIO; 142use IO::AIO;
143use AnyEvent::AIO;
136 144
137IO::AIO::min_parallel $::AIO_PARALLEL; 145IO::AIO::min_parallel $::AIO_PARALLEL;
138
139our $AIO_WATCHER = EV::io IO::AIO::poll_fileno, EV::READ, \&IO::AIO::poll_cb;
140 146
141our %conn; # $conn{ip}{self} => connobj 147our %conn; # $conn{ip}{self} => connobj
142our %uri; # $uri{ip}{uri}{self} 148our %uri; # $uri{ip}{uri}{self}
143our %blocked; 149our %blocked;
144our %mimetype; 150our %mimetype;
… …
204 for (keys %blocked) { 210 for (keys %blocked) {
205 delete $blocked{$_} unless $blocked{$_}[0] > $::NOW; 211 delete $blocked{$_} unless $blocked{$_}[0] > $::NOW;
206 } 212 }
207} 213}
208 214
209our $PRUNE_WATCHER = EV::timer 60, 60, \&prune_caches; 215our $PRUNE_WATCHER = AE::timer 60, 60, \&prune_caches;
210 216
211sub slog { 217sub slog {
212 my $self = shift; 218 my $self = shift;
213 main::slog($_[0], "$self->{remote_id}> $_[1]"); 219 main::slog($_[0], "$self->{remote_id}> $_[1]");
214} 220}
… …
557 $self->err (416, "not satisfiable", $hdr, ""); 563 $self->err (416, "not satisfiable", $hdr, "");
558 564
559satisfiable: 565satisfiable:
560 # check for segmented downloads 566 # check for segmented downloads
561 if ($l && $::NO_SEGMENTED) { 567 if ($l && $::NO_SEGMENTED) {
562 my $timeout = $::NOW + 15; 568 my $timeout = $::NOW + 60;
563 while (keys %{$uri{$self->{remote_id}}{$self->{uri}}} > 1) { 569 while (keys %{$uri{$self->{remote_id}}{$self->{uri}}} > 1) {
564 if ($timeout <= $::NOW) { 570 if ($timeout <= $::NOW) {
565 $self->block ($::BLOCKTIME, "segmented downloads are forbidden");
566 #$self->err_segmented_download; 571 $self->err_segmented_download;
567 } else { 572 } else {
568 $httpevent->wait; 573 $httpevent->wait;
569 } 574 }
570 } 575 }
571 } 576 }
… …
587 592
588 if ($self->{method} eq "GET") { 593 if ($self->{method} eq "GET") {
589 $self->{time} = $::NOW; 594 $self->{time} = $::NOW;
590 $self->{written} = 0; 595 $self->{written} = 0;
591 596
592 open my $fh, "<", $self->{path} 597 my $fh = Coro::AIO::aio_open $self->{path}, Fcntl::O_RDONLY, 0
593 or die "$self->{path}: late open failure ($!)"; 598 or die "$self->{path}: late open failure ($!)";
594 599
595 $h -= $l - 1; 600 $h -= $l - 1;
596 601
597 my $transfer = $queue->start_transfer ($h); 602 my $transfer = $queue->start_transfer ($h);
598 my $locked; 603 my $locked;
599 my $bufsize = $::WAIT_BUFSIZE; # initial buffer size 604 my $bufsize = $::WAIT_BUFSIZE; # initial buffer size
600 605
601 while ($h > 0) { 606 while ($h > 0) {
607 Coro::cede;
608 my $transfer_lock = $TRANSFER_LOCK->guard ($self->{remote_id});
609
602 unless ($locked) { 610 unless ($locked) {
603 if ($locked ||= $transfer->try ($::WAIT_INTERVAL)) { 611 if ($locked ||= $transfer->try ($::WAIT_INTERVAL)) {
604 $bufsize = $::BUFSIZE; 612 $bufsize = $::BUFSIZE;
605 $self->{time} = $::NOW; 613 $self->{time} = $::NOW;
606 $self->{written} = 0; 614 $self->{written} = 0;
… …
613 } 621 }
614 622
615 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
616 or last; 624 or last;
617 625
626 # readahead to work around rijk disk issues
627 IO::AIO::aio_readahead $fh, $l + $bufsize, $bufsize;
628
618 $tbf->request (length $buf); 629 $tbf->request (length $buf);
619 my $w = $self->{fh}->syswrite ($buf) 630 my $w = $self->{fh}->syswrite ($buf)
620 or last; 631 or last;
621 $::written += $w; 632 $::written += $w;
622 $self->{written} += $w; 633 $self->{written} += $w;

Diff Legend

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