Trying to implement a setInterval() that I can cancel using:

Tid tid = setInterval(2000, (){ writeln("hello");})

And then I can do:

stopInterval(tid);


With something like this:

stopInterval(Tid tid) {
    send(tid, "cancel");
}

import std.stdio : writeln;
import std.concurrency : receive, spawn, thisTid, Tid;

auto setInterval(long milliseconds, void delegate() callback)
{
    auto worker(long mls, void delegate() cb)
    {
        import core.thread.osthread : Thread;
        import std.datetime : seconds, msecs;

        writeln("Starting ", thisTid, "...");

        bool done = false;

        receive((string text) {
           writeln("Received string: ", text);
           done = true;
        });

        do
        {
            // or receive() comes here?

            Thread.sleep(mls.msecs);
            cb();
        }
        while (!done);
    }

    // I guess issue is with the callback
    Tid id = spawn(&worker, milliseconds, &callback);

    return id;
}



Getting error:

Error: template std.concurrency.spawn cannot deduce function from argument types !()(void delegate(Tid id) @system, Tid), candidates are:
/usr/include/dmd/phobos/std/concurrency.d(460,5):        spawn(F, T...)(F fn, T args)
  with F = void delegate(Tid) @system,
       T = (Tid)
  must satisfy the following constraint:
       isSpawnable!(F, T)


Am I even using the right tool here?

On Sunday, 28 June 2020 at 13:29:08 UTC, aberba wrote:

> Getting error:
>
> Error: template std.concurrency.spawn cannot deduce function from argument types !()(void delegate(Tid id) @system, Tid), candidates are:
> /usr/include/dmd/phobos/std/concurrency.d(460,5):        spawn(F, T...)(F fn, T args)
>   with F = void delegate(Tid) @system,
>        T = (Tid)
>   must satisfy the following constraint:
>        isSpawnable!(F, T)


The error you're getting is because you're passing a pointer to a delegate instead of a delegate.

>     Tid id = spawn(&worker, milliseconds, &callback);
                                            ^ here

But fixing that still won't compile, because when you want to pass a delegate to `spawn`, it needs to be a shared delegate.

If I understood your intent correctly, here's how you can do it:

import std.stdio : writeln;
import std.concurrency;
import core.thread.osthread : Thread;
import std.datetime.stopwatch;

auto setInterval(long milliseconds, void function() callback)
{
    static void worker(Duration d, void function() cb)
    {
        writeln("Starting ", thisTid, "...");

        bool done = false;

        StopWatch sw;
        sw.start;
        while (true)
        {
            // wait for messages for a timespan of at least `d`
            receiveTimeout(
                d,
                (string text) {
                   writeln("Received string: ", text);
                   if (text == "cancel")
                       done = true;
               });

            if (done)
                break;

            // a non-cancelling message might've been received before timeout,
            // so test if it's really time for the callback
            if (sw.peek >= d)
            {
                cb();
                sw.reset;
            }
        }
    }

    Tid id = spawn(&worker, milliseconds.msecs, callback);

    return id;
}

void stopInterval(Tid tid) {
    send(tid, "cancel");
}

void main()
{
    auto tid = setInterval(1000, { writeln("tick"); });
    Thread.sleep(2.seconds);
    send(tid, "not cancel");
    Thread.sleep(5.seconds);
    stopInterval(tid);
}

On Sunday, 28 June 2020 at 14:23:01 UTC, Stanislav Blinov wrote:
> On Sunday, 28 June 2020 at 13:29:08 UTC, aberba wrote:
 Thanks.

I believe this:

StopWatch sw;
sw.start;


works becuse D structs are initialized by default, right?
I've never actually done it this way. Little details.

On Sunday, 28 June 2020 at 23:02:26 UTC, aberba wrote:
> On Sunday, 28 June 2020 at 14:23:01 UTC, Stanislav Blinov wrote:
>> On Sunday, 28 June 2020 at 13:29:08 UTC, aberba wrote:
>  Thanks.
>
> I believe this:
>
> StopWatch sw;
> sw.start;
>
>
> works becuse D structs are initialized by default, right?
> I've never actually done it this way. Little details.

So I checked receiveTimeout() when I was looking for what I could use. I wish there was an example in the docs.

https://dlang.org/library/std/concurrency/receive_timeout.html

On Sunday, 28 June 2020 at 23:02:26 UTC, aberba wrote:

> I believe this:
>
> StopWatch sw;
> sw.start;
>
> works becuse D structs are initialized by default, right?
> I've never actually done it this way. Little details.

Yup. You can also do a

auto sw = StopWatch(AutoStart.yes);

and not have to call `start` explicitly.

On Sunday, 28 June 2020 at 23:39:07 UTC, Stanislav Blinov wrote:
> On Sunday, 28 June 2020 at 23:02:26 UTC, aberba wrote:
>
>> I believe this:
>>
>> StopWatch sw;
>> sw.start;
>>
>> works becuse D structs are initialized by default, right?
>> I've never actually done it this way. Little details.
>
> Yup. You can also do a
>
> auto sw = StopWatch(AutoStart.yes);
>
> and not have to call `start` explicitly.

Interesting. I should look into Phobos more.

On 6/28/20 4:08 PM, aberba wrote:

> So I checked receiveTimeout() when I was looking for what I could use. I wish there was an example in the docs.
> 
> https://dlang.org/library/std/concurrency/receive_timeout.html

I have an example of it:

  http://ddili.org/ders/d.en/concurrency.html#ix_concurrency.receiveTimeout

Ali

I'm doing this in an X11 application in order to send a timer event every 100 milliseconds to the main event queue.

class Application
{
  shared private bool s_tick;

  void clock_task (shared X11.Display* disp, X11.Atom atom, X11.Window win)
  {
    for (;;)
    {
      try
      {
        receiveTimeout (100.msecs);

        if (disp && atomicLoad(s_tick))
        {
          // disable ticking until it is allowed again at the end of the event loop
          atomicStore(s_tick, false);

          X11.XClientMessageEvent event;
          event.type         = X11.ClientMessage;
          event.window       = win;
          event.message_type = atom;
          event.format       = 32;
          event.data.l       = [0, 0, 0, 0, 0];

          X11.XSendEvent (cast (X11.Display*) disp, win, 0, 0,  cast(X11.XEvent*)&event);
          X11.XFlush (cast (X11.Display*) disp);
        }
      }
      catch (Throwable)
      {
        return;
      }
    }
  }

  this ()
  {
...
    spawn (&clock_task, cast(shared)x11Display, x11SigClockAtom, _x11_proxyWindow);
  }

  run ()
  {
    while (true)
    {
...
      // event processing starts here: read in X11 event and convert it to a wit Event
      X11.XEvent x11_event;
      X11.XNextEvent (_x11.display, &x11_event);
...
      atomicStore(s_tick, true);
    }
  }
}

On Sunday, 28 June 2020 at 14:23:01 UTC, Stanislav Blinov wrote:
> On Sunday, 28 June 2020 at 13:29:08 UTC, aberba wrote:
>
>>        [...]
>
>
> The error you're getting is because you're passing a pointer to a delegate instead of a delegate.
>
> [...]

So with this, without the Thread.sleep() to block main from exiting, the spawned thread  will terminate immediately. How do I keep it from happening? Keep it running continuously?

From the docs, it says OwnerTerminated exception gets thrown when the sending thread (e.i. main) is terminated.

> Thrown on calls to receive if the thread that spawned the receiving thread has terminated and no more messages exist.

On 6/29/20 4:34 PM, aberba wrote:

> So with this, without the Thread.sleep() to block main from exiting, the
> spawned thread  will terminate immediately.

You can call core.thread.thread_joinAll at the end of main.

Another way would be to wait for a worker's exit by looking for LinkTerminated but you need to start the thread with spawnLinked:

  http://ddili.org/ders/d.en/concurrency.html#ix_concurrency.LinkTerminated

Ali