// Copyright 2025 International Digital Economy Academy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

///|
priv struct Waiter {
  coro : @coroutine.Coroutine
  /// For intrusive cancellable queue
  mut index : Int
  mut next : Int
  mut prev : Int
}

///|
impl @cancellable_queue.Element for Waiter with fn get_index(self) {
  self.index
}

///|
impl @cancellable_queue.Element for Waiter with fn set_index(self, value) {
  self.index = value
}

///|
impl @cancellable_queue.Element for Waiter with fn get_next(self) {
  self.next
}

///|
impl @cancellable_queue.Element for Waiter with fn set_next(self, value) {
  self.next = value
}

///|
impl @cancellable_queue.Element for Waiter with fn get_prev(self) {
  self.prev
}

///|
impl @cancellable_queue.Element for Waiter with fn set_prev(self, value) {
  self.prev = value
}

///|
extend Waiter with @cancellable_queue.Element::{is_queued}

///|
/// A condition variable that can be used for synchronization between tasks.
#valtype
struct Cond {
  waiters : @cancellable_queue.Queue[Waiter]
}

///|
/// Create a new condition variable.
#alias(new, deprecated)
pub fn Cond::Cond() -> Cond {
  { waiters: Queue(), }
}

///|
/// Wait for the condition to get signaled.
/// `wait` itself never fail, and will wait indefinitely.
/// However, since `wait` is a blocking point,
/// the task running `wait` may be cancelled,
/// in this case, an cancellation will be raised from `wait`.
pub async fn Cond::wait(cond : Cond) -> Unit noraise {
  let coro = @coroutine.current_coroutine()
  let waiter = { coro, index: -1, next: -1, prev: -1, }
  cond.waiters.push(waiter)
  match @coroutine.suspend_check_cancel() {
    Continue => ()
    Cancelled if !waiter.is_queued() =>
      // In the rare case where:
      //
      // - `signal` wake up a waiter, the piece of resource is assigned to that waiter
      // - the waiter is immediately cancelled after woken up by `signal`,
      //   but before the waiter's coroutine actually resumes execution
      //
      // we must swallow the cancellation signal,
      // as otherwise the signal will be gone forever.
      // So we set `woken = true` after waking up a waiter,
      // and use this flag to protect against dangerous cancellation.
      ()
    Cancelled => {
      cond.waiters.remove(waiter)
      @coroutine.raise_cancellation_signal()
    }
  }
}

///|
/// Issue a single signal through the condition variable,
/// waking up the first task waiting for the condition variable.
/// If no task is waiting for the condition variable, nothing will happen.
pub fn Cond::signal(cond : Cond) -> Unit {
  if cond.waiters.pop() is Some(waiter) {
    waiter.coro.wake()
  }
}

///|
/// Send a broadcast signal to the condition variable,
/// waking up all tasks waiting for the condition variable.
/// If no task is waiting for the condition variable, nothing will happen.
pub fn Cond::broadcast(cond : Cond) -> Unit {
  while cond.waiters.pop() is Some(waiter) {
    waiter.coro.wake()
  }
}