Added new command message: Cancel #122

ahitacat · 2023-05-11T15:14:38Z

Proposal

Add a new control message of type command that can be used to cancel a work that is already in progress in a worker.
yggd will receive this command and forward it to the appropriate worker. If this worker implements the cancellation feature it will stop or cancel its work. But it is up to the worker to implement this functionality.

This PR also provides an example of how the cancellation message can be implemented in the echo worker. It uses a map of synchronized channels to manage the current work, and to send the cancellation message to the goroutines that are handle it.

This commit adds:

A new command message that can be use to cancel work in progress that is done in the workers.Is send as a command to yggdrasil client. Yggdrasil will send this message to the workers Cancel method.
A new function to the worker, CancelRX. This function will handle the cancel signal in workers side. If it is null the worker doesn't support cancellation of messages.
echo worker implements this cancel function to handle cancellation when it runs in slow mode.

How to test it

Follow the Quickstart steps to set the environment. With the following changes:

Terminal 2
Run the worker with the loop and sleep parameters, this will allow some time to launch the cancel message.

go run ./worker/echo --log-level trace --loop 10 --sleep 10s

Terminal 4
Run the hello world, that will send a new job to the worker.

echo "hello world"| \
go run ./cmd/yggctl generate data-message --directive echo - | \
pub -broker tcp://localhost:1883 -topic yggdrasil/$(hostname)/data/in

In the worker it will appear the id of that message:

2023/10/06 12:21:58 addr = echo
2023/10/06 12:21:58 id = 67c40762-2343-48ad-9bf1-23a9b98bdacb
2023/10/06 12:21:58 responseTo = 
2023/10/06 12:21:58 metadata = map[string]string{}
2023/10/06 12:21:58 data = [104 101 108 108 111 32 119 111 114 108 100 10]
2023/10/06 12:21:58 emitting event BEGIN

Use the id 67c40762-2343-48ad-9bf1-23a9b98bdacb to send the cancellation message:

echo '{"command":"cancel", "arguments":{"directive":"echo","messageID":"67c40762-2343-48ad-9bf1-23a9b98bdacb"}}' |\
go run ./cmd/yggctl generate control-message --type command - |\
pub -broker tcp://test.mosquitto.org:1883 -topic yggdrasil/$(hostname)/control/in

subpop

This looks great. I'll give this a try to see how it performs too.

ipc/com.redhat.Yggdrasil1.Worker1.xml

internal/work/dispatcher.go

worker/echo/main.go

worker/worker.go

jirihnidek

Hi, Thanks for the PR. 👍 I have few, requests, comments and suggestions.

internal/work/dispatcher.go

worker/echo/main.go

jirihnidek · 2023-05-25T14:38:25Z

worker/echo/main.go

+		default:
+			delete(cancelChan, id)
+		}
+		chanMutex.Unlock()


Again this looks like as something that will be duplicate in all workers. Shouldn't we create something that will help all developers?

As previously mentioned, channels are specific to Golang, and this implementation can be different depending on each worker. How to handle the cancellation should be up to each worker.

worker/worker.go

ahitacat · 2023-05-30T11:35:13Z

Summary of changes made from reviews:

fix typos in code and comments
fix logs to follow the conventions on this project and set better debug levels
echo worker use RWMutexMap from internal/sync to create a table of channels to handle current pending work.
- Those pending works are related to sent echo message through yggdrasil, echo worker running in sleep mode will have the opportunity to cancel that work that has not being finished.
add doc-string comments

jirihnidek

Thanks for updates. I have only two small requests/comments.

jirihnidek · 2023-06-12T09:02:02Z

internal/work/dispatcher.go

@@ -328,3 +328,26 @@ func (d *Dispatcher) senderName(sender dbus.Sender) (string, error) {

 	return "", fmt.Errorf("cannot get name for sender: %v", sender)
 }
+
+// CancelMessage implements the dispatch of a cancel method to the worker.
+func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {


Following comment seems little bit off.

This method signature feels odd to me. It strikes me as odd that we expect an
entire yggdrasil.Command structure as the first parameter, but then only the
message ID from the encapsulating message as the second parameter. Perhaps the
only parameters that get passed to the dispatcher via this method are the
unpacked values from the command?

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) CancelMessage(directive string, message_id string, cancel_id string) error {

This forces all the unmarshalling and parameter validation into the client and
makes it very clear which message ID is to be cancelled.

Another option is to follow the pattern of the Dispatch method above, and expect
the entire yggdrasil.Control struct in the method:

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) CancelMessage(control yggdrasil.Control) error {

This has the downside of unmarshalling the Command struct out of the control
message a second time (the first having occurred in the client when processing
the command name). I'm less inclined to take this approach, because this
introduces the potential to scope-creep the method into a larger "control
message handler" method. It's important to preserve it's responsibility as much
as possible.

So that being said, I think we should push all the message unpacking up to the
client and change the signature of this method as follows:

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) Cancel(directive string, message_id string, cancel_id string) error {

I wanted to avoid the CancelMessage to unmarshall again the control message. I'm OK to move this logic into the client. It seems a good idea.

jirihnidek · 2023-06-12T09:03:44Z

internal/work/dispatcher.go

+// CancelMessage implements the dispatch of a cancel method to the worker.
+func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {
+	// Send the message through the cancel interface
+	directive := command.Arguments["directive"]


Shouldn't we check the existence of directive as we do it for messageID?

You are right 😄

ahitacat · 2023-06-19T15:08:31Z

Sumary of changes

Addressed all the change from @jirihnidek comments:

Improve the description of CancelMessage function.
Check that directive is not empty

subpop · 2023-07-21T15:24:21Z

worker/echo/main.go has changed; could you resolve those conflicts and we'll see if we can revisit this PR again.

subpop

Overall, this is a great approach. We're blazing a new trail here with the introduction of control message data being handled and passed to workers, so I left a few comments about how to maybe improve the API boundary between client and dispatcher. I think we could probably offer a more functional worker example too; considering how the implementation isn't obvious (using channels and sync maps immediately increases the complexity of the example significantly). Having a clear example to use as guidance and documentation is going to be important.

internal/work/dispatcher.go

cmd/yggd/client.go

subpop · 2023-08-08T16:17:57Z

internal/work/dispatcher.go

@@ -328,3 +328,26 @@ func (d *Dispatcher) senderName(sender dbus.Sender) (string, error) {

 	return "", fmt.Errorf("cannot get name for sender: %v", sender)
 }
+
+// CancelMessage implements the dispatch of a cancel method to the worker.
+func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {


This method signature feels odd to me. It strikes me as odd that we expect an
entire yggdrasil.Command structure as the first parameter, but then only the
message ID from the encapsulating message as the second parameter. Perhaps the
only parameters that get passed to the dispatcher via this method are the
unpacked values from the command?

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) CancelMessage(directive string, message_id string, cancel_id string) error {

This forces all the unmarshalling and parameter validation into the client and
makes it very clear which message ID is to be cancelled.

Another option is to follow the pattern of the Dispatch method above, and expect
the entire yggdrasil.Control struct in the method:

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) CancelMessage(control yggdrasil.Control) error {

This has the downside of unmarshalling the Command struct out of the control
message a second time (the first having occurred in the client when processing
the command name). I'm less inclined to take this approach, because this
introduces the potential to scope-creep the method into a larger "control
message handler" method. It's important to preserve it's responsibility as much
as possible.

So that being said, I think we should push all the message unpacking up to the
client and change the signature of this method as follows:

Suggested change

func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {

func (d *Dispatcher) Cancel(directive string, message_id string, cancel_id string) error {

worker/worker.go

subpop · 2023-08-08T16:29:15Z

worker/echo/main.go

@@ -12,16 +12,23 @@ import (

 	"git.sr.ht/~spc/go-log"

+	"github.com/redhatinsights/yggdrasil/internal/sync"


This is okay because the echo worker is part of the yggdrasil repository, but I
don't know if Go would permit this import from an external project. I think the
'internal' package has special value to the Go compiler. Might be worth testing.

We could easily solve this by moving the sync package to the top of the
yggdrasil project, but it seems weird for the yggdrasil package to export a
sync package.

I think this sync package might be useful for worker development, in case other workers want to use this utilty.

What do you think to move this pieces of code into the pkg directory?

yggdrasil . ├── build-aux ├── builddir ├── cmd ├── data ├── dbus ├── dist ├── doc ├── internal ├── ipc ├── lib ├── pkg └── worker

That way, we have such an standard way to do the same things. I'm thinking as well in the http implementation of the client we have in yggdrasil, it can be also be used in rhc.

Yes, we could move sync out of internal and into the top level of the project. I would prefer to avoid creating a pkg package. It is unnecessary and creates an extraneous entry in import paths:

With pkg

import "github.com/redhatinsights/yggdrasil/pkg/sync"

Without pkg

import "github.com/redhatinsights/yggdrasil/sync"

worker/echo/main.go

subpop · 2023-08-08T16:40:41Z

worker/echo/main.go

+func cancelEcho(w *worker.Worker, addr string, id string, cancelID string) error {
+	log.Infof("cancelling message with id %v", cancelID)
+	if cancelChan, exists := sycMapCancelChan.Get(cancelID); exists {
+		close(cancelChan)


If we've closed the channel because a cancellation was received, the channel
should get deleted from the map too.

The channel is deleted from the map when the close message is used, that is in the echo function. The new echo worker functionality may change this.

ahitacat · 2023-10-06T15:07:42Z

Changelog

Minor fixed spoted by @subpop
Reformulate CancelMessage signature. I have move the handle of the parameters into the client, so this function needs to recieve CancelMessage(directive string, message_id string, cancel_id string)
Open a new discussion about move some internal modules to other path and allow them to be imported into new projects.
Added a DBus Error and open a new discussion. [TODO: Open a discussion on how to treat better DBUs API errors]
Added a new example, using the loop functionality of the worker (added in this PR).
Provide some steps on the description of this PR on how it can be tested.
Rebased and resolve conflicts.

jirihnidek

Thanks for updates 👍 It is better and I think that we are reaching end of PR review process. I have some small suggestions and some requests.

internal/work/dispatcher.go

worker/echo/main.go

worker/worker.go

worker/echo/main.go

worker/worker.go

subpop

I saw a few typos as well. I'm going to take a second pass over this to experiment with the functionality next.

cmd/yggd/client.go

worker/worker.go

ahitacat · 2023-10-10T09:40:26Z

@subpop @jirihnidek I have fixed the typos and include your suggestions 😄

subpop

I found one last typo. But the functionality works great!

worker/echo/main.go

This commit adds: - A new command message that can be use to cancel work in progress that is done in the workers.Is send as a command to yggdrasil client. Yggdrasil will send this message to the workers Cancel method. - A new function to the worker, CancelRX. This function will handle the cancel signal in workers side. If it is null the worker doesn't support cancellation of messages. - echo worker implements this cancel function to handle cancellation when it runs in slow mode. Signed-off-by: Alba Hita Catala <ahitacat@redhat.com>

jirihnidek

LGTM! Thanks for all updates.

ahitacat requested a review from subpop May 11, 2023 15:14

ahitacat force-pushed the feat-cancel-control-msg branch 2 times, most recently from 767d994 to 7f843d9 Compare May 15, 2023 10:02

ahitacat mentioned this pull request May 15, 2023

Added cancel message option to make data message cancellables #114

Closed

subpop requested changes May 19, 2023

View reviewed changes

ahitacat force-pushed the feat-cancel-control-msg branch from 7f843d9 to 2bc0bd1 Compare May 23, 2023 14:21

ahitacat requested a review from subpop May 24, 2023 09:19

jirihnidek requested changes May 25, 2023

View reviewed changes

ahitacat force-pushed the feat-cancel-control-msg branch from 2bc0bd1 to 56cb0ab Compare May 30, 2023 11:35

ahitacat force-pushed the feat-cancel-control-msg branch 2 times, most recently from d8cfbeb to 357f77b Compare June 1, 2023 09:53

ahitacat force-pushed the feat-cancel-control-msg branch 2 times, most recently from 0f83e01 to 60350fc Compare June 7, 2023 16:57

jirihnidek requested changes Jun 12, 2023

View reviewed changes

ahitacat force-pushed the feat-cancel-control-msg branch from 60350fc to 429142a Compare June 19, 2023 15:06

ahitacat requested a review from jirihnidek June 19, 2023 15:08

jirihnidek approved these changes Jun 19, 2023

View reviewed changes

ahitacat force-pushed the feat-cancel-control-msg branch from 429142a to ff9f125 Compare July 24, 2023 16:01

subpop force-pushed the feat-cancel-control-msg branch from ff9f125 to 5c6f2c2 Compare August 8, 2023 15:33

subpop requested changes Aug 8, 2023

View reviewed changes

ahitacat force-pushed the feat-cancel-control-msg branch from 5c6f2c2 to 8dda469 Compare September 8, 2023 09:31

ahitacat force-pushed the feat-cancel-control-msg branch 3 times, most recently from 545c8c5 to 0197614 Compare October 6, 2023 15:06

ahitacat requested a review from subpop October 6, 2023 15:38

jirihnidek requested changes Oct 9, 2023

View reviewed changes

subpop requested changes Oct 9, 2023

View reviewed changes

cmd/yggd/client.go Outdated Show resolved Hide resolved

worker/worker.go Outdated Show resolved Hide resolved

ahitacat force-pushed the feat-cancel-control-msg branch 2 times, most recently from 1d46c38 to ec0d919 Compare October 10, 2023 09:40

ahitacat requested review from subpop and jirihnidek October 10, 2023 09:40

subpop requested changes Oct 10, 2023

View reviewed changes

worker/echo/main.go Outdated Show resolved Hide resolved

ahitacat force-pushed the feat-cancel-control-msg branch from ec0d919 to 17d8850 Compare October 10, 2023 15:25

ahitacat requested a review from subpop October 10, 2023 15:25

jirihnidek approved these changes Oct 10, 2023

View reviewed changes

subpop approved these changes Oct 11, 2023

View reviewed changes

subpop merged commit 5130298 into RedHatInsights:main Oct 11, 2023
6 checks passed

ahitacat mentioned this pull request Oct 11, 2023

Implement cancellation message RedHatInsights/libygg#9

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added new command message: Cancel #122

Added new command message: Cancel #122

ahitacat commented May 11, 2023 •

edited

Loading

subpop left a comment

jirihnidek left a comment

jirihnidek May 25, 2023

ahitacat May 25, 2023

ahitacat commented May 30, 2023

jirihnidek left a comment

jirihnidek Jun 12, 2023

subpop Aug 8, 2023

ahitacat Aug 17, 2023

jirihnidek Jun 12, 2023

ahitacat Jun 19, 2023

ahitacat commented Jun 19, 2023 •

edited

Loading

subpop commented Jul 21, 2023

subpop left a comment

subpop Aug 8, 2023

subpop Aug 8, 2023

ahitacat Aug 17, 2023

ahitacat Oct 5, 2023

subpop Oct 5, 2023

subpop Aug 8, 2023

ahitacat Oct 6, 2023

ahitacat commented Oct 6, 2023

jirihnidek left a comment

subpop left a comment

ahitacat commented Oct 10, 2023

subpop left a comment

jirihnidek left a comment

	func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {
	func (d *Dispatcher) CancelMessage(directive string, message_id string, cancel_id string) error {

	func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {
	func (d *Dispatcher) CancelMessage(control yggdrasil.Control) error {

	func (d *Dispatcher) CancelMessage(command yggdrasil.Command, id string) error {
	func (d *Dispatcher) Cancel(directive string, message_id string, cancel_id string) error {

		@@ -12,16 +12,23 @@ import (

		"git.sr.ht/~spc/go-log"

		"github.com/redhatinsights/yggdrasil/internal/sync"

Added new command message: Cancel #122

Added new command message: Cancel #122

Conversation

ahitacat commented May 11, 2023 • edited Loading

Proposal

This commit adds:

How to test it

subpop left a comment

Choose a reason for hiding this comment

jirihnidek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahitacat commented May 30, 2023

Summary of changes made from reviews:

jirihnidek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahitacat commented Jun 19, 2023 • edited Loading

Sumary of changes

subpop commented Jul 21, 2023

subpop left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

With pkg

Without pkg

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ahitacat commented Oct 6, 2023

Changelog

jirihnidek left a comment

Choose a reason for hiding this comment

subpop left a comment

Choose a reason for hiding this comment

ahitacat commented Oct 10, 2023

subpop left a comment

Choose a reason for hiding this comment

jirihnidek left a comment

Choose a reason for hiding this comment

ahitacat commented May 11, 2023 •

edited

Loading

ahitacat commented Jun 19, 2023 •

edited

Loading

With `pkg`

Without `pkg`