Inheritance diagram for Bio::EnsEMBL::Hive::GuestProcess:

Collaboration diagram for Bio::EnsEMBL::Hive::GuestProcess:

Public Member Functions
public String	get_protocol_version ()

public Bio::EnsEMBL::Hive::GuestProcess	new ()

protected String	_get_wrapper_for_language ()

public void	DESTROY ()

public void	print_debug ()

public IO::Handle	child_in ()

public IO::Handle	child_out ()

public Int	child_pid ()

public Instance	json_formatter ()

public void	send_message ()

public void	send_response ()

public Perl	read_message ()

public void	wait_for_OK ()

public Hashref	param_defaults ()

public Hashref	life_cycle ()

public String	worker_temp_directory_name ()

Public Member Functions inherited from Bio::EnsEMBL::Hive::Process
public	new ()

public	life_cycle ()

public	say_with_header ()

public	enter_status ()

public	warning ()

public	param_defaults ()

public	fetch_input ()

public	run ()

public	write_output ()

public Bio::EnsEMBL::Hive::Worker	worker ()

public Boolean	execute_writes ()

public Bio::EnsEMBL::Hive::DBSQL::DBAdaptor	db ()

public Bio::EnsEMBL::Hive::DBSQL::DBConnection	dbc ()

public Bio::EnsEMBL::Hive::DBSQL::DBConnection	data_dbc ()

public Bio::EnsEMBL::Hive::AnalysisJob	input_job ()

public	input_id ()

public	param ()

public	param_required ()

public	param_is_defined ()

public	param_substitute ()

public	dataflow_output_id ()

public	throw ()

public	complete_early ()

public Int	debug ()

public	worker_temp_directory ()

public	worker_temp_directory_name ()

public	cleanup_worker_temp_directory ()

Detailed Description

Synopsis

This is a variant of Bio::EnsEMBL::Hive::Process that forks into a wrapper that can itself

run jobs (runnables) written in a different language

Description

Upon initialisation, GuestProcess forks, and the child process executes the wrapper that
will allow running Runnables of the other language. The communication is ensured by two
pipes and is schematically similar to running "| wrapper |", except that GuestProcess
uses non-standard file descriptors, thus allowing the Runnable to still use std{in,out,err}.

The wrapper receives the two file-numbers that it is meant to use (one for reading data
from GuestProcess, and one to send data to GuestProcess). All the messages are passed
around in single-line JSON structures. The protocol is described below using the convention:
    ---> represents a message sent to the child process,
    <--- represents a message sent by the child process

The initialisation (in the constructor) consists in checking that both sides spek the same
version of the protocol:
    <--- { "version": "XXX" }
    ---> "OK"
GuestProcess will bail out if the response is not "OK"

Then, the child process (i.e. the runnable) will send its default parameters to GuestProcess.
This fills the usual param_defaults() section of the Runnable:
    <--- { ... param_defaults ... }
    ---> "OK"

The child process then goes to sleep, waiting for jobs to be seeded. Meanwhile,
GuestProcess enters a number of life_cycle() executions (as triggered by Worker).
Each one first sends a JSON object to the child process to initialize the job parameters
    ---> {
           "input_job": {
             "parameters": { ... the unsubstituted job parameters as compiled by Worker ... },
             // followed by several attributes of the job
             "input_id": { ...  },
             "dbID": XXX,
             "retry_count": XXX
           },
           "execute_writes": [1|0],
           "debug": XXX
         }
    <--- "OK"

From this point, GuestProcess acts as a server, listening to events sent by the child.
Events are JSON objects composed of an "event" field (the name of the event) and a
"content" field (the payload). Events can be of the following kinds (with the expected
response from GuestProcess):

    <--- JOB_STATUS_UPDATE
         // The content is one of "PRE_CLEANUP", "FETCH_INPUT", "RUN", "WRITE_OUTPUT", "POST_CLEANUP"
    ---> "OK"

    <--- WARNING
         // The content is a JSON object:
            {
              "message": "XXX",
              "is_error": [true|false],
            }
    ---> "OK"

    <--- DATAFLOW
         // The content is a JSON object:
            {
              "branch_name_or_code": XXX,
              "output_ids": an array or a hash,
              "params": {
                "substituted": { ... the parameters that are currently substituted ... }
                "unsubstituted": { ... the parameters that have not yet been substituted ... }
              }
            }
    ---> dbIDs of the jobs that have been created

    <--- WORKER_TEMP_DIRECTORY
         // The content is the "worker_temp_directory_name" as defined in the runnable (or null otherwise)
    ---> returns the temporary directory of the worker

    <--- JOB_END
         // The content is a JSON object describing the final state of the job
            {
              "complete": [true|false],
              "job": {
                "autoflow": [true|false],
                "lethal_for_worker": [true|false],
                "transient_error": [true|false],
              },
              "params": {
                "substituted": { ... the parameters that are currently substituted ... }
                "unsubstituted": { ... the parameters that have not yet been substituted ... }
              }
            }
    ---> "OK"

Member Function Documentation

◆ _get_wrapper_for_language()

protected String Bio::EnsEMBL::Hive::GuestProcess::_get_wrapper_for_language ( )

  Example     :

Bio::EnsEMBL::Hive::GuestProcess::_get_wrapper_for_language('python3');

  Description : Finds the wrapper that understands the given language
  Returntype  : String
  Exceptions  : Can die if the wrapper doesn't exist

Code:

click to view

◆ child_in()

public IO::Handle Bio::EnsEMBL::Hive::GuestProcess::child_in ( )

  Example     :

my $child_in = $process->child_in();

  Example     :

$process->child_in(*CHILD_WTR);

  Description : Getter/Setter for the file handle that allows talking to the
                child process.
  Returntype  : IO::Handle
  Exceptions  : none

Code:

click to view

◆ child_out()

public IO::Handle Bio::EnsEMBL::Hive::GuestProcess::child_out ( )

  Example     :

my $child_out = $process->child_out();

  Example     :

$process->child_out(*CHILD_RDR);

  Description : Getter/Setter for the file handle that allows receiving data
                from the child process.
  Returntype  : IO::Handle
  Exceptions  : none

Code:

click to view

◆ child_pid()

public Int Bio::EnsEMBL::Hive::GuestProcess::child_pid ( )

  Example     :

my $child_pid = $process->child_pid();

  Example     :

$process->child_pid($child_pid);

  Description : Getter/Setter for the process ID of the child
  Returntype  : integer
  Exceptions  : none

Code:

click to view

◆ DESTROY()

public void Bio::EnsEMBL::Hive::GuestProcess::DESTROY ( )

  Description : Destructor: tells the child to exit by sending an empty JSON object
  Returntype  : none

Code:

click to view

◆ get_protocol_version()

public String Bio::EnsEMBL::Hive::GuestProcess::get_protocol_version ( )

  Example     :

print Bio::EnsEMBL::Hive::GuestProcess->get_protocol_version(), "\n";

  Description : Returns the version number of the communication protocol
  Returntype  : String

Code:

click to view

◆ json_formatter()

public Instance Bio::EnsEMBL::Hive::GuestProcess::json_formatter ( )

  Example     :

my $json_formatter = $object_name->json_formatter();

  Example     :

$object_name->json_formatter($json_formatter);

  Description : Getter/Setter for the JSON formatter.
  Returntype  : instance of JSON
  Exceptions  : none

Code:

click to view

◆ life_cycle()

public Hashref Bio::EnsEMBL::Hive::GuestProcess::life_cycle ( )

  Example     :

my $partial_timings = $runnable->life_cycle();

  Description : Runs the life-cycle of the input job and returns the timings
                of each Runnable method (fetch_input, run, etc).
                See the description of this module for details about the protocol
  Returntype  : Hashref
  Exceptions  : none

Code:

click to view

sub life_cycle {
    my $self = shift;
    $self->print_debug("LIFE_CYCLE");
    my $job = $self->input_job();
    my $partial_stopwatch = Bio::EnsEMBL::Hive::Utils::Stopwatch->new();
    my %job_partial_timing = ();
    my %struct = (
        input_job => {
            parameters => $job->{_unsubstituted_param_hash},
            input_id => $job->input_id,
            dbID => $job->dbID + 0,
            retry_count => $job->retry_count + 0,
        },
        execute_writes => $self->execute_writes || 0,
        debug => $self->debug || 0,
    );
    $self->print_debug("SEND JOB PARAM");
    $self->send_message(\%struct);
    $self->wait_for_OK();
    # A simple event loop
    while (1) {
        $self->print_debug("WAITING IN LOOP");
        my $msg = $self->read_message;
        my $event = $msg->{event};
        my $content = $msg->{content};
        $self->print_debug("processing event '$event'");
        if ($event eq 'JOB_STATUS_UPDATE') {
            $job_partial_timing{$job->status} = $partial_stopwatch->get_elapsed() if ($job->status ne 'READY') and ($job->status ne 'CLAIMED');
            $self->enter_status(uc $content);
            $partial_stopwatch->restart();
            $self->send_response('OK');
        } elsif ($event eq 'WARNING') {
            $self->warning($content->{message}, $content->{is_error});
            $self->send_response('OK');
        } elsif ($event eq 'DATAFLOW') {
            $job->{_param_hash} = $content->{params}->{substituted};
            $job->{_unsubstituted_param_hash} = $content->{params}->{unsubstituted};
            my $d = $self->dataflow_output_id($content->{output_ids}, $content->{branch_name_or_code});
            $self->send_message($d);
        } elsif ($event eq 'WORKER_TEMP_DIRECTORY') {
            $self->{worker_temp_directory_name} = $content;
            my $wtd = $self->worker_temp_directory;
            $self->send_response($wtd);
        } elsif ($event eq 'JOB_END') {
            # Especially here we need to be careful about boolean values
            # They are coded as JSON::true and JSON::false which have
            # different meanings in text / number contexts
            $job->autoflow($job->autoflow and $content->{job}->{autoflow});
            $job->lethal_for_worker($content->{job}->{lethal_for_worker}?1:0);
            $job->transient_error($content->{job}->{transient_error}?1:0);
            $job->{_param_hash} = $content->{params}->{substituted};
            $job->{_unsubstituted_param_hash} = $content->{params}->{unsubstituted};
            # This piece of code is duplicated from Process
            if ($content->{complete}) {
                if( $self->execute_writes and $job->autoflow ) {    # AUTOFLOW doesn't have its own status so will have whatever previous state of the job
                    $self->say_with_header( ': AUTOFLOW input->output' );
                    $job->dataflow_output_id();
                }
                my @zombie_funnel_dataflow_rule_ids = keys %{$job->fan_cache};
                if( scalar(@zombie_funnel_dataflow_rule_ids) ) {
                    $job->transient_error(0);
                    die "There are cached semaphored fans for which a funnel job (dataflow_rule_id(s) ".join(',',@zombie_funnel_dataflow_rule_ids).") has never been dataflown";
                }
            } else {
                $job->died_somewhere(1);
            }
            $self->send_response('OK');
            return \%job_partial_timing;
        } else {
            die "Unknown event '$event' coming from the child";
        }
    }
}

◆ new()

public Bio::EnsEMBL::Hive::GuestProcess Bio::EnsEMBL::Hive::GuestProcess::new ( )

  Arg[1]      : $language: the programming language the external runnable is in
  Arg[2]      : $module: the name of the runnable (usually a package name)
  Example     :

Bio::EnsEMBL::Hive::GuestProcess->new();

  Description : Constructor
  Returntype  : Bio::EnsEMBL::Hive::GuestProcess
  Exceptions  : if $language or $module is not defined properly or if the pipes /
                child process could not be created

Code:

click to view

sub new {
    my ($class, $language, $module) = @_;
    die "GuestProcess must be told which language to interface with" unless $language;
    my $wrapper = _get_wrapper_for_language($language);
    die "GuestProcess must be told which module to run" unless $module;
    my ($PARENT_RDR, $PARENT_WTR, $CHILD_WTR,$CHILD_RDR);
    pipe($PARENT_RDR, $CHILD_WTR) or die 'Could not create a pipe to send data to the child !';
    pipe($CHILD_RDR,  $PARENT_WTR) or die 'Could not create a pipe to get data from the child !';;
    print STDERR "PARENT_RDR is ", fileno($PARENT_RDR), "\n";
    print STDERR "PARENT_WTR is ", fileno($PARENT_WTR), "\n";
    print STDERR "CHILD_RDR is ", fileno($CHILD_RDR), "\n";
    print STDERR "CHILD_WTR is ", fileno($CHILD_WTR), "\n";
    my $pid;
    if ($pid = fork()) {
        # In the parent
        close $PARENT_RDR;
        close $PARENT_WTR;
        print STDERR "parent is PID $$\n";
    } else {
        die "cannot fork: $!" unless defined $pid;
        # In the child
        close $CHILD_RDR;
        close $CHILD_WTR;
        print STDERR "child is PID $$\n";
        # Do not close the non-standard file descriptors on exec(): the child process will need them !
        use Fcntl;
        my $flags = fcntl($PARENT_RDR, F_GETFD, 0);
        fcntl($PARENT_RDR, F_SETFD, $flags & ~FD_CLOEXEC);
        $flags = fcntl($PARENT_WTR, F_GETFD, 0);
        fcntl($PARENT_WTR, F_SETFD, $flags & ~FD_CLOEXEC);
        exec($wrapper, 'run', $module, fileno($PARENT_RDR), fileno($PARENT_WTR));
    }
    $CHILD_WTR->autoflush(1);
    my $self = bless {}, $class;
    $self->child_out($CHILD_RDR);
    $self->child_in($CHILD_WTR);
    $self->child_pid($pid);
    $self->json_formatter( JSON->new()->indent(0) );
    $self->print_debug('CHECK VERSION NUMBER');
    my $other_version = $self->read_message()->{content};
    if ($other_version ne $VERSION) {
        $self->send_response('NO');
        die "eHive's protocol version is '$VERSION' but the wrapper's is '$other_version'\n";
    } else {
        $self->send_response('OK');
    }
    $self->print_debug("BEFORE READ PARAM_DEFAULTS");
    $self->param_defaults( $self->read_message()->{content} );
    $self->send_response('OK');
    $self->print_debug("INIT DONE");
    return $self;
}

◆ param_defaults()

public Hashref Bio::EnsEMBL::Hive::GuestProcess::param_defaults ( )

  Example     :

my $param_defaults = $runnable->param_defaults();

  Example     :

$runnable->param_defaults($param_defaults);

  Description : Getter/Setter for the default parameters of this runnable.
                Hive only uses it as a getter, but here, we need a setter to
                define the parameters at the Perl layer once they've been
                retrieved from the child process.
  Returntype  : Hashref
  Exceptions  : none

Code:

click to view

◆ print_debug()

public void Bio::EnsEMBL::Hive::GuestProcess::print_debug ( )

  Example     :

$process->print_debug("debug message");

  Description : Prints a message if $self->debug is 2 or above
  Returntype  : none

Code:

click to view

◆ read_message()

public Perl Bio::EnsEMBL::Hive::GuestProcess::read_message ( )

  Example     :

my $msg = $process->read_message();

  Description : Wait for and read the next message coming from the child.
                Again, the message itself is serialized and transmitted
                via the pipe
  Returntype  : Perl structure
  Exceptions  : raised by JSON / IO::Handle

Code:

click to view

◆ send_message()

public void Bio::EnsEMBL::Hive::GuestProcess::send_message ( )

  Example     :

$process->send_message($perl_structure);

  Description : Send the Perl structure to the child process via the pipe (and
                serialized in JSON).
  Returntype  : none
  Exceptions  : raised by JSON / IO::Handle

Code:

click to view

◆ send_response()

public void Bio::EnsEMBL::Hive::GuestProcess::send_response ( )

  Example     :

$process->send_response('OK');

  Description : Wrapper around send_message to send a response to the child.
  Returntype  : none
  Exceptions  : raised by JSON / IO::Handle

Code:

click to view

◆ wait_for_OK()

public void Bio::EnsEMBL::Hive::GuestProcess::wait_for_OK ( )

  Example     :

$process->wait_for_OK();

  Description : Wait for the child process to send the OK signal
  Returntype  : none
  Exceptions  : dies if the response is not OK, or anything raised by read_message()

Code:

click to view

◆ worker_temp_directory_name()

public String Bio::EnsEMBL::Hive::GuestProcess::worker_temp_directory_name ( )

  Example     :

$process->worker_temp_directory_name();

  Description : Returns the name of the temp directory for this module
                The value in $self is initialized at the WORKER_TEMP_DIRECTORY
                event above and returned to the caller if defined. This allows
                runnables to redefine the name
  Returntype  : string
  Exceptions  : none

Code:

click to view

The documentation for this class was generated from the following file:

modules/Bio/EnsEMBL/Hive/GuestProcess.pm

Public Member Functions

Detailed Description

Synopsis

Description

Member Function Documentation

◆ _get_wrapper_for_language()

◆ child_in()

◆ child_out()

◆ child_pid()

◆ DESTROY()

◆ get_protocol_version()

◆ json_formatter()

◆ life_cycle()

◆ new()

◆ param_defaults()

◆ print_debug()

◆ read_message()

◆ send_message()

◆ send_response()

◆ wait_for_OK()

◆ worker_temp_directory_name()