i#6938 sched migrate: Add input lock to queue pop result (#7000)

Adds a missing input lock to the result of popping from the queue.

Changes set_cur_input to set the prior input's last_run_time, etc.
*before* adding to the ready queue.

Tested by running ThreadSanitizer on the same internal test where it
reported a race in reading input_info_t.last_run_time by
pop_from_ready_queue_hold_locks while not holding a lock (during a
rebalance), versus set_cur_input writing to that field.

Issue: #6938

clients/drcachesim/scheduler/scheduler.cpp

@@ -2618,6 +2618,7 @@ scheduler_tmpl_t<RecordType, ReaderType>::pop_from_ready_queue_hold_locks(
             res = outputs_[from_output].ready_queue.queue.top();
             outputs_[from_output].ready_queue.queue.pop();
         }
+        std::lock_guard<mutex_dbg_owned> input_lock(*res->lock);
         assert(!res->unscheduled ||
                res->blocked_time > 0); // Should be in unscheduled_priority_.
         if (res->binding.empty() || for_output == INVALID_OUTPUT_ORDINAL ||
@@ -2691,6 +2692,8 @@ scheduler_tmpl_t<RecordType, ReaderType>::pop_from_ready_queue_hold_locks(
         std::lock_guard<mutex_dbg_owned> input_lock(*save->lock);
         add_to_ready_queue_hold_locks(from_output, save);
     }
+    auto res_lock = (res == nullptr) ? std::unique_lock<mutex_dbg_owned>()
+                                     : std::unique_lock<mutex_dbg_owned>(*res->lock);
     VDO(this, 1, {
         static int output_heartbeat;
         // We are ok with races as the cadence is approximate.
@@ -2815,10 +2818,6 @@ scheduler_tmpl_t<RecordType, ReaderType>::set_cur_input(output_ordinal_t output,
     assert(input < static_cast<input_ordinal_t>(inputs_.size()));
     int prev_input = outputs_[output].cur_input;
     if (prev_input >= 0) {
-        if (options_.mapping == MAP_TO_ANY_OUTPUT && prev_input != input &&
-            !inputs_[prev_input].at_eof) {
-            add_to_ready_queue(output, &inputs_[prev_input]);
-        }
         if (prev_input != input) {
             input_info_t &prev_info = inputs_[prev_input];
             std::lock_guard<mutex_dbg_owned> lock(*prev_info.lock);
@@ -2831,6 +2830,15 @@ scheduler_tmpl_t<RecordType, ReaderType>::set_cur_input(output_ordinal_t output,
                     return status;
             }
         }
+        // Now that we've updated prev_info, add it to the ready queue (once on the
+        // queues others can see it and pop it off, though they can't access/modify its
+        // fields (for identifying if it can be migrated, e.g.) until we release the
+        // input lock, so it should be safe to add first, but this order is more
+        // straightforward).
+        if (options_.mapping == MAP_TO_ANY_OUTPUT && prev_input != input &&
+            !inputs_[prev_input].at_eof) {
+            add_to_ready_queue(output, &inputs_[prev_input]);
+        }
     } else if (options_.schedule_record_ostream != nullptr &&
                outputs_[output].record.back().type == schedule_record_t::IDLE) {
         input_info_t unused;