code: def acquire(self) -> "PortsLock":
      assert not self.sockets
      assigned_ports = set()
  
      # mapped by user
      for remote_port, local_port in self.restrictions.items():
          if not local_port:  # None or 0
              continue
          if local_port in assigned_ports:
              raise PortUsedError(
                  f"Port {local_port} is already in use"
              )
          sock = self._listen(local_port)
          if sock is None:
              raise PortUsedError(
                  f"Port {local_port} is already in use"
              )
          self.sockets[remote_port] = sock
          assigned_ports.add(local_port)
  
      # mapped automatically
      for remote_port, local_port in self.restrictions.items():
          if local_port:
              continue
          local_port = remote_port
          while True:
              if local_port not in assigned_ports:
                  sock = self._listen(local_port)
                  if sock is not None:
                      break
              local_port += 1
          self.sockets[remote_port] = sock
          assigned_ports.add(local_port)
      return self
  
file_path: dstack/src/dstack/_internal/core/services/ssh/ports.py
imports: import errno
  import socket
  from typing import Dict, List, Optional
  from dstack._internal.core.errors import DstackError
  from dstack._internal.core.models.configurations import PortMapping
label_desc: Use image node_id 2 for calling the PortsLock obj's underlying member method code with example usage: obj.acquire() and returns: self