File Coverage

blib/lib/PDL/Parallel/threads.pm

Criterion	Covered	Total	%
statement	75	92	81.5
branch	16	30	53.3
condition	2	6	33.3
subroutine	11	11	100.0
pod	3	5	60.0
total	107	144	74.3

package PDL::Parallel::threads;

use strict;
use warnings;
use Carp;
use PDL::LiteF;
use PDL::Exporter;
our $VERSION = '0.07';

our @ISA = ( 'PDL::Exporter' );
our @EXPORT_OK = qw(share_pdls retrieve_pdls free_pdls);
our %EXPORT_TAGS = (Func=>\@EXPORT_OK);

my $can_use_threads;
BEGIN {
	$can_use_threads = eval {
		require threads;
		threads->import();
		require threads::shared;
		threads::shared->import();
		1;
	};
}

# These are the means by which we share data across Perl threads. Note that
# we cannot share ndarrays directly across threads, but we can share arrays
# of scalars, scalars whose integer values are the pointers to ndarray data,
# etc.
my %datasv_pointers :shared;
my %dataref_svs;
my %dim_arrays :shared;
my %types :shared;
my %badflag :shared;
my %badvalue :shared;
my %nbytes :shared;
my %originating_tid :shared;

# PDL data should not be naively copied by Perl. Tell PDL we know about this
$PDL::no_clone_skip_warning = 1;

sub auto_package_name {
	my $name = shift;
	my ($package_name) = caller(1);
	$name = "$package_name/$name" if $name =~ /^\w+$/;
	return $name;
}

sub share_pdls {
	croak("share_pdls: expected key/value pairs")
		unless @_ % 2 == 0;
	my %to_store = @_;

	while (my ($name, $to_store) = each %to_store) {
		$name = auto_package_name($name);

		# Make sure we're not overwriting already shared data
		if (exists $datasv_pointers{$name}) {
			croak("share_pdls: you already have data associated with '$name'");
		}

		if ( eval{$to_store->isa("PDL")} ) {
			# Integers (which can be cast to and from
			# pointers) are easily shared using threads::shared
			# in a shared hash. This method provides a
			# way to obtain the pointer to the datasv for
			# the incoming ndarray, and it increments the
			# SV's refcount.
			$dataref_svs{$name} = eval{
			  croak("the ndarray does not have any allocated memory\n")
			    if !$to_store->allocated;
			  $to_store->get_dataref;
			};
			if ($@) {
				my $error = $@;
				chomp $error;
				delete $datasv_pointers{$name};
				croak('share_pdls: Could not share an ndarray under '
					. "name '$name' because $error");
			}
			$datasv_pointers{$name} = 0+$dataref_svs{$name};
			$to_store->set_donttouchdata($nbytes{$name} = $to_store->nbytes); # protect its memory
			if ($can_use_threads) {
				$dim_arrays{$name} = shared_clone([$to_store->dims]);
				$originating_tid{$name} = threads->tid;
			}
			else {
				$dim_arrays{$name} = [$to_store->dims];
			}
			$types{$name} = $to_store->get_datatype;
			$badflag{$name} = $to_store->badflag;
			my $badval = $to_store->badvalue->sclr;
			$badval = shared_clone([$badval->Re,$badval->Im]) if ref $badval ;
			$badvalue{$name} = $badval;
		}
		else {
			croak("share_pdls passed data under '$name' that it doesn't "
				. "know how to store");
		}
	}
}



# Frees the memory associated with the given names.
sub free_pdls {
	# Keep track of each name that is successfully freed
	my @removed;

	for my $short_name (@_) {
		my $name = auto_package_name($short_name);

		# If it's a regular ndarray, decrement the memory's refcount
		if (exists $datasv_pointers{$name}) {
			delete $dataref_svs{$name};
			delete $datasv_pointers{$name};
			delete $dim_arrays{$name};
			delete $types{$name};
			delete $badflag{$name};
			delete $badvalue{$name};
			delete $nbytes{$name};
			delete $originating_tid{$name};
			push @removed, $name;
		}
		# If its none of the above, indicate that we didn't free anything
		else {
			push @removed, '';
		}
	}

	return @removed;
}

# PDL method to share an individual ndarray
sub PDL::share_as {
	my ($self, $name) = @_;
	share_pdls(auto_package_name($name) => $self);
	return $self;
}

# Method to get an ndarray that points to the shared data associated with the
# given name(s).
sub retrieve_pdls {
	return if @_ == 0;

	my @to_return;
	for my $short_name (@_) {
		my $name = auto_package_name($short_name);

		if (exists $datasv_pointers{$name}) {
			# Make sure that the originating thread still exists, or the
			# data will be gone.
			if ($can_use_threads and $originating_tid{$name} > 0
				and not defined (threads->object($originating_tid{$name}))
			) {
				croak("retrieve_pdls: '$name' was created in a thread that "
						. "has ended or is detached");
			}

			# Create the new thinly wrapped ndarray
			my $new_ndarray = PDL->new_around_datasv($datasv_pointers{$name});
			$new_ndarray->set_datatype($types{$name});
			$new_ndarray->badflag($badflag{$name});
			$new_ndarray->badvalue(ref $badvalue{$name}
				? Math::Complex->make(@{$badvalue{$name}})
				: $badvalue{$name});
			$new_ndarray->setdims(\@{$dim_arrays{$name}});
			$new_ndarray->set_donttouchdata($nbytes{$name}); # protect its memory
			push @to_return, $new_ndarray;
		}
		else {
			croak("retrieve_pdls could not find data associated with '$name'");
		}
	}

	# In list context, return all the ndarrays
	return @to_return if wantarray;

	# Scalar context only makes sense if they asked for a single name
	return $to_return[0] if @_ == 1;

	# We're here if they asked for multiple names but assigned the result
	# to a single scalar, which is probably not what they meant:
	carp("retrieve_pdls: requested many ndarrays... in scalar context?");
	return $to_return[0];
}

1;

__END__