root/lang/perl/Moobal/trunk/lib/Moobal/Component/Proxy.pm @ 11968

Revision 11968, 53.5 kB (checked in by daisuke, 5 years ago)

shuffle shuffle, that's all we do

  • Property svn:keywords set to Id
Line 
1######################################################################
2# HTTP Connection from a reverse proxy client
3#
4# Copyright 2004, Danga Interactive, Inc.
5# Copyright 2005-2007, Six Apart, Ltd.
6#
7package Moobal::Component::Proxy;
8use Moose;
9
10extends 'Moobal::Component::HTTP';
11
12use Moobal::ChunkedUploadState;
13use Moobal::Util;
14
15has 'backend' => (
16    is => 'rw',
17    isa => 'Moobal::Component::BackendHTTP'
18);
19
20has 'is_buffering' => (
21    is => 'rw',
22    isa => 'Bool',
23    default => 0
24);
25
26has 'spawn_lock' => (
27    is => 'rw',
28    isa => 'Bool',
29    default => 0
30);
31
32has 'pending_connect_count' => (
33    is => 'rw',
34    isa => 'Int',
35    default => 0
36);
37
38has 'bored_backends' => (
39    is => 'rw',
40    isa => 'ArrayRef[Moobal::Component::BackendHTP]',
41    default => sub { [] }
42);
43
44has 'waiting_client_count' => (
45    is => 'rw',
46    isa => 'Int',
47    default => 0
48);
49
50has 'connect_ahead' => (
51    is => 'rw',
52    isa => 'Int',
53    default => 0
54);
55
56has 'pending_connects' => (
57    is => 'rw',
58    isa => 'HashRef',
59    default => sub { +{} }
60);
61
62# called by backend connection after it becomes writable
63my @BoredBackends;
64sub register_boredom {
65    my ($self, $be) = @_;
66
67    # note that this backend is no longer pending a connect,
68    # if we thought it was before.  but not if it's a persistent
69    # connection asking to be re-used.
70    unless ($be->use_count) {
71        $self->clear_pending_connect($be);
72    }
73
74    # it is possible that this backend is part of a different pool that we're
75    # no longer using... if that's the case, we want to close it
76    return unless $self->verify_generation($be);
77
78    # now try to fetch a client for it
79    my $cp = $self->get_client;
80    if ($cp) {
81        return if $be->assign_client($cp);
82
83        # don't want to lose client, so we (unfortunately)
84        # stick it at the end of the waiting queue.
85        # fortunately, assign_client shouldn't ever fail.
86        $self->request_backend_connection($cp);
87    }
88
89    # don't hang onto more bored, persistent connections than
90    # has been configured for connect-ahead
91    if ($be->use_count) {
92        my $current_bored = scalar @BoredBackends;
93        if ($current_bored >= $self->{backend_persist_cache}) {
94            $be->close('too_many_bored');
95            return;
96        }
97    }
98
99    # put backends which are known to be bound to processes
100    # and not to TCP stacks at the beginning where they'll
101    # be used first
102    if ($be->{has_attention}) {
103        unshift @BoredBackends, $be;
104    } else {
105        push @BoredBackends, $be;
106    }
107}
108
109
110my @fields = (
111# use fields (
112            'backend',             # Moobal::Component::BackendHTTP object (or undef if disconnected)
113            'backend_requested',   # true if we've requested a backend for this request
114            'reconnect_count',     # number of times we've tried to reconnect to backend
115            'high_priority',       # boolean; 1 if we are or were in the high priority queue
116            'low_priority',        # boolean; 1 if we are or were in the low priority queue
117            'reproxy_uris',        # arrayref; URIs to reproxy to, in order
118            'reproxy_expected_size', # int: size of response we expect to get back for reproxy
119            'currently_reproxying',  # arrayref; the host info and URI we're reproxying right now
120            'content_length_remain', # int: amount of data we're still waiting for
121            'responded',           # bool: whether we've already sent a response to the user or not
122            'last_request_time',   # int: time that we last received a request
123            'primary_res_hdrs',  # if defined, we are doing a transparent reproxy-URI
124                                 # and the headers we get back aren't necessarily
125                                 # the ones we want.  instead, get most headers
126                                 # from the provided res headers object here.
127            'is_buffering',        # bool; if we're buffering some/all of a request to memory/disk
128            'is_writing',          # bool; if on, we currently have an aio_write out
129            'start_time',          # hi-res time when we started getting data to upload
130            'bufh',                # buffered upload filehandle object
131            'bufilename',          # string; buffered upload filename
132            'bureason',            # string; if defined, the reason we're buffering to disk
133            'buoutpos',            # int; buffered output position
134            'backend_stalled',   # boolean:  if backend has shut off its reads because we're too slow.
135            'unread_data_waiting',  # boolean:  if we shut off reads while we know data is yet to be read from client
136            'chunked_upload_state', # bool/obj:  if processing a chunked upload, Moobal::ChunkedUploadState object, else undef
137            'request_body_length',  # integer:  request's body length, either as-declared,
138                                    #           or calculated after chunked upload is complete
139
140            # for perlbal sending out UDP packets related to upload status (for xmlhttprequest upload bar)
141            'last_upload_packet',  # unixtime we last sent a UDP upload packet
142            'upload_session',      # client's self-generated upload session
143
144            # error-retrying stuff
145            'retry_count',         # number of times we've retried this request so far after getting 500 errors
146            );
147
148use constant READ_SIZE         => 131072;    # 128k, ~common TCP window size?
149use constant READ_AHEAD_SIZE   =>  32768;    # kinda arbitrary.  sum of these two is max stored per connection while waiting for backend.
150use Errno qw( EPIPE ENOENT ECONNRESET EAGAIN );
151use POSIX qw( O_CREAT O_TRUNC O_RDWR O_RDONLY );
152use Time::HiRes qw( gettimeofday tv_interval );
153
154my $udp_sock;
155
156sub new_from_base {
157    my $class = shift;
158    my $cb = shift; # Moobal::ClientHTTPBase
159    Moobal::Util::rebless($cb, $class);
160    $cb->init;
161    $cb->watch_read(1);
162    $cb->handle_request;
163    return $cb;
164}
165
166sub init {
167    my $self = $_[0]; # Moobal::Component::Proxy
168
169    $self->{last_request_time} = 0;
170
171    $self->{backend} = undef;
172    $self->{high_priority} = 0;
173    $self->{low_priority} = 0;
174
175    $self->{responded} = 0;
176    $self->{unread_data_waiting} = 0;
177    $self->{content_length_remain} = undef;
178    $self->{backend_requested} = 0;
179
180    $self->{is_buffering} = 0;
181    $self->{is_writing} = 0;
182    $self->{start_time} = undef;
183    $self->{bufh} = undef;
184    $self->{bufilename} = undef;
185    $self->{buoutpos} = 0;
186    $self->{bureason} = undef;
187    $self->{chunked_upload_state} = undef;
188    $self->{request_body_length} = undef;
189
190    $self->{reproxy_uris} = undef;
191    $self->{reproxy_expected_size} = undef;
192    $self->{currently_reproxying} = undef;
193
194    $self->{retry_count} = 0;
195}
196
197# given a service name, re-request (GET/HEAD only) to that service, even though
198# you've already done a request to your original service
199sub start_reproxy_service {
200    my $self = $_[0]; # Moobal::Component::Proxy
201    my $primary_res_hdrs = $_[1]; # Moobal::HTTPHeaders
202    my $svc_name = $_[2];
203
204    my $svc = $svc_name ? Moobal->service($svc_name) : undef;
205    unless ($svc) {
206        $self->_simple_response(404, "Vhost twiddling not configured for requested pair.");
207        return 1;
208    }
209
210    $self->{backend_requested} = 0;
211    $self->{backend} = undef;
212    $self->{res_headers} = $primary_res_hdrs;
213
214    $svc->adopt_base_client($self);
215}
216
217# call this with a string of space separated URIs to start a process
218# that will fetch the item at the first and return it to the user,
219# on failure it will try the second, then third, etc
220sub start_reproxy_uri {
221    my $self = $_[0]; # Moobal::Component::Proxy
222    my $primary_res_hdrs = $_[1]; # Moobal::HTTPHeaders
223    my $urls = $_[2];
224
225    # at this point we need to disconnect from our backend
226    $self->{backend} = undef;
227
228    # failure if we have no primary response headers
229    return unless $self->{primary_res_hdrs} ||= $primary_res_hdrs;
230
231    # construct reproxy_uri list
232    if (defined $urls) {
233        my @uris = split /\s+/, $urls;
234        $self->{currently_reproxying} = undef;
235        $self->{reproxy_uris} = [];
236        foreach my $uri (@uris) {
237            next unless $uri =~ m!^http://(.+?)(?::(\d+))?(/.*)?$!;
238            push @{$self->{reproxy_uris}}, [ $1, $2 || 80, $3 || '/' ];
239        }
240    }
241
242    # if we get in here and we have currently_reproxying defined, then something
243    # happened and we want to retry that one
244    if ($self->{currently_reproxying}) {
245        unshift @{$self->{reproxy_uris}}, $self->{currently_reproxying};
246        $self->{currently_reproxying} = undef;
247    }
248
249    # if we have no uris in our list now, tell the user 404
250    return $self->_simple_response(503)
251        unless @{$self->{reproxy_uris} || []};
252
253    # set the expected size if we got a content length in our headers
254    if ($primary_res_hdrs && (my $expected_size = $primary_res_hdrs->header('X-REPROXY-EXPECTED-SIZE'))) {
255        $self->{reproxy_expected_size} = $expected_size;
256    }
257
258    # pass ourselves off to the reproxy manager
259    $self->state('wait_backend');
260    Moobal::ReproxyManager::do_reproxy($self);
261}
262
263# called by the reproxy manager when we can't get to our requested backend
264sub try_next_uri {
265    my $self = $_[0]; # Moobal::Component::Proxy
266
267    shift @{$self->{reproxy_uris}};
268    $self->{currently_reproxying} = undef;
269    $self->start_reproxy_uri();
270}
271
272# returns true if this ClientProxy is too many bytes behind the backend
273sub too_far_behind_backend {
274    my $self    = $_[0]; # Moobal::Component::Proxy
275    my $backend = $self->{backend}   or return 0; # Moobal::Component::BackendHTTP
276
277    # if a backend doesn't have a service, it's a
278    # ReproxyManager-created backend, and thus it should use the
279    # 'buffer_size_reproxy_url' parameter for acceptable buffer
280    # widths, and not the regular 'buffer_size'.  this lets people
281    # tune buffers depending on the types of webservers.  (assumption
282    # being that reproxied-to webservers are event-based and it's okay
283    # to tie the up longer in favor of using less buffer memory in
284    # perlbal)
285    my $max_buffer = defined $backend->{service} ?
286        $self->{service}->{buffer_size} :
287        $self->{service}->{buffer_size_reproxy_url};
288
289    return $self->{write_buf_size} > $max_buffer;
290}
291
292# this is a callback for when a backend has been created and is
293# ready for us to do something with it
294sub use_reproxy_backend {
295    my $self = $_[0]; # Moobal::Component::Proxy
296    my $be = $_[1]; # Moobal::Component::BackendHTTP
297
298    # get a URI
299    my $datref = $self->{currently_reproxying} = shift @{$self->{reproxy_uris}};
300    unless (defined $datref) {
301        # return error and close the backend
302        $be->close('invalid_uris');
303        return $self->_simple_response(503);
304    }
305
306    # now send request
307    $self->{backend} = $be;
308    $be->{client} = $self;
309
310    my $extra_hdr = "";
311    if (my $range = $self->{req_headers}->header("Range")) {
312        $extra_hdr .= "Range: $range\r\n";
313    }
314    if (my $host = $self->{req_headers}->header("Host")) {
315        $extra_hdr .= "Host: $host\r\n";
316    }
317
318    my $req_method = $self->{req_headers}->request_method eq 'HEAD' ? 'HEAD' : 'GET';
319    my $headers = "$req_method $datref->[2] HTTP/1.0\r\nConnection: keep-alive\r\n${extra_hdr}\r\n";
320
321    $be->{req_headers} = Moobal::HTTPHeaders->new(\$headers);
322    $be->state('sending_req');
323    $self->state('backend_req_sent');
324    $be->write($be->{req_headers}->to_string_ref);
325    $be->watch_read(1);
326    $be->watch_write(1);
327}
328
329# this is called when a transient backend getting a reproxied URI has received
330# a response from the server and is ready for us to deal with it
331sub backend_response_received {
332    my $self = $_[0]; # Moobal::Component::Proxy
333    my $be = $_[1]; # Moobal::Component::BackendHTTP
334
335    # a response means that we are no longer currently waiting on a reproxy, and
336    # don't want to retry this URI
337    $self->{currently_reproxying} = undef;
338
339    # we fail if we got something that's NOT a 2xx code, OR, if we expected
340    # a certain size and got back something different
341    my $code = $be->{res_headers}->response_code + 0;
342
343    my $bad_code = sub {
344        return 0 if $code >= 200 && $code <= 299;
345        return 0 if $code == 416;
346        return 1;
347    };
348
349    my $bad_size = sub {
350        return 0 unless defined $self->{reproxy_expected_size};
351        return $self->{reproxy_expected_size} != $be->{res_headers}->header('Content-length');
352    };
353
354    if ($bad_code->() || $bad_size->()) {
355        # fall back to an alternate URL
356        $be->{client} = undef;
357        $be->close('non_200_reproxy');
358        $self->try_next_uri;
359        return 1;
360    }
361    return 0;
362}
363
364sub start_reproxy_file {
365    my $self = shift; # Moobal::Component::Proxy
366    my $file = shift; # filename to reproxy
367    my $hd = shift; # Moobal::HTTPHeaders headers from backend, in need of cleanup
368    # at this point we need to disconnect from our backend
369    $self->{backend} = undef;
370
371    # call hook for pre-reproxy
372    return if $self->{service}->run_hook("start_file_reproxy", $self, \$file);
373
374    # set our expected size
375    if (my $expected_size = $hd->header('X-REPROXY-EXPECTED-SIZE')) {
376        $self->{reproxy_expected_size} = $expected_size;
377    }
378
379    # start an async stat on the file
380    $self->state('wait_stat');
381    Moobal::AIO::aio_stat($file, sub {
382
383        # if the client's since disconnected by the time we get the stat,
384        # just bail.
385        return if $self->{closed};
386
387        my $size = -s _;
388
389        unless ($size) {
390            # FIXME: POLICY: 404 or retry request to backend w/o reproxy-file capability?
391            return $self->_simple_response(404);
392        }
393        if (defined $self->{reproxy_expected_size} && $self->{reproxy_expected_size} != $size) {
394            # 404; the file size doesn't match what we expected
395            return $self->_simple_response(404);
396        }
397
398        # if the thing we're reproxying is indeed a file, advertise that
399        # we support byteranges on it
400        if (-f _) {
401            $hd->header("Accept-Ranges", "bytes");
402        }
403
404        my ($status, $range_start, $range_end) = $self->{req_headers}->range($size);
405        my $not_satisfiable = 0;
406
407        if ($status == 416) {
408            $hd = Moobal::HTTPHeaders->new_response(416);
409            $hd->header("Content-Range", $size ? "bytes */$size" : "*");
410            $not_satisfiable = 1;
411        }
412
413        # change the status code to 200 if the backend gave us 204 No Content
414        $hd->code(200) if $hd->response_code == 204;
415
416        # fixup the Content-Length header with the correct size (application
417        # doesn't need to provide a correct value if it doesn't want to stat())
418        if ($status == 200) {
419            $hd->header("Content-Length", $size);
420        } elsif ($status == 206) {
421            $hd->header("Content-Range", "bytes $range_start-$range_end/$size");
422            $hd->header("Content-Length", $range_end - $range_start + 1);
423            $hd->code(206);
424        }
425
426        # don't send this internal header to the client:
427        $hd->header('X-REPROXY-FILE', undef);
428
429        # rewrite some other parts of the header
430        $self->setup_keepalive($hd);
431
432        # just send the header, now that we cleaned it.
433        $self->{res_headers} = $hd;
434        $self->write($hd->to_string_ref);
435
436        if ($self->{req_headers}->request_method eq 'HEAD' || $not_satisfiable) {
437            $self->write(sub { $self->http_response_sent; });
438            return;
439        }
440
441        $self->state('wait_open');
442        Moobal::AIO::aio_open($file, 0, 0 , sub {
443            my $fh = shift;
444
445            # if client's gone, just close filehandle and abort
446            if ($self->{closed}) {
447                CORE::close($fh) if $fh;
448                return;
449            }
450
451            # handle errors
452            if (! $fh) {
453                # FIXME: do 500 vs. 404 vs whatever based on $! ?
454                return $self->_simple_response(500);
455            }
456
457            # seek if partial content
458            if ($status == 206) {
459                sysseek($fh, $range_start, &POSIX::SEEK_SET);
460                $size = $range_end - $range_start + 1;
461            }
462
463            $self->reproxy_fh($fh, $size);
464            $self->watch_write(1);
465        });
466    });
467}
468
469# Client
470# get/set backend proxy connection
471sub backend {
472    my $self = shift; # Moobal::Component::Proxy
473    return $self->{backend} unless @_;
474
475    my $backend = shift;
476    $self->state('draining_res') unless $backend;
477    return $self->{backend} = $backend;
478}
479
480# invoked by backend when it wants us to start watching for reads again
481# and feeding it data (if we have any)
482sub backend_ready {
483    my Moobal::Component::Proxy $self = $_[0];
484    my Moobal::Component::BackendHTTP $be = $_[1];
485
486    # if we'd turned ourselves off while we waited for a backend, turn
487    # ourselves back on, because the backend is ready for data now.
488    if ($self->{unread_data_waiting}) {
489        $self->watch_read(1);
490    }
491
492    # normal, not-buffered-to-disk case:
493    return $self->drain_read_buf_to($be) unless $self->{bureason};
494
495    # buffered-to-disk case.
496
497    # tell the backend it has to go into buffered_upload_mode,
498    # which makes it inform us of its writable availability
499    $be->invoke_buffered_upload_mode;
500}
501
502# our backend enqueues a call to this method in our write buffer, so this is called
503# right after we've finished sending all of the results to the user.  at this point,
504# if we were doing keep-alive, we don't close and setup for the next request.
505sub backend_finished {
506    my $self = shift; # Moobal::Component::Proxy
507    print "ClientProxy::backend_finished\n" if &Moobal::DEBUG >= 3;
508
509    # mark ourselves as having responded (presumeably if we're here,
510    # the backend has responded already)
511    $self->{responded} = 1;
512
513    # our backend is done with us, so we disconnect ourselves from it
514    $self->{backend} = undef;
515
516    # backend is done sending data to us, so we can recycle this clientproxy
517    # if we don't have any data yet to read
518    return $self->http_response_sent unless $self->{unread_data_waiting};
519
520    # if we get here (and we do, rarely, in practice) then that means
521    # the backend read was empty/disconected (or otherwise messed up),
522    # and the only thing we can really do is close the client down.
523    $self->close("backend_finished_while_unread_data");
524}
525
526# called when we've sent a response to a user fully and we need to reset state
527sub http_response_sent {
528    my Moobal::Component::Proxy $self = $_[0];
529
530    # persistence logic is in ClientHTTPBase
531    return 0 unless $self->SUPER::http_response_sent;
532
533    print "ClientProxy::http_response_sent -- resetting state\n" if &Moobal::DEBUG >= 3;
534
535    if (my $be = $self->{backend}) {
536        $self->{backend} = undef;
537        $be->forget_client;
538    }
539
540    # if we get here we're being persistent, reset our state
541    $self->{backend_requested} = 0;
542    $self->{high_priority} = 0;
543    $self->{reproxy_uris} = undef;
544    $self->{reproxy_expected_size} = undef;
545    $self->{currently_reproxying} = undef;
546    $self->{content_length_remain} = undef;
547    $self->{primary_res_hdrs} = undef;
548    $self->{responded} = 0;
549    $self->{is_buffering} = 0;
550    $self->{is_writing} = 0;
551    $self->{start_time} = undef;
552    $self->{bufh} = undef;
553    $self->{bufilename} = undef;
554    $self->{buoutpos} = 0;
555    $self->{bureason} = undef;
556    $self->{upload_session} = undef;
557    $self->{chunked_upload_state} = undef;
558    $self->{request_body_length} = undef;
559    return 1;
560}
561
562# to request a backend connection AFTER you've already done so, if you
563# didn't like the results from the first one.  (like after a 500 error)
564sub rerequest_backend {
565    my $self = shift; # Moobal::Component::Proxy
566
567    $self->{backend_requested} = 0;
568    $self->{backend} = undef;
569    $self->request_backend;
570}
571
572sub request_backend {
573    my $self = shift; # Moobal::Component::Proxy
574    return if $self->{backend_requested};
575    $self->{backend_requested} = 1;
576
577    $self->state('wait_backend');
578    $self->request_backend_connection();
579    $self->tcp_cork(1);  # cork writes to self
580}
581
582sub request_backend_connection {
583    my $self = shift;
584
585    return if $self->closed;
586
587    my $hi_pri = $self->{high_priority};  # load values from the client proxy object
588    my $low_pri = $self->{low_priority};  # they are initialized as 0 during object creation, but hooks can override them
589
590    # is there a defined high-priority cookie?
591    my $service = $self->service;
592    if (my $cname = $service->param('high_priority_cookie')) {
593        # decide what priority class this request is in
594        my $hd = $self->{req_headers};
595        my %cookie;
596        foreach (split(/;\s+/, $hd->header("Cookie") || '')) {
597            next unless ($_ =~ /(.*)=(.*)/);
598            $cookie{Moobal::Util::durl($1)} = Moobal::Util::durl($2);
599        }
600        my $hicookie = $cookie{$cname} || "";
601        $hi_pri = index($hicookie, $service->param('high_priority_cookie_contents')) != -1;
602    }
603
604    # now, call hook to see if this should be high priority
605    $hi_pri = $service->run_hook('make_high_priority', $self)
606        unless $hi_pri; # only if it's not already
607
608    # and then, call hook to see about low priority
609    $low_pri = $service->run_hook('make_low_priority', $self)
610        unless $hi_pri || $low_pri; # only if it's not high or low already
611
612    $self->{high_priority} = 1 if $hi_pri;
613    $self->{low_priority} = 1 if $low_pri;
614
615    # before we even consider spawning backends, let's see if we have
616    # some bored (pre-connected) backends that'd take this client
617    my $be;
618    my $now = time;
619    while ($be = shift @{$self->{bored_backends}}) {
620        next if $be->{closed};
621
622        # now make sure that it's still in our pool, and if not, close it
623        next unless $self->verify_generation($be);
624
625        # don't use connect-ahead connections when we haven't
626        # verified we have their attention
627        if (! $be->{has_attention} && $be->{create_time} < $now - 5) {
628            $be->close("too_old_bored");
629            next;
630        }
631
632        # don't use keep-alive connections if we know the server's
633        # just about to kill the connection for being idle
634        if ($be->{disconnect_at} && $now + 2 > $be->{disconnect_at}) {
635            $be->close("too_close_disconnect");
636            next;
637        }
638
639        # give the backend this client
640        if ($be->assign_client($self)) {
641            # and make some extra bored backends, if configured as such
642            $self->spawn_backends;
643            return;
644        }
645
646        # assign client can end up closing the connection, so check for that
647        return if $self->{closed};
648    }
649
650    if ($hi_pri) {
651        push @{$self->{waiting_clients_highpri}}, $self;
652    } elsif ($low_pri) {
653        push @{$self->{waiting_clients_lowpri}}, $self;
654    } else {
655        push @{$self->{waiting_clients}}, $self;
656    }
657
658    $self->{waiting_client_count}++;
659    $self->{waiting_client_map}{$self->socket->fileno} = 1;
660
661    $self->spawn_backends;
662}
663
664# sees if it should spawn one or more backend connections
665sub spawn_backends {
666    my $self = shift;
667    my $service = $self->service;
668
669    # check our lock and set it if we can
670    return if $self->spawn_lock;
671    $self->spawn_lock( 1 );
672
673    # sanity checks on our bookkeeping
674    if ($self->pending_connect_count < 0) {
675        Moobal::log('crit', "Bogus: service @{[$service->name]} has pending connect ".
676                     "count of @{[$self->{pending_connect_count}]}?!  Resetting.");
677        $self->pending_connect_count(
678            scalar
679                map { $_ && ! $_->closed } values %{$self->pending_connects}
680        );
681    }
682
683    # keep track of the sum of existing_bored + bored_created
684    my $backends_created = scalar(@{$self->bored_backends}) + $self->pending_connect_count;
685    my $backends_needed = $self->waiting_client_count + $self->connect_ahead;
686    my $to_create = $backends_needed - $backends_created;
687
688    my $pool = $self->context->pool( $service->param('pool') );
689
690    # can't create more than this, assuming one pending connect per node
691    my $max_creatable = $pool ? ($pool->node_count - $self->{pending_connect_count}) : 1;
692    $to_create = $max_creatable if $to_create > $max_creatable;
693
694    # cap number of attempted connects at once
695    $to_create = 10 if $to_create > 10;
696
697    my $now = time;
698
699    while ($to_create > 0) {
700        $to_create--;
701
702        # spawn processes if not a pool, else whine.
703        unless ($pool) {
704            if (my $sp = $self->{server_process}) {
705                warn "To create = $to_create...\n";
706                warn "  spawing $sp\n";
707                my $be = Moobal::Component::BackendHTTP->new_process($self, $sp);
708                return;
709            }
710            warn "No pool! Can't spawn backends.\n";
711            return;
712        }
713
714        my ($ip, $port) = $pool->get_backend_endpoint;
715        unless ($ip) {
716            Moobal::log('crit', "No backend IP for service $self->{name}");
717            # FIXME: register desperate flag, so load-balancer module can callback when it has a node
718            $self->spawn_lock( 0 );
719            return;
720        }
721
722        # handle retry timeouts so we don't spin
723        next if $self->{backend_no_spawn}->{"$ip:$port"};
724
725        # if it's pending, verify the pending one is still valid
726        if (my $be = $self->{pending_connects}{"$ip:$port"}) {
727            my $age = $now - $be->create_time;
728            if ($age >= 5 && $be->state eq "connecting") {
729                $be->close('connect_timeout');
730            } elsif ($age >= 60 && $be->state eq "verifying_backend") {
731                # after 60 seconds of attempting to verify, we're probably already dead
732                $be->close('verify_timeout');
733            } elsif (! $be->closed) {
734                next;
735            }
736        }
737
738        # now actually spawn a backend and add it to our pending list
739        my $be = Moobal::Component::BackendHTTP->new(
740            service => $self->service,
741            ip => $ip,
742            port => $port,
743            pool => $pool,
744            context => $self->context,
745        );
746        if ($be) {
747            $self->add_pending_connect($be);
748        }
749    }
750
751    # clear our spawn lock
752    $self->spawn_lock( 0 );
753}
754
755
756
757
758# Client (overrides and calls super)
759sub close {
760    my $self = shift; # Moobal::Component::Proxy
761    my $reason = shift;
762
763    warn sprintf(
764                    "Moobal::Component::Proxy closed %s%s.\n",
765                    ( $self->{closed} ? "again " : "" ),
766                    (defined $reason ? "saying '$reason'" : "for an unknown reason")
767    ) if &Moobal::DEBUG >= 2;
768
769    # don't close twice
770    return if $self->{closed};
771
772    # signal that we're done
773    $self->{service}->run_hooks('end_proxy_request', $self);
774
775    # kill our backend if we still have one
776    if (my $backend = $self->{backend}) {
777        print "Client ($self) closing backend ($backend)\n" if &Moobal::DEBUG >= 1;
778        $self->backend(undef);
779        $backend->close($reason ? "proxied_from_client_close:$reason" : "proxied_from_client_close");
780    } else {
781        # if no backend, tell our service that we don't care for one anymore
782        $self->{service}->note_client_close($self);
783    }
784
785    # call ClientHTTPBase's close
786    $self->SUPER::close($reason);
787}
788
789sub client_disconnected { # : void
790    my $self = shift; # Moobal::Component::Proxy
791    print "ClientProxy::client_disconnected\n" if &Moobal::DEBUG >= 2;
792
793    # if client disconnected, then we need to turn off watching for
794    # further reads and purge the existing upload if any. also, we
795    # should just return and do nothing else.
796
797    $self->watch_read(0);
798    $self->purge_buffered_upload if $self->{bureason};
799    return $self->close('user_disconnected');
800}
801
802# Client
803sub event_write {
804    my $self = shift; # Moobal::Component::Proxy
805    print "ClientProxy::event_write\n" if &Moobal::DEBUG >= 3;
806
807    $self->SUPER::event_write;
808
809    # obviously if we're writing the backend has processed our request
810    # and we are responding/have responded to the user, so mark it so
811    $self->{responded} = 1;
812
813    # trigger our backend to keep reading, if it's still connected
814    if ($self->{backend_stalled} && (my $backend = $self->{backend})) {
815        print "  unstalling backend\n" if &Moobal::DEBUG >= 3;
816
817        $self->{backend_stalled} = 0;
818        $backend->watch_read(1);
819    }
820}
821
822# ClientProxy
823sub event_read {
824    my $self = shift; # Moobal::Component::Proxy
825    print "ClientProxy::event_read\n" if &Moobal::DEBUG >= 3;
826
827    # mark alive so we don't get killed for being idle
828    $self->{alive_time} = time;
829
830    # if we have no headers, the only thing we can do is try to get some
831    if (! $self->{req_headers}) {
832        print "  no headers.  reading.\n" if &Moobal::DEBUG >= 3;
833        $self->handle_request if $self->read_request_headers;
834        return;
835    }
836
837    # if we're buffering to disk or haven't read too much from this client, keep reading,
838    # otherwise shut off read notifications
839    unless ($self->is_buffering || $self->read_ahead < READ_AHEAD_SIZE) {
840        # our buffer is full, so turn off reads for now
841        print "  disabling reads.\n" if &Moobal::DEBUG >= 3;
842        $self->watch_read(0);
843        return;
844    }
845
846    # deal with chunked uploads
847    if (my $cus = $self->{chunked_upload_state}) {
848        $cus->on_readable($self);
849
850        # if we got more than 1MB not flushed to disk,
851        # stop reading for a bit until disk catches up
852        if ($self->{read_ahead} > 1024*1024) {
853            $self->watch_read(0);
854        }
855        return;
856    }
857
858    # read more data if we're still buffering or if our current read buffer
859    # is not full to the max READ_AHEAD_SIZE which is how much data we will
860    # buffer in from the user before passing on to the backend
861
862    # read the MIN(READ_SIZE, content_length_remain)
863    my $read_size = READ_SIZE;
864    my $remain = $self->{content_length_remain};
865
866    $read_size = $remain if $remain && $remain < $read_size;
867    print "  reading $read_size bytes (", (defined $remain ? $remain : "(undef)"), " bytes remain)\n" if &Moobal::DEBUG >= 3;
868
869    my $bref = $self->read($read_size);
870
871    # if the read returned undef, that means the connection was closed
872    # (see: Danga::Socket::read)
873    return $self->client_disconnected unless defined $bref;
874
875    # if they didn't declare a content body length and we just got a
876    # readable event that's not a disconnect, something's messed up.
877    # they're overflowing us.  disconnect!
878    if (! $remain) {
879        $self->_simple_response(400, "Can't pipeline to HTTP/1.0");
880        $self->close("pipelining_to_http10");
881        return;
882    }
883
884    # now that we know we have a defined value, determine how long it is, and do
885    # housekeeping to keep our tracking numbers up to date.
886    my $len = length($$bref);
887    print "  read $len bytes\n" if &Moobal::DEBUG >= 3;
888
889    # when run under the program "trickle", epoll speaks the truth to
890    # us, but then trickle interferes and steals our reads/writes, so
891    # this fails.  normally this check isn't needed.
892    return unless $len;
893
894    $self->{read_size} += $len;
895    $self->{content_length_remain} -= $len if $remain;
896
897    my $done_reading = defined $self->{content_length_remain} && $self->{content_length_remain} <= 0;
898    my $backend = $self->backend;
899    print("  done_reading = $done_reading, backend = ", ($backend || "<undef>"), "\n") if &Moobal::DEBUG >= 3;
900
901    # upload tracking
902    if (my $session = $self->{upload_session}) {
903        my $cl = $self->{req_headers}->content_length;
904        my $remain = $self->{content_length_remain};
905        my $now = time();  # FIXME: more efficient?
906        if ($cl && $remain && ($self->{last_upload_packet} || 0) != $now) {
907            my $done = $cl - $remain;
908            $self->{last_upload_packet} = $now;
909            $udp_sock ||= IO::Socket::INET->new(Proto => 'udp');
910            my $since = $self->{last_request_time};
911            my $send = "UPLOAD:$session:$done:$cl:$since:$now";
912            if ($udp_sock) {
913                foreach my $ep (@{ $self->{service}{upload_status_listeners_sockaddr} }) {
914                    my $rv = $udp_sock->send($send, 0, $ep);
915                }
916            }
917        }
918    }
919
920    # just dump the read into the nether if we're dangling. that is
921    # the case when we send the headers to the backend and it responds
922    # before we're done reading from the client; therefore further
923    # reads from the client just need to be sent nowhere, because the
924    # RFC2616 section 8.2.3 says: "the server SHOULD NOT close the
925    # transport connection until it has read the entire request"
926    if ($self->{responded}) {
927        print "  already responded.\n" if &Moobal::DEBUG >= 3;
928        # in addition, if we're now out of data (clr == 0), then we should
929        # either close ourselves or get ready for another request
930        return $self->http_response_sent if $done_reading;
931
932        print "  already responded [2].\n" if &Moobal::DEBUG >= 3;
933        # at this point, if the backend has responded then we just return
934        # as we don't want to send it on to them or buffer it up, which is
935        # what the code below does
936        return;
937    }
938
939    # if we have no data left to read, stop reading.  all that can
940    # come later is an extra \r\n which we handle later when parsing
941    # new request headers.  and if it's something else, we'll bail on
942    # the next request, not this one.
943    if ($done_reading) {
944        Carp::confess("content_length_remain less than zero: self->{content_length_remain}")
945            if $self->{content_length_remain} < 0;
946        $self->{unread_data_waiting} = 0;
947        $self->watch_read(0);
948    }
949
950    # now, if we have a backend, then we should be writing it to the backend
951    # and not doing anything else
952    if ($backend) {
953        print "  got a backend.  sending write to it.\n" if &Moobal::DEBUG >= 3;
954        $backend->write($bref);
955        # TODO: monitor the backend's write buffer depth?
956        return;
957    }
958
959    # now, we know we don't have a backend, so we have to push this data onto our
960    # read buffer... it's not going anywhere yet
961    push @{$self->{read_buf}}, $bref;
962    $self->{read_ahead} += $len;
963    print "  no backend.  read_ahead = $self->{read_ahead}.\n" if &Moobal::DEBUG >= 3;
964
965    # if we know we've already started spooling a file to disk, then continue
966    # to do that.
967    print "  bureason = $self->{bureason}\n" if &Moobal::DEBUG >= 3 && $self->{bureason};
968    return $self->buffered_upload_update if $self->{bureason};
969
970    # if we are under our buffer-to-memory size, just continue buffering here and
971    # don't fall through to the backend request call below
972    return if
973        ! $done_reading &&
974        $self->{read_ahead} < $self->{service}->{buffer_backend_connect};
975
976    # over the buffer-to-memory size, see if we should start spooling to disk.
977    return if $self->{service}->{buffer_uploads} && $self->decide_to_buffer_to_disk;
978
979    # give plugins a chance to act on the request before we request a backend
980    # (added by Chris Hondl <chris@imvu.com>, March 2006)
981    my $svc = $self->{service};
982    return if $svc->run_hook('proxy_read_request', $self);
983
984    # if we fall through to here, we need to ensure that a backend is on the
985    # way, because no specialized handling took over above
986    print "  finally requesting a backend\n" if &Moobal::DEBUG >= 3;
987    return $self->request_backend;
988}
989
990sub handle_request {
991    my $self = shift; # Moobal::Component::Proxy
992    my $req_hd = $self->{req_headers};
993
994    unless ($req_hd) {
995        $self->close("handle_request without headers");
996        return;
997    }
998
999    $self->check_req_headers;
1000
1001    my $svc = $self->{service};
1002    # give plugins a chance to force us to bail
1003    return if $svc->run_hook('start_proxy_request', $self);
1004    return if $svc->run_hook('start_http_request',  $self);
1005
1006    if ($self->handle_chunked_upload) {
1007        # handled in method.
1008    } else {
1009        # if defined we're waiting on some amount of data.  also, we have to
1010        # subtract out read_size, which is the amount of data that was
1011        # extra in the packet with the header that's part of the body.
1012        $self->{request_body_length} =
1013            $self->{content_length_remain} =
1014            $req_hd->content_length;
1015        $self->{unread_data_waiting} = 1 if $self->{content_length_remain};
1016    }
1017
1018    # upload-tracking stuff.  both starting a new upload track session,
1019    # and checking on status of ongoing one
1020    return if $svc->{upload_status_listeners} && $self->handle_upload_tracking;
1021
1022    # note that we've gotten a request
1023    $self->{requests}++;
1024    $self->{last_request_time} = $self->{alive_time};
1025
1026    # either start buffering some of the request to memory, or
1027    # immediately request a backend connection.
1028    if ($self->{chunked_upload_state}) {
1029        $self->{request_body_length} = 0;
1030        $self->is_buffering( 1 );
1031        $self->{bureason} = 'chunked';
1032        $self->buffered_upload_update;
1033    } elsif ($self->{content_length_remain} && $self->{service}->{buffer_backend_connect}) {
1034        # the deeper path
1035        $self->start_buffering_request;
1036    } else {
1037        # get the backend request process moving, since we aren't buffering
1038        $self->is_buffering( 0 );
1039
1040        # if reproxy-caching is enabled, we can often bypass needing to allocate a BackendHTTP connection:
1041        return if $svc->{reproxy_cache} && $self->satisfy_request_from_cache;
1042
1043        $self->request_backend;
1044    }
1045}
1046
1047sub handle_chunked_upload {
1048    my $self = shift; # Moobal::Component::Proxy
1049    my $req_hd = $self->{req_headers};
1050    my $te = $req_hd->header("Transfer-Encoding");
1051    return unless $te && $te eq "chunked";
1052    return unless $self->{service}->{buffer_uploads};
1053
1054    $req_hd->header("Transfer-Encoding", undef); # remove it (won't go to backend)
1055
1056    my $eh = $req_hd->header("Expect");
1057    if ($eh && $eh =~ /\b100-continue\b/) {
1058        $self->write(\ "HTTP/1.1 100 Continue\r\n\r\n");
1059        $req_hd->header("Expect", undef); # remove it (won't go to backend)
1060    }
1061
1062    my $max_size = $self->{service}{max_chunked_request_size};
1063
1064    my $args = {
1065        on_new_chunk => sub {
1066            my $cref = shift;
1067            my $len = length($$cref);
1068            push @{$self->{read_buf}}, $cref;
1069            $self->{read_ahead}          += $len;
1070            $self->{request_body_length} += $len;
1071
1072            # if too large, disconnect them...
1073            if ($max_size && $self->{request_body_length} > $max_size) {
1074                $self->purge_buffered_upload;
1075                $self->close;
1076                return;
1077            }
1078            $self->buffered_upload_update;
1079        },
1080        on_disconnect => sub {
1081            $self->client_disconnected;
1082        },
1083        on_zero_chunk => sub {
1084            $self->send_buffered_upload;
1085        },
1086    };
1087
1088    $self->{chunked_upload_state} = Moobal::ChunkedUploadState->new(%$args);
1089    return 1;
1090}
1091
1092sub satisfy_request_from_cache {
1093    my $self = shift; # Moobal::Component::Proxy
1094
1095    my $req_hd = $self->{req_headers};
1096    my $svc    = $self->{service};
1097    my $cache  = $svc->{reproxy_cache};
1098    $svc->{_stat_requests}++;
1099
1100    my $requri   = $req_hd->request_uri    || '';
1101    my $hostname = $req_hd->header("Host") || '';
1102
1103    my $key      = "$hostname|$requri";
1104
1105    my $reproxy  = $cache->get($key) or
1106        return 0;
1107
1108    my ($timeout, $headers, $urls) = @$reproxy;
1109    return 0 if time() > $timeout;
1110
1111    $svc->{_stat_cache_hits}++;
1112    my %headers = map { ref $_ eq 'SCALAR' ? $$_ : $_ } @{$headers || []};
1113
1114    if (my $ims = $req_hd->header("If-Modified-Since")) {
1115        my ($lm_key) = grep { uc($_) eq "LAST-MODIFIED" } keys %headers;
1116        my $lm = $headers{$lm_key} || "";
1117
1118        # remove the IE length suffix
1119        $ims =~ s/; length=(\d+)//;
1120
1121        # If 'Last-Modified' is same as 'If-Modified-Since', send a 304
1122        if ($ims eq $lm) {
1123            my $res_hd = Moobal::HTTPHeaders->new_response(304);
1124            $res_hd->header("Content-Length", "0");
1125            $self->tcp_cork(1);
1126            $self->state('xfer_resp');
1127            $self->write($res_hd->to_string_ref);
1128            $self->write(sub { $self->http_response_sent; });
1129            return 1;
1130        }
1131    }
1132
1133    my $res_hd = Moobal::HTTPHeaders->new_response(200);
1134    $res_hd->header("Date", HTTP::Date::time2str(time()));
1135    while (my ($key, $value) = each %headers) {
1136        $res_hd->header($key, $value);
1137    }
1138
1139    $self->start_reproxy_uri($res_hd, $urls);
1140    return 1;
1141}
1142
1143# return 1 to steal this connection (when they're asking status of an
1144# upload session), return 0 to return it to handle_request's control.
1145sub handle_upload_tracking {
1146    my $self = shift; # Moobal::Component::Proxy
1147    my $req_hd = $self->{req_headers};
1148
1149    return 0 unless
1150        $req_hd->request_uri =~ /[\?&]client_up_sess=(\w{5,50})\b/;
1151
1152    my $sess = $1;
1153
1154    # getting status?
1155    if ($req_hd->request_uri =~ m!^/__upload_status\?!) {
1156        my $status = Moobal::UploadListener::get_status($sess);
1157        my $now = time();
1158        my $body = $status ?
1159            "{done:$status->{done},total:$status->{total},starttime:$status->{starttime},nowtime:$now}" :
1160            "{}";
1161
1162        my $res = $self->{res_headers} = Moobal::HTTPHeaders->new_response(200);
1163        $res->header("Content-Type", "text/plain");
1164        $res->header('Content-Length', length $body);
1165        $self->setup_keepalive($res);
1166        $self->tcp_cork(1);  # cork writes to self
1167        $self->write($res->to_string_ref);
1168        $self->write(\ $body);
1169        $self->write(sub { $self->http_response_sent; });
1170        return 1;
1171    }
1172
1173    # otherwise just tagging this upload as a new upload session
1174    $self->{upload_session} = $sess;
1175    return 0;
1176}
1177
1178# continuation of handle_request, in the case where we need to start buffering
1179# a bit of the request body to memory, either hoping that's all of it, or to
1180# make a determination of whether or not we should save it all to disk first
1181sub start_buffering_request {
1182    my $self = shift; # Moobal::Component::Proxy
1183
1184    # buffering case:
1185    $self->is_buffering( 1 );
1186
1187    # shortcut: if we know that we're buffering by size, and the size
1188    # of this upload is bigger than that value, we can just turn on spool
1189    # to disk right now...
1190    if ($self->{service}->{buffer_uploads} && $self->{service}->{buffer_upload_threshold_size}) {
1191        my $req_hd = $self->{req_headers};
1192        if ($req_hd->content_length >= $self->{service}->{buffer_upload_threshold_size}) {
1193            $self->{bureason} = 'size';
1194            if ($ENV{PERLBAL_DEBUG_BUFFERED_UPLOADS}) {
1195                $self->{req_headers}->header('X-PERLBAL-BUFFERED-UPLOAD-REASON', 'size');
1196            }
1197            $self->state('buffering_upload');
1198            $self->buffered_upload_update;
1199            return;
1200        }
1201    }
1202
1203    # well, we're buffering, but we're not going to disk just yet (but still might)
1204    $self->state('buffering_request');
1205
1206    # only need time if we are using the buffer to disk functionality
1207    $self->{start_time} = [ gettimeofday() ]
1208        if $self->{service}->{buffer_uploads};
1209}
1210
1211# looks at our states and decides if we should start writing to disk
1212# or should just go ahead and blast this to the backend.  returns 1
1213# if the decision was made to buffer to disk
1214sub decide_to_buffer_to_disk {
1215    my $self = shift; # Moobal::Component::Proxy
1216    return unless $self->is_buffering;
1217    return $self->{bureason} if defined $self->{bureason};
1218
1219    # this is called when we have enough data to determine whether or not to
1220    # start buffering to disk
1221    my $dur = tv_interval($self->{start_time}) || 1;
1222    my $rate = $self->{read_ahead} / $dur;
1223    my $etime = $self->{content_length_remain} / $rate;
1224
1225    # see if we have enough data to make the determination
1226    my $reason = undef;
1227
1228    # see if we blow the rate away
1229    if ($self->{service}->{buffer_upload_threshold_rate} > 0 &&
1230            $rate < $self->{service}->{buffer_upload_threshold_rate}) {
1231        # they are slower than the minimum rate
1232        $reason = 'rate';
1233    }
1234
1235    # and finally check estimated time exceeding
1236    if ($self->{service}->{buffer_upload_threshold_time} > 0 &&
1237            $etime > $self->{service}->{buffer_upload_threshold_time}) {
1238        # exceeds
1239        $reason = 'time';
1240    }
1241
1242    unless ($reason) {
1243        $self->is_buffering( 0 );
1244        return 0;
1245    }
1246
1247    # start saving it to disk
1248    $self->state('buffering_upload');
1249    $self->buffered_upload_update;
1250    $self->{bureason} = $reason;
1251
1252    if ($ENV{PERLBAL_DEBUG_BUFFERED_UPLOADS}) {
1253        $self->{req_headers}->header('X-PERLBAL-BUFFERED-UPLOAD-REASON', $reason);
1254    }
1255
1256    return 1;
1257}
1258
1259# take ourselves and send along our buffered data to the backend
1260sub send_buffered_upload {
1261    my $self = shift; # Moobal::Component::Proxy
1262
1263    # make sure our buoutpos is the same as the content length...
1264    return if $self->{is_writing};
1265
1266    # set the content-length that goes to the backend...
1267    if ($self->{chunked_upload_state}) {
1268        $self->{req_headers}->header("Content-Length", $self->{request_body_length});
1269    }
1270
1271    my $clen = $self->{req_headers}->content_length;
1272    if ($clen != $self->{buoutpos}) {
1273        Moobal::log('critical', "Content length of $clen declared but $self->{buoutpos} bytes written to disk");
1274        return $self->_simple_response(500);
1275    }
1276
1277    # reset our position so we start reading from the right spot
1278    $self->{buoutpos} = 0;
1279    sysseek($self->{bufh}, 0, 0) if ($self->{bufh}); # But only if it exists at all
1280
1281    # notify that we want the backend so we get the ball rolling
1282    $self->request_backend;
1283}
1284
1285sub continue_buffered_upload {
1286    my $self = shift; # Moobal::Component::Proxy
1287    my $be = shift; # Moobal::Component::BackendHTTP
1288    return unless $self && $be;
1289
1290    # now send the data
1291    my $clen = $self->{request_body_length};
1292
1293    if ($self->{buoutpos} < $clen) {
1294        my $sent = Moobal::Socket::sendfile($be->{fd}, fileno($self->{bufh}), $clen - $self->{buoutpos});
1295        if ($sent < 0) {
1296            return $self->close("epipe") if $! == EPIPE;
1297            return $self->close("connreset") if $! == ECONNRESET;
1298            print STDERR "Error w/ sendfile: $!\n";
1299            return $self->close('sendfile_error');
1300        }
1301        $self->{buoutpos} += $sent;
1302    }
1303
1304    # if we're done, purge the file and move on
1305    if ($self->{buoutpos} >= $clen) {
1306        $be->{buffered_upload_mode} = 0;
1307        $self->purge_buffered_upload;
1308        return;
1309    }
1310
1311    # we will be called again by the backend since buffered_upload_mode is on
1312}
1313
1314# write data to disk
1315sub buffered_upload_update {
1316    my $self = shift; #  Moobal::Component::Proxy
1317    return if $self->{is_writing};
1318    return unless $self->is_buffering && $self->{read_ahead};
1319
1320    # so we're not writing now and we have data to write...
1321    unless ($self->{bufilename}) {
1322        # create a filename and see if it exists or not
1323        $self->{is_writing} = 1;
1324        my $fn = join('-', $self->{service}->name, $self->{service}->listenaddr, "client", $self->{fd}, int(rand(0xffffffff)));
1325        $fn = $self->{service}->{buffer_uploads_path} . '/' . $fn;
1326
1327        # good, now we need to create the file
1328        Moobal::AIO::aio_open($fn, O_CREAT | O_TRUNC | O_RDWR, 0644, sub {
1329            $self->{is_writing} = 0;
1330            $self->{bufh} = shift;
1331
1332            # throw errors back to the user
1333            if (! $self->{bufh}) {
1334                Moobal::log('critical', "Failure to open $fn for buffered upload output");
1335                return $self->_simple_response(500);
1336            }
1337
1338            # save state and info and bounce it back to write data
1339            $self->{bufilename} = $fn;
1340            $self->buffered_upload_update;
1341        });
1342
1343        return;
1344    }
1345
1346    # can't proceed if we have no disk file to async write to
1347    # people reported seeing this crash rarely in production...
1348    # must be a race between previously in-flight's write
1349    # re-invoking a write immediately after something triggered
1350    # a buffered upload purge.
1351    unless ($self->{bufh}) {
1352        $self->close;
1353        return;
1354    }
1355
1356    # at this point, we want to do some writing
1357    my $bref = shift(@{$self->{read_buf}});
1358    my $len = length $$bref;
1359    $self->{read_ahead} -= $len;
1360
1361    # so at this point we have a valid filename and file handle and should write out
1362    # the buffer that we have
1363    $self->{is_writing} = 1;
1364    Moobal::AIO::aio_write($self->{bufh}, $self->{buoutpos}, $len, $$bref, sub {
1365        my $bytes = shift;
1366        $self->{is_writing} = 0;
1367
1368        # check for error
1369        unless ($bytes) {
1370            Moobal::log('critical', "Error writing buffered upload: $!.  Tried to do $len bytes at $self->{buoutpos}.");
1371            return $self->_simple_response(500);
1372        }
1373
1374        # update our count of data written
1375        $self->{buoutpos} += $bytes;
1376
1377        # now check if we wrote less than we had in this chunk of buffer.  if that's
1378        # the case then we need to reenqueue the part of the chunk that wasn't
1379        # written out and update as appropriate.
1380        if ($bytes < $len) {
1381            my $diff = $len - $bytes;
1382            unshift @{$self->{read_buf}}, \ substr($$bref, $bytes, $diff);
1383            $self->{read_ahead} += $diff;
1384        }
1385
1386        # if we're processing a chunked upload, ...
1387        if ($self->{chunked_upload_state}) {
1388            # turn reads back on, if we haven't hit the end yet.
1389            if ($self->{unread_data_waiting} && $self->{read_ahead} < 1024*1024) {
1390                $self->watch_read(1);
1391                $self->{unread_data_waiting} = 0;
1392            }
1393
1394            if ($self->{read_ahead} == 0 && $self->{chunked_upload_state}->hit_zero_chunk) {
1395                $self->watch_read(0);
1396                $self->send_buffered_upload;
1397                return;
1398            }
1399        }
1400
1401        # if we're done (no clr and no read ahead!) then send it
1402        elsif ($self->{read_ahead} <= 0 && $self->{content_length_remain} <= 0) {
1403            $self->send_buffered_upload;
1404            return;
1405        }
1406
1407        # spawn another writer!
1408        $self->buffered_upload_update;
1409    });
1410}
1411
1412# destroy any files we've created
1413sub purge_buffered_upload {
1414    my $self = shift; # Moobal::Component::Proxy
1415
1416    # Main reason for failure below is a 0-length chunked upload, where the file is never created.
1417    return unless $self->{bufh};
1418
1419    # FIXME: it's reported that sometimes the two now-in-eval blocks
1420    # fail, hence the eval blocks and warnings.  the FIXME is to
1421    # figure this out, why it happens sometimes.
1422
1423    # first close our filehandle... not async
1424    eval {
1425        CORE::close($self->{bufh});
1426    };
1427    if ($@) { warn "Error closing file in ClientProxy::purge_buffered_upload: $@\n"; }
1428
1429    $self->{bufh} = undef;
1430
1431    eval {
1432        # now asyncronously unlink the file
1433        Moobal::AIO::aio_unlink($self->{bufilename}, sub {
1434            if ($!) {
1435                # note an error, but whatever, we'll either overwrite the file later (O_TRUNC | O_CREAT)
1436                # or a cleaner will come through and do it for us someday (if the user runs one)
1437                Moobal::log('warning', "Unable to link $self->{bufilename}: $!");
1438              }
1439        });
1440    };
1441    if ($@) { warn "Error unlinking file in ClientProxy::purge_buffered_upload: $@\n"; }
1442}
1443
1444# returns bool; whether backend should hide the 500 error from the client
1445#   and have us try a new backend.  return true to retry, false to get a 500 error.
1446sub should_retry_after_500 {
1447    my $self = shift; # Moobal::Component::Proxy
1448    my $be   = shift; # Moobal::Component::BackendHTTP
1449    my $svc = $be->{service};
1450    return 0 unless $svc->{enable_error_retries};
1451    my @sched = split(/\s*,\s*/, $svc->{error_retry_schedule});
1452    return 0 if ++$self->{retry_count} > @sched;
1453    return 1;
1454}
1455
1456# called by Backend to tell us it got a 500 error and we should retry another backend.
1457sub retry_after_500 {
1458    my $self = shift; # Moobal::Component::Proxy
1459    my $svc  = shift; # Moobal::Service     
1460
1461    my @sched = split(/\s*,\s*/, $svc->{error_retry_schedule});
1462    my $delay = $sched[$self->{retry_count} - 1];
1463
1464    if ($delay) {
1465        Danga::Socket->AddTimer($delay, sub {
1466            return if $self->{closed};
1467            $self->rerequest_backend;
1468        });
1469    } else {
1470        $self->rerequest_backend;
1471    }
1472
1473}
1474
1475sub as_string {
1476    my $self = shift; # Moobal::Component::Proxy
1477
1478    my $ret = $self->SUPER::as_string;
1479    if ($self->{backend}) {
1480        my $ipport = $self->{backend}->{ipport};
1481        $ret .= "; backend=$ipport";
1482    } else {
1483        $ret .= "; write_buf_size=$self->{write_buf_size}"
1484            if $self->{write_buf_size} > 0;
1485    }
1486    $ret .= "; highpri" if $self->{high_priority};
1487    $ret .= "; lowpri" if $self->{low_priority};
1488    $ret .= "; responded" if $self->{responded};
1489    $ret .= "; waiting_for=" . $self->{content_length_remain}
1490        if defined $self->{content_length_remain};
1491    $ret .= "; reproxying" if $self->{currently_reproxying};
1492
1493    return $ret;
1494}
1495
1496sub set_queue_low {
1497    my $self = shift; # Moobal::Component::Proxy
1498    $self->{low_priority} = 1;
1499    return;
1500}
1501
1502sub set_queue_high {
1503    my $self = shift; # Moobal::Component::Proxy
1504    $self->{high_priority} = 1;
1505    return;
1506}
1507
1508# take a backend we've created and mark it as pending if we do not
1509# have another pending backend connection in this slot
1510sub add_pending_connect {
1511    my $self = shift;
1512    my $be = shift;
1513
1514    # error if we already have a pending connection for this ipport
1515    my $pending_connects = $self->pending_connects;
1516    if (defined $pending_connects->{$be->ipport}) {
1517        Moobal::log('warning', "Warning: attempting to spawn backend connection that already existed.");
1518
1519        # now dump a backtrace so we know how we got here
1520        my $depth = 0;
1521        while (my ($package, $filename, $line, $subroutine) = caller($depth++)) {
1522            Moobal::log('warning', "          -- [$filename:$line] $package::$subroutine");
1523        }
1524
1525        # we're done now, just return
1526        return;
1527    }
1528
1529    # set this connection up in the pending connection list
1530    $pending_connects->{$be->ipport} = $be;
1531    $self->pending_connect_count( $self->pending_connect_count + 1 );
1532}
1533
15341;
1535
1536
1537# Local Variables:
1538# mode: perl
1539# c-basic-indent: 4
1540# indent-tabs-mode: nil
1541# End:
Note: See TracBrowser for help on using the browser.