File Coverage

blib/lib/IO/Lambda/Thread.pm

Criterion	Covered	Total	%
statement	21	77	27.2
branch	0	26	0.0
condition	0	9	0.0
subroutine	7	16	43.7
pod	4	5	80.0
total	32	133	24.0

# $Id: Thread.pm,v 1.25 2010/03/27 19:51:24 dk Exp $
package IO::Lambda::Thread;
use base qw(IO::Lambda);
use strict;
use warnings;
use Exporter;
use Socket;
use IO::Handle;
use IO::Lambda qw(:all :dev swap_frame);

our $DISABLED;
eval { require threads; };
$DISABLED = $@ if $@;

our $DEBUG = $IO::Lambda::DEBUG{thread};

our @EXPORT_OK = qw(threaded new_thread);
our %EXPORT_TAGS = (all => \@EXPORT_OK);

sub _d { "threaded(" . _o($_[0]) . ")" }

sub thread_init
{
	my ( $r, $cb, $pass_handle, @param) = @_;

	$SIG{KILL} = sub { threads-> exit(0) };
	$SIG{PIPE} = 'IGNORE';
	eval "END { IO::Lambda::__end(); };";
	warn "thread(", threads->tid, ") started\n" if $DEBUG;

	my @ret;
	eval { @ret = $cb-> (( $pass_handle ? $r : ()), @param) if $cb };

	warn "thread(", threads->tid, ") ended: [@ret]\n" if $DEBUG;
	close($r);
	undef $r;
	die $@ if $@;

	return @ret;
}

sub new_thread
{
	return undef, $DISABLED if $DISABLED;

	my ( @args, $cb, $pass_handle, @param);
	@args = shift if $_[0] and ref($_[0]) and ref($_[0]) eq 'HASH';
	( $cb, $pass_handle, @param) = @_;

	my $r = IO::Handle-> new;
	my $w = IO::Handle-> new;
	socketpair( $r, $w, AF_UNIX, SOCK_STREAM, PF_UNSPEC);
	$w-> blocking(0);

	my ($t) = threads-> create(
		@args,
		\&thread_init,
		$r, $cb, $pass_handle, @param
	);

	close($r);

	warn "new thread(", $t->tid, ")\n" if $DEBUG;
	return ($t, $w);
}

# overridden IO::Lambda methods

sub DESTROY
{
	my $self = shift;

	return if defined($self->{tid}) and $self->{tid} != threads-> tid;

	close($self->{socket}) if $self-> {socket};
	delete @{$self}{qw(socket thread)};

	$self-> SUPER::DESTROY;
}

sub thread { $_[0]-> {thread} }
sub socket { $_[0]-> {socket} }

sub threaded(&)
{
	my $cb = shift;

	# use overridden IO::Lambda, because we need
	# give the caller a chance to join
	# for it, if the lambda gets terminated
	__PACKAGE__-> new( sub {
		# Save context. This is needed because the caller
		# may have his own this. lambda(&) does the same
		# protection
		my $this  = shift;
		my @frame = swap_frame($this);

		warn _d($this), " started\n" if $DEBUG;

		# can start a thread?
		my ( $t, $r) = new_thread( $cb, 1 );
		return $r unless $t;

		# save this
		$this-> {tid}    = threads-> tid;
		$this-> {thread} = $t;
		$this-> {socket} = $r;

		# now wait
		context $this-> {socket};
		readable {
			my $this = this;
			delete $this-> {thread};
			close($this-> {socket});
			delete @{$this}{qw(socket thread)};
			$this-> clear;
			warn _d($this), " joining\n" if $DEBUG;
			$t-> join;
		};

		# restore context
		swap_frame(@frame);
	});
}

1;

__DATA__