view release on metacpan or  search on metacpan

lib/AnyEvent/Redis/Federated.pm  view on Meta::CPAN

	return 1 if $self->{server_status}{"$server:down"};
 	return 0;
}

sub isServerUp {
	my ($self, $server) = @_;
	return 0 if $self->{server_status}{"$server:down"};
	return 1;
}

sub nextServer {
	my ($self, $server, $node) = @_;
	return $server unless $self->{config}->{nodes}->{$node}->{addresses};
	$self->{config}->{nodes}->{$node}->{address} = shift(@{$self->{config}->{nodes}->{$node}->{addresses}});
	push @{$self->{config}->{nodes}->{$node}->{addresses}}, $self->{config}->{nodes}->{$node}->{address};
	warn "redis server for $node changed from $server to $self->{config}->{nodes}->{$node}->{address} selected\n" if $self->{debug};
	return $self->{config}->{nodes}->{$node}->{address};
}

## return only on-line/up servers?

sub allServers {
	my ($self, $node) = @_;
	my $hosts = [ grep { $self->isServerUp($_) } @{$self->{config}->{nodes}->{$node}->{addresses}} ];
	return $hosts;
}

sub markServerUp {
	my ($self, $server) = @_;
	if ($self->{server_status}{"$server:down"}) {
		my $down_since = localtime($self->{server_status}{"$server:down_since"});
		delete $self->{server_status}{"$server:down"};
		delete $self->{server_status}{"$server:retries"};
		delete $self->{server_status}{"$server:down_since"};
		delete $self->{server_status}{"$server:retry_pending"};
		delete $self->{server_status}{"$server:retry_interval"};
 		warn "redis server $server back up (down since $down_since)\n" if $self->{debug};
	}
	return 1;
}

sub markServerDown {
	my ($self, $server, $delay) = @_;
	$delay ||= $self->{base_retry_interval};
	warn "redis server $server seems down\n" if $self->{debug};

	# first time?
	if (not $self->{server_status}{"$server:down"}) {
		warn "server $server down, first time\n" if $self->{debug};
		$self->{server_status}{"$server:down"} = 1;
		$self->{server_status}{"$server:retries"}++;
		$self->{server_status}{"$server:down_since"} = time();
		$self->{server_status}{"$server:retry_interval"} ||= $self->{base_retry_interval};
	}

	if ($self->{server_status}{"$server:retry_pending"}) {
		warn "retry already pending for $server, skipping\n" if $self->{debug};
		return 1;
	}

	# ok, schedule the timer to re-check. this should NOT be a
	# recurring timer, otherwise we end up with a bunch of pending
	# retries since the "interval" is likely shorter than TCP timeout.
	# eventually this will error out, in which case we'll try it again
	# by calling markServerDown() after clearying retry_pending, or
	# it'll work and we're good to go.
	my $t;
	my $r;
	$t = AnyEvent->timer(
		after => $delay,
		cb => sub {
			warn "timer callback triggered for $server" if $self->{debug};
			my ($host, $port) = split /:/, $server;
			print "attempting reconnect to $server\n" if $self->{debug};
			$r = AnyEvent::Redis->new(
				host => $host,
				port => $port,
				on_error => sub {
					warn @_ unless $self->{quiet};
					$self->{server_status}{"$server:retry_pending"} = 0;
					$self->markServerDown($server); # schedule another try
				}
			);

			$r->ping(sub{
				my $val = shift; # should be 'PONG'
				if ($val ne 'PONG') {
					warn "retry ping got $val instead of PONG" if $self->{debug};
				}
				$self->{conn}->{$server} = $r;
				$self->{server_status}{"$server:retry_pending"} = 0;
				$self->markServerUp($server);
				undef $t; # we need to keep a ref to the timer here so it runs at all
			 });
		}
	);
	warn "scheduled health check of $server in $delay secs\n" if $self->{debug};
	$self->{server_status}{"$server:retry_pending"} = 1;
	return 1;
}

our $AUTOLOAD;

sub AUTOLOAD {
	my $self = shift;
	my $call = lc $AUTOLOAD;
	$call =~ s/.*:://;
	print "method [$call] autoloaded\n" if $self->{debug};

	my $key = $_[0];
	my $hk  = $key;
	my $cb  = sub { };

	if (ref $_[-1] eq 'CODE') {
		$cb = pop @_;
	}

	# key group?
	if (ref($_[0]) eq 'ARRAY') {
		$hk  = $_[0]->[0];
		$key = $_[0]->[1];
		$_[0] = $key;
	}

	my $node = $self->keyToNode($hk);
	my $query_all = $self->{query_all};

	if ($call =~ s/_all$//) {
		$query_all = 1;
	}

	## The normal single-server case...
	if (not $query_all) {
		my $server = $self->nodeToHost($node);
		print "server [$server] of node [$node] for key [$key] hashkey [$hk]\n" if $self->{debug};

		if ($self->isServerDown($server)) {
			# try another if we can
			if ($self->{config}->{nodes}->{$node}->{addresses}) {
				print "server [$server] seems down\n" if $self->{debug};
				$server = $self->nextServer($server, $node);
				print "trying next server in line [$server] for node [$node]\n" if $self->{debug};
			}
			# bail otherwise
			else {
				print "server $server down.  abandoning call.\n" if $self->{debug};
				$cb->(undef);
				return $self;
			}
		}

		return $self->scheduleCall($server, $call, [@_], $cb);
	}

	## Need to fire this one at all up servers in the node group...
	else {
		my $servers = $self->allServers($node);
		for my $server (@$servers) {
			$self->scheduleCall($server, $call, [@_], $cb);
		}
		return $self;
	}
}

sub poll {
	my ($self) = @_;
	#return if $self->{pending_requests} < 1;
	return if not defined $self->{cv};
	my $rid = $self->{request_serial};
	my $timeout = $self->{command_timeout};

	my $w;
	if ($timeout) {
		$w = AnyEvent->signal(signal => "ALRM", cb => sub {
			warn "AnyEvent::Redis::Federated::poll alarm timeout! ($rid)\n" if $self->{debug};

			# check the state of requests, marking remaining as cancelled
			while (my ($rid, $state) = each %{$self->{request_state}}) {
				if ($self->{request_state}->{$rid}) {
					print "found pending request to cancel: $rid\n" if $self->{debug};
					$self->{request_state}->{$rid} = 0;
					$self->{cv}->end;
					undef $w;
				}
			}
		});
		print "scheduling alarm timer in poll() for $timeout\n" if $self->{debug};
		alarm($timeout);
	}

	$self->{cv}->recv;
	$self->{cv} = undef;
	alarm(0);
	undef $w;
}

sub scheduleCall {
	my ($self, $server, $call, $args, $cb) = @_;

	# have a non-idle connection already?
	my $r;
	if ($self->{conn}->{$server}) {
		if ($self->{idle_timeout}) {
			if ($self->{last_used}->{$server} > time - $self->{idle_timeout}) {
				$r = $self->{conn}->{$server};
			}
		}
		else {
			$r = $self->{conn}->{$server};
		}
	}

	# otherwise create a new connection
	if (not defined $r) {
		my ($host, $port) = split /:/, $server;
		print "attempting new connection to $server\n" if $self->{debug};
		$r = AnyEvent::Redis->new(
			host => $host,
			port => $port,
			on_error => sub {
				warn @_ unless $self->{quiet};
				$self->markServerDown($server);
				$self->{cv}->end;
			}
		);

		$self->{conn}->{$server} = $r;
	}

	if (not defined $self->{cv}) {
		$self->{cv} = AnyEvent->condvar;
	}

	$self->{cv}->begin;
	$self->{request_serial}++;
	my $rid = $self->{request_serial};
	$self->{request_state}->{$rid} = 1; # open request; 0 is cancelled
	print "scheduling request $rid: $_[0]\n" if $self->{debug};

	if ($call eq 'multi' or $call eq 'exec') {
		@$args = (); # these don't really take args
	}

	$r->$call(@$args, sub {
		if (not $self->{request_state}->{$rid}) {
			print "call found request $rid cancelled\n" if $self->{debug};
			delete $self->{request_state}->{$rid};