code: def test_performance_ema_threadsafe(
      max_workers: int,
      interval: float = 0.01,
      num_updates: int = 100,
      alpha: float = 0.05,
      bias_power: float = 0.7,
      tolerance: float = 0.05,
  ):
      def run_task(ema):
          task_size = random.randint(1, 4)
          with ema.update_threadsafe(task_size):
              time.sleep(
                  task_size * interval * (0.9 + 0.2 * random.random())
              )
              return task_size
  
      with ThreadPoolExecutor(max_workers) as pool:
          ema = PerformanceEMA(alpha=alpha)
          start_time = time.perf_counter()
          futures = [
              pool.submit(run_task, ema) for i in range(num_updates)
          ]
          total_size = sum(future.result() for future in futures)
          end_time = time.perf_counter()
          target = total_size / (end_time - start_time)
          assert ema.samples_per_second >= (
              1 - tolerance
          ) * target * max_workers ** (bias_power - 1)
          assert ema.samples_per_second <= (1 + tolerance) * target
  
imports: import asyncio
  import concurrent.futures
  import multiprocessing
  import random
  import time
  from concurrent.futures import ThreadPoolExecutor
  import numpy
  import pytest
  import torch
  import hivemind
  from hivemind.compression import deserialize_torch_tensor, serialize_torch_tensor
  from hivemind.proto.runtime_pb2 import CompressionType
  from hivemind.utils import BatchTensorDescriptor, DHTExpiration, HeapEntry, MSGPackSerializer, ValueWithExpiration
  from hivemind.utils.asyncio import achain, aenumerate, aiter_with_timeout, amap_in_executor, anext, as_aiter, asingle, attach_event_on_finished, azip, cancel_and_wait, enter_asynchronously
  from hivemind.utils.mpfuture import InvalidStateError
  from hivemind.utils.performance_ema import PerformanceEMA
label_desc: Use image node_id 12 for calling a global function with example usage: test_performance_ema_threadsafe(max_workers, interval, num_updates, alpha, bias_power, tolerance) and returns: task_size