ENT-14195 2. cf-reactor data structure and polling

[victormlg]

Spec

In order to track all the events promises, we use two datastructures: a global list of "Watcher", which is a struct associated with an event type and the promise name (also called key) and a global hashmap mapping this key to a bundle which is parsed from the policy.

cf-reactor reads the policy periodically, and when it does, rebuilds the list of watchers and the hashmap using the single function WatcherRegister(key, event_type, state, bundle, interval). Each events promise is associated with an event type, which is defined in when bodies:

body when file_deleted(filename)
{
    file_deleted => "$(filename)";
}

Every event type must have defined:

• A check function (called check_callback in Watcher): This is a function defined specifically for the event that checks if the conditions holds. For example, in case of file deletion, we check if the file doesn't exist anymore compare to the last time we checked. If yes, then it returns true.
• A state: This is a struct whose interpretation depends on the event type (thus being declared as void *). We typically need some state that we compare between each event-check. In the case of file deletion, we need to know the name of the file we are watching, and whether the file existed last time we checked.
• A state destroying function (called destroy_state): This is simply a function to free the state associated with the event type.

Also, we need a function that will create the state. That's what FileWatcherPayloadNew() does.

So each event type we add in the future just need to have these four things defined, and we need to create the Watcher object with the right functions inside WatcherRegister() and also call the right "...PayloadNew()" function.

Polling

ReactorContextInitialize() sets up all the necessary data structures for polling, and then starts WatcherThreadMain, which polls for events as follows:

• It iterates through each watcher in the global list of watchers.
• If the elapsed time exceeds the watcher's interval, it runs check_callback to determine whether an event has been triggered.
• If an event was triggered, it pushes the watcher's key (the promise name) onto a thread-safe queue, then signals the file descriptor via WakeupChannelNotify, which select(2) will pick up on its next iteration. It then goes back to sleep.

In parallel, EventWatcherHandleEvents, called from within ReactorContextHandleEvents, reads from the file descriptor with WakeupChannelReadFd() once notified that an event has occurred, and pops the thread-safe queue until it's empty. Each key popped from the queue is looked up in the global hashmap to retrieve the corresponding bundle, which cf-reactor then runs (in another thread or subprocess)

[cf-bottom]

Thank you for submitting a pull request! Maybe @craigcomstock can review this?

[olehermanse]

@victormlg please rebase this one :)

[larsewi]

Did not review the whole thing yet. Here is what I have so far

[olehermanse]

On an agent run, cf-agent makes cf-reactor read the policy, and rebuilds the list of watchers and the hashmap using the single function

I commented on this in the README PR as well - why should cf-agent be involved? cf-reactor can do this on its own.

[victormlg]

On an agent run, cf-agent makes cf-reactor read the policy, and rebuilds the list of watchers and the hashmap using the single function

I commented on this in the README PR as well - why should cf-agent be involved? cf-reactor can do this on its own.

I forgot to update the description in the PR. The README doesn't mention this anymore.

[olehermanse]

@cf-bottom jenkins, please

[cf-bottom]

Alright, I triggered a build:

Jenkins: https://ci.cfengine.com/job/pr-pipeline/14671/

Packages: http://buildcache.cfengine.com/packages/testing-pr/jenkins-pr-pipeline-14671/

[craigcomstock]

I read through about half of the changes and researched claude's "low" effort code review results.

[craigcomstock]

thanks for the signals refactor.

All that I'd like to see is a few TODO comments.

[olehermanse]

@cf-bottom jenkins, please

[cf-bottom]

Sure, I triggered a build:

Jenkins: https://ci.cfengine.com/job/pr-pipeline/14678/

Packages: http://buildcache.cfengine.com/packages/testing-pr/jenkins-pr-pipeline-14678/

[larsewi]

Some smaller comments

[olehermanse]

@cf-bottom please test this in jenkins.

[cf-bottom]

Alright, I triggered a build:

Jenkins: https://ci.cfengine.com/job/pr-pipeline/14692/

Packages: http://buildcache.cfengine.com/packages/testing-pr/jenkins-pr-pipeline-14692/