///|
/// Cooperative workers sharing memory. Exchange is data parallel; execution is
/// single-threaded. Network transport and cross-process progress are separate work.
pub struct WorkerGroup {
  priv workers : Array[Worker]
  priv mut next : Int32
}

///|
pub fn WorkerGroup::WorkerGroup(
  peers : Int32,
  clock? : () -> Duration,
) -> WorkerGroup {
  guard peers > 0 else { abort("worker group needs at least one peer") }
  let gate = Ref(false)
  let network = ProgressNetwork()
  {
    workers: Array::makei(peers, fn(_) {
      Worker::with_gate(gate, network, clock?)
    }),
    next: 0,
  }
}

///|
pub fn WorkerGroup::peers(self : WorkerGroup) -> Int32 {
  self.workers.length()
}

///|
pub fn WorkerGroup::worker(self : WorkerGroup, index : Int32) -> Worker {
  self.workers[index]
}

///|
pub fn WorkerGroup::step(self : WorkerGroup) -> Bool {
  for ;; {
    for _ in 0.. RunResult {
  guard max_steps >= 0 else { abort("step limit must be nonnegative") }
  for step in 0.. Unit {
  for worker in self.workers {
    worker.dispose()
  }
}

///|
priv struct GroupProgress[T, S] {
  members : Array[ScopeProgress[T, S]]
  schedulers : Array[Scheduler]
  local_counts : Array[ChangeBatch[(Int32, T)]]
  mut tracker : ProgressTracker[T, S]?
}

///|
fn[T : Lattice, S : Compare] GroupProgress::pump(
  self : GroupProgress[T, S],
) -> Unit {
  if self.members.is_empty() {
    return
  }
  let mut finished = true
  for index, core in self.members {
    if !core.ready_to_reclaim(self.schedulers[index]) {
      finished = false
      break
    }
  }
  if finished {
    // Local positive and negative counts can cancel across peers. Remove all
    // members together only after the global frontier and every task finish.
    for core in self.members {
      core.dispose()
    }
    self.tracker = None
    self.local_counts.clear()
    self.members.clear()
    self.schedulers.clear()
    return
  }
  let first = self.tracker is None
  let tracker = match self.tracker {
    Some(tracker) => tracker
    None => {
      let graph = self.members[0].graph
      for index in 0.. peer_tracker
        None => {
          let peer_tracker = ProgressTracker(tracker.reachability)
          core.tracker = Some(peer_tracker)
          peer_tracker
        }
      }
      peer_tracker.counts.clear()
      peer_tracker.counts.update_all(tracker.counts.to_array())
      core.dirty = true
    }
  }
}

///|
/// Identical operator graphs with globally counted pointstamps across peer workers.
pub struct ScopeGroup[T, S] {
  priv scopes : Array[Scope[T, S]]
  priv coordinator : GroupProgress[T, S]
}

///|
pub fn[T : Lattice, S : Compare] WorkerGroup::scopes(
  self : WorkerGroup,
  minimum : T,
  domain : SummaryDomain[T, S],
) -> ScopeGroup[T, S] {
  let scopes = self.workers.map(fn(worker) {
    worker.make_scope(minimum, domain, false)
  })
  coordinate_scopes(scopes)
}

///|
fn[T : Lattice, S : Compare] coordinate_scopes(
  scopes : Array[Scope[T, S]],
) -> ScopeGroup[T, S] {
  let progress : GroupProgress[T, S] = {
    members: scopes.map(fn(scope) { scope.progress }),
    schedulers: scopes.map(fn(scope) { scope.worker.scheduler }),
    local_counts: Array::makei(scopes.length(), fn(_) { ChangeBatch() }),
    tracker: None,
  }
  for scope in scopes {
    scope.worker.progress.push(fn() { progress.pump() })
    let core = scope.progress
    scope.worker.cleanup.push(fn() {
      core.disposed = true
      if !progress.members.iter().any(fn(core) { !core.disposed }) {
        progress.tracker = None
        progress.local_counts.clear()
        progress.members.clear()
        progress.schedulers.clear()
      }
    })
  }
  { scopes, coordinator: progress, }
}

///|
pub fn[T, S] ScopeGroup::scope(
  self : ScopeGroup[T, S],
  index : Int32,
) -> Scope[T, S] {
  guard !self.coordinator.members.is_empty() ||
    self.scopes[index].worker.scheduler.active else {
    abort("worker group is disposed")
  }
  self.scopes[index]
}

///|
pub fn[D, T : Lattice, R, S] ScopeGroup::input(
  self : ScopeGroup[T, S],
) -> Array[(InputSession[D, T, R], Collection[D, T, R])] {
  self.scopes.map(fn(scope) { scope.input() })
}

///|
/// Route each record to hash(data) % peers. Connectivity and message counts are
/// shared across peers, so an idle peer cannot report another peer's time complete.
pub fn[D, T : Lattice, R, S] ScopeGroup::exchange(
  self : ScopeGroup[T, S],
  inputs : Array[Collection[D, T, R]],
  hash : (D) -> UInt64,
) -> Array[Collection[D, T, R]] {
  guard inputs.length() == self.scopes.length() else {
    abort("exchange needs one collection per peer")
  }
  for index, source in inputs {
    guard source.channel.context is Some(context) &&
      physical_equal(context, self.scopes[index].context) else {
      abort("exchange input is not in the matching peer scope")
    }
  }
  let outputs : Array[Channel[D, T, R]] = inputs.map(fn(source) {
    source.channel.output()
  })
  let peers = outputs.length().to_uint64()
  for index, source in inputs {
    let input : Mailbox[D, T, R] = Mailbox()
    input.task = source.worker.register(fn() {
      for frame in input.drain() {
        if frame is Records(envelope, updates) {
          let buckets = Array::makei(outputs.length(), fn(_) { [] })
          for update in updates {
            let destination = (hash(update.0) % peers).to_int()
            buckets[destination].push(update)
          }
          for peer, bucket in buckets {
            outputs[peer].data_at(envelope, bucket)
          }
        }
      }
    })
    source.channel.subscribe(input)
    outputs[index].depends_on(input)
  }
  Array::makei(inputs.length(), fn(index) {
    { worker: inputs[index].worker, channel: outputs[index], }
  })
}