Make events per-parser

This makes the following changes:

1. Events in background threads are executed in those threads, instead of
being silently dropped

2. Blocked events are now per-parser instead of global

3. Events are posted in builtin_set instead of within the environment stack

The last one means that we no longer support event handlers for implicit
sets like (example) argv. Instead only the `set` builtin (and also `cd`)
post variable-change events.

Events from universal variable changes are still not fully rationalized.

src/builtin_cd.cpp

@@ -87,6 +87,10 @@ int builtin_cd(parser_t &parser, io_streams_t &streams, wchar_t **argv) {
         return STATUS_CMD_ERROR;
     }
 
-    parser.vars().set_one(L"PWD", ENV_EXPORT | ENV_GLOBAL, std::move(norm_dir));
+    std::vector<event_t> evts;
+    parser.vars().set_one(L"PWD", ENV_EXPORT | ENV_GLOBAL, std::move(norm_dir), &evts);
+    for (const auto &evt : evts) {
+        event_fire(parser, evt);
+    }
     return STATUS_CMD_OK;
 }

src/builtin_emit.cpp

@@ -31,6 +31,6 @@ int builtin_emit(parser_t &parser, io_streams_t &streams, wchar_t **argv) {
 
     const wchar_t *eventname = argv[optind];
     wcstring_list_t args(argv + optind + 1, argv + argc);
-    event_fire_generic(eventname, &args);
+    event_fire_generic(parser, eventname, &args);
     return STATUS_CMD_OK;
 }

src/builtin_read.cpp

@@ -223,7 +223,7 @@ static int read_interactive(parser_t &parser, wcstring &buff, int nchars, bool s
     reader_set_buffer(commandline, std::wcslen(commandline));
     proc_push_interactive(1);
 
-    event_fire_generic(L"fish_prompt");
+    event_fire_generic(parser, L"fish_prompt");
     auto mline = reader_readline(nchars);
     proc_pop_interactive();
     if (mline) {

src/builtin_set.cpp

@@ -341,12 +341,12 @@ static void handle_env_return(int retval, const wchar_t *cmd, const wchar_t *key
 /// description of the problem to stderr.
 static int env_set_reporting_errors(const wchar_t *cmd, const wchar_t *key, int scope,
                                     const wcstring_list_t &list, io_streams_t &streams,
-                                    env_stack_t &vars) {
+                                    env_stack_t &vars, std::vector<event_t> *evts) {
     if (is_path_variable(key) && !validate_path_warning_on_colons(cmd, key, list, streams, vars)) {
         return STATUS_CMD_ERROR;
     }
 
-    int retval = vars.set(key, scope | ENV_USER, list);
+    int retval = vars.set(key, scope | ENV_USER, list, evts);
     handle_env_return(retval, cmd, key, streams);
 
     return retval;
@@ -649,8 +649,9 @@ static int builtin_set_erase(const wchar_t *cmd, set_cmd_opts_t &opts, int argc,
         return STATUS_INVALID_ARGS;
     }
 
+    std::vector<event_t> evts;
     if (idx_count == 0) {  // unset the var
-        retval = parser.vars().remove(dest, scope);
+        retval = parser.vars().remove(dest, scope, &evts);
         // When a non-existent-variable is unset, return ENV_NOT_FOUND as $status
         // but do not emit any errors at the console as a compromise between user
         // friendliness and correctness.
@@ -663,7 +664,12 @@ static int builtin_set_erase(const wchar_t *cmd, set_cmd_opts_t &opts, int argc,
         wcstring_list_t result;
         dest_var->to_list(result);
         erase_values(result, indexes);
-        retval = env_set_reporting_errors(cmd, dest, scope, result, streams, parser.vars());
+        retval = env_set_reporting_errors(cmd, dest, scope, result, streams, parser.vars(), &evts);
+    }
+
+    // Fire any events.
+    for (const auto &evt : evts) {
+        event_fire(parser, evt);
     }
 
     if (retval != STATUS_CMD_OK) return retval;
@@ -675,7 +681,6 @@ static int set_var_array(const wchar_t *cmd, set_cmd_opts_t &opts, const wchar_t
                          wcstring_list_t &new_values, int argc, wchar_t **argv, parser_t &parser,
                          io_streams_t &streams) {
     UNUSED(cmd);
-    UNUSED(parser);
     UNUSED(streams);
 
     if (opts.prepend || opts.append) {
@@ -771,7 +776,14 @@ static int builtin_set_set(const wchar_t *cmd, set_cmd_opts_t &opts, int argc, w
     }
     if (retval != STATUS_CMD_OK) return retval;
 
-    retval = env_set_reporting_errors(cmd, varname, scope, new_values, streams, parser.vars());
+    std::vector<event_t> evts;
+    retval =
+        env_set_reporting_errors(cmd, varname, scope, new_values, streams, parser.vars(), &evts);
+    // Fire any events.
+    for (const auto &evt : evts) {
+        event_fire(parser, evt);
+    }
+
     if (retval != STATUS_CMD_OK) return retval;
     return check_global_scope_exists(cmd, opts, varname, streams, parser.vars());
 }

src/env.cpp

@@ -1212,7 +1212,8 @@ maybe_t<env_var_t> env_stack_t::get(const wcstring &key, env_mode_flags_t mode)
 
 wcstring_list_t env_stack_t::get_names(int flags) const { return acquire_impl()->get_names(flags); }
 
-int env_stack_t::set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t vals) {
+int env_stack_t::set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t vals,
+                     std::vector<event_t> *out_events) {
     // Historical behavior.
     if (vals.size() == 1 && (key == L"PWD" || key == L"HOME")) {
         path_make_canonical(vals.front());
@@ -1223,7 +1224,9 @@ int env_stack_t::set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t
     if (ret == ENV_OK) {
         // Important to not hold the lock here.
         env_dispatch_var_change(key, *this);
-        event_fire(event_t::variable(key, {L"VARIABLE", L"SET", key}));
+        if (out_events) {
+            out_events->push_back(event_t::variable(key, {L"VARIABLE", L"SET", key}));
+        }
     }
     if (needs_uvar_sync) {
         universal_barrier();
@@ -1231,23 +1234,27 @@ int env_stack_t::set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t
     return ret;
 }
 
-int env_stack_t::set_one(const wcstring &key, env_mode_flags_t mode, wcstring val) {
+int env_stack_t::set_one(const wcstring &key, env_mode_flags_t mode, wcstring val,
+                         std::vector<event_t> *out_events) {
     wcstring_list_t vals;
     vals.push_back(std::move(val));
-    return set(key, mode, std::move(vals));
+    return set(key, mode, std::move(vals), out_events);
 }
 
-int env_stack_t::set_empty(const wcstring &key, env_mode_flags_t mode) {
-    return set(key, mode, {});
+int env_stack_t::set_empty(const wcstring &key, env_mode_flags_t mode,
+                           std::vector<event_t> *out_events) {
+    return set(key, mode, {}, out_events);
 }
 
-int env_stack_t::remove(const wcstring &key, int mode) {
+int env_stack_t::remove(const wcstring &key, int mode, std::vector<event_t> *out_events) {
     bool needs_uvar_sync = false;
     int ret = acquire_impl()->remove(key, mode, &needs_uvar_sync);
     if (ret == ENV_OK) {
         // Important to not hold the lock here.
         env_dispatch_var_change(key, *this);
-        event_fire(event_t::variable(key, {L"VARIABLE", L"ERASE", key}));
+        if (out_events) {
+            out_events->push_back(event_t::variable(key, {L"VARIABLE", L"ERASE", key}));
+        }
     }
     if (needs_uvar_sync) {
         universal_barrier();

src/env.h

@@ -16,6 +16,8 @@
 extern size_t read_byte_limit;
 extern bool curses_initialized;
 
+struct event_t;
+
 /// Character for separating two array elements. We use 30, i.e. the ascii record separator since
 /// that seems logical.
 #define ARRAY_SEP (wchar_t)0x1e
@@ -235,13 +237,18 @@ class env_stack_t final : public environment_t {
     wcstring_list_t get_names(int flags) const override;
 
     /// Sets the variable with the specified name to the given values.
-    int set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t vals);
+    /// If \p out_events is supplied, populate it with any events generated through setting the
+    /// variable.
+    int set(const wcstring &key, env_mode_flags_t mode, wcstring_list_t vals,
+            std::vector<event_t> *out_events = nullptr);
 
     /// Sets the variable with the specified name to a single value.
-    int set_one(const wcstring &key, env_mode_flags_t mode, wcstring val);
+    int set_one(const wcstring &key, env_mode_flags_t mode, wcstring val,
+                std::vector<event_t> *out_events = nullptr);
 
     /// Sets the variable with the specified name to no values.
-    int set_empty(const wcstring &key, env_mode_flags_t mode);
+    int set_empty(const wcstring &key, env_mode_flags_t mode,
+                  std::vector<event_t> *out_events = nullptr);
 
     /// Update the PWD variable based on the result of getcwd.
     void set_pwd_from_getcwd();
@@ -252,9 +259,11 @@ class env_stack_t final : public environment_t {
     /// \param mode should be ENV_USER if this is a remove request from the user, 0 otherwise. If
     /// this is a user request, read-only variables can not be removed. The mode may also specify
     /// the scope of the variable that should be erased.
+    /// \param out_events if non-null, populate it with any events generated from removing this
+    /// variable.
     ///
     /// \return zero if the variable existed, and non-zero if the variable did not exist
-    int remove(const wcstring &key, int mode);
+    int remove(const wcstring &key, int mode, std::vector<event_t> *out_events = nullptr);
 
     /// Push the variable stack. Used for implementing local variables for functions and for-loops.
     void push(bool new_scope);

src/env_dispatch.cpp

@@ -45,6 +45,7 @@
 #include "input_common.h"
 #include "maybe.h"
 #include "output.h"
+#include "parser.h"
 #include "proc.h"
 #include "reader.h"
 #include "screen.h"
@@ -211,7 +212,9 @@ static void universal_callback(env_stack_t *stack, const callback_data_t &cb) {
 
     env_dispatch_var_change(cb.key, *stack);
     stack->mark_changed_exported();
-    event_fire(event_t::variable(cb.key, {L"VARIABLE", op, cb.key}));
+
+    // TODO: eliminate this principal_parser. Need to rationalize how multiple threads work here.
+    event_fire(parser_t::principal_parser(), event_t::variable(cb.key, {L"VARIABLE", op, cb.key}));
 }
 
 void env_universal_callbacks(env_stack_t *stack, const callback_data_list_t &callbacks) {

src/event.cpp

@@ -86,10 +86,6 @@ static pending_signals_t s_pending_signals;
 /// List of event handlers.
 static event_handler_list_t s_event_handlers;
 
-/// List of events that have been sent but have not yet been delivered because they are blocked.
-using event_list_t = std::vector<shared_ptr<event_t>>;
-static event_list_t blocked;
-
 /// Variables (one per signal) set when a signal is observed. This is inspected by a signal handler.
 static volatile bool s_observed_signals[NSIG] = {};
 static void set_signal_observed(int sig, bool val) {
@@ -130,11 +126,9 @@ static bool handler_matches(const event_handler_t &classv, const event_t &instan
 }
 
 /// Test if specified event is blocked.
-static int event_is_blocked(const event_t &e) {
+static int event_is_blocked(parser_t &parser, const event_t &e) {
     (void)e;
     const block_t *block;
-    parser_t &parser = parser_t::principal_parser();
-
     size_t idx = 0;
     while ((block = parser.block_at_index(idx++))) {
         if (event_block_list_blocks_type(block->event_blocks)) return true;
@@ -246,9 +240,8 @@ bool event_is_signal_observed(int sig) {
 /// Perform the specified event. Since almost all event firings will not be matched by even a single
 /// event handler, we make sure to optimize the 'no matches' path. This means that nothing is
 /// allocated/initialized unless needed.
-static void event_fire_internal(const event_t &event) {
-    ASSERT_IS_MAIN_THREAD();
-    auto &ld = parser_t::principal_parser().libdata();
+static void event_fire_internal(parser_t &parser, const event_t &event) {
+    auto &ld = parser.libdata();
     assert(ld.is_event >= 0 && "is_event should not be negative");
     scoped_push<decltype(ld.is_event)> inc_event{&ld.is_event, ld.is_event + 1};
 
@@ -293,18 +286,14 @@ static void event_fire_internal(const event_t &event) {
 }
 
 /// Handle all pending signal events.
-void event_fire_delayed() {
-    // Hack: only allow events on the main thread.
-    // TODO: rationalize how events work with multiple threads.
-    if (!is_main_thread()) return;
-
-    auto &parser = parser_t::principal_parser();
+void event_fire_delayed(parser_t &parser) {
+    auto &ld = parser.libdata();
     // Do not invoke new event handlers from within event handlers.
-    if (parser.libdata().is_event) return;
+    if (ld.is_event) return;
 
-    event_list_t to_send;
-    to_send.swap(blocked);
-    assert(blocked.empty());
+    std::vector<shared_ptr<event_t>> to_send;
+    to_send.swap(ld.blocked_events);
+    assert(ld.blocked_events.empty());
 
     // Append all signal events to to_send.
     auto signals = s_pending_signals.acquire_pending();
@@ -321,10 +310,10 @@ void event_fire_delayed() {
 
     // Fire or re-block all events.
     for (const auto &evt : to_send) {
-        if (event_is_blocked(*evt)) {
-            blocked.push_back(evt);
+        if (event_is_blocked(parser, *evt)) {
+            ld.blocked_events.push_back(evt);
         } else {
-            event_fire_internal(*evt);
+            event_fire_internal(parser, *evt);
         }
     }
 }
@@ -334,18 +323,14 @@ void event_enqueue_signal(int signal) {
     s_pending_signals.mark(signal);
 }
 
-void event_fire(const event_t &event) {
-    // Hack: only allow events on the main thread.
-    // TODO: rationalize how events work with multiple threads.
-    if (!is_main_thread()) return;
-
+void event_fire(parser_t &parser, const event_t &event) {
     // Fire events triggered by signals.
-    event_fire_delayed();
+    event_fire_delayed(parser);
 
-    if (event_is_blocked(event)) {
-        blocked.push_back(std::make_shared<event_t>(event));
+    if (event_is_blocked(parser, event)) {
+        parser.libdata().blocked_events.push_back(std::make_shared<event_t>(event));
     } else {
-        event_fire_internal(event);
+        event_fire_internal(parser, event);
     }
 }
 
@@ -440,13 +425,13 @@ void event_print(io_streams_t &streams, maybe_t<event_type_t> type_filter) {
     }
 }
 
-void event_fire_generic(const wchar_t *name, const wcstring_list_t *args) {
+void event_fire_generic(parser_t &parser, const wchar_t *name, const wcstring_list_t *args) {
     assert(name && "Null name");
 
     event_t ev(event_type_t::generic);
     ev.desc.str_param1 = name;
     if (args) ev.arguments = *args;
-    event_fire(ev);
+    event_fire(parser, ev);
 }
 
 event_description_t event_description_t::signal(int sig) {

src/event.h

@@ -90,6 +90,8 @@ struct event_t {
     static event_t variable(wcstring name, wcstring_list_t args);
 };
 
+class parser_t;
+
 /// Add an event handler.
 void event_add_handler(std::shared_ptr<event_handler_t> eh);
 
@@ -103,11 +105,11 @@ event_handler_list_t event_get_function_handlers(const wcstring &name);
 /// a signal handler.
 bool event_is_signal_observed(int signal);
 
-/// Fire the specified event \p event.
-void event_fire(const event_t &event);
+/// Fire the specified event \p event, executing it on \p parser.
+void event_fire(parser_t &parser, const event_t &event);
 
-/// Fire all delayed eents.
-void event_fire_delayed();
+/// Fire all delayed events attached to the given parser.
+void event_fire_delayed(parser_t &parser);
 
 /// Enqueue a signal event. Invoked from a signal handler.
 void event_enqueue_signal(int signal);
@@ -119,7 +121,7 @@ void event_print(io_streams_t &streams, maybe_t<event_type_t> type_filter);
 wcstring event_get_desc(const event_t &e);
 
 /// Fire a generic event with the specified name.
-void event_fire_generic(const wchar_t *name, const wcstring_list_t *args = NULL);
+void event_fire_generic(parser_t &parser, const wchar_t *name, const wcstring_list_t *args = NULL);
 
 /// Return the event type for a given name, or none.
 maybe_t<event_type_t> event_type_for_name(const wcstring &name);

src/fish.cpp

@@ -522,11 +522,12 @@ int main(int argc, char **argv) {
 
     // TODO: The generic process-exit event is useless and unused.
     // Remove this in future.
-    event_fire(proc_create_event(L"PROCESS_EXIT", event_type_t::exit, getpid(), exit_status));
+    event_fire(parser,
+               proc_create_event(L"PROCESS_EXIT", event_type_t::exit, getpid(), exit_status));
 
     // Trigger any exit handlers.
     wcstring_list_t event_args = {to_string(exit_status)};
-    event_fire_generic(L"fish_exit", &event_args);
+    event_fire_generic(parser, L"fish_exit", &event_args);
 
     restore_term_mode();
     restore_term_foreground_process_group();

src/input.cpp

@@ -241,14 +241,16 @@ void input_mapping_set_t::add(const wchar_t *sequence, const wchar_t *command, c
     input_mapping_set_t::add(sequence, &command, 1, mode, sets_mode, user);
 }
 
-/// Handle interruptions to key reading by reaping finshed jobs and propagating the interrupt to the
-/// reader.
+/// Handle interruptions to key reading by reaping finished jobs and propagating the interrupt to
+/// the reader.
 static maybe_t<char_event_t> interrupt_handler() {
     // Fire any pending events.
-    event_fire_delayed();
+    // TODO: eliminate this principal_parser().
+    auto &parser = parser_t::principal_parser();
+    event_fire_delayed(parser);
     // Reap stray processes, including printing exit status messages.
     // TODO: shouldn't need this parser here.
-    if (job_reap(parser_t::principal_parser(), true)) reader_repaint_needed();
+    if (job_reap(parser, true)) reader_repaint_needed();
     // Tell the reader an event occured.
     if (reader_reading_interrupted()) {
         auto vintr = shell_modes.c_cc[VINTR];

src/parse_execution.cpp

@@ -729,7 +729,7 @@ parse_execution_result_t parse_execution_context_t::handle_command_not_found(
             event_args.insert(event_args.begin(), cmd_str);
         }
 
-        event_fire_generic(L"fish_command_not_found", &event_args);
+        event_fire_generic(*parser, L"fish_command_not_found", &event_args);
 
         // Here we want to report an error (so it shows a backtrace), but with no text.
         this->report_error(statement, L"");

src/parser.h

@@ -119,6 +119,7 @@ struct profile_item_t {
 
 class parse_execution_context_t;
 class completion_t;
+struct event_t;
 
 /// Miscelleneous data used to avoid recursion and others.
 struct library_data_t {
@@ -160,6 +161,9 @@ struct library_data_t {
     /// The current filename we are evaluating, either from builtin source or on the command line.
     /// This is an intern'd string.
     const wchar_t *current_filename{};
+
+    /// List of events that have been sent but have not yet been delivered because they are blocked.
+    std::vector<shared_ptr<event_t>> blocked_events;
 };
 
 class parser_t : public std::enable_shared_from_this<parser_t> {

src/proc.cpp

@@ -618,7 +618,7 @@ static bool process_clean_after_marking(parser_t &parser, bool allow_interactive
 
     // Post pending exit events.
     for (const auto &evt : exit_events) {
-        event_fire(evt);
+        event_fire(parser, evt);
     }
 
     if (printed) {

src/reader.cpp

@@ -2242,7 +2242,7 @@ static int read_i(parser_t &parser) {
     data->prev_end_loop = 0;
 
     while (!shell_is_exiting()) {
-        event_fire_generic(L"fish_prompt");
+        event_fire_generic(parser, L"fish_prompt");
         run_count++;
 
         if (parser.libdata().is_breakpoint && function_exists(DEBUG_PROMPT_FUNCTION_NAME, parser)) {
@@ -2271,9 +2271,9 @@ static int read_i(parser_t &parser) {
             data->command_line.text.clear();
             data->command_line_changed(&data->command_line);
             wcstring_list_t argv(1, command);
-            event_fire_generic(L"fish_preexec", &argv);
+            event_fire_generic(parser, L"fish_preexec", &argv);
             reader_run_command(parser, command);
-            event_fire_generic(L"fish_postexec", &argv);
+            event_fire_generic(parser, L"fish_postexec", &argv);
             // Allow any pending history items to be returned in the history array.
             if (data->history) {
                 data->history->resolve_pending();

tests/cd.err

@@ -10,3 +10,6 @@
 
 ####################
 # CDPATH
+
+####################
+# on-variable PWD

tests/cd.in

@@ -79,3 +79,10 @@ test $PWD = $base; and echo No crash with ./ CDPATH
 # cd back before removing the test directory again.
 cd $oldpwd
 rm -Rf $base
+
+logmsg on-variable PWD
+# Verify that PWD on-variable events are sent
+function __fish_test_changed_pwd --on-variable PWD
+   echo "Changed to $PWD"
+end
+cd /

tests/cd.out

@@ -30,3 +30,7 @@ Gone to linkhome via CDPATH
 Gone to base
 Gone to linkhome via implicit . in CDPATH
 No crash with ./ CDPATH
+
+####################
+# on-variable PWD
+Changed to /