E10050: this operation needs a bounded source; an unbounded source (Stdin, UDP, TCPConnection) or an infinite one (an iterator whose hasNext() always returns true) must be bounded first with 'head N'

Category: E10xxx IR generation · Reported in phase: FULL_RESOLUTION · Classification: BUFFERING_REQUIRES_BOUNDED · All error codes

What went wrong

A buffering stream operation was used on an UNBOUNDED source. Buffering operations must see (or accumulate state over) ALL of the input before they can produce output, so they require a BOUNDED (finite) source. An unbounded source is an infinite IO stream — Stdin, a UDP socket, or a TCPConnection — that has no defined end, so a buffering stage would consume memory without bound and never emit. The message names the source and the stage: 'unbounded source: 'stdin' is a Stdin, which never ends, so 'sort' would wait for an end that never comes'.

Buffering operations that require a bounded source:

  • sort: must see all items to order them
  • group: must see all items of each (sorted) key run
  • tail: must see all items to know the last N
  • uniq: accumulates a set of seen keys without bound
  • join: folds every item before it emits its single result
  • split: reads all of its input before emitting its groups (a 'head' AFTER join or split cannot stop them; put it before)

Worked examples

The [EXAMPLE] section below shows 'cat stdin | sort > stdout' (this error), and the same pipeline with 'head MAX_LINES' bounding the stream before the sort. Reading in bounded windows works the same way: collect 'head 100' of the source into a List, then sort that List.

Head converts unbounded to bounded

'head N' takes the first N items and is the mechanism that bounds an otherwise unbounded stream. After a 'head', buffering operations are allowed.

Streaming operations are fine on unbounded sources

Operations that process each item as it arrives - filter, reject, map, flatten, call, async, skip, and tee into an output stream - do NOT require a bounded source and compose freely with Stdin/UDP/TCP:

//Each item is processed as it arrives: no bound is needed
cat stdin | filter by validLine | map with shout > stdout

Infinite sources

A source the compiler can PROVE never ends is refused the same way (reported in PRE_IR_CHECKS, with the message starting 'infinite source:' and saying why):

  • a class iterated directly whose hasNext() always returns true
  • a hasNext() that returns a Boolean property declared true and never changed
  • an iterator() or a generator function that always returns such an iterator
  • a multi-source 'cat a, b' where ANY source is infinite

Only an operation that waits for the END of the stream is refused - and on an infinite source that end never comes, so a later 'head' cannot help: sort, tail, join, group, split, the stage after 'async | head N', '| collect as ...', and '>' into anything that keeps what it is given (a collection; '> stdout' writes through and is fine). uniq and tee into a list are NOT refused on an infinite source: they stream, and a later 'head' ends the pipeline (cat Forever() | uniq | head 2 terminates). 'cat Forever() | sort > stdout' is this error; a 'head' BEFORE the waiting stage bounds it:

//'head' bounds the never-ending source before anything waits for its end
cat Forever() | head 10 | sort > stdout
firstTen <- cat Forever() | head 10 | collect as List of Integer

A class declared 'as open' is never proved infinite (a subclass may override hasNext()), and nor is any source the compiler cannot prove - those keep the rules above.

How to fix it

  1. Add '| head N' before the buffering operation to bound the stream, OR
  2. Replace the buffering operation with a streaming one (filter/map/tee), OR
  3. Read the source in bounded windows (cat source | head N | collect ...).

See also

  • E10050: buffering operation requires a bounded source (this error)
  • E10040: group requires preceding sort
  • E10020: void in stream pipeline

Examples

Every example is compiled by the EK9 compiler: the first fails exactly as captioned, the others fix it.

Fails with E10050: this operation needs a bounded source; an unbounded source (Stdin, UDP, TCPConnection) or an infinite one (an iterator whose hasNext() always returns true) must be bounded first with 'head N'

#!ek9
defines module error.e10050.bad

  defines program

    SortInput()
      stdin <- Stdin()
      stdout <- Stdout()
      cat stdin | sort > stdout

Fixed: Bound the source with head first

#!ek9
defines module error.e10050.good

  defines constant

    MAX_LINES <- 1000

  defines program

    SortInput()
      stdin <- Stdin()
      stdout <- Stdout()
      cat stdin | head MAX_LINES | sort > stdout

Fixed: Streaming stages need no bound; 'head' bounds even a source that never ends

#!ek9
defines module error.e10050.good.streaming

  defines class
    //hasNext() always answers true: the compiler proves this source never ends
    Forever
      count <- 0
      hasNext() as pure
        <- rtn as Boolean: true
      next()
        <- rtn as Integer: Integer(count)
        count++
      default operator ?

  defines function
    validLine() as pure
      -> line as String
      <- rtn as Boolean: not line.empty()

    shout() as pure
      -> line as String
      <- rtn as String: line.upperCase()

  defines program
    EchoInput()
      stdin <- Stdin()
      stdout <- Stdout()
      //Each item is processed as it arrives: no bound is needed
      cat stdin | filter by validLine | map with shout > stdout

    FirstTen()
      stdout <- Stdout()
      //'head' bounds the never-ending source before anything waits for its end
      cat Forever() | head 10 | sort > stdout
      firstTen <- cat Forever() | head 10 | collect as List of Integer
      stdout.println($firstTen)

Questions and answers

These Q&A entries show E10050 and its fix: