package MogileFS::Worker;
use strict;
use fields ('psock',              # socket for parent/child communications
            'last_bcast_state',   # "{device|host}-$devid" => [$time, {alive|dead}]
            'readbuf',            # unparsed data from parent
            'monitor_has_run',    # true once we've heard of the monitor job being alive
            'last_ping',          # time we last said we're alive
            'woken_up',           # bool: if we've been woken up
            'last_wake',          # hashref: { $class -> time() } when we last woke up a certain job class
            'queue_depth',        # depth of a queue we queried
            'queue_todo',         # aref of hrefs of work sent from parent
            );

use MogileFS::Util qw(error eurl decode_url_args apply_state_events);
use MogileFS::Server;

use vars (
          '$got_live_vs_die',    # local'ized scalarref flag for whether we've
                                 # gotten a live-vs-die instruction from parent
          );

sub new {
    my ($self, $psock) = @_;
    $self = fields::new($self) unless ref $self;

    $self->{psock}            = $psock;
    $self->{readbuf}          = '';
    $self->{last_bcast_state} = {};
    $self->{monitor_has_run}  = MogileFS::ProcManager->is_monitor_good;
    $self->{last_ping}        = 0;
    $self->{last_wake}        = {};
    $self->{queue_depth}      = {};
    $self->{queue_todo}       = {};

    IO::Handle::blocking($psock, 0);
    return $self;
}

sub psock_fd {
    my $self = shift;
    return fileno($self->{psock});
}

sub psock {
    my $self = shift;
    return $self->{psock};
}

sub validate_dbh {
    return Mgd::validate_dbh();
}

sub monitor_has_run {
    my $self = shift;
    return $self->{monitor_has_run} ? 1 : 0;
}

sub forget_that_monitor_has_run {
    my $self = shift;
    $self->{monitor_has_run} = 0;
}

sub wait_for_monitor {
    my $self = shift;
    while (! $self->monitor_has_run) {
        $self->read_from_parent(1);
        $self->still_alive;
    }
}

# method that workers can call just to write something to the parent, so worker
# doesn't get killed.  (during idle/slow operation, say)
# returns current time, so caller can avoid a time() call as well, for its loop
sub still_alive {
    my $self = shift;
    my $now = time();
    if ($now > $self->{last_ping} + ($self->watchdog_timeout / 4)) {
        $self->send_to_parent(":still_alive");  # a no-op, just for the watchdog
        $self->{last_ping} = $now;
    }
    return $now;
}

sub send_to_parent {
    my $self = shift;

    # can be called as package method:  MogileFS::Worker->send_to_parent...
    unless (ref $self) {
        $self = MogileFS::ProcManager->is_child
            or return;
    }

    my $write = "$_[0]\r\n";
    my $totallen = length $write;
    my $rv = syswrite($self->{psock}, $write);
    return 1 if defined $rv && $rv == $totallen;
    die "Error writing to parent process: $!" if $! && ! $!{EAGAIN};

    $rv ||= 0;  # could've been undef, if EAGAIN immediately.
    my $remain = $totallen - $rv;
    my $offset = $rv;
    while ($remain > 0) {
        MogileFS::Util::wait_for_writeability(fileno($self->{psock}), 30)
            or die "Parent not writable in 30 seconds";

        $rv = syswrite($self->{psock}, $write, $remain, $offset);
        die "Error writing to parent process (in loop): $!" if $! && ! $!{EAGAIN};
        if ($rv) {
            $remain -= $rv;
            $offset += $rv;
        }
    }
    die "remain is negative:  $remain" if $remain < 0;
    return 1;
}

# override in children
sub watchdog_timeout {
    return 10;
}

# should be overridden by workers to process worker-specific directives
# from the parent process.  return 1 if you recognize the command, 0 otherwise.
sub process_line {
    my ($self, $lineref) = @_;
    return 0;
}

sub read_from_parent {
    my $self    = shift;
    my $timeout = shift || 0;
    my $psock = $self->{psock};

    # while things are immediately available,
    # (or optionally sleep a bit)
    while (MogileFS::Util::wait_for_readability(fileno($psock), $timeout)) {
        $timeout = 0; # only wait on the timeout for the first read.
        my $buf;
        my $rv = sysread($psock, $buf, Mgd::UNIX_RCVBUF_SIZE());
        if (!$rv) {
            if (defined $rv) {
                die "While reading pipe from parent, got EOF.  Parent's gone.  Quitting.\n";
            } else {
                die "Error reading pipe from parent: $!\n";
            }
        }

        if ($Mgd::POST_SLEEP_DEBUG) {
            my $out = $buf;
            $out =~ s/\s+$//;
            warn "proc ${self}[$$] read: [$out]\n"
        }
        $self->{readbuf} .= $buf;

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

            next if $self->process_generic_command(\$line);
            my $ok = $self->process_line(\$line);
            unless ($ok) {
                error("Unrecognized command from parent: $line");
            }
        }
    }
}

sub parent_ping {
    my $self = shift;
    my $psock = $self->{psock};
    $self->send_to_parent(':ping');

    my $got_reply = 0;
    die "recursive parent_ping!" if $got_live_vs_die;
    local $got_live_vs_die = \$got_reply;

    my $loops = 0;

    while (!$got_reply) {
        $self->read_from_parent;
        return if $got_reply;

        $loops++;
        select undef, undef, undef, 0.20;
        if ($loops > 5) {
            warn "No simple reply from parent to child $self [$$] in $loops 0.2second loops.\n";
            die "No answer in 4 seconds from parent to child $self [$$], dying" if $loops > 20;
        }
    }
}

# tries to parse generic (not job-specific) commands sent from parent
# to child.  returns 1 on success, or 0 if command given isn't generic,
# and child should parse.
# lineref doesn't have \r\n at end.
sub process_generic_command {
    my ($self, $lineref) = @_;
    return 0 unless $$lineref =~ /^:/;  # all generic commands start with colon

    if ($$lineref =~ /^:shutdown/) {
        $$got_live_vs_die = 1 if $got_live_vs_die;
        exit 0;
    }

    if ($$lineref =~ /^:stay_alive/) {
        $$got_live_vs_die = 1 if $got_live_vs_die;
        return 1;
    }

    if ($$lineref =~ /^:monitor_events/) {
        apply_state_events($lineref);
        return 1;
    }

    if ($$lineref =~ /^:monitor_has_run/) {
        $self->{monitor_has_run} = 1;
        return 1;
    }

    if ($$lineref =~ /^:wake_up/) {
        $self->{woken_up} = 1;
        return 1;
    }

    if ($$lineref =~ /^:set_config_from_parent (\S+) (.+)/) {
        # the 'no_broadcast' API keeps us from looping forever.
        MogileFS::Config->set_config_no_broadcast($1, $2);
        return 1;
    }

    # queue_name depth
    if ($$lineref =~ /^:queue_depth (\w+) (\d+)/) {
        $self->queue_depth($1, $2);
        return 1;
    }

    # queue_name encoded_item
    if ($$lineref =~ /^:queue_todo (\w+) (.+)/) {
        # TODO: Use the accessor.
        push(@{$self->{queue_todo}->{$1}}, decode_url_args(\$2));
        return 1;
    }

    # TODO: warn on unknown commands?

    return 0;
}

sub queue_depth {
    my MogileFS::Worker $self = shift;
    my $type = shift;
    $self->{queue_depth}->{$type} ||= 0;
    return $self->{queue_depth}->{$type} unless @_;
    return $self->{queue_depth}->{$type} = shift;
}

sub queue_todo {
    my MogileFS::Worker $self = shift;
    my $type = shift;
    $self->{queue_todo}->{$type} ||= [];
    push(@{$self->{queue_todo}->{$type}}, @_) if @_;
    return $self->{queue_todo}->{$type};
}

sub was_woken_up {
    my MogileFS::Worker $self = shift;
    return $self->{woken_up};
}

sub forget_woken_up {
    my MogileFS::Worker $self = shift;
    $self->{woken_up} = 0;
}

# don't wake processes more than once a second... not necessary.
sub wake_a {
    my ($self, $class) = @_;
    my $now = time();
    return if ($self->{last_wake}{$class}||0) == $now;
    $self->{last_wake}{$class} = $now;
    $self->send_to_parent(":wake_a $class");
}

1;

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