Is there a good way to stop work-intensive threads via thread communication (instead of using a shared variable)? The example below is very basic and naive and only meant to exemplify the basic problem.

I want to stop (and abort) the worker as soon as new input arrives. However, while executing the function that contains the foreach-loop the worker thread doesn't listen, because it's busy, of course. I've tried a few solutions with send and receive in this block, but somehow none of them work perfectly.

//=======
import std.stdio : readln, writefln, writeln;
import std.string : strip;
import std.concurrency;
import core.thread;

Tid thread1;

struct Exit {}

void main()
{
  string input;
  bool exists;
  while ((input = readln.strip) != null)
  {
    if (exists)
    {
      thread1.send(Exit());
    }
    thread1 = spawn(&worker);
    exists = true;
    thread1.send(input.idup);
  }
}

void worker()
{
  bool run = true;
  while (run)
  {
    receive(
      (string input)
      {
        foreach (ref i; 0..10)
        {
          writefln("%d.\tDoing something with input %s", i+1, input);
          Thread.sleep(500.msecs);
        }
        run = false;
      },
      (Exit exit)
      {
        run = false;
      }
    );
  }
  writeln("End of thread worker");
}
//=======

receiveTimeout

On 08/04/2015 09:19 AM, Dicebot wrote:
> receiveTimeout

I think the problem here is that the worker is busy, not even able to call that.

This sounds like sending a signal to the specific thread (with pthread_kill()) but I don't know the details of it nor whether Phobos supports it.

Ali

On Tuesday, 4 August 2015 at 15:19:51 UTC, Chris wrote:

> I want to stop (and abort) the worker as soon as new input arrives. However, while executing the function that contains the foreach-loop the worker thread doesn't listen, because it's busy, of course.

I think this is a matter of architecture. If you want to use message-passing and you want the worker to react quickly to new events, this means it needs to check for new messages (via receiveTimeout) often enough, there's no way around it.

On Tuesday, 4 August 2015 at 18:15:08 UTC, Ali Çehreli wrote:
> On 08/04/2015 09:19 AM, Dicebot wrote:
>> receiveTimeout
>
> I think the problem here is that the worker is busy, not even able to call that.
>
> This sounds like sending a signal to the specific thread (with pthread_kill()) but I don't know the details of it nor whether Phobos supports it.
>
> Ali

The problem is that it works up to a certain extent with receiveTimeout. However, if the input arrives in very short intervals, all the solutions I've come up with so far (including data sharing) fail sooner or later. New threads are spawned faster than old ones can be given the abort signal. There are ways to wait, till a given thread dies, say with a shared variable isAlive `while (isAlive) {}`, but even here I've come across problems when the input comes very fast.

I don't know how to solve this problem, because message passing follows a linear protocol (as far as I understand it) and shared variables give rise to data races. Something like pthread_kill() would indeed be useful, to terminate a thread at random. I wonder if fibers would be an option.

D-threads seem to be based on the assumption that there is no need to abort threads at random, any time. Or am I mistaken?

On Wednesday, 5 August 2015 at 11:23:28 UTC, Chris wrote:
> The problem is that it works up to a certain extent with receiveTimeout. However, if the input arrives in very short intervals, all the solutions I've come up with so far (including data sharing) fail sooner or later. New threads are spawned faster than old ones can be given the abort signal. There are ways to wait, till a given thread dies, say with a shared variable isAlive `while (isAlive) {}`, but even here I've come across problems when the input comes very fast.

You could use a thread pool, thereby limiting the number of threads that can run at any one time. But I guess you want the processing of new data to start as soon as possible, in which case that wouldn't help you.

>
> I don't know how to solve this problem, because message passing follows a linear protocol (as far as I understand it) and shared variables give rise to data races. Something like pthread_kill() would indeed be useful, to terminate a thread at random. I wonder if fibers would be an option.
>
> D-threads seem to be based on the assumption that there is no need to abort threads at random, any time. Or am I mistaken?

It was a conscious decision not to provide a kill method for threads, because it is impossible to guarantee that your program is still consistent afterwards. Maybe we can lift this restriction if we know that the thread's main function is pure and takes no references to mutable data, because then it can by definition never mess up the program's state. OTOH, the GC might be running at the time the thread is killed, which could again lead to inconsistencies...

On Wednesday, 5 August 2015 at 14:31:20 UTC, Marc Schütz wrote:
> Maybe we can lift this restriction if we know that the thread's main function is pure and takes no references to mutable data, because then it can by definition never mess up the program's state.

That'd be a pretty useless thread; how would it communicate results back to the main thread (or wherever it should go)?

On Wednesday, 5 August 2015 at 14:34:42 UTC, Alex Parrill wrote:
> On Wednesday, 5 August 2015 at 14:31:20 UTC, Marc Schütz wrote:
>> Maybe we can lift this restriction if we know that the thread's main function is pure and takes no references to mutable data, because then it can by definition never mess up the program's state.
>
> That'd be a pretty useless thread; how would it communicate results back to the main thread (or wherever it should go)?

It could return something. `std.concurrency.Tid` would have to be extended with a `join()` method that returns its result. Or we could somehow allow sending and receiving data.

On Wednesday, 5 August 2015 at 14:31:20 UTC, Marc Schütz wrote:
> It was a conscious decision not to provide a kill method for threads, because it is impossible to guarantee that your program is still consistent afterwards.

What about the situation where we want to kill worker threads off when closing a program? For example, I have a program with a thread that does some heavy computation in the background. When the application is closed, I want it to abort that computation, however I can't just slap a receiveTimeout in the worker thread because it is doing its work in a parallel foreach loop.

On Tuesday, 4 August 2015 at 15:19:51 UTC, Chris wrote:
>         foreach (ref i; 0..10)
>         {
>           writefln("%d.\tDoing something with input %s", i+1, input);
>           Thread.sleep(500.msecs);
>         }

AFAIK, boost does it by integrating support for interruption into various functions, so IO, waits and locks reply to interrupt requests appropriately. You can do something similar.