/

module iter

module lib/iter.mu

import "iter"

iter provides lazy, pull-driven iterators built on Mu tasks and channels.

An iterator producer receives a yield function. Calling yield(value) publishes one value and suspends the producer until the consumer asks for another. yield returns false when the iterator has been cancelled; producers must return when that happens.

== Writing iterators: foreach and emit ==

Two macros carry the two halves of that protocol, so neither has to be written out by hand:

foreach(item, source, fn() { ... })   pull from source until it is dry
emit(yield, value)                    publish, or leave a cancelled producer

Both have to be macros rather than functions, because the return they perform belongs to the CALLER's function and a call cannot return on its caller's behalf -- the same reason try is a macro. See docs/Macros.md.

fn double(source) {
  return iterator(fn(yield) {
    foreach(item, source, fn() {
      emit(yield, item * 2)
    })
    return nil
  }, source)
}

== Cancelling ==

An iterator that is neither consumed to exhaustion nor stopped leaves its producer parked forever. That is not an error -- a task still blocked when main returns is abandoned work and the program exits normally (see docs/Specification.md §4.4) -- but it does hold the task and its channels until the program ends, so a consumer that stops early should say so:

value := next(source)
stop(source)

stop() cascades. Every combinator registers the iterators it reads as its UPSTREAM, so stopping the end of a chain stops all of it.

Imports

Functions

any#

fn any(source, predicate)

any returns true if predicate holds for any value.

Source lib/iter.mu:510
fn any(source, predicate) {
  foreach(item, source, fn() {
    if predicate(item) {
      stop(source)
      return true
    }
  })

  return false
}

chain#

fn chain(first, second)

chain yields all of first, then all of second.

Source lib/iter.mu:397
fn chain(first, second) {
  return concat(first, second)
}

collect#

fn collect(source)

collect consumes source and returns all yielded values as an array.

Source lib/iter.mu:456
fn collect(source) {
  return reduce(source, [], fn(acc, value) {
    return append(acc, value)
  })
}

combinations#

fn combinations(source, size)

combinations lazily yields every size-element combination from finite source.

Source is consumed once into a pool when the first value is requested. The combinations themselves are then generated lazily in lexicographic position order, so only O(len(source) + size) iterator state is retained.

Example:

combinations(from_array([1, 2, 3, 4]), 2)

yields:

[1, 2]
[1, 3]
[1, 4]
[2, 3]
[2, 4]
[3, 4]

size == 0 yields one empty combination. A negative size, or a size larger than the finite source, yields nothing.

Source lib/iter.mu:610
fn combinations(source, size) {
  return iterator(fn(yield) {
    if size < 0 {
      stop(source)
      return nil
    }

    if size == 0 {
      stop(source)
      emit(yield, [])
      return nil
    }

    // A generic iterator does not expose its length, so retain the finite
    // source once. Results remain lazy; we never materialise the combination
    // set itself.
    pool := collect(source)
    n := len(pool)

    if size > n {
      return nil
    }

    // The current combination is represented by increasing positions into
    // pool. Start with [0, 1, ..., size-1].
    indices := []
    i := 0
    while i < size {
      indices = append(indices, i)
      i = i + 1
    }

    while true {
      result := []
      i = 0
      while i < size {
        result = append(result, pool[indices[i]])
        i = i + 1
      }

      emit(yield, result)

      // Find the rightmost index that can still advance. For position i the
      // largest legal value is i + n - size.
      i = size - 1
      while i >= 0 && indices[i] == i + n - size {
        i = i - 1
      }

      if i < 0 {
        return nil
      }

      indices[i] = indices[i] + 1

      // Reset the suffix to the smallest strictly increasing continuation.
      j := i + 1
      while j < size {
        indices[j] = indices[j - 1] + 1
        j = j + 1
      }
    }
  }, source)
}

concat#

fn concat(sources...)

concat yields all values from each source iterator in order.

Sources are consumed lazily: the next source is not read until the current source is exhausted.

Example:

concat(
  from_array([1, 2]),
  from_array([3, 4]),
  from_array([5, 6])
)

yields:

1
2
3
4
5
6
Source lib/iter.mu:819
fn concat(sources...) {
  produce := fn(yield) {
    i := 0

    while i < len(sources) {
      foreach(item, sources[i], fn() {
        emit(yield, item)
      })

      i = i + 1
    }

    return nil
  }

  // The upstream list is assigned rather than passed, because sources is
  // already an array and a call cannot spread one back into a variadic. The
  // worker is parked on its first request until next(), and stop() reads the
  // list only when it runs, so setting it after construction is in time.
  it := iterator(produce)
  it["upstream"] = sources
  return it
}

count#

fn count(source)

count consumes source and returns the number of yielded values.

Source lib/iter.mu:463
fn count(source) {
  return reduce(source, 0, fn(n, _) {
    return n + 1
  })
}

drop#

fn drop(source, count)

drop skips count values, then yields the rest of source.

Source lib/iter.mu:342
fn drop(source, count) {
  return iterator(fn(yield) {
    skipped := 0

    foreach(item, source, fn() {
      if skipped < count {
        skipped = skipped + 1
        continue
      }

      emit(yield, item)
    })

    return nil
  }, source)
}

drop_while#

fn drop_while(source, predicate)

drop_while skips values while predicate is true, then yields the remainder.

Source lib/iter.mu:378
fn drop_while(source, predicate) {
  return iterator(fn(yield) {
    dropping := true

    foreach(item, source, fn() {
      if dropping && predicate(item) {
        continue
      }

      dropping = false

      emit(yield, item)
    })

    return nil
  }, source)
}

each#

fn each(source, f)

each consumes source, calling f for every value.

Source lib/iter.mu:440
fn each(source, f) {
  foreach(item, source, fn() {
    f(item)
  })
  return nil
}

enumerate#

fn enumerate(source, start...)

enumerate yields [index, value] pairs, starting at start (default 0).

Source lib/iter.mu:402
fn enumerate(source, start...) {
  first := 0
  if len(start) > 0 {
    first = start[0]
  }

  return iterator(fn(yield) {
    index := first

    foreach(item, source, fn() {
      emit(yield, [index, item])
      index = index + 1
    })

    return nil
  }, source)
}

every#

fn every(source, predicate)

every returns true only if predicate holds for every value.

Source lib/iter.mu:522
fn every(source, predicate) {
  foreach(item, source, fn() {
    if !predicate(item) {
      stop(source)
      return false
    }
  })

  return true
}

filter#

fn filter(source, predicate)

filter lazily yields values from source that satisfy predicate.

Source lib/iter.mu:304
fn filter(source, predicate) {
  return iterator(fn(yield) {
    foreach(item, source, fn() {
      if predicate(item) {
        emit(yield, item)
      }
    })
    return nil
  }, source)
}

find#

fn find(source, predicate)

find returns the first value satisfying predicate, or nil if none does.

Source lib/iter.mu:498
fn find(source, predicate) {
  foreach(item, source, fn() {
    if predicate(item) {
      stop(source)
      return item
    }
  })

  return nil
}

from_array#

fn from_array(arr)

from_array lazily yields the values in arr.

Source lib/iter.mu:214
fn from_array(arr) {
  return iterator(fn(yield) {
    i := 0
    while i < len(arr) {
      emit(yield, arr[i])
      i = i + 1
    }
    return nil
  })
}

iterate#

fn iterate(seed, transform)

iterate yields seed, transform(seed), transform(transform(seed)), ... forever.

Source lib/iter.mu:235
fn iterate(seed, transform) {
  return iterator(fn(yield) {
    value := seed
    while true {
      emit(yield, value)
      value = transform(value)
    }
  })
}

iterator#

fn iterator(produce, upstream...)

iterator creates a lazy iterator from produce(yield).

Iterators passed after produce are this one's upstream: stop() cancels them too. They have to be declared here rather than cancelled from inside the producer body, because the body has not run yet when the iterator is stopped before its first next() -- a stop(source) written in there would simply never execute, and the source's task would be left parked with nothing to free it.

Source lib/iter.mu:112
fn iterator(produce, upstream...) {
  requests := chan()
  responses := chan()

  worker := task(fn() {
    // Do not run the producer until the first value is requested.
    demand := recv(requests)
    if !demand {
      return nil
    }

    cancelled := false

    yield := fn(value) {
      if cancelled {
        return false
      }

      // Answer the outstanding next() request.
      send(responses, [true, value])

      // Suspend until the consumer requests another value or cancels.
      demand = recv(requests)
      if !demand {
        cancelled = true
        return false
      }

      return true
    }

    result := produce(yield)

    // If the producer ended normally, answer the outstanding next() request
    // with the end-of-stream marker. Cancellation needs no response because
    // stop() waits directly for the worker.
    if !cancelled {
      send(responses, [false, nil])
    }

    return result
  })

  return {
    "requests": requests,
    "responses": responses,
    "worker": worker,
    "upstream": upstream,
    "done": false
  }
}

map#

fn map(source, transform)

map lazily transforms every value from source.

Source lib/iter.mu:294
fn map(source, transform) {
  return iterator(fn(yield) {
    foreach(item, source, fn() {
      emit(yield, transform(item))
    })
    return nil
  }, source)
}

next#

fn next(it)

next returns [true, value], or [false, nil] at end of stream.

Source lib/iter.mu:165
fn next(it) {
  if it["done"] {
    return [false, nil]
  }

  send(it["requests"], true)
  item := recv(it["responses"])

  if !item[0] {
    it["done"] = true
    wait(it["worker"])
  }

  return item
}

nth#

fn nth(source, index)

nth returns the zero-based nth value and cancels the remaining source.

Source lib/iter.mu:477
fn nth(source, index) {
  if index < 0 {
    stop(source)
    return error("iter.nth expects a non-negative index")
  }

  i := 0

  foreach(item, source, fn() {
    if i == index {
      stop(source)
      return item
    }

    i = i + 1
  })

  return error("iter.nth index is past the end of the sequence")
}

once#

fn once(value)

once yields exactly one value.

Source lib/iter.mu:206
fn once(value) {
  return iterator(fn(yield) {
    emit(yield, value)
    return nil
  })
}

permutations#

fn permutations(source, size...)

permutations lazily yields every ordered selection from finite source.

If size is omitted, full-length permutations are produced. Elements are selected by source position, so duplicate source values can produce duplicate-looking permutations, matching the usual positional semantics. Source is consumed once into a pool on first demand; permutation results are generated lazily with O(len(source) + size) iterator state.

Example:

permutations(from_array([1, 2, 3]), 2)

yields:

[1, 2]
[1, 3]
[2, 1]
[2, 3]
[3, 1]
[3, 2]

size == 0 yields one empty permutation. A negative size, or a size larger than the finite source, yields nothing.

Source lib/iter.mu:698
fn permutations(source, size...) {
  requested := nil
  if len(size) > 0 {
    requested = size[0]
  }

  return iterator(fn(yield) {
    // An explicit zero/negative size needs no values from source at all.
    if requested != nil && requested < 0 {
      stop(source)
      return nil
    }

    if requested == 0 {
      stop(source)
      emit(yield, [])
      return nil
    }

    pool := collect(source)
    n := len(pool)

    r := n
    if requested != nil {
      r = requested
    }

    if r > n {
      return nil
    }

    // The sole full permutation of an empty source is the empty permutation.
    if r == 0 {
      emit(yield, [])
      return nil
    }

    // This is the positional indices/cycles algorithm used for lexicographic
    // r-permutations. indices holds a permutation of source positions; cycles
    // controls when each prefix position rotates or swaps.
    indices := []
    i := 0
    while i < n {
      indices = append(indices, i)
      i = i + 1
    }

    cycles := []
    i = 0
    while i < r {
      cycles = append(cycles, n - i)
      i = i + 1
    }

    while true {
      result := []
      i = 0
      while i < r {
        result = append(result, pool[indices[i]])
        i = i + 1
      }

      emit(yield, result)

      i = r - 1
      advanced := false

      while i >= 0 {
        cycles[i] = cycles[i] - 1

        if cycles[i] == 0 {
          // Rotate indices[i:] one position to the left.
          first := indices[i]
          j := i
          while j < n - 1 {
            indices[j] = indices[j + 1]
            j = j + 1
          }
          indices[n - 1] = first
          cycles[i] = n - i
          i = i - 1
        } else {
          // Swap the current position with the element selected from the end
          // by the remaining cycle count.
          j := n - cycles[i]
          tmp := indices[i]
          indices[i] = indices[j]
          indices[j] = tmp
          advanced = true
          break
        }
      }

      if !advanced {
        return nil
      }
    }
  }, source)
}

range#

fn range(start, end, step...)

range lazily yields start (inclusive) to end (exclusive), stepping by step.

Source lib/iter.mu:264
fn range(start, end, step...) {
  step_value := 1
  if len(step) > 0 {
    step_value = step[0]
  }

  return iterator(fn(yield) {
    if step_value == 0 {
      return nil
    }

    value := start

    if step_value > 0 {
      while value < end {
        emit(yield, value)
        value = value + step_value
      }
    } else {
      while value > end {
        emit(yield, value)
        value = value + step_value
      }
    }

    return nil
  })
}

reduce#

fn reduce(source, acc, combiner)

reduce consumes source from the left using acc and combiner.

Source lib/iter.mu:448
fn reduce(source, acc, combiner) {
  foreach(item, source, fn() {
    acc = combiner(acc, item)
  })
  return acc
}

repeat#

fn repeat(value)

repeat yields value forever.

Source lib/iter.mu:226
fn repeat(value) {
  return iterator(fn(yield) {
    while true {
      emit(yield, value)
    }
  })
}

sliding#

fn sliding(source, size)

sliding lazily yields overlapping windows of size values.

Example:

sliding(from_array([1, 2, 3, 4, 5]), 3)

yields:

[1, 2, 3]
[2, 3, 4]
[3, 4, 5]

Each yielded window is a fresh array. If source contains fewer than size values, no windows are yielded.

Source lib/iter.mu:559
fn sliding(source, size) {
  return iterator(fn(yield) {
    if size <= 0 {
      stop(source)
      return nil
    }

    window := []

    foreach(item, source, fn() {
      // Still filling the first window: nothing to yield until it is full.
      if len(window) < size {
        window = append(window, item)

        if len(window) < size {
          continue
        }

        emit(yield, window)
        continue
      }

      window = _slide(window, item)
      emit(yield, window)
    })

    return nil
  }, source)
}

stop#

fn stop(it)

stop cancels an iterator and everything upstream of it. It is safe to call more than once, and safe to call on an iterator that was never started.

Source lib/iter.mu:183
fn stop(it) {
  if it["done"] {
    return nil
  }

  it["done"] = true
  send(it["requests"], false)
  result := wait(it["worker"])
  _stop_upstream(it)
  return result
}

sum#

fn sum(source)

sum consumes source and adds all yielded values.

Source lib/iter.mu:470
fn sum(source) {
  return reduce(source, 0, fn(total, value) {
    return total + value
  })
}

take#

fn take(source, count)

take yields at most count values, then cancels source.

Source lib/iter.mu:316
fn take(source, count) {
  return iterator(fn(yield) {
    if count <= 0 {
      stop(source)
      return nil
    }

    taken := 0

    foreach(item, source, fn() {
      emit(yield, item)
      taken = taken + 1

      if taken >= count {
        // Taken all we were asked for: source is free to go, even though
        // nobody has cancelled us.
        stop(source)
        return nil
      }
    })

    return nil
  }, source)
}

take_while#

fn take_while(source, predicate)

take_while yields values while predicate remains true, then cancels source.

Source lib/iter.mu:360
fn take_while(source, predicate) {
  return iterator(fn(yield) {
    foreach(item, source, fn() {
      if !predicate(item) {
        // The predicate decided we are done, so nothing downstream will ask
        // for more: release source now rather than waiting to be stopped.
        stop(source)
        return nil
      }

      emit(yield, item)
    })

    return nil
  }, source)
}

unfold#

fn unfold(state, step)

unfold builds an iterator from evolving state. step(state) must return [value, next_state], or nil to end the sequence.

Source lib/iter.mu:247
fn unfold(state, step) {
  return iterator(fn(yield) {
    current := state
    while true {
      item := step(current)
      if item == nil {
        return nil
      }

      emit(yield, item[0])

      current = item[1]
    }
  })
}

zip#

fn zip(left, right)

zip yields pairs until either iterator is exhausted.

Source lib/iter.mu:421
fn zip(left, right) {
  return iterator(fn(yield) {
    foreach(a, left, fn() {
      b := next(right)
      if !b[0] {
        stop(left)
        return nil
      }

      emit(yield, [a, b[1]])
    })

    // Left ran out, so right can never be paired again.
    stop(right)
    return nil
  }, left, right)
}

Macros

emit#

macro emit(yield, value)

emit(yield, value) publishes one value, and returns from the producer if the consumer has cancelled the iterator.

It expands to the guard every producer owes its consumer:

if !yield(value) { return nil }

The producer's yield function is named rather than reached for. A quoted free yield would in fact resolve at the call site and work, but hygiene is absolute here for a reason: a macro that silently captures a name the caller never mentioned changes meaning the moment that name means something else. Naming it also keeps mu -check honest -- the checker runs before macros are expanded, so a yield that only an expansion used would be reported as an unused parameter in every producer anyone writes.

Source lib/iter.mu:97
macro.define("emit", fn(yield, value) {
  return quote {
    if !(unquote yield)(unquote value) {
      return nil
    }
  }
})

foreach#

macro foreach(name, source, body)

foreach(name, source, fn() { ... }) pulls values from source, binding each one to name, until source is exhausted.

The body's statements are SPLICED into the caller rather than called, which is what makes return, break and continue mean what they look like: return leaves the enclosing function, break leaves the loop, continue moves to the next value. A callback could offer none of the three, which is why iter.each cannot replace this.

fn first_even(source) {
  foreach(value, source, fn() {
    if value % 2 == 0 {
      stop(source)
      return value          // returns from first_even
    }
  })
  return nil
}

The source expression is bound to a temporary FIRST and pulled from after, so it is evaluated exactly once. Splicing it into the loop instead would re-evaluate it on every pass -- foreach(x, from_array(xs), ...) would build a fresh iterator each time round and hand back its first value forever.

Source lib/iter.mu:68
macro.define("foreach", fn(name, source, body) {
  stmts := body["body"]["statements"]
  return quote {
    src := unquote source
    while true {
      pulled := next(src)
      if !pulled[0] {
        break
      }
      unquote name := pulled[1]
      unquote stmts...
    }
  }
})

Internal helpers

Underscore-prefixed names are implementation detail. They are listed so the module's source reads without surprises, not as API — they may change at any time.

_slide(window, value)_slide returns a fresh array holding window without its first element, plus value.
_stop_upstream(it)—