Revert "Fix co-assignment for binary operations"

[gjoseph92]

This reverts commit 7ebd1d9.

Re-opens #6597. See coiled/benchmarks#295 for motivation.

• Tests added / passed
• Passes pre-commit run --all-files

[gjoseph92]

For reference, what it looked like before #6985:

distributed/distributed/scheduler.py

Lines 1850 to 1868 in f07f384

	ws = tg.last_worker
	if not (ws and tg.last_worker_tasks_left and ws.address in self.workers):
	# Last-used worker is full or unknown; pick a new worker for the next few tasks
	ws = min(
	(self.idle or self.workers).values(),
	key=partial(self.worker_objective, ts),
	)
	assert ws
	tg.last_worker_tasks_left = math.floor(
	(len(tg) / self.total_nthreads) * ws.nthreads
	)
	# Record `last_worker`, or clear it on the final task
	tg.last_worker = (
	ws if tg.states["released"] + tg.states["waiting"] > 1 else None
	)
	tg.last_worker_tasks_left -= 1
	return ws

Note there's still a slight change here—previously we'd schedule onto any worker if none were idle, now we only schedule onto running workers.

[jrbourbeau]

Thanks @gjoseph92! Have we had a chance to confirm this fixes the observed regression in coiled/benchmarks#295 as expected? Also cc @ian-r-rose in case he's looked into it

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       14 files   -        1         14 suites   - 1   6h 9m 1s ⏱️ - 23m 51s
  3 082 tests  -        4    2 997 ✔️  -        5    84 💤 ±  0  1 ❌ +1 
21 569 runs   - 1 265  20 572 ✔️  - 1 277  996 💤 +11  1 ❌ +1 

For more details on these failures, see this check.

Results for commit 3f215fd. ± Comparison against base commit bfc5cfe.

[gjoseph92]

• Flaky distributed/tests/test_client_executor.py::test_wait #6986

Have we had a chance to confirm this fixes the observed regression

I have not, but will try now.

[gjoseph92]

I've run it manually and confirmed it fixes the regression. I can't figure out how to get it to render with the dashboard script, but here are memory profiles. Notice the difference in peak values:

Before this PR:

After this PR:

[fjetter]

Thanks @gjoseph92

[jrbourbeau]

Yeah, thanks for handling this @gjoseph92