Is there a way to stop actors system with discarding messages in the message boxes?

[svetlyak40wt]

I want to do a shutdown which will let actors finish their current tasks but will not let them to start new.

This is a small example of what I mean:

(defun test-stopping (&key (wait t))
  (let* ((system (make-actor-system))
         (actor (flet ((func (msg)
                         (log:warn "Sleeping 10 seconds" msg)
                         (sleep 10)
                         (log:warn "Done")))
                  (actor-of system
                            :receive #'func
                            ;; :other-args '(:queue-size 1)
                            ))))
    (log:warn "Posting messages to the actor")
    (sento.actor:tell actor "first-time")
    (sento.actor:tell actor "second-time")
    (log:warn "Stopping the system")
    (sento.actor-context:shutdown system
                                  :wait wait)
    (log:warn "System was stopped")
    (values)))

Currently, this code will wait 20 seconds while actor will process both messages sent to it before the call to shutdown function. But I want to shutdown as soon as possible, discarding both or at least the last message, sent to the actor.

P.S. – I've tried to call (sento.actor-cell:stop actor) before shutdown, but it has no desired effect :(

[mdbergmann]

There are currently two ways how to stop an actor:

1. sending :stop to the actor. This will just enqueue this as a normal message and the actor is stopped once the message was processed.
2. executing actor-cell:stop. This disallows more message to be enqueued, but also 'just' normally enqueues a message that it will first need to process to be fully stopped. This is what ac:shutdown does.

Unfortunately your scenario is not implemented. I didn't need it yet, but it would be good to be there and it's not hard to do (I think).
I will provide a fix over the weekend (maybe today).

[svetlyak40wt]

Ok. Here is where I need such feature.

I'm building a cron-like addon for Sento. There is a macro defschedule, which can be used to define a schedule – right now it is a delay and a fbound symbol:

(defun test-func ()
  (with-log-unhandled ()
    (let ((run-num (incf *run-num*))
          (num-seconds 120))
      (log:info "Sleeping" run-num num-seconds)
      (sleep num-seconds)
      (log:info "Sleeping Done" run-num))))

(defschedule *schedule*
  (30 test-func))

Also, there are two functions:

• start-scheduler - creates a new actor system, an actor for each scheduled function and uses sento.wheel-timer:schedule-recurring to schedule a helper which will tell an actor to do it's job.
• stop-scheduler - should stop everything and cleanup resources.

And here are two moments:

• an actor should have a limited to 1 queue size. Otherwise if job will require more time than a scheduling interval, the new messages will be stacked in the queue.
• shutdown does not work as I want and I have to wait while an actor will execute it's task a few times after shutdown was called.

Probably I'm overcomplicating and there is a simpler way to implement a cron-like system on actors?

[mdbergmann]

Well, usually the message processing of actors should not be blocked.
But it depends, in your case it is necessary.

Actor system comes with a scheduler, which can be used by actors. Here is an example:

ACTOR-CELL-TEST> (ac:actor-of *sys*
                              :init (lambda (self)
                                      (wt:schedule-recurring 
                                       (asys:scheduler (ac:system 
                                                        (act:context self)))
                                       1 1 (lambda () (act:! self "Foobar"))))
                              :receive (lambda (msg)
                                         (format t "msg: ~a~%" msg)))
msg: Foobar
msg: Foobar
msg: Foobar
msg: Foobar

This sets up a recurring timer in actors 'init' function. The timer function then just sends message to the actor which should do some action. This is to avoid race-condition on handling the timer function if actor state is involved.

The granularity of the scheduler is on a second basis.

Maybe this could be used instead of blocking message processing.

[svetlyak40wt]

Indeed! I'm using the similar code but call schedule-recurring not in the init.

What I'm talking about is that if you will add a (sleep 5) into the "receive" function, simulating some real job. Then queue of the actor will grow infinitely. And when you'll try to shutdown the system with :wait t argument, it will wait infinitely (until all items in the queue will be processed).

[mdbergmann]

Yeah, that's right.
I'm working on something to discard existing messages after 'stop'.

[svetlyak40wt]

Ok. I'll live without a stop function for now. Just wanted to let you know there is at least one more user of this feature :)

[mdbergmann]

Actually, if you try without wait it should already discard remaining messages, but doesn't wait until the current one is fully processed.

[mdbergmann]

So, please check the latest code. This should work now.

[svetlyak40wt]

Cool! Now it works as expected.

By the way, did you make a force push? I see only one last commit with the fix.

[mdbergmann]

Awesome.
Yes, squashed and force pushed.

[svetlyak40wt]

It is not the best practice for something pushed to the master branch. Probably the next time we should work on some branch until the feature or fix will be done.

Anyway, thank you for a quick fix!

[mdbergmann]

Yes, that would be better. And a proper PR also and just squash in the end.

[svetlyak40wt]

Oh, you've reminded me that I've promised to create a PR with proposal, how to pass queue size to the actor's mailbox. I'll try to do this during the week.

[mdbergmann]

Message discarding implemented as part of 94a9591