External trigger "plugin" functions. #2423

hjoliver · 2017-09-09T12:32:20Z

Trigger tasks off of anything by having the suite daemon call arbitrary "xtrigger functions" that return True/False and some result - e.g. path to a new external dataset - to broadcast to the ~~cycle point~~ dependent task.

This should replace in-task-proxy clock-triggers, client/server external triggers, and polling tasks.

xtrigger functions are called in the process pool with asynchronous callback (except for clock xtriggers which are guaranteed to be quick)
a function won't be called again before the previous call comes back
once an xtrigger is satisfied, the result is remembered and the function not called again
~~if a task has a clock xtrigger, don't bother calling other xtriggers until the clock time is up~~ [not compatible with use of clock triggers in conditional expressions]
xtrigger functions are tracked by unique "call signature" (function name and args). If multiple tasks depend on the same xtrigger, it gets shared among them
function call frequency is configurable, default PT10S
variables like suite name, run directory, task cycle point, etc. can be substituted into function arg values via string templates (example below)
to extend functionality (trigger off of other external events) simply put a new function in /lib/cylc/xtriggers/ (or anywhere in sys.path) - and use it in the suite definition immediately

hjoliver · 2017-09-09T12:32:56Z

I've implemented some xtrigger functions under lib/cylc/xtriggers/, with example suites under dev/suites/xtrigger/:

~~dummy(...) - a test trigger that sleeps while, and "fails" a few times before succeeding~~
wall_clock(...) - trigger off of the wall clock (can replace the in-task-proxy clock triggers)
suite_state(...) - trigger off of a task in another suite (can replace suite-state polling tasks)
~~file_exists(...) - trigger if a specific file exists~~
~~match_file(...) - trigger if a file is found to match some pattern (and report the matched file)~~
xrandom(..) - random trigger with configurable success rate.

[UPDATE: file triggering functions removed for the moment - a general solution to that problem is tricky]

hjoliver · 2017-09-11T12:59:24Z

Satisfied xtriggers are now stored and reloaded from the DB on restart, so the corresponding functions won't be called again (for the same function signature). This is needed because some xtriggers will only report in satisfied once. E.g. for triggering off of a message in a broker like RabbitMQ (but not Kafka) in which a message disappears once consumed; and for triggering off of detection of a new external dataset that does not have cycle point in the filename (e.g. satellite datasets) so the xtrigger function probably needs to rename the file on detection to ensure that it doesn't get re-detected in the next suite cycle.

oliver-sanders · 2017-09-19T14:24:30Z

Looks good! Having external triggers appear in suite visualisations should make things much clearer.

This opens up using clock triggers with conditional statements which could be interesting e.g:

early | ( @clock_0 & data ) => process

In combination with #2334:

( ensemble:succeed-90% & @clock_0 ) | ( ensemble:succeed-80% & @clock_1 ) => process

Would it be possible to move a little more configuration into the graph by specifying selected arguments there? E.G:

[[[PT3H]]]
    graph = @clock => foo
[[[PT6H]]]
    graph = foo | @clock(+PT3H) => bar

hjoliver · 2017-09-28T00:47:39Z

@oliver-sanders - good suggestions, I'll address them as soon as possible. Unfortunately these new triggers can't be used in conditional expressions yet, but I hope it is easy to do - it may need #2413

hjoliver · 2017-10-02T22:59:40Z

@oliver-sanders - after a bit of thought I'd prefer to leave your suggestions to possible future enhancements.

xtriggers in conditional expressions: this is obviously desirable, but I think it requires unifying the various trigger types Universal trigger handling #2413. This PR is still a significant improvement without that (and validation catches any attempt to do it)
allowing some xtrigger function args in the graph: this may be desirable as a small efficiency enhancement in suites with lots of xtriggers, as you'd only have to define the function call once. However, it doesn't really add more info to the graph if you use labels carefully (e.g. clock_PT0H => foo vs clock(PT0H) => foo) and on the downside it breaks the useful one-to-one mapping between labels and unique function signatures, which is used to qualify functional results in case a single task has multiple xtriggers of the same type, e.g.:

x1 = xtrig(file=a)
x2 = xtrig(file=b)
...
graph = "x1 & x2 => foo"

if the function xtrig(file=a) returns {name: 'bob'} and xtrig(file=b) returns {name: 'alice'} then the suite daemon will broadcast to task foo environment, x1_name=bob and x2_name=alice. If instead we had this:

x = xtrig()
...
graph = "x(file=a) & x(file=b) => foo"

then it's not clear how to pass function results on to triggered tasks.

hjoliver · 2017-10-05T05:46:44Z

Revised example suite:

[cylc]
    cycle point format = %Y
[scheduling]
    initial cycle point = 2020
    [[xtriggers]]
         suite_up = suite_state(suite=upstream, task=foo, point=%(point)s):PT10S
         clock_0 = wall_clock(offset=PT0H) 
   [[dependencies]]
        [[[P1Y]]]
           graph = "@clock_0 & @suite_up => FAM"
[runtime]
    [[FAM]]
    [[f_1, f_2, f_3, f_4, f_5]]
          inherit = FAM

Note:

xtrigger configuration is a direct representation of the function call.
The optional :INTERVAL in label = func(args):INTERVAL determines how frequently the suite daemon calls the function until it returns True.
clock triggers are treated differently - they are checked synchronously (not in the process pool with a callback), they don't have a configurable re-check delay, and the cycle point of the waiting task is automatically passed to the function.

hjoliver · 2017-10-05T06:01:47Z

An interesting quirk to note: unlike tasks, xtriggers do not necessarily have an associated cycle point (hence they are not labeled with cycle point in the graph - see image above - but for convenience they are plotted per cycle point). Whether or not they are cycle-point specific depends on whether or not the cycle point of the dependent task is passed to the function (see the template %(point)s in the example suite above). The daemon groups xtriggers by unique function call signature, so if the cycle point does not appear in the argument list, then a single call will satisfy all cycle points as the suite runs. Similarly, if task name appears in the argument list (via template) xtrigger calls will be task-name specific (i.e. separate calls will be made to satisfy tasks with different names).

hjoliver · 2017-10-05T06:07:33Z

I will add some tests, then request a code review...

hjoliver · 2017-10-12T20:22:55Z

A note on the trigger function interface. The function signature is arbitrary, except that only keyword args are allowed. The suite-specific use of a trigger function is defined by the suite.rc syntax. Some suite and task variables (such as cycle point) are made available for use in the function args (in the suite.rc) - these are what can make a trigger cycle-point-, or task-name-, or whatever-, specific (because the template values are determined by the dependent task in each case). To illustrate, use of task name in the arg list below makes family members trigger at different random times off of (seemingly) the same xtrigger (in fact, on evaluation it creates 5 different xtriggers):

[scheduling]
   [[xtriggers]]
        rnd_by_name = xrandom(percent=50, secs=2, _=%(name)s)
   [[dependencies]]
        graph = "@rnd_by_name => FAM" 
[runtime]
   [[FAM]]
   [[f_1, f_2, f_3, f_4, f_5]]
         inherit = FAM

hjoliver · 2017-11-06T04:28:25Z

Branch rebased onto master.

hjoliver · 2017-11-06T22:50:07Z

@matthewrmshin @oliver-sanders - I think this is ready for review. I have not added user guide documentation yet in case review results in significant changes (although I'm pretty confident that the suite.rc interface is good) ... so let me know if the functionality and usage is not clear after reading the above comments, the docstring in lib/cylc/xtrigger_mgr.py, functions in lib/cylc/xtriggers/ and suites in /dev/suites/xtrigger/ and tests/xtriggers/.

matthewrmshin · 2017-11-07T08:51:01Z

We can ignore Codacy's failure here. The SQL statement is safe in the context it is used - adding a class level constant to the statement. The use of pickle is also safe in the context it is used. (I wonder if we should switch to using JSON instead of pickle for (de-)serializing data? I don't think Codacy complains against JSON.)

hjoliver · 2017-11-07T10:02:09Z

Re codacy: yes, I guess JSoN is not a security risk like pickle, although slower (where that matters).

BTW I still intend to go through all issues flagged by codacy and "ignore" or address them (particularly security ones) ... but have not found the time for that yet.

matthewrmshin

Some initial comments.

matthewrmshin · 2018-06-28T14:37:50Z

bin/cylc-help

@@ -292,6 +292,7 @@ admin_commands['profile-battery'] = ['profile-battery']
 admin_commands['import-examples'] = ['import-examples']
 admin_commands['upgrade-run-dir'] = ['upgrade-run-dir']
 admin_commands['check-software'] = ['check-software']
+admin_commands['wrap-func'] = ['wrap-func']


More like call-func than wrap-func?

Well, "call" suggests a normal function call, doesn't it? This wraps a function into a minimal executable program, to execute in the process pool.

Changed to function-run - less ambiguous, and no abbrev similarity to other commands (cylc warranty was a problem with wrap-func; and cylc run-func is a no-go thanks to cylc run).

matthewrmshin · 2018-06-28T14:38:34Z

bin/cylc-submit

+    db_mgr = SuiteDatabaseManager()
+    b_mgr = BroadcastMgr(db_mgr)
+    te_mgr = TaskEventsManager(suite, pool, db_mgr, b_mgr)
+    task_job_mgr = TaskJobManager(suite, pool, db_mgr, suite_srv_mgr, te_mgr)


Is this change still necessary?

What do you mean exactly by "this change"? Oh, maybe just that the te_mgr and b_bgr variables are unnecessary as only used once? I'll change that...

matthewrmshin · 2018-06-28T14:42:38Z

bin/cylc-wrap-func

+    func_name = sys.argv[1]
+    func_args = json.loads(sys.argv[2])
+    func_kwargs = json.loads(sys.argv[3])
+    run_func(func_name, func_args, func_kwargs)


It may be good to move the json-load => call function => json-dump logic to a location under lib/ (in the future?). (See also protocols such as JSON-RPC.)

Could do (pref "for the future" as I haven't time to look at JSON-RPC right now)

Fair enough.

I've taken your advice on move into lib/cylc.

matthewrmshin · 2018-06-28T14:44:29Z

lib/cylc/cfgspec/suite.py

@@ -29,8 +29,9 @@
 from isodatetime.parsers import TimePointParser, DurationParser

 from cylc.cfgspec.utils import (
-    coerce_interval, coerce_interval_list, DurationFloat)
+    coerce_interval, coerce_xtrig, coerce_interval_list, DurationFloat)


I may move this in #2609!

Yes, I thought you might 👍

matthewrmshin · 2018-06-28T14:45:47Z

lib/cylc/cfgspec/utils.py

-    IllegalValueError
-)
+    IllegalValueError)
+from cylc.mp_pool import SuiteFuncContext


Awful name SuiteFuncContext. Probably my fault with SuiteProcContext!

Agreed, and yes, just following the precedent!

matthewrmshin · 2018-06-28T14:48:09Z

lib/cylc/mp_pool.py

+import time
+import json
+import traceback
+from signal import signal, alarm, SIGALRM


New imports still used?

Yes, I'll remove those. That was from the original function timeout mechanism, not needed now after your process pool refactor.

matthewrmshin · 2018-06-28T14:49:20Z

lib/cylc/suite_db_mgr.py

@@ -306,6 +309,13 @@ def put_task_event_timers(self, task_events_mgr):
                    "delay": timer.delay,
                    "timeout": timer.timeout})

+    def put_xtriggers(self, sat_xtrig):


Add docstring?

matthewrmshin · 2018-06-28T14:52:47Z

lib/cylc/task_state.py

+            xclock_satisfied = False
+        else:
+            xclock_satisfied = True
+        return xclock_satisfied and all(self.xtriggers.values())


return self.xclock and self.xclock[1] and all(self.xtriggers.values())

Addressed (slightly differently - "None and bool" returns None)

matthewrmshin · 2018-06-28T14:57:36Z

lib/cylc/xtrigger_mgr.py

+            if sig not in self.all_xtrig:
+                rem.append(sig)
+        for r in rem:
+            del self.sat_xtrig[r]


May be?

for sig in list(self.sat_xtrig): if sig not in self.all_xtrig: del self.sat_xtrig[sig] # OR: self.sat_xtrig = [sig for sig in self.sat_xtrig if sig in self.all_xtrig]

Done. (The list comprehension doesn't work as sat_xtrig is a dict - and can't use dict comprehensions till we drop support for Python 2.6)

hjoliver · 2018-06-29T01:23:24Z

Latest review comments addressed.

hjoliver · 2018-06-29T06:32:22Z

Done (I hope). Documentation completed (I'm sure it could be improved, but I'm pretty happy with it).

matthewrmshin

Test OK in my environment.

matthewrmshin · 2018-06-29T09:17:49Z

Good to have this in early for the next-release cycle.

oliver-sanders

Everything seems in good shape and well documented. A few notes probably comments for future work, I don't actually care about spelling (second comment).

oliver-sanders · 2018-06-28T11:38:29Z

bin/cylc-wrap-func

+
+# Function return value to stdout: for compat with the command process pool.
+
+import importlib


Unused import?

I think you forgot to fetch my updates. This command has changed name to cylc-function-run, and the unused import removed.

oliver-sanders · 2018-06-28T11:51:37Z