///|
priv enum TicketState {
  Waiting
  Granted
  Refused
}

///|
/// One request's claim on a connection slot.
priv struct Ticket {
  origin : String
  mut state : TicketState
  cond : @cond_var.Cond
}

///|
/// Connections in use by an exchange, from its slot wait until its body is
/// read or abandoned, in all and per origin, under optional caps. Parked
/// connections hold no slot. A request that finds no slot waits; slots are
/// granted in arrival order to the waiters they fit, so a waiter whose origin
/// is at its cap does not hold up one for another origin. Waiters leave the
/// queue when they give up, so it holds only the requests waiting right now.
priv struct Slots {
  max_total : Int32?
  max_per_origin : Int32?
  mut in_use : Int32
  per_origin : Map[String, Int32]
  waiting : Array[Ticket]
}

///|
/// Caps below 1 are taken as 1, since a cap of 0 would refuse every request.
fn Slots::new(max_total? : Int32, max_per_origin? : Int32) -> Slots {
  {
    max_total: max_total.map(cap => cap.max(1)),
    max_per_origin: max_per_origin.map(cap => cap.max(1)),
    in_use: 0,
    per_origin: {},
    waiting: [],
  }
}

///|
fn Slots::fits(self : Slots, origin : String) -> Bool {
  let total_ok = match self.max_total {
    Some(cap) => self.in_use < cap
    None => true
  }
  let origin_ok = match self.max_per_origin {
    Some(cap) => self.per_origin.get(origin).unwrap_or(0) < cap
    None => true
  }
  total_ok && origin_ok
}

///|
fn Slots::take(self : Slots, origin : String) -> Unit {
  self.in_use += 1
  self.per_origin[origin] = self.per_origin.get(origin).unwrap_or(0) + 1
}

///|
/// Hands freed slots to the waiters they fit, earliest first.
fn Slots::grant_waiting(self : Slots) -> Unit {
  let mut i = 0
  while i < self.waiting.length() {
    let ticket = self.waiting[i]
    if self.fits(ticket.origin) {
      ignore(self.waiting.remove(i))
      self.take(ticket.origin)
      ticket.state = Granted
      ticket.cond.signal()
    } else {
      i += 1
    }
  }
}

///|
/// A ticket for `origin`: granted at once when a slot is free, otherwise
/// queued behind the requests already waiting. A free slot never has a
/// waiter it fits, since releases hand slots out as they free them.
fn Slots::enqueue(self : Slots, origin : String) -> Ticket {
  let ticket = { origin, state: Waiting, cond: @cond_var.Cond(), }
  if self.fits(origin) {
    self.take(origin)
    ticket.state = Granted
  } else {
    self.waiting.push(ticket)
  }
  ticket
}

///|
/// Frees a slot of `origin` and hands it on.
fn Slots::release(self : Slots, origin : String) -> Unit {
  self.in_use -= 1
  match self.per_origin.get(origin) {
    Some(count) if count > 1 => self.per_origin[origin] = count - 1
    _ => self.per_origin.remove(origin)
  }
  self.grant_waiting()
}

///|
/// Gives up `ticket`: a granted slot is released, a waiting ticket leaves the
/// queue without touching any slot.
fn Slots::abandon(self : Slots, ticket : Ticket) -> Unit {
  match ticket.state {
    Granted => self.release(ticket.origin)
    Waiting => {
      self.waiting.retain(other => !physical_equal(other, ticket))
      ticket.state = Refused
    }
    Refused => ()
  }
}

///|
/// Refuses every waiting ticket, as when the transport closes.
fn Slots::refuse_waiting(self : Slots) -> Unit {
  for ticket in self.waiting {
    ticket.state = Refused
    ticket.cond.signal()
  }
  self.waiting.clear()
}

///|
/// Waits until the ticket is granted; a refused ticket raises
/// `Connect("transport closed")`.
async fn Ticket::wait(self : Ticket) -> Unit raise @http.HttpError {
  while self.state is Waiting {
    self.cond.wait()
  }
  if self.state is Refused {
    raise @http.HttpError::Connect("transport closed")
  }
}

///|
/// Takes a slot for `origin`, waiting for one as the `Pool` phase of `timer`
/// when none is free. On a timeout, a failure or cancellation the request
/// leaves the queue, releasing a slot only if it had been granted one.
async fn Slots::acquire(
  self : Slots,
  origin : String,
  timer : PhaseTimer,
) -> Unit raise @http.HttpError {
  let ticket = self.enqueue(origin)
  errdefer self.abandon(ticket)
  if ticket.state is Waiting {
    timer.run(Pool, () => ticket.wait())
  }
}