code: def test_gather_priority(ws):
      """Test that tasks are fetched in the following order:
  
      1. by task priority
      2. in case of tie, from local workers first
      3. in case of tie, from the worker with the most tasks queued
      4. in case of tie, from a random worker (which is actually deterministic).
      """
      ws.transfer_incoming_count_limit = 4
  
      instructions = ws.handle_stimulus(
          PauseEvent(stimulus_id="pause"),
          # Note: tasks fetched by acquire-replicas always have priority=(1, )
          AcquireReplicasEvent(
              who_has={
                  # Remote + local
                  "x1": ["127.0.0.2:1", "127.0.0.1:2"],
                  # Remote. After getting x11 from .1, .2  will have less tasks than .3
                  "x2": ["127.0.0.2:1"],
                  "x3": ["127.0.0.3:1"],
                  "x4": ["127.0.0.3:1"],
                  # It will be a random choice between .2, .4, .5, .6, and .7
                  "x5": ["127.0.0.4:1"],
                  "x6": ["127.0.0.5:1"],
                  "x7": ["127.0.0.6:1"],
                  # This will be fetched first because it's on the same worker as y
                  "x8": ["127.0.0.7:1"],
              },
              # Substantial nbytes prevents transfer_incoming_count_limit to be
              # overridden by transfer_incoming_bytes_throttle_threshold,
              # but it's less than transfer_message_bytes_limit
              nbytes={f"x{i}": 4 * 2**20 for i in range(1, 9)},
              stimulus_id="compute1",
          ),
          # A higher-priority task, even if scheduled later, is fetched first
          ComputeTaskEvent.dummy(
              key="z",
              who_has={"y": ["127.0.0.7:1"]},
              priority=(0,),
              stimulus_id="compute2",
          ),
          UnpauseEvent(stimulus_id="unpause"),
      )
  
      assert instructions == [
          # Highest-priority task first. Lower priority tasks from the same worker are
          # shoved into the same instruction (up to 50MB worth)
          GatherDep(
              stimulus_id="unpause",
              worker="127.0.0.7:1",
              to_gather={"y", "x8"},
              total_nbytes=1 + 4 * 2**20,
          ),
          # Followed by local workers
          GatherDep(
              stimulus_id="unpause",
              worker="127.0.0.1:2",
              to_gather={"x1"},
              total_nbytes=4 * 2**20,
          ),
          # Followed by remote workers with the most tasks
          GatherDep(
              stimulus_id="unpause",
              worker="127.0.0.3:1",
              to_gather={"x3", "x4"},
              total_nbytes=8 * 2**20,
          ),
          # Followed by other remote workers, randomly.
          # Determinism is guaranteed by a statically-seeded random number generator.
          # FIXME It would have not been deterministic if we instead of multiple keys we
          #       had used a single key with multiple workers, because sets
          #       (like TaskState.who_has) change order at every interpreter restart.
          GatherDep(
              stimulus_id="unpause",
              worker="127.0.0.4:1",
              to_gather={"x5"},
              total_nbytes=4 * 2**20,
          ),
      ]
      expected_bytes = (
          1 + 4 * 2**20 + 4 * 2**20 + 8 * 2**20 + 4 * 2**20
      )
      assert ws.transfer_incoming_bytes == expected_bytes
      assert ws.transfer_incoming_count == 4
      assert ws.transfer_incoming_count_total == 4
  
imports: from __future__ import annotations
  import asyncio
  import gc
  import pickle
  import sys
  from collections import defaultdict
  from collections.abc import Iterator
  from time import sleep
  import pytest
  from tlz import first
  from dask.sizeof import sizeof
  import distributed.profile
  from distributed import Nanny, Worker, wait
  from distributed.compatibility import MACOS, WINDOWS
  from distributed.protocol.serialize import Serialize
  from distributed.scheduler import TaskState
  from distributed.utils import recursive_to_dict
  from distributed.utils_test import NO_AMM, _LockedCommPool, assert_story, async_poll_for, freeze_data_fetching, gen_cluster, inc, wait_for_state
  from distributed.worker_state_machine import AcquireReplicasEvent, AddKeysMsg, ComputeTaskEvent, Execute, ExecuteFailureEvent, ExecuteSuccessEvent, FreeKeysEvent, GatherDep, GatherDepBusyEvent, GatherDepFailureEvent, GatherDepNetworkFailureEvent, GatherDepSuccessEvent, Instruction, InvalidTaskState, InvalidTransition, PauseEvent, RecommendationsConflict, RefreshWhoHasEvent, ReleaseWorkerDataMsg, RemoveReplicasEvent, RemoveWorkerEvent, RequestRefreshWhoHasMsg, RescheduleEvent, RescheduleMsg, RetryBusyWorkerEvent, RetryBusyWorkerLater, SecedeEvent, StateMachineEvent, TaskErredMsg, TaskState, TransitionCounterMaxExceeded, UnpauseEvent, UpdateDataEvent, merge_recs_instructions
label_desc: Use image node_id 25 for calling a global function with example usage: test_gather_priority(ws) without return types