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