class_docstr: Simulate distributed training.
class_docstr_tok: ["Simulate","distributed","training","."]
docstr: Randomly assign a port for training to each worker and save all ports to mlist.txt.
docstr_tok: ["Randomly","assign","a","port","for","training","to","each","worker","and","save","all","ports","to","mlist.txt","."]
code: def _set_ports(self) -> None:
      """Randomly assign a port for training to each worker and save all ports to mlist.txt."""
      ports = set(_generate_n_ports(self.n_workers))
      i = 0
      max_tries = 100
      while i < max_tries and len(ports) < self.n_workers:
          n_ports_left = self.n_workers - len(ports)
          candidates = _generate_n_ports(n_ports_left)
          ports.update(candidates)
          i += 1
      if i == max_tries:
          raise RuntimeError("Unable to find non-colliding ports.")
      self.listen_ports = list(ports)
      with open(TESTS_DIR / "mlist.txt", "wt") as f:
          for port in self.listen_ports:
              f.write(f"127.0.0.1 {port}\n")
  
imports: import copy
  import io
  import socket
  import subprocess
  from concurrent.futures import ThreadPoolExecutor
  from pathlib import Path
  from typing import Any, Dict, Generator, List
  import numpy
  import pytest
  from sklearn.datasets import make_blobs, make_regression
  from sklearn.metrics import accuracy_score
label_desc: Use image node_id 3 for calling the DistributedMockup obj's underlying member method code with example usage: obj._set_ports() without return types