view release on metacpan or  search on metacpan

lib/App/Context/POE/ClusterNode.pm  view on Meta::CPAN

    &App::sub_exit() if ($App::trace);
}

sub send_node_status {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;
    my $controller_host = $self->{controller_host};
    my $controller_port = $self->{controller_port};
    my $node_host       = $self->{host};
    my $node_port       = $self->{port};

    my $remote_server_name = "poe_${controller_host}_${controller_port}";
    my $remote_session_alias = $self->{poe_session_name};  # remote is same as local
    my $remote_session_state = "poe_receive_node_status";
    my $sys_info = $self->get_sys_info();
    my $memfree = $sys_info->{memfree} + $sys_info->{buffers} + $sys_info->{cached};
    my $s_info = {
        load => $sys_info->{load},
        system_load => $sys_info->{load},
        memfree => $memfree,
        memtotal => $sys_info->{memtotal},
        swapfree => $sys_info->{swapfree},
        swaptotal => $sys_info->{swaptotal},
        max_async_events => $self->{max_async_events}
    };

    if ($self->{controller_up}) {
        my $kernel = $self->{poe_kernel};
        $kernel->post("IKC", "post", "poe://$remote_server_name/$remote_session_alias/$remote_session_state",
            [ "$node_host:$node_port", $s_info ]);
    }

    &App::sub_exit() if ($App::trace);
}

sub alarm_noop {
    &App::sub_entry if ($App::trace);
    &App::sub_exit() if ($App::trace);
    return();
}

sub state {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;

    my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
    my $state = "Cluster Node: Node: $self->{host}:$self->{port}  procs[$self->{num_procs}/$self->{max_procs}:max]  async_events[$self->{num_async_events}/$self->{max_async_events}:max]\n[$datetime]\n";
    $state .= "\n";
    $state .= $self->_state();

    &App::sub_exit($state) if ($App::trace);
    return($state);
}

sub _state {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;

    my $state = "";

    $state .= $self->SUPER::_state();

    &App::sub_exit($state) if ($App::trace);
    return($state);
}

sub poe_cancel_async_event {
    &App::sub_entry if ($App::trace);
    my ( $self, $kernel, $heap, $arg0 ) = @_[ OBJECT, KERNEL, HEAP, ARG0 ];
    $self->profile_start("poe_cancel_async_event") if $self->{poe_profile};
    my ($runtime_event_token) = @$arg0;

    $self->log({level=>3},"poe_cancel_async_event BEGIN runtime_event_token=[$runtime_event_token]\n");
    ### Find if running
    my ($event_token);
    for my $pid (keys %{$self->{running_async_event}}) {
        $event_token = $self->{running_async_event}{$pid}[0]{event_token};
        if ($runtime_event_token eq $event_token) {

            ### Kill it
            if ($pid =~ /^[0-9]+$/) {
                kill(9, $pid);
            }

            ### Remove from pending
            delete $self->{running_async_event}{$pid};
            $self->log({level=>3},"poe_cancel_async_event FOUND RUNNING event_token=[$event_token] pid=[$pid]\n");

            last;
        }
    }

    ### Find if pending
    for (my $i = 0; $i < @{$self->{pending_async_events}}; $i++) {
        $event_token = $self->{pending_async_events}[$i][0]{event_token};
        if ($runtime_event_token eq $event_token) {
            splice(@{$self->{pending_async_events}}, $i, 1);
            $self->log({level=>3},"poe_cancel_async_event FOUND PENDING event_token=[$event_token]\n");
        }
    }
    $self->log({level=>3},"poe_cancel_async_event END   event_token=[$event_token]\n");
    $self->profile_stop("poe_cancel_async_event") if $self->{poe_profile};

    &App::sub_exit() if ($App::trace);
}

1;