channel.ltl

// =======================================================================
// LATERALUS Standard Library — channel.ltl
// CSP-style channel primitives for concurrent communication
// =======================================================================
// Channels are typed, bounded message queues used for safe communication
// between concurrent tasks. Inspired by Go channels and Rust's mpsc.
//
// Usage:
//   let ch = channel_new(10)        // buffered channel, capacity 10
//   channel_send(ch, 42)            // send a value
//   let val = channel_recv(ch)      // receive a value
//   channel_close(ch)               // close the channel
//
// Select:
//   let result = select([ch1, ch2]) // wait on multiple channels
// =======================================================================

module channel

// -- Channel -------------------------------------------------------------

/// Create a new buffered channel with the given capacity.
/// A capacity of 0 creates an unbuffered (synchronous) channel.
pub fn channel_new(capacity: int) -> map {
    return {
        "type":     "channel",
        "buffer":   [],
        "capacity": max(capacity, 0),
        "closed":   false,
        "send_count": 0,
        "recv_count": 0,
    }
}

/// Create an unbuffered channel (synchronous send/recv).
pub fn unbuffered() -> map {
    return channel_new(0)
}

/// Send a value into the channel.
/// Returns true if sent, false if the channel is closed.
/// Blocks (busy-waits) if the buffer is full.
pub fn channel_send(ch: map, value) -> bool {
    if ch["closed"] {
        return false
    }

    // For buffered channels, wait until there's space
    if ch["capacity"] > 0 {
        while len(ch["buffer"]) >= ch["capacity"] {
            if ch["closed"] {
                return false
            }
            yield()
        }
    }

    ch["buffer"] = ch["buffer"] + [value]
    ch["send_count"] = ch["send_count"] + 1
    return true
}

/// Receive a value from the channel.
/// Returns the value, or none if the channel is closed and empty.
/// Blocks (busy-waits) if the buffer is empty.
pub fn channel_recv(ch: map) {
    while len(ch["buffer"]) == 0 {
        if ch["closed"] {
            return none
        }
        yield()
    }

    let value = ch["buffer"][0]
    ch["buffer"] = ch["buffer"][1:]
    ch["recv_count"] = ch["recv_count"] + 1
    return value
}

/// Try to send without blocking. Returns true if sent.
pub fn try_send(ch: map, value) -> bool {
    if ch["closed"] {
        return false
    }
    if ch["capacity"] > 0 && len(ch["buffer"]) >= ch["capacity"] {
        return false
    }
    ch["buffer"] = ch["buffer"] + [value]
    ch["send_count"] = ch["send_count"] + 1
    return true
}

/// Try to receive without blocking. Returns the value or none.
pub fn try_recv(ch: map) {
    if len(ch["buffer"]) == 0 {
        return none
    }
    let value = ch["buffer"][0]
    ch["buffer"] = ch["buffer"][1:]
    ch["recv_count"] = ch["recv_count"] + 1
    return value
}

/// Close the channel. No more values can be sent.
/// Remaining buffered values can still be received.
pub fn channel_close(ch: map) {
    ch["closed"] = true
}

/// Check if the channel is closed.
pub fn is_closed(ch: map) -> bool {
    return ch["closed"]
}

/// Check if the channel has data available to receive.
pub fn has_data(ch: map) -> bool {
    return len(ch["buffer"]) > 0
}

/// Check if the channel is full (buffered) or always true (unbuffered).
pub fn is_full(ch: map) -> bool {
    if ch["capacity"] == 0 {
        return true
    }
    return len(ch["buffer"]) >= ch["capacity"]
}

/// Get the number of values currently buffered.
pub fn pending(ch: map) -> int {
    return len(ch["buffer"])
}

// -- Select --------------------------------------------------------------

/// Wait on multiple channels. Returns a map with:
///   "index": which channel fired
///   "value": the received value
///   "channel": the channel itself
///
/// Polls channels in order and returns the first with data.
/// If all channels are closed and empty, returns none.
pub fn select(channels: list) {
    loop {
        let all_closed = true
        for i in range(0, len(channels)) {
            let ch = channels[i]
            if !ch["closed"] || len(ch["buffer"]) > 0 {
                all_closed = false
            }
            if len(ch["buffer"]) > 0 {
                let value = channel_recv(ch)
                return {
                    "index":   i,
                    "value":   value,
                    "channel": ch,
                }
            }
        }
        if all_closed {
            return none
        }
        yield()
    }
}

/// Select with a timeout (in milliseconds).
/// Returns the select result, or none if timed out.
pub fn select_timeout(channels: list, timeout_ms: int) {
    let start = clock_ms()
    loop {
        let all_closed = true
        for i in range(0, len(channels)) {
            let ch = channels[i]
            if !ch["closed"] || len(ch["buffer"]) > 0 {
                all_closed = false
            }
            if len(ch["buffer"]) > 0 {
                let value = channel_recv(ch)
                return {
                    "index":   i,
                    "value":   value,
                    "channel": ch,
                }
            }
        }
        if all_closed {
            return none
        }
        if clock_ms() - start >= timeout_ms {
            return none
        }
        yield()
    }
}

// -- Fan-out / Fan-in ----------------------------------------------------

/// Fan-out: distribute items from a source channel to N worker channels.
/// Returns a list of worker channels.
pub fn fan_out(source: map, n: int, capacity: int) -> list {
    let workers = []
    for i in range(0, n) {
        workers = workers + [channel_new(capacity)]
    }

    // Round-robin distribution
    let idx = 0
    while !is_closed(source) || has_data(source) {
        let val = try_recv(source)
        if val != none {
            channel_send(workers[idx % n], val)
            idx = idx + 1
        } else if is_closed(source) {
            break
        } else {
            yield()
        }
    }

    // Close all worker channels
    for w in workers {
        channel_close(w)
    }
    return workers
}

/// Fan-in: merge multiple channels into one output channel.
/// Returns the merged channel.
pub fn fan_in(sources: list, capacity: int) -> map {
    let out = channel_new(capacity)

    let done = 0
    while done < len(sources) {
        done = 0
        for src in sources {
            if is_closed(src) && !has_data(src) {
                done = done + 1
            } else {
                let val = try_recv(src)
                if val != none {
                    channel_send(out, val)
                }
            }
        }
        if done < len(sources) {
            yield()
        }
    }

    channel_close(out)
    return out
}

// -- Pipeline ------------------------------------------------------------

/// Create a pipeline stage: reads from `input`, applies `transform`,
/// writes to a new output channel. Returns the output channel.
pub fn pipeline(input: map, transform, capacity: int) -> map {
    let out = channel_new(capacity)

    while !is_closed(input) || has_data(input) {
        let val = try_recv(input)
        if val != none {
            let result = transform(val)
            channel_send(out, result)
        } else if is_closed(input) {
            break
        } else {
            yield()
        }
    }

    channel_close(out)
    return out
}

/// Drain all remaining values from a channel into a list.
pub fn drain(ch: map) -> list {
    let result = []
    loop {
        let val = try_recv(ch)
        if val == none {
            break
        }
        result = result + [val]
    }
    return result
}

/// Collect all values from a channel (blocking) until it's closed.
pub fn collect(ch: map) -> list {
    let result = []
    loop {
        if is_closed(ch) && !has_data(ch) {
            break
        }
        let val = try_recv(ch)
        if val != none {
            result = result + [val]
        } else {
            yield()
        }
    }
    return result
}

// -- Statistics ----------------------------------------------------------

/// Get channel statistics.
pub fn stats(ch: map) -> map {
    return {
        "capacity":   ch["capacity"],
        "pending":    len(ch["buffer"]),
        "sent":       ch["send_count"],
        "received":   ch["recv_count"],
        "closed":     ch["closed"],
    }
}