CODEWITHSUNDEEP
multithreadingperl
bob_saginowski
bob_saginowski

Reputation: 1429

Perl resume a thread

I am trying to start a thread that I could suspend/ resume any time. Here is how I create the thread: 

use strict;
use warnings;
use threads;
use threads::shared;
use Thread::Suspend;
use Thread::Semaphore;

sub createThread { 
    my $semaphore = Thread::Semaphore->new();
    my $thr = threads->create(sub {
        local $SIG{KILL} = sub { 
            die "Thread killed\n";
        };
        local $SIG{STOP} = sub { 
            print "sig stop\n";
            $semaphore->down();
        };
        local $SIG{CONT} = sub {
            $semaphore->up();
            print "sig cont\n";
        };

        my $test = sub {
            while (1) {
                $semaphore->down();
                print STDERR "Working process\n";
                sleep(2);
                $semaphore->up();
            }
        };

        return $test->();
    });

    return $thr->tid();
}

After retrieving the thread id (with return $thr->tid();). Then I try to pause it and the message sig stop is being printed and later when I try to resume it sig cont is not printed. Here is the code that suspends/ resumes the thread: 

sub pause {
    my $class = shift;
    my $threadId = shift;
    my $thr = threads->object($threadId);

    if ($thr) {
        if ($thr->is_suspended() == 0) {
            $thr->kill('STOP');
            $thr->suspend();
            return "Process $threadId paused\n";
        } else {
            return "Process $threadId has to be resumed\n";
        }
    } else {
        return "Process $threadId not found\n";
    }
}

sub resume {
    my $class = shift;
    my $threadId = shift;
    my $thr = threads->object($threadId);

    if ($thr) {
        if ($thr->is_suspended() != 0) {
            $thr->resume();
            $thr->kill('CONT');
            return "Operation $threadId resumed\n";
        } else {
            return "Operation $threadId has not been paused\n";
        }
    } else {
        return "Process $threadId not found\n";
    }
}

After I resume a suspended thread, the message Operation X was resumed but sig cont isn't and the thread function is also not being resumed.

Upvotes: 8

Views: 257

Answers (2)

Borodin
Borodin

Reputation: 126722

It's unclear whether you need Thread::Semaphore for a separate purpose, but it's not required for the functioning of Thread::Suspend

I suspect the reason your suspend/resume isn't working is that you've overridden the signal handler that Thread::Suspend sets up for its own purposes

If I remove all the signal handlers and the Thread::Semaphore stuff, then your code works fine:

use strict;
use warnings 'all';
use threads;

use Thread::Suspend;

STDOUT->autoflush;


my $tid = create_thread();

for ( 1 .. 10 ) {
    sleep 5;
    print pause($tid);
    sleep 5;
    print resume($tid);
}


sub create_thread { 

    my $thr = threads->create( sub {
        while () {
            print "Working thread\n";
            sleep 1;
        }
    } );

    return $thr->tid;
}


sub pause {
    my ($tid) = @_;
    my $thr = threads->object($tid);

    return "Thread $tid not found\n" unless $thr;

    return "Thread $tid is already suspended\n" if $thr->is_suspended;

    $thr->suspend;
    return "Thread $tid paused\n";
}

sub resume {
    my ($tid) = @_;

    my $thr = threads->object($tid);

    return "Thread $tid not found\n" unless $thr;
    return "Thread $tid has not been paused\n" unless $thr->is_suspended;

    $thr->resume;
    return "Thread $tid resumed\n";
}

output

Working thread
Working thread
Working thread
Working thread
Working thread
Thread 1 paused
Thread 1 resumed
Working thread
Working thread
Working thread
Working thread
Working thread
Thread 1 paused
Thread 1 resumed
Working thread
Working thread
...


Update

There's no real need for your subroutines either. Here's a bare implementation

use strict;
use warnings 'all';
use threads;

use Thread::Suspend;

STDOUT->autoflush;

sub thread_sub {

    while () {
        printf "Working thread %d\n", threads->self->tid;
        sleep 1;
    }
}

my $thr = threads->create(\&thread_sub);

for ( 1 .. 10 ) {

    sleep 5;

    if ( my $suspended = $thr->suspend ) {
        printf "Thread %d suspended\n", $suspended->tid;
    }

    sleep 5;

    if ( my $resumed = $thr->resume ) {
        printf "Thread %d resumed\n", $resumed->tid;
    }
}

output

Working thread 1
Working thread 1
Working thread 1
Working thread 1
Working thread 1
Working thread 1
Thread 1 suspended
Thread 1 resumed
Working thread 1
Working thread 1
Working thread 1
Working thread 1
Working thread 1
Working thread 1
Thread 1 suspended
Thread 1 resumed
Working thread 1
Working thread 1
Working thread 1

Upvotes: 5

ikegami
ikegami

Reputation: 385789

No actual signals are being used here. Signals can only be sent to processes, not threads. From $thread->kill's documentation:

CAVEAT: The thread signalling capability provided by this module does not actually send signals via the OS. It emulates signals at the Perl-level such that signal handlers are called in the appropriate thread. For example, sending $thr->kill('STOP') does not actually suspend a thread (or the whole process), but does cause a $SIG{'STOP'} handler to be called in that thread (as illustrated above).

Since no actual signals are being used here, you're not mixing signals and threads. Good.

But it's an overly complicated design. Simply call $sem->down_force() in pause and $sem->up() in resume. There's no need for this to happen in the thread.

use strict;
use warnings;
use threads;
use Thread::Semaphore qw( );

{
    package Worker;

    sub new { 
        my $class = shift;
        return bless({ @_ }, $class);
    }

    sub thr    { return $_[0]{thr}        }
    sub tid    { return $_[0]{thr}->tid() }
    sub join   { $_[0]{thr}->join()       }
    sub pause  { $_[0]{sem}->down_force() }
    sub resume { $_[0]{sem}->up()         }
}

sub createThread {
    my $sem = Thread::Semaphore->new();
    my $thr = async {
        while (1) {
            $sem->down();
            ...
            $sem->up();
        }
    };

    return Worker->new( thr => $thr, sem => $sem );
}

sub pause  { my ($worker) = @_;  $worker->pause();  }
sub resume { my ($worker) = @_;  $worker->resume(); }

Of course, that assumes you only want to suspend the thread between work units. If you want to suspend the thread immediately, you don't need semaphores or "signals" at all[1].

use strict;
use warnings;
use threads;
use Thread::Suspend;  # Must call import!

{
    package Worker;

    sub new { 
        my $class = shift;
        return bless({ @_ }, $class);
    }

    sub thr    { return $_[0]{thr}                           }
    sub tid    { return $_[0]{thr}->tid()                    }
    sub join   { $_[0]{thr}->join()                          }
    sub pause  { $_[0]{suspended}++ || $_[0]{thr}->suspend() }
    sub resume { --$_[0]{suspended} || $_[0]{thr}->resume()  }
}

sub createThread {
    my $thr = async {
        ...
    };

    return Worker->new( thr => $thr );
}

sub pause  { my ($worker) = @_;  $worker->pause();  }
sub resume { my ($worker) = @_;  $worker->resume(); }

Bonus: $worker->pause; $worker->pause; $worker->resume; $worker->resume; works fine for both of these methods (unlike the version in the question).

If you want to keep working using a tid instead of an object, just store the object in a hash keyed by tid.

  1. At least, not in addition to Thread::Suspend, which uses a "signal" and what amounts to semaphores internally.

Upvotes: 4

Related Questions