Can ithreads be used with Moose lazy attributes?

Question

For the below program I'm getting this error message:

Thread 2 terminated abnormally: Invalid value for shared scalar at reader Foo::bar (defined at ... line 9) line 10.

The program consists of a pipeline where the first thread creates some Moose-based objects and puts them in the queue, which are then picked up in the second thread. The problem seems to be that the attribute is lazy because the error disappears if I remove the lazy setting.

package Foo;
use Moose;

has 'bar' => (
    is      => 'ro',
    isa     => 'HashRef', # the error doesn't happen with simpler datatypes
    lazy    => 1, # this line causes the error
    default => sub { return { map {$_ => $_} (1 .. 10) } },
);

package main;

use threads;
use Thread::Queue;

my $threadq = Thread::Queue->new;

sub create {
    # $_ doesn't seem to be thread-safe
    # this resolved another problem I had with a custom Moose type constraint 
    # where the 'where' clause used $_
    local $_;

    $threadq->enqueue( Foo->new ) foreach 1 .. 5;
    $threadq->enqueue( undef );
    return;
}

sub process {
    local $_;
    while (my $f = $threadq->dequeue) {
        print keys %{$f->bar}, "\n";
    }
    return;
}

threads->create( \&create )->join;
threads->create( \&process )->join;

Can anyone shed some light on this problem? Is Moose itself thread-safe (I couldn't find much in the documentation about this)?

Answer 1

Thread::Queue replaces all the hashes and arrays in your object with shared hashes and arrays with the same content, and does so recursively.

At the same time, you are attempting to change the object. Yeah, that's not gonna end well. (And not because of any bug in Moose, Thread::Queue or threads.)

A solution is to pass around the object in a serialised form. You could handle serialisation and deserialisation yourself, or you could use Thread::Queue::Any to have it done implicitly.

use threads;
use Thread::Queue::Any;

my $threadq = Thread::Queue::Any->new;

sub create {
    $threadq->enqueue( Foo->new ) for 1 .. 5;
    $threadq->enqueue( );
}

sub process {
    while ( my ($f) = $threadq->dequeue ) {
       print sort keys %{$f->bar};
       print "\n";
    }
}

Note there are subtle but important differences in the usage of enqueue and dequeue. Most importantly, dequeue must be called in list context when using T::Q::A. This is because enqueue's argument list is passed as one message, and dequeue returns that argument list.