· 8 years ago · Aug 22, 2018, 02:34 AM
1package POE::Component::Client::Keepalive;
2BEGIN {
3 $POE::Component::Client::Keepalive::VERSION = '0.268';
4 sub POE::Kernel::CATCH_EXCEPTIONS () { 0 }
5}
6
7use warnings;
8use strict;
9
10use Carp qw(croak cluck);
11use Errno qw(ETIMEDOUT EBADF);
12use Socket qw(SOL_SOCKET SO_LINGER);
13
14use POE;
15use POE::Wheel::SocketFactory;
16use POE::Component::Connection::Keepalive;
17use POE::Component::Resolver;
18use Net::IP qw(ip_is_ipv4);
19use Log::Fu;
20use Scalar::Util qw(weaken);
21
22my $ssl_available;
23eval {
24 require POE::Component::SSLify;
25 $ssl_available = 1;
26};
27
28use constant DEBUG => 1;
29use constant DEBUG_DNS => DEBUG || 0;
30use constant DEBUG_DEALLOCATE => DEBUG || 0;
31
32# Manage connection request IDs.
33
34my $CurrentID = 0;
35
36my $default_resolver;
37my $instances = 0;
38
39my @LookupKeys;
40BEGIN {
41 @LookupKeys = qw(
42 ATTR_CONNKEY_FREE
43 ATTR_CONNKEY_BUSY
44 KEY_SOCKET
45 KEY_CONNKEY
46 ATTR_CONNKEY
47 ATTR_SOCKET_USED
48 ATTR_SOCKET_FREE
49 KEY_WID
50 KEY_REQID
51 KEY_WHEELOBJ
52 KEY_REQOBJ
53 KEY_CONNOBJ
54 );
55 foreach (@LookupKeys) {
56 no strict 'refs';
57 my $tmp = $_;
58 *{$tmp} = sub () { "$tmp-"; };
59 }
60}
61
62use Hash::Registry::PP;
63
64sub _new_lookup {
65 my $lookup = Hash::Registry::PP->new();
66 no strict 'refs';
67 $lookup->register_kt("$_-", "$_-") foreach @LookupKeys;
68 log_info("Initializing lookup $lookup");
69 return $lookup;
70}
71
72
73
74# The connection manager uses a number of data structures, most of
75# them arrays. These constants define offsets into those arrays, and
76# the comments document them.
77
78
79use constant {
80 SKI_SOCKET => 0,
81 SKI_KEY => 1,
82 SKI_ATIME => 2,
83 SKI_TIMER => 3
84};
85
86
87 # @$self = (
88#use constant SF_POOL => 0; # \%socket_pool, UNUSED!
89use constant SF_QUEUE => 1; # \@request_queue,
90#use constant SF_USED => 2; # \%sockets_in_use, UNUSED!
91use constant SF_WHEELS => 3; # H::R,
92use constant SF_USED_EACH => 4; # \%count_by_triple,
93use constant SF_MAX_OPEN => 5; # $max_open_count,
94use constant SF_MAX_HOST => 6; # $max_per_host,
95use constant SF_SOCKETS => 7; # H::R,
96use constant SF_KEEPALIVE => 8; # $keep_alive_secs,
97use constant SF_TIMEOUT => 9; # $default_request_timeout,
98use constant SF_RESOLVER => 10; # $poco_client_dns_object,
99use constant SF_SHUTDOWN => 11; # $shutdown_flag,
100use constant SF_REQ_INDEX => 12; # H::R,
101use constant SF_BIND_ADDR => 13; # $bind_address,
102 # );
103
104# @request_queue = (
105# $request,
106# $request,
107# ....
108# );
109
110 # $request = [
111use constant RQ_SESSION => 0; # $request_session,
112use constant RQ_EVENT => 1; # $request_event,
113use constant RQ_SCHEME => 2; # $request_scheme,
114use constant RQ_ADDRESS => 3; # $request_address,
115use constant RQ_IP => 4; # $request_ip,
116use constant RQ_PORT => 5; # $request_port,
117use constant RQ_CONN_KEY => 6; # $request_connection_key,
118use constant RQ_CONTEXT => 7; # $request_context,
119use constant RQ_TIMEOUT => 8; # $request_timeout,
120use constant RQ_START => 9; # $request_start_time,
121use constant RQ_TIMER_ID => 10; # $request_timer_id,
122use constant RQ_WHEEL_ID => 11; # $request_wheel_id,
123use constant RQ_ACTIVE => 12; # $request_is_active,
124use constant RQ_ID => 13; # $request_id,
125use constant RQ_ADDR_FAM => 14; # $request_address_family,
126use constant RQ_FOR_SCHEME => 15; # $request_address_family,
127use constant RQ_FOR_ADDRESS => 16; # $request_address_family,
128use constant RQ_FOR_PORT => 17; # $request_address_family,
129 # ];
130
131# Create a connection manager.
132
133sub new {
134 my $class = shift;
135 croak "new() needs an even number of parameters" if @_ % 2;
136 my %args = @_;
137
138 my $max_per_host = delete($args{max_per_host}) || 4;
139 my $max_open = delete($args{max_open}) || 128;
140 my $keep_alive = delete($args{keep_alive}) || 4;
141 my $timeout = delete($args{timeout}) || 6;
142 my $resolver = delete($args{resolver});
143 my $bind_address = delete($args{bind_address});
144
145 my @unknown = sort keys %args;
146 if (@unknown) {
147 croak "new() doesn't accept: @unknown";
148 }
149
150 my $self = bless [
151 undef, # SF_POOL
152 [ ], # SF_QUEUE
153 undef, # SF_USED
154 _new_lookup(), # SF_WHEELS
155 { }, # SF_USED_EACH
156 $max_open, # SF_MAX_OPEN
157 $max_per_host, # SF_MAX_HOST
158 _new_lookup(), # SF_SOCKETS
159 $keep_alive, # SF_KEEPALIVE
160 $timeout, # SF_TIMEOUT
161 undef, # SF_RESOLVER
162 undef, # SF_SHUTDOWN
163 _new_lookup(), # SF_REQ_INDEX
164 $bind_address, # SF_BIND_ADDR
165 ], $class;
166
167 $default_resolver = $resolver
168 if $resolver && eval { $resolver->isa('POE::Component::Resolver') };
169
170 $self->[SF_RESOLVER] = (
171 $default_resolver ||= POE::Component::Resolver->new()
172 );
173
174 POE::Session->create(
175 object_states => [
176 $self => {
177 _start => "_ka_initialize",
178 _stop => "_ka_stopped",
179 ka_add_to_queue => "_ka_add_to_queue",
180 ka_cancel_dns_response => "_ka_cancel_dns_response",
181 ka_conn_failure => "_ka_conn_failure",
182 ka_conn_success => "_ka_conn_success",
183 ka_deallocate => "_ka_deallocate",
184 ka_dns_response => "_ka_dns_response",
185 ka_keepalive_timeout => "_ka_keepalive_timeout",
186 ka_reclaim_socket => "_ka_reclaim_socket",
187 ka_relinquish_socket => "_ka_relinquish_socket",
188 ka_request_timeout => "_ka_request_timeout",
189 ka_resolve_request => "_ka_resolve_request",
190 ka_set_timeout => "_ka_set_timeout",
191 ka_shutdown => "_ka_shutdown",
192 ka_socket_activity => "_ka_socket_activity",
193 ka_wake_up => "_ka_wake_up",
194 },
195 ],
196 );
197
198 return $self;
199}
200
201# Initialize the hidden session behind this component.
202# Set an alias so the public methods can send it messages easily.
203
204sub _ka_initialize {
205 my ($object, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
206 $instances++;
207 $heap->{resolve} = { };
208 $kernel->alias_set("$object");
209}
210
211# When programs crash, the session may stop in a non-shutdown state.
212# _ka_stopped and DESTROY catch this either way the death occurs.
213
214sub _ka_stopped {
215 $_[OBJECT][SF_SHUTDOWN] = 1;
216}
217
218sub DESTROY {
219 my $self = shift;
220 $self->shutdown();
221}
222
223# Request to wake up. This should only happen during the edge
224# condition where the component's request queue goes from empty to
225# having one item.
226#
227# It also happens during free(), to see if there are more sockets to
228# deal with.
229#
230# TODO - Make the _ka_wake_up stuff smart enough not to post duplicate
231# messages to the queue.
232
233
234
235sub _ka_wake_up {
236 my ($self, $kernel) = @_[OBJECT, KERNEL];
237
238 # Scan the list of requests, until we find one that can be met.
239 # Fire off POE::Wheel::SocketFactory to begin the connection
240 # process.
241 use Data::Dumper;
242 my $request_index = 0;
243 my $currently_open = scalar $self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_FREE);
244 $currently_open += ($self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_USED) || 0);
245 log_err("Currently open: $currently_open");
246
247 my @splice_list;
248
249 QUEUED:
250 foreach my $request (@{$self->[SF_QUEUE]}) {
251 DEBUG and warn "WAKEUP: checking for $request->[RQ_CONN_KEY]";
252
253 # Sweep away inactive requests.
254
255 unless ($request->[RQ_ACTIVE]) {
256 push @splice_list, $request_index;
257 next;
258 }
259
260 # Skip this request if its scheme/address/port triple is maxed
261 # out.
262 my $req_key = $request->[RQ_CONN_KEY];
263 next if (
264 ($self->[SF_USED_EACH]{$req_key} || 0) >= $self->[SF_MAX_HOST]
265 );
266
267 # Honor the request from the free pool, if possible. The
268 # currently open socket count does not increase.
269
270 my $existing_connection = $self->_check_free_pool($req_key);
271 if ($existing_connection) {
272 push @splice_list, $request_index;
273
274 _respond(
275 $request, {
276 connection => $existing_connection,
277 from_cache => "deferred",
278 }
279 );
280
281 # Remove the wheel-to-request index.
282
283 #NB we don't need to really do this, assuming that the request itself
284 #only exists in the queue, this entry should be garbage collected.
285 $self->[SF_REQ_INDEX]->delete_value($request);
286 next;
287 }
288
289
290 # we can't easily take this out of the outer loop since _check_free_pool
291 # can change it from under us
292 my @free_sockets = $self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_FREE);
293
294
295 #At this point we have a valid request, but we need to make sure
296 #that we don't have too many open requests..
297
298
299 # Try to free over-committed (but unused) sockets until we're back
300 # under SF_MAX_OPEN sockets. Bail out if we can't free enough.
301 # TODO - Consider removing @free_sockets in least- to
302 # most-recently used order.
303 while ($currently_open >= $self->[SF_MAX_OPEN]) {
304 last QUEUED unless @free_sockets;
305 my $next_to_go = $free_sockets[rand(@free_sockets)];
306 $self->_remove_socket_from_pool($next_to_go);
307 $currently_open--;
308 }
309
310 # Start the request. Create a wheel to begin the connection.
311 # Move the wheel and its request into SF_WHEELS.
312 log_warnf("Found enqueued request: %d Creating wheel...", $request->[RQ_ID]);
313 DEBUG and warn "WAKEUP: creating wheel for $req_key";
314
315 my $addr = ($request->[RQ_IP] or $request->[RQ_ADDRESS]);
316 my $wheel = POE::Wheel::SocketFactory->new(
317 (
318 defined($self->[SF_BIND_ADDR])
319 ? (BindAddress => $self->[SF_BIND_ADDR])
320 : ()
321 ),
322 RemoteAddress => $addr,
323 RemotePort => $request->[RQ_PORT],
324 SuccessEvent => "ka_conn_success",
325 FailureEvent => "ka_conn_failure",
326 SocketDomain => $request->[RQ_ADDR_FAM],
327 );
328 #Make the connecting wheel dependent on the request ID..
329
330 #XXX: We need to 'strongify' the reqid somehow..
331
332 $self->[SF_WHEELS]->store_kt($request, KEY_REQOBJ, $wheel, StrongValue => 1);
333 $self->[SF_WHEELS]->store_kt($wheel->ID, KEY_WID, $wheel);
334 $self->[SF_REQ_INDEX]->store_kt($wheel, KEY_WHEELOBJ, $request);
335
336 # store the wheel's ID in the request object
337 $request->[RQ_WHEEL_ID] = $wheel->ID;
338
339 # Count it as used, so we don't over commit file handles.
340 $currently_open++;
341 $self->[SF_USED_EACH]{$req_key}++;
342
343 # Mark the request index as one to splice out.
344
345 push @splice_list, $request_index;
346 }
347 continue {
348 $request_index++;
349 }
350
351 # The @splice_list is a list of element indices that need to be
352 # spliced out of the request queue. We scan in backwards, from
353 # highest index to lowest, so that each splice does not affect the
354 # indices of the other.
355 #
356 # This removes the request from the queue. It's vastly important
357 # that the request be entered into SF_WHEELS before now.
358
359 my $splice_index = @splice_list;
360 while ($splice_index--) {
361 splice @{$self->[SF_QUEUE]}, $splice_list[$splice_index], 1;
362 }
363}
364
365sub allocate {
366 my $self = shift;
367 croak "allocate() needs an even number of parameters" if @_ % 2;
368 my %args = @_;
369
370 # TODO - Validate arguments.
371
372 my $scheme = delete $args{scheme};
373 croak "allocate() needs a 'scheme'" unless $scheme;
374 my $address = delete $args{addr};
375 croak "allocate() needs an 'addr'" unless $address;
376 my $port = delete $args{port};
377 croak "allocate() needs a 'port'" unless $port;
378 my $event = delete $args{event};
379 croak "allocate() needs an 'event'" unless $event;
380 my $context = delete $args{context};
381 croak "allocate() needs a 'context'" unless $context;
382 my $timeout = delete $args{timeout};
383 $timeout = $self->[SF_TIMEOUT] unless $timeout;
384
385 my $for_scheme = delete($args{for_scheme}) || $scheme;
386 my $for_address = delete($args{for_addr}) || $address;
387 my $for_port = delete($args{for_port}) || $port;
388
389 croak "allocate() on shut-down connection manager" if $self->[SF_SHUTDOWN];
390
391 my @unknown = sort keys %args;
392 if (@unknown) {
393 croak "allocate() doesn't accept: @unknown";
394 }
395
396 my $conn_key = (
397 "$scheme $address $port for $for_scheme $for_address $for_port"
398 );
399 log_err("Request called for '$conn_key'");
400 # If we have a connection pool for the scheme/address/port triple,
401 # then we can maybe post an available connection right away.
402
403 my $existing_connection = $self->_check_free_pool($conn_key);
404 if (defined $existing_connection) {
405 log_warn("Request for $conn_key immediately allocated");
406 $poe_kernel->post(
407 $poe_kernel->get_active_session,
408 $event => {
409 addr => $address,
410 context => $context,
411 port => $port,
412 scheme => $scheme,
413 connection => $existing_connection,
414 from_cache => "immediate",
415 }
416 );
417 return;
418 }
419
420 # We can't honor the request immediately, so it's put into a queue.
421 DEBUG and warn "ALLOCATE: enqueuing request for $conn_key";
422 my $rqid = ++$CurrentID;
423 my $request = [
424 $poe_kernel->get_active_session(), # RQ_SESSION
425 $event, # RQ_EVENT
426 $scheme, # RQ_SCHEME
427 $address, # RQ_ADDRESS
428 undef, # RQ_IP
429 $port, # RQ_PORT
430 $conn_key, # RQ_CONN_KEY
431 $context, # RQ_CONTEXT
432 $timeout, # RQ_TIMEOUT
433 time(), # RQ_START
434 undef, # RQ_TIMER_ID
435 undef, # RQ_WHEEL_ID
436 1, # RQ_ACTIVE
437 $rqid, # RQ_ID
438 undef, # RQ_ADDR_FAM
439 $for_scheme, # RQ_FOR_SCHEME
440 $for_address, # RQ_FOR_ADDRESS
441 $for_port, # RQ_FOR_PORT
442 ];
443
444 $self->[SF_REQ_INDEX]->store_kt($rqid, KEY_REQID, $request, StrongValue => 1);
445
446 $poe_kernel->refcount_increment(
447 $request->[RQ_SESSION]->ID(),
448 "poco-client-keepalive"
449 );
450
451 $poe_kernel->call("$self", ka_set_timeout => $request);
452 $poe_kernel->call("$self", ka_resolve_request => $request);
453
454 return $request->[RQ_ID];
455}
456
457sub deallocate {
458 my ($self, $req_id) = @_;
459
460 my $request = $self->[SF_REQ_INDEX]->delete_value_by_key_kt($req_id, KEY_REQID);
461
462 croak "deallocate() requires a request ID" unless defined $request;
463
464 # Now pass the vetted request & its ID into our manager session.
465 $poe_kernel->call("$self", "ka_deallocate", $request, $req_id);
466}
467
468sub _ka_deallocate {
469 my ($self, $heap, $request, $req_id) = @_[OBJECT, HEAP, ARG0, ARG1];
470
471 my $conn_key = $request->[RQ_CONN_KEY];
472 my $existing_connection = $self->_check_free_pool($conn_key);
473
474 # Existing connection. Remove it from the pool, and delete the socket.
475 if (defined $existing_connection) {
476 $self->_remove_socket_from_pool($existing_connection->{socket});
477 DEBUG_DEALLOCATE and warn(
478 "deallocate called, deleted already-connected socket"
479 );
480 return;
481 }
482
483 # No connection yet. Cancel the request.
484 DEBUG_DEALLOCATE and warn(
485 "deallocate called without an existing connection. ",
486 "cancelling connection request"
487 );
488
489 unless (exists $heap->{resolve}->{$request->[RQ_ADDRESS]}) {
490 DEBUG_DEALLOCATE and warn(
491 "deallocate cannot cancel dns -- no pending request"
492 );
493 return;
494 }
495
496 if ($heap->{resolve}->{$request->[RQ_ADDRESS]} eq 'cancelled') {
497 DEBUG_DEALLOCATE and warn(
498 "deallocate cannot cancel dns -- request already cancelled"
499 );
500 return;
501 }
502
503 $poe_kernel->call( "$self", ka_cancel_dns_response => $request );
504 return;
505}
506
507sub _ka_cancel_dns_response {
508 my ($self, $kernel, $heap, $request) = @_[OBJECT, KERNEL, HEAP, ARG0];
509
510 my $address = $request->[RQ_ADDRESS];
511 DEBUG_DNS and warn "DNS: canceling request for $address\n";
512 my $requests = $heap->{resolve}{$address};
513
514 # Remove the resolver request for the address of this connection
515 # request
516
517 my $req_index = @$requests;
518 while ($req_index--) {
519 next unless $requests->[$req_index] == $request;
520 splice(@$requests, $req_index, 1);
521 last;
522 }
523
524 # Clean up the structure for the address if there are no more
525 # requests to resolve that address.
526
527 unless (@$requests) {
528 DEBUG_DNS and warn "DNS: canceled all requests for $address";
529 $heap->{resolve}{$address} = 'cancelled';
530 }
531
532 # cancel our attempt to connect
533 $poe_kernel->alarm_remove( $request->[RQ_TIMER_ID] );
534 $poe_kernel->refcount_decrement(
535 $request->[RQ_SESSION]->ID(), "poco-client-keepalive"
536 );
537}
538
539# Set the request's timeout, in the component's context.
540
541sub _ka_set_timeout {
542 my ($kernel, $request) = @_[KERNEL, ARG0];
543 $request->[RQ_TIMER_ID] = $kernel->delay_set(
544 ka_request_timeout => $request->[RQ_TIMEOUT], $request
545 );
546}
547
548# The request has timed out. Mark it as defunct, and respond with an
549# ETIMEDOUT error.
550
551sub _ka_request_timeout {
552 my ($self, $kernel, $request) = @_[OBJECT, KERNEL, ARG0];
553 log_warnf("Request (ID=%d) timed out", $request->[RQ_ID]);
554 DEBUG and warn(
555 "CON: request from session ", $request->[RQ_SESSION]->ID,
556 " for address ", $request->[RQ_ADDRESS], " timed out"
557 );
558 $! = ETIMEDOUT;
559
560 # The easiest way to do this? Simulate an error from the wheel
561 # itself.
562
563 if (defined $request->[RQ_WHEEL_ID]) {
564 @_[ARG0..ARG3] = ("connect", $!+0, "$!", $request->[RQ_WHEEL_ID]);
565 goto &_ka_conn_failure;
566 }
567
568 # But what if there is no wheel?
569 _respond_with_error($request, "connect", $!+0, "$!"),
570}
571
572# Connection failed. Remove the SF_WHEELS record corresponding to the
573# request. Remove the SF_USED placeholder record so it won't count
574# anymore. Send a failure notice to the requester.
575
576sub _ka_conn_failure {
577 my ($self, $func, $errnum, $errstr, $wheel_id) = @_[OBJECT, ARG0..ARG3];
578
579 DEBUG and warn "CON: sending $errstr for function $func";
580 # Remove the SF_WHEELS record.
581
582 my $wheel = $self->[SF_WHEELS]->delete_value_by_key_kt($wheel_id, KEY_WID);
583 my $ski = $self->[SF_SOCKETS]->fetch_kt($wheel, KEY_WHEELOBJ);
584 my $request = $self->[SF_REQ_INDEX]->delete_value_by_key_kt($wheel, KEY_WHEELOBJ);
585
586 $self->_ski_remove($ski);
587
588 # Discount the use by request key, removing the SF_USED record
589 # entirely if it's now moot.
590 my $request_key = $request->[RQ_CONN_KEY];
591 $self->_decrement_used_each($request_key);
592
593 # Tell the requester about the failure.
594 _respond_with_error($request, $func, $errnum, $errstr),
595}
596
597# Connection succeeded. Remove the SF_WHEELS record corresponding to
598# the request. Flesh out the placeholder SF_USED record so it counts.
599
600sub _ka_conn_success {
601 my ($self, $socket, $wheel_id) = @_[OBJECT, ARG0, ARG3];
602
603 my $wheel = $self->[SF_WHEELS]->delete_value_by_key_kt($wheel_id, KEY_WID);
604 my $request = $self->[SF_REQ_INDEX]->delete_value_by_key_kt($wheel, KEY_WHEELOBJ);
605 # Remove the SF_WHEELS record.
606
607
608
609 if ($request->[RQ_SCHEME] eq 'https') {
610 unless ($ssl_available) {
611 die "There is no SSL support, please install POE::Component::SSLify";
612 }
613 eval {
614 $socket = POE::Component::SSLify::Client_SSLify($socket);
615 };
616 if ($@) {
617 _respond_with_error($request, "sslify", undef, "$@");
618 return;
619 }
620 }
621
622 my $ski = [
623 $socket, #SKI_SOCKET,
624 $request->[RQ_CONN_KEY], #SKI_KEY,
625 time(), #SKI_ATIME,
626 undef, #SKI_TIMER
627 ];
628
629 #weaken($ski->[SKI_SOCKET]);
630 $self->[SF_SOCKETS]->store_kt($socket, KEY_SOCKET, $ski, StrongValue => 1);
631 $self->_ski_mark_used($ski);
632
633 DEBUG and warn(
634 "CON: posting... to $request->[RQ_SESSION] . $request->[RQ_EVENT]"
635 );
636
637 # Build a connection object around the socket.
638 my $connection = POE::Component::Connection::Keepalive->new(
639 socket => $socket,
640 manager => $self,
641 );
642 $self->[SF_SOCKETS]->store_kt($connection, KEY_CONNOBJ,
643 $socket, StrongValue => 1);
644 # Give the socket to the requester.
645 _respond(
646 $request, {
647 connection => $connection,
648 }
649 );
650}
651
652# The user is done with a socket. Make it available for reuse.
653
654sub free {
655 my ($self, $socket) = @_;
656 return if $self->[SF_SHUTDOWN];
657 DEBUG and warn "FREE: freeing socket";
658 log_err("Calling fetch_kt=$socket");
659 my $ski;
660 eval {
661 $ski = $self->[SF_SOCKETS]->fetch_kt($socket, KEY_SOCKET);
662 };
663 if($@) {
664 log_err($@);
665 }
666 log_err("fetch_kt done");
667 log_err("SKI=$ski");
668 # Remove the accompanying SF_USED record.
669 croak "can't free() undefined socket" unless defined $ski;
670
671 # Reclaim the socket.
672 log_err("calling reclaim..");
673 $poe_kernel->call("$self", "ka_reclaim_socket", $ski);
674
675 # Avoid returning things by mistake.
676 return;
677}
678
679# A sink for deliberately unhandled events.
680
681sub _ka_ignore_this_event {
682 # Do nothing.
683}
684
685# An internal method to fetch a socket from the free pool, if one
686# exists.
687sub _ski_mark_used {
688 my ($self,$ski) = @_;
689 #cluck("Telling you how we got here");
690 log_warn("Marking $ski as used");
691 my $table = $self->[SF_SOCKETS];
692 my $key = $ski->[SKI_KEY];
693 $table->store_a(1, ATTR_SOCKET_USED, $ski);
694 $table->store_a($key, ATTR_CONNKEY_BUSY, $ski);
695
696 $table->delete_attr_from_value(1, ATTR_SOCKET_FREE, $ski);
697 $table->delete_attr_from_value($key, ATTR_CONNKEY_FREE, $ski);
698
699}
700
701sub _ski_mark_free {
702 my ($self,$ski) = @_;
703 log_warn("Marking $ski as free");
704 my $table = $self->[SF_SOCKETS];
705 my $key = $ski->[SKI_KEY];
706 $table->store_a(1, ATTR_SOCKET_FREE, $ski);
707 $table->store_a($key, ATTR_CONNKEY_FREE, $ski);
708
709 $table->delete_attr_from_value(1, ATTR_SOCKET_USED, $ski);
710 $table->delete_attr_from_value($key, ATTR_CONNKEY_BUSY, $ski);
711
712}
713
714sub _check_free_pool {
715 my ($self, $conn_key) = @_;
716
717 #Get all free sockets for this connection
718 log_err("Fetching free list");
719 my @free = $self->[SF_SOCKETS]->fetch_a($conn_key, ATTR_CONNKEY_FREE);
720 log_err("Done!");
721 return unless @free;
722
723 #log_warn(Dumper(@free));
724 my $ski = shift @free;
725
726 #mark as used
727 $self->_ski_mark_used($ski);
728
729 DEBUG and warn "CHECK: reusing $conn_key";
730
731 # _check_free_pool() may be operating in another session, so we call
732 # the correct one here.
733
734
735 $ski->[SKI_ATIME] = time();
736 $self->[SF_USED_EACH]{$conn_key}++;
737 $poe_kernel->call("$self", "ka_relinquish_socket", $ski);
738
739 # Build a connection object around the socket.
740 my $connection = POE::Component::Connection::Keepalive->new(
741 socket => $ski->[SKI_SOCKET],
742 manager => $self,
743 );
744
745 return $connection;
746}
747
748sub _decrement_used_each {
749 my ($self, $request_key) = @_;
750 unless (--$self->[SF_USED_EACH]{$request_key}) {
751 delete $self->[SF_USED_EACH]{$request_key};
752 log_err("Nothing left for $request_key");
753 }
754}
755
756# Reclaim a socket. Put it in the free socket pool, and wrap it with
757# select_read() to discard any data and detect when it's closed.
758
759sub _ka_reclaim_socket {
760 my ($self, $kernel, $ski) = @_[OBJECT, KERNEL, ARG0];
761 log_err("Reclaim=$ski");
762 my $socket = $ski->[SKI_SOCKET];
763
764 # Decrement the usage counter for the given connection key.
765 my $request_key = $ski->[SKI_KEY];
766 $self->_decrement_used_each($request_key);
767
768 if(!defined fileno $socket) {
769 DEBUG and warn "RECLAIM: freed socket has previously been closed";
770 $self->_ski_remove($ski);
771 goto &_ka_wake_up;
772 }
773
774 # Socket is still open. Check for lingering data.
775 DEBUG and warn "RECLAIM: checking if socket still works";
776
777 # Check for data on the socket, which implies that the server
778 # doesn't know we're done. That leads to desynchroniziation on the
779 # protocol level, which strongly implies that we can't reuse the
780 # socket. In this case, we'll make a quick attempt at fetching all
781 # the data, then close the socket.
782
783 my $rin = '';
784 vec($rin, fileno($socket), 1) = 1;
785 my ($rout, $eout);
786 my $socket_is_active = select ($rout=$rin, undef, $eout=$rin, 0);
787
788 if ($socket_is_active) {
789 DEBUG and warn "RECLAIM: socket is still active; trying to drain";
790 use bytes;
791
792 my $socket_had_data = sysread($socket, my $buf = "", 65536) || 0;
793 DEBUG and warn "RECLAIM: socket had $socket_had_data bytes. 0 means EOF";
794 DEBUG and warn "RECLAIM: Giving up on socket.";
795
796 # Avoid common FIN_WAIT_2 issues, but only for valid sockets.
797 #if ($socket_had_data and fileno($socket)) {
798 if ($socket_had_data) {
799 my $opt_result = setsockopt(
800 $socket, SOL_SOCKET, SO_LINGER, pack("sll",1,0,0)
801 );
802 die "setsockopt: " . ($!+0) . " $!" if (not $opt_result and $! != EBADF);
803 }
804 $self->_ski_remove($ski);
805 goto &_ka_wake_up;
806 }
807
808 # Socket is alive and has no data, so it's in a quiet, theoretically
809 # reclaimable state.
810
811 DEBUG and warn "RECLAIM: reclaiming socket";
812
813 # Watch the socket, and set a keep-alive timeout.
814 $kernel->select_read($socket, "ka_socket_activity");
815 my $ski_to = $ski;
816 weaken($ski_to);
817
818 if(!defined $ski->[SKI_TIMER]) {
819 $ski->[SKI_TIMER] = $kernel->delay_set(
820 ka_keepalive_timeout => $self->[SF_KEEPALIVE], $ski_to
821 );
822 }
823
824
825 $self->_ski_mark_free($ski);
826
827 goto &_ka_wake_up;
828}
829
830# Socket timed out. Discard it.
831
832sub _ka_keepalive_timeout {
833 my ($self, $ski) = @_[OBJECT, ARG0];
834 log_err("Timeout triggered!");
835 $self->_ski_remove($ski);
836}
837
838# Relinquish a socket. Stop selecting on it.
839
840sub _ka_relinquish_socket {
841 my ($kernel, $ski) = @_[KERNEL, ARG0];
842 my $sock = $ski->[SKI_SOCKET];
843 log_warn("RELINQUISH: $ski $sock");
844 $kernel->alarm_remove($ski->[SKI_TIMER]) if defined $ski->[SKI_TIMER];
845 $ski->[SKI_TIMER] = undef;
846
847 $kernel->select_read($sock, undef) if defined $sock;
848}
849
850# Shut down the component. Release any sockets we're currently
851# holding onto. Clean up any timers. Remove the alias it's known by.
852
853sub shutdown {
854 my $self = shift;
855 return if $self->[SF_SHUTDOWN];
856 $poe_kernel->call("$self", "ka_shutdown");
857}
858
859sub _ka_shutdown {
860 my ($self, $kernel, $heap) = @_[OBJECT, KERNEL, HEAP];
861
862 return if $self->[SF_SHUTDOWN];
863
864 $instances--;
865
866 # Clean out the request queue.
867 foreach my $request (@{$self->[SF_QUEUE]}) {
868 $self->_shutdown_request($kernel, $request);
869 }
870 $self->[SF_QUEUE] = [ ];
871 # Clean out the socket pool.
872
873 #TODO: Implement 'all values' or whatever API function for HR
874
875 my @skis = $self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_FREE);
876 foreach my $ski (@skis) {
877 $self->_ski_remove($ski);
878 }
879 log_err("Have SKIs", @skis);
880 @skis = ();
881
882 my @open = $self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_FREE);
883 log_err("REMAINING SKIs", @open);
884 log_err(Dumper($self->[SF_SOCKETS]));
885
886 # Stop any pending resolver requests.
887 foreach my $host (keys %{$heap->{resolve}}) {
888 if ($heap->{resolve}{$host} eq 'cancelled') {
889 DEBUG and warn "SHT: Skipping shutdown for $host (already cancelled)";
890 next;
891 }
892 DEBUG and warn "SHT: Shutting down resolver requests for $host";
893 foreach my $request (@{$heap->{resolve}{$host}}) {
894 $self->_shutdown_request($kernel, $request);
895 }
896 }
897 $heap->{resolve} = { };
898
899 # Shut down the resolver.
900 DEBUG and warn "SHT: Shutting down resolver";
901 if ( $self->[SF_RESOLVER] != $default_resolver ) {
902 $self->[SF_RESOLVER]->shutdown();
903 }
904 $self->[SF_RESOLVER] = undef;
905
906 if ( $default_resolver and !$instances ) {
907 $default_resolver->shutdown();
908 $default_resolver = undef;
909 }
910
911 # Finish keepalive's shutdown.
912 $kernel->alias_remove("$self");
913 $self->[SF_SHUTDOWN] = 1;
914
915 return;
916}
917
918sub _shutdown_request {
919 my ($self, $kernel, $request) = @_;
920 $self->[SF_REQ_INDEX]->delete_value($request);
921
922 if (defined $request->[RQ_TIMER_ID]) {
923 DEBUG and warn "SHT: Shutting down resolver timer $request->[RQ_TIMER_ID]";
924 $kernel->alarm_remove($request->[RQ_TIMER_ID]);
925 }
926
927 if (defined $request->[RQ_WHEEL_ID]) {
928 DEBUG and warn "SHT: Shutting down resolver wheel $request->[RQ_TIMER_ID]";
929 delete $self->[SF_WHEELS]{$request->[RQ_WHEEL_ID]};
930 }
931
932 if (defined $request->[RQ_SESSION]) {
933 my $session_id = $request->[RQ_SESSION]->ID;
934 DEBUG and warn "SHT: Releasing session $session_id";
935 $kernel->refcount_decrement($session_id, "poco-client-keepalive");
936 }
937}
938
939# A socket in the free pool has activity. Read from it and discard
940# the output. Discard the socket on error or remote closure.
941
942sub _ka_socket_activity {
943 my ($self, $kernel, $socket) = @_[OBJECT, KERNEL, ARG0];
944 my $ski = $self->[SF_SOCKETS]->fetch_kt($socket, KEY_SOCKET);
945
946 if (DEBUG) {
947 my $key = $ski->[SKI_KEY];
948 if(!$key) {
949 print Dumper($ski);
950 die "SKI without key!";
951 }
952 warn "CON: Got activity on socket for $key";
953 }
954
955 # Any socket activity on a kept-alive socket implies that the socket
956 # is no longer reusable.
957
958 use bytes;
959 my $socket_had_data = sysread($socket, my $buf = "", 65536) || 0;
960 DEBUG and warn "CON: socket had $socket_had_data bytes. 0 means EOF";
961 DEBUG and warn "CON: Removing socket from the pool";
962 $self->_ski_remove($ski);
963}
964
965sub _ka_resolve_request {
966 my ($self, $kernel, $heap, $request) = @_[OBJECT, KERNEL, HEAP, ARG0];
967
968 my $host = $request->[RQ_ADDRESS];
969
970 # Skip DNS resolution if it's already a dotted quad.
971 # ip_is_ipv4() doesn't require quads, so we count the dots.
972 #
973 # TODO - Do the same for IPv6 addresses containing colons?
974 # TODO - Would require AF_INET6 support around the SocketFactory.
975 if ((($host =~ tr[.][.]) == 3) and ip_is_ipv4($host)) {
976 DEBUG_DNS and warn "DNS: $host is a dotted quad; skipping lookup";
977 $kernel->call("$self", ka_add_to_queue => $request);
978 return;
979 }
980
981 # It's already pending DNS resolution. Combine this with previous.
982 if (exists $heap->{resolve}->{$host}) {
983 DEBUG_DNS and warn "DNS: $host is piggybacking on a pending lookup.\n";
984 push @{$heap->{resolve}->{$host}}, $request;
985 return;
986 }
987
988 # New request. Start lookup.
989 $heap->{resolve}->{$host} = [ $request ];
990
991 my $response = $self->[SF_RESOLVER]->resolve(
992 event => 'ka_dns_response',
993 host => $host,
994 service => $request->[RQ_SCHEME],
995 );
996
997 DEBUG_DNS and warn "DNS: looking up $host in the background.\n";
998}
999
1000sub _ka_dns_response {
1001 my ($self, $kernel, $heap, $response_error, $addresses, $request) = @_[
1002 OBJECT, KERNEL, HEAP, ARG0..ARG2
1003 ];
1004
1005 # We've shut down. Nothing to do here.
1006 return if $self->[SF_SHUTDOWN];
1007
1008 my $request_address = $request->{host};
1009 my $requests = delete $heap->{resolve}->{$request_address};
1010
1011 DEBUG_DNS and warn "DNS: got response for request address $request_address";
1012
1013 # Requests on record.
1014 if (defined $requests) {
1015 # We can receive responses for canceled requests. Ignore them: we
1016 # cannot cancel PoCo::Client::DNS requests, so this is how we reap
1017 # them when they're canceled.
1018 if ($requests eq 'cancelled') {
1019 DEBUG_DNS and warn "DNS: reaping cancelled request for $request_address";
1020 return;
1021 }
1022 unless (ref $requests eq 'ARRAY') {
1023 die "DNS: got an unknown requests for $request_address: $requests";
1024 }
1025 }
1026 else {
1027 die "DNS: Unexpectedly undefined requests for $request_address";
1028 }
1029
1030 # This is an error. Cancel all requests for the address.
1031 # Tell everybody that their requests failed.
1032 if ($response_error) {
1033 DEBUG_DNS and warn "DNS: resolver error = $response_error";
1034 foreach my $request (@$requests) {
1035 _respond_with_error($request, "resolve", undef, $response_error),
1036 }
1037 return;
1038 }
1039
1040 DEBUG_DNS and warn "DNS: got a response";
1041
1042 # A response!
1043 foreach my $address_rec (@$addresses) {
1044 my $numeric = $self->[SF_RESOLVER]->unpack_addr($address_rec);
1045
1046 DEBUG_DNS and warn "DNS: $request_address resolves to $numeric";
1047
1048 foreach my $request (@$requests) {
1049 # Don't bother continuing inactive requests.
1050 next unless $request->[RQ_ACTIVE];
1051 $request->[RQ_IP] = $numeric;
1052 $request->[RQ_ADDR_FAM] = $address_rec->{family};
1053 log_warn("Adding to queue: $request");
1054 $kernel->yield(ka_add_to_queue => $request);
1055 }
1056
1057 # Return after the first good answer.
1058 return;
1059 }
1060
1061 # Didn't return here. No address record for the host?
1062 foreach my $request (@$requests) {
1063 DEBUG_DNS and warn "DNS: $request_address does not resolve";
1064 _respond_with_error($request, "resolve", undef, "Host has no address."),
1065 }
1066}
1067
1068
1069sub _ka_add_to_queue {
1070 my ($self, $kernel, $request) = @_[OBJECT, KERNEL, ARG0];
1071
1072 push @{ $self->[SF_QUEUE] }, $request;
1073
1074 # If the queue has more than one request in it, then it already has
1075 # a wakeup event pending. We don't need to send another one.
1076 my $qsize = @{$self->[SF_QUEUE]};
1077 log_err("Queue size is $qsize");
1078 return if @{$self->[SF_QUEUE]} > 1;
1079
1080 # If the component's allocated socket count is maxed out, then it
1081 # will check the queue when an existing socket is released. We
1082 # don't need to wake it up here.
1083 my $use_count = $self->[SF_SOCKETS]->fetch_a(1, ATTR_SOCKET_USED) || 0;
1084 log_errf("Use count: %d, MAX=%d", $use_count, $self->[SF_MAX_OPEN]);
1085
1086 return if $use_count >= $self->[SF_MAX_OPEN];
1087
1088 # Likewise, we shouldn't awaken the session if there are no
1089 # available slots for the given scheme/address/port triple. "|| 0"
1090 # to avoid an undef error.
1091 my $conn_key = $request->[RQ_CONN_KEY];
1092
1093 log_errf("Max per host: %d", $self->[SF_MAX_HOST]);
1094 log_errf("Current for key '%s': %d", $conn_key, $self->[SF_USED_EACH]{$conn_key});
1095 return if (
1096 ($self->[SF_USED_EACH]{$conn_key} || 0) >= $self->[SF_MAX_HOST]
1097 );
1098
1099 # Wake the session up, and return nothing, signifying sound and fury
1100 # yet to come.
1101 DEBUG and warn "posting wakeup for $conn_key";
1102 $poe_kernel->post("$self", "ka_wake_up");
1103 return;
1104}
1105
1106# Remove a socket from the free pool, by the socket handle itself.
1107sub _ski_remove {
1108 my ($self,$ski) = @_;
1109 log_err("Removing $ski");
1110 $self->[SF_SOCKETS]->delete_value($ski);
1111 my $sock = delete $ski->[SKI_SOCKET];
1112 $poe_kernel->alarm_remove($ski->[SKI_TIMER]) if defined $ski->[SKI_TIMER];
1113 if(defined $sock and defined fileno $sock) {
1114 $poe_kernel->select_read($sock, undef);
1115 close($sock);
1116 } else {
1117 log_err("Socket for $ski is invalid!");
1118 }
1119 log_warn("Socket is $sock");
1120}
1121
1122sub _remove_socket_from_pool {
1123 my ($self, $socket) = @_;
1124 my $ski = $self->[SF_SOCKETS]->fetch_kt($socket, KEY_SOCKET);
1125 $self->_ski_remove($ski);
1126
1127 # Avoid common FIN_WAIT_2 issues.
1128 # Commented out because fileno() will return true for closed
1129 # sockets, which makes setsockopt() highly unhappy. Also, SO_LINGER
1130 # will cause te socket closure to block, which is less than ideal.
1131 # We need to revisit this another way, or just let sockets enter
1132 # FIN_WAIT_2.
1133
1134# if (fileno $socket) {
1135# setsockopt($socket, SOL_SOCKET, SO_LINGER, pack("sll",1,0,0)) or die(
1136# "setsockopt: $!"
1137# );
1138# }
1139}
1140
1141# Internal function. NOT AN EVENT HANDLER.
1142
1143sub _respond_with_error {
1144 my ($request, $func, $num, $string) = @_;
1145 _respond(
1146 $request,
1147 {
1148 connection => undef,
1149 function => $func,
1150 error_num => $num,
1151 error_str => $string,
1152 }
1153 );
1154}
1155
1156sub _respond {
1157 my ($request, $fields) = @_;
1158
1159 # Bail out early if the request isn't active.
1160 return unless $request->[RQ_ACTIVE] and $request->[RQ_SESSION];
1161
1162 $poe_kernel->post(
1163 $request->[RQ_SESSION],
1164 $request->[RQ_EVENT],
1165 {
1166 addr => $request->[RQ_ADDRESS],
1167 context => $request->[RQ_CONTEXT],
1168 port => $request->[RQ_PORT],
1169 scheme => $request->[RQ_SCHEME],
1170 for_addr => $request->[RQ_FOR_ADDRESS],
1171 for_scheme => $request->[RQ_FOR_SCHEME],
1172 for_port => $request->[RQ_FOR_PORT],
1173 %$fields,
1174 }
1175 );
1176
1177 # Drop the extra refcount.
1178 $poe_kernel->refcount_decrement(
1179 $request->[RQ_SESSION]->ID(),
1180 "poco-client-keepalive"
1181 );
1182
1183 # Remove associated timer.
1184 if ($request->[RQ_TIMER_ID]) {
1185 $poe_kernel->alarm_remove($request->[RQ_TIMER_ID]);
1186 $request->[RQ_TIMER_ID] = undef;
1187 }
1188
1189 # Deactivate the request.
1190 $request->[RQ_ACTIVE] = undef;
1191}
1192
11931;