#!/usr/bin/perl
#
# Danga's Distributed Lock Daemon
#
# Copyright 2004, Danga Interactive
# Copyright 2005-2006, Six Apart, Ltd.
#
# Authors:
#   Brad Fitzpatrick <brad@danga.com>
#   Jonathan Steinert <jsteinert@sixapart.com>
#
# License:
#   terms of Perl itself.
#

use strict;
use Getopt::Long;
use Carp;
use Danga::Socket;
use IO::Socket::INET;
use POSIX ();

## the storage load intervals: %IVs = ( time_iv => load, ... )
our %IVs = ();
## haw many load buckets we want
our $Bucket_qty = 100; #
## the total duration of load measurement
our $Total_duration = 60 * 5; # 5mn
## the duration handled by one bucket
our $IV_duration = int($Total_duration / $Bucket_qty); 
## how many bucket held before we purge the old ones
our $Bucket_limit =  2 * $Bucket_qty;

use vars qw($DEBUG);
$DEBUG = 0;

my (
    $daemonize,
    $nokeepalive,
    $hostname,
    $table,
   );
my $conf_port = 7002;
my $lock_type = "internal";

Getopt::Long::GetOptions(
    'd|daemon'       => \$daemonize,
    'p|port=i'       => \$conf_port,
    'debug=i'        => \$DEBUG,
    'n|no-keepalive' => \$nokeepalive,
    't|type=s'       => \$lock_type,
    'h|hostname=s'   => \$hostname,
    'T|table=s'  => \$table,
   );

# Statistics counters
my $lock_successes = 0;
my $lock_failures  = 0;

my $client_class;
my @client_options;
if ($lock_type eq 'internal') {
    $client_class = "Client::Internal";
}
elsif($lock_type eq 'dlmfs') {
    $client_class = "Client::DLMFS";
}
elsif($lock_type eq 'dbi') {
    length( $hostname ) or die( "-h (--hostname) must be included with a hostname in dbi mode\n" );
    length( $table ) or die( "-T (--table) must be included with a table name in dbi mode\n" );
    $client_class = "Client::DBI";
    @client_options = ( $hostname, $table );
}
else {
    die( "Unknown lock type of '$lock_type' specified.\n" );
}

$client_class->_setup( @client_options );

daemonize() if $daemonize;

use Socket qw(IPPROTO_TCP SO_KEEPALIVE TCP_NODELAY SOL_SOCKET);

# Linux-specific:
use constant TCP_KEEPIDLE  => 4; # Start keeplives after this period
use constant TCP_KEEPINTVL => 5; # Interval between keepalives
use constant TCP_KEEPCNT   => 6; # Number of keepalives before death

$SIG{'PIPE'} = "IGNORE";  # handled manually

# establish SERVER socket, bind and listen.
my $server = IO::Socket::INET->new(LocalPort => $conf_port,
                                   Type      => SOCK_STREAM,
                                   Proto     => IPPROTO_TCP,
                                   Blocking  => 0,
                                   Reuse     => 1,
                                   Listen    => 10 )
    or die "Error creating socket: $@\n";

# Not sure if I'm crazy or not, but I can't see in strace where/how
# Perl 5.6 sets blocking to 0 without this.  In Perl 5.8, IO::Socket::INET
# obviously sets it from watching strace.
IO::Handle::blocking($server, 0);

my $accept_handler = sub {
    my $csock = $server->accept();
    return unless $csock;

    printf("Listen child making a Client for %d.\n", fileno($csock))
        if $DEBUG;

    IO::Handle::blocking($csock, 0);
    setsockopt($csock, IPPROTO_TCP, TCP_NODELAY, pack("l", 1)) or die;

    # Enable keep alive
    unless ( $nokeepalive ) {
        (setsockopt($csock, SOL_SOCKET, SO_KEEPALIVE,  pack("l", 1)) &&
         setsockopt($csock, IPPROTO_TCP, TCP_KEEPIDLE,  pack("l", 30)) &&
         setsockopt($csock, IPPROTO_TCP, TCP_KEEPCNT,   pack("l", 10)) &&
         setsockopt($csock, IPPROTO_TCP, TCP_KEEPINTVL, pack("l", 30)) &&
         1
        ) || die "Couldn't set keep-alive settings on socket (Not on Linux?)";
    }

    my $client = $client_class->new($csock);
    $client->watch_read(1);
};

Client->OtherFds(fileno($server) => $accept_handler);
Client->EventLoop();

sub daemonize {
    my($pid, $sess_id, $i);

    ## Fork and exit parent
    if ($pid = fork) { exit 0; }

    ## Detach ourselves from the terminal
    croak "Cannot detach from controlling terminal"
        unless $sess_id = POSIX::setsid();

    ## Prevent possibility of acquiring a controling terminal
    $SIG{'HUP'} = 'IGNORE';
    if ($pid = fork) { exit 0; }

    ## Change working directory
    chdir "/";

    ## Clear file creation mask
    umask 0;

    ## Close open file descriptors
    close(STDIN);
    close(STDOUT);
    close(STDERR);

    ## Reopen stderr, stdout, stdin to /dev/null
    open(STDIN,  "+>/dev/null");
    open(STDOUT, "+>&STDIN");
    open(STDERR, "+>&STDIN");
}

#####################################################################
### C L I E N T   C L A S S
#####################################################################
package Client;

use Danga::Socket;
use base 'Danga::Socket';
use fields (
            'locks',  # hashref of locks held by this connection. values are 1
            'read_buf',
            );

# TODO: out %waiters, lock -> arrayref of client waiters (waker should check not closed)

sub new {
    my Client $self = shift;
    $self = fields::new($self) unless ref $self;
    $self->SUPER::new( @_ );

    $self->{locks} = {};
    $self->{read_buf} = '';
    return $self;
}

# Client
sub event_read {
    my Client $self = shift;

    my $bref = $self->read(1024);
    return $self->close() unless defined $bref;
    $self->{read_buf} .= $$bref;

    if ($self->{read_buf} =~ s/^(.+?)\r?\n//) {
        my $line = $1;
        $self->process_line( $line );
    }
}

sub process_line {
    my Client $self = shift;
    my $line = shift;

    if ($line =~ /^(\w+)\s*(.*)/) {
        my ($cmd, $args) = ($1, $2);
        $cmd = lc($cmd);

        no strict 'refs';
        my $cmd_handler = *{"cmd_$cmd"}{CODE};
        if ($cmd_handler) {
            my $args = decode_url_args(\$args);
            $cmd_handler->($self, $args);
            next;
        }
    }

    return $self->err_line('unknown_command');
}

sub close {
    my Client $self = shift;

    foreach my $lock (keys %{$self->{locks}}) {
        $self->_release_lock($lock);
    }

    $self->SUPER::close;
}


# Client
sub event_err { my $self = shift; $self->close; }
sub event_hup { my $self = shift; $self->close; }

sub cmd_status {
    my Client $self = shift;

    my $runtime = time - $^T;

    my $load = current_load();
    $self->write("STATUS: OK\n");
    $self->write("SUCCESSES: $lock_successes\n");
    $self->write("FAILURES: $lock_failures\n");
    $self->write("RUNTIME: $runtime\n");
    $self->write("LOAD: $load\n");
    $self->write("\n");

    return 1;
}

sub iv_for_time {
    my $time = shift;
    return $time - ($time % $main::IV_duration);
}

sub latest_ivs {
    my $time = shift || time;
    my $current_iv = iv_for_time($time) + $main::IV_duration;
    return map { $current_iv -= $main::IV_duration } ( 0 .. $main::Bucket_qty - 1);  
}


sub current_load {
    my $sum = 0; 
    for (grep { $_ } @main::IVs{latest_ivs()}) {
         $sum += $_;
    }
    return $sum;
}

sub cmd_load {
    my Client $self = shift;
    my $load = current_load();
    $self->write("LOAD: $load\n");
    $self->write("\n");

    return 1;
}

sub increase_load {
    my $time = time; 
    $main::IVs{ iv_for_time($time) }++;
    if (scalar( keys %main::IVs ) > $main::Bucket_limit) {
        # purge IVs 
        %main::IVs = map { $_ => $main::IVs{$_} } latest_ivs($time);
    }
}

# gets a lock or fails with 'taken'
sub cmd_trylock {
    my Client $self = shift;
    my $args = shift;

    my $lock = $args->{lock};
    my $lockstate = $self->_trylock( $lock );

    increase_load();
    if ($lockstate) {
        $lock_successes++;
    } else {
        $lock_failures++;
    }

    return $lockstate;
}

# releases a lock or fails with 'didnthave'
sub cmd_releaselock {
    my Client $self = shift;
    my $args = shift;

    my $lock = $args->{lock};
    return $self->err_line("empty_lock") unless length($lock);
    return $self->err_line("didnthave") unless $self->{locks}{$lock};

    $self->_release_lock($lock);
    return $self->ok_line;
}

# shows current locks
sub cmd_locks {
    my Client $self = shift;
    my $args = shift;

    $self->write("LOCKS:\n");
    $self->write( join( "\n", $self->_get_locks ) );
    $self->write("\n");

    return 1;
}

sub cmd_noop {
    my Client $self = shift;
    # TODO: set self's last activity time so it isn't cleaned in a purge
    #       of stale connections?
    return $self->ok_line;
}

sub ok_line {
    my Client $self = shift;
    my $args = shift || {};
    my $argline = join('&', map { eurl($_) . "=" . eurl($args->{$_}) } keys %$args);
    $self->write("OK $argline\r\n");
    return 1;
}

sub err_line {
    my Client $self = shift;
    my $err_code = shift;
    my $err_text = {
        'unknown_command' => "Unknown server command",
    }->{$err_code};

    $self->write("ERR $err_code " . eurl($err_text) . "\r\n");
    return 0;
}

sub eurl
{
    my $a = $_[0];
    $a =~ s/([^a-zA-Z0-9_\,\-.\/\\\: ])/uc sprintf("%%%02x",ord($1))/eg;
    $a =~ tr/ /+/;
    return $a;
}

sub durl
{
    my ($a) = @_;
    $a =~ tr/+/ /;
    $a =~ s/%([a-fA-F0-9][a-fA-F0-9])/pack("C", hex($1))/eg;
    return $a;
}

sub decode_url_args
{
    my $a = shift;
    my $buffer = ref $a ? $a : \$a;
    my $ret = {};

    my $pair;
    my @pairs = split(/&/, $$buffer);
    my ($name, $value);
    foreach $pair (@pairs)
    {
        ($name, $value) = split(/=/, $pair);
        $value =~ tr/+/ /;
        $value =~ s/%([a-fA-F0-9][a-fA-F0-9])/pack("C", hex($1))/eg;
        $name =~ tr/+/ /;
        $name =~ s/%([a-fA-F0-9][a-fA-F0-9])/pack("C", hex($1))/eg;
        $ret->{$name} .= $ret->{$name} ? "\0$value" : $value;
    }
    return $ret;
}

package Client::Internal;

use base 'Client';

our (%holder);  # hash of lock -> Client object holding it
# TODO: out %waiters, lock -> arrayref of client waiters (waker should check not closed)

sub _setup {
    # Nothing to set up.
}

sub _trylock {
    my Client::Internal $self = shift;
    my $lock = shift;

    return $self->err_line("empty_lock") unless length($lock);
    return $self->err_line("taken") if defined $holder{$lock};

    $holder{$lock} = $self;
    $self->{locks}{$lock} = 1;

    return $self->ok_line();
}

sub _release_lock {
    my Client::Internal $self = shift;
    my $lock = shift;

    # TODO: notify waiters
    delete $self->{locks}{$lock};
    delete $holder{$lock};
    return 1;
}

sub _get_locks {
    return map { "  $_ = " . $holder{$_}->as_string } (sort keys %holder);
}

package Client::DLMFS;

use base 'Client';
use Fcntl;
use Errno qw(EEXIST ETXTBSY);

sub FLAGS () { O_NONBLOCK | O_RDWR | O_CREAT | O_EXCL }
sub PATH () { "/dlm/ddlockd" };

sub _setup {
    -d "/dlm" or die( "DLMFS mount at /dlm not found\n" );
    mkdir PATH;
}

sub _trylock {
    my Client::Internal $self = shift;
    my $lock = shift;

    return $self->err_line("empty_lock") unless length($lock);

    if (sysopen( my $handle, PATH . "/$lock", FLAGS )) {
        $self->{locks}{$lock} = 1;
        return $self->ok_line();
    }
    else {
        if ($! == EEXIST) {
            return $self->err_line( "local taken" );
        }
        elsif( $! == ETXTBSY) {
            unlink( PATH . "/$lock" );
            return $self->err_line( "remote taken" );
        }
        else {
            return $self->err_line( "unknown: $!" );
        }
    }
}

sub _release_lock {
    my Client::Internal $self = shift;
    my $lock = shift;

    # TODO: notify waiters
    delete $self->{locks}{$lock};
    unlink( PATH . "/$lock" );
    return 1;
}

sub _get_locks {
# TODO
#    return map { "  $_ = " . $holder{$_}->as_string } (sort keys %holder);
}

package Client::DBI;

# CREATE TABLE

use base 'Client';
use fields qw(dbh);

my $hostname;
my $table;
my $dbh;

sub _setup {
    my Client::DBI $self = shift;
    ($hostname, $table) = @_;
    eval "use DBI; 1" or die "No DBI available?";
    $dbh = DBI->connect( 'dbi:mysql:dbname=sixalock', '', '', {AutoCommit => 0} ) or die;
}

sub _trylock {
    my Client::DBI $self = shift;
    my $lock = shift;

    my $local_locks = $self->{locks};
    exists( $local_locks->{$lock} ) and return $self->err_line( "local taken" );

    $dbh->do( 'START TRANSACTION' ) or return $self->err_line( "Transaction failed to start" );

    my $sth = $dbh->prepare( "SELECT * FROM $table WHERE name=?" );
    $sth->execute( $lock )
        or return $self->err_line( "STH->execute failed" );
    my $ary = $sth->fetchall_arrayref;
    ref($ary) eq 'ARRAY'
        or return $self->err_line( "DBI->selectall_arrayref returned non-arrayref" );
    scalar @$ary == 0
        or return $self->err_line( "remote taken" );

    $dbh->do( "INSERT INTO $table (name) VALUES (?)", {}, $lock )
        or return $self->err_line( "INSERT failed" );
    $dbh->do( 'COMMIT' )
        or return $self->err_line( "COMMIT failed" );
    return $self->ok_line();
}

sub _release_lock {
    my Client::DBI $self = shift;
    my $lock = shift;

    my $locks = $self->{locks};
    if (exists( $locks->{$lock} )) {
        delete $locks->{$lock};
        $dbh->do( "SELECT AND DELETE * FROM $table WHERE name=?", {}, $lock );
        return 1;
    }
    else {
        return 0;
    }
}

sub _get_locks {
    my Client::DBI $self = shift;

    my $ary = $dbh->selectall_arrayref( "SELECT name FROM $table" );
    return map { $_->[0] } @$ary;
}

# Local Variables:
# mode: perl
# c-basic-indent: 4
# indent-tabs-mode: nil
# End: