coroutine xxx apply {{} {
    set id1 [async::exec sort]
    chan puts $id1 b\na\nc
    set id2 [async::exec sort]
    async::close -direction stdin $id1
    chan puts $id2 z\na\nc
    async::close -direction stdin $id2
    chan puts [async::get $id2]
    chan puts [async::read -all -dropNewline $id2]
    chan puts [async::read -all -dropNewline $id1]
    async::close $id2
    async::close $id1
    set ::end 1
}}
vwait end

% [pipeline {exec rev | tac}] flow -shutdown -all foo\nbar\n
rab
oof

% [pipeline {exec rev} {exec tac}] flow -shutdown -all foo\nbar\n

% [pipeline {loop -buffer -trim -command string reverse} {exec tac}] flow -shutdown -all foo\nbar\n

# async.tcl
# Andy Goth <[email protected]>

# Require [dict], [throw], [chan pipe], [chan close dir], and [coroutine].
package require Tcl 8.6

# Create namespace.
namespace eval ::async {}

# Asynchronous channel internal data.
set ::async::Channels {}

# ::async::exec --
# Executes a child process command and returns a channel identifier which may be
# used to asynchronously communicate with the child process via [async::*] and
# [chan].  The valid [chan] subcommands are: puts, flush, copy, configure, push,
# and pop.  With [chan copy], the channel identifier may only be the outputChan
# argument.
#
# The channel is configured to be nonblocking with binary translation and line
# buffering.  Do not change the blocking mode using [chan configure].  To block,
# instead omit the -noWait switch to [async::get] and [async::read].
#
# If [async::get] or [async::read] produce an empty result and are given both
# -eof empty and -noWait, [chan eof] is necessary to disambiguate between end of
# file versus a complete line or block not being immediately available.  This
# only works when reading from stdout, not from stderr.  [chan eof] may not be
# used outside of this specific circumstance.
proc ::async::exec {args} {
    variable Channels

    # Check if stdin, stdout, and/or stderr are being redirected.
    foreach arg $args {
        if {[info exists redir(next)]} {
            unset redir(next)
        } elseif {[regexp {^(<[<@]?|>[>&]?|>&?@|>>&|2>[>@]?)(.*)} $arg _\
                opcode operand]} {
            switch $opcode {
                <  - <<  - <@  {set redir(stdin)  {}}
                >  - >>  - >@  {set redir(stdout) {}}
                >& - >>& - >&@ {set redir(stdout) {}; set redir(stderr) {}}
                2> - 2>> - 2>@ {set redir(stderr) {}}
            }
            if {$operand eq {}} {
                set redir(next) {}
            }
        }
    }

    # Determine the access mode depending on the redirections.
    if {[info exists redir(stdin)]} {
        set mode r
    } elseif {[info exists redir(stdout)]} {
        set mode w
    } else {
        set mode a+
    }

    # Unless stderr is redirected, collect stderr from the child process.
    if {![info exists redir(stderr)]} {
        lassign [chan pipe] chanStderr chanStderrWrite
        lappend args 2>@ $chanStderrWrite
    }

    # Execute the child process pipeline.
    set chanInOut [open |$args $mode]
    if {![info exists redir(stderr)]} {
        chan close $chanStderrWrite
    }

    # Configure stdin/stdout and stderr channels.
    chan configure $chanInOut -blocking 0 -translation binary -buffering line
    chan configure $chanStderr -blocking 0 -translation binary -buffering line

    # Initialize the channel buffers.
    dict set Channels $chanInOut {open {} eof {} buf {stdout {} stderr {}}}
    foreach channel {stdin stdout stderr} {
        if {![info exists redir($channel)]} {
            dict set Channels $chanInOut open $channel {}
        }
    }
    dict set Channels $chanInOut chan stdout $chanInOut
    if {![info exists redir(stderr)]} {
        dict set Channels $chanInOut chan stderr $chanStderr
    }

    # Return the input/output channel to the caller.
    return $chanInOut
}

# ::async::channels --
# Returns a list of all currently open asynchronous channel IDs.
proc ::async::channels {} {
    variable Channels
    dict keys $Channels
}

# ::async::close --
# Closes a currently open asynchronous channel.  A channel may be partially
# closed, for example to send the sort(1) program EOF on its stdin after which
# it will start writing results to its stdout.  Unread data will be discarded,
# possibly except for stderr.  If stderr is being closed and -stderr ignore was
# not used, any unread stderr data is thrown with ASYNC STDERR.
#
# async::close ?-direction all|stdin|stdout|stderr? ?-stderr ignore|throw? id
# -direction all: Fully closes channel.  The default.
# -direction stdin: Closes stdin component of channel.
# -direction stdout: Closes stdout component of channel.
# -direction stderr: Closes stderr component of channel.
# -stderr ignore: Ignores any unread data on stderr.
# -stderr throw: Throws any stderr data with ASYNC STDERR.  The default.
# id: The asynchronous channel identifier returned by [async::exec].
proc ::async::close {args} {
    variable Channels

    # Parse arguments.
    set direction all
    set stderrMode throw
    while {$args ne {}} {
        set args [lassign $args arg]
        if {[info exists id]} {
            return -code error "wrong # args: should be \"async::close\
                    ?-direction all|stdin|stdout|stderr?\
                    ?-stderr ignore|throw? id"
        } elseif {$arg eq "-direction"} {
            if {$args eq {} || [lindex $args 0] ni {all stdin stdout stderr}} {
                return -code error "-direction switch must be followed by:\
                        all, stdin, stdout, or stderr"
            }
            set args [lassign $args direction]
        } elseif {$arg eq "-stderr"} {
            if {$args eq {} || [lindex $args 0] ni {ignore throw}} {
                return -code error "-stderr switch must be followed by:\
                        ignore or throw"
            }
            set args [lassign $args stderrMode]
        } elseif {![dict exists $Channels $arg]} {
            return -code error "not an open asynchronous channel: $arg"
        } else {
            set id $arg
        }
    }

    # When closing stderr and using -stderr throw, check for unread stderr.
    if {$stderrMode eq "throw" && $direction in {all stderr}} {
        flow $id stderr
        set stderrData [dict get $Channels $id buf stderr]
    }

    if {$direction eq "all"} {
        # If requested, close all open channels.
        if {[dict exists $Channels $id $open stderr]} {
            chan close [dict get $Channels $id chan stderr]
        }
        if {[dict exists $Channels $id $open stdin]
         || [dict exists $Channels $id $open stdout]} {
            chan close [dict get $Channels $id chan stdout]
        }
        dict unset Channels $id
    } elseif {![dict exists $Channels $id $open $direction]} {
        # Check for double close.
        return -code error "direction \"$direction\" already closed for\
                asynchronous channel: $id"
    } else {
        # Close only the requested direction.
        if {$direction eq "stdin"} {
            chan close [dict get $Channels $id chan stdout] write
        } elseif {$direction eq "stdout"} {
            chan close [dict get $Channels $id chan stdout] read
        } else {
            chan close [dict get $Channels $id chan stderr]
        }

        # Clean up the asynchronous channel data as appropriate.
        if {$open eq {}} {
            dict unset Channels $id
        } else {
            dict unset Channels $id open $direction
        }
    }

    # Throw any unread stderr data.
    if {[info exists stderrData] && $stderrData ne {}} {
        throw {ASYNC STDERR} $stderrData
    }
}

# ::async::get --
# Gets one line of text from the stdout and/or stderr of an asynchronous child
# process, according to the value of the -stderr switch, which defaults to
# -stderr throw.
#
# The trailing newline is stripped, unless the -keepNewline switch is used.
#
# Yields until the requested data is available, unless -noWait is used.
#
# The result is returned or written into a variable, in which case the number of
# characters read is returned.  In the case of -noWait with a variable, -1 is
# returned rather than 0 if the requested data is not immediately available.
#
# If EOF is reached on the channel (or both the channels) being read from, and
# -eof throw is used, ASYNC EOF is thrown.  With -stderr ignore, EOF on stdout
# causes ASYNC EOF to be thrown.  With -stderr read, EOF on stderr causes ASYNC
# EOF to be thrown.  In all other cases, stdout and stderr must both be at EOF
# (or be closed) for ASYNC EOF to be thrown.
#
# At end of file, the last line of stdout or stderr is considered complete even
# if it does not end in a newline.  When called at end of file after all data
# has already been read and returned, the -eof switch determines whether the
# result is empty string or (the default behavior) ASYNC EOF is thrown.
#
# When using -stderr ignore or throw (the default), the closure of stderr does
# not result in ASYNC EOF being thrown.  When using -stderr read and -eof throw,
# stderr closing does cause ASYNC EOF to be thrown, but stdout closing does not.
# When using -stderr merge or tag and -eof throw, ASYNC EOF is thrown when both
# stdout and stderr close.
#
# If -eof empty and -noWait are both used, [chan eof] must be used to determine
# if an empty result signifies end of file or the data not being available.
# [chan eof] can only be used to check the EOF status of stdout, not stderr.
#
# async::get ?-noWait? ?-keepNewline? ?-variable varName? ?-stderr mode?
#       ?-eof mode? id
# -noWait: Returns immediately rather than yield if data is not available.
# -keepNewline: Does not strip the trailing newline.
# -variable varName: Name of variable into which to write the result.
# -stderr ignore: Ignores stderr and reads only stdout.
# -stderr merge: Reads from either stderr or stdout, whichever is available.
# -stderr tag: Like merge, except the return value is a tagged list.
# -stderr read: Ignores stdout and reads only stderr.
# -stderr throw: Returns stdout normally and throws stderr with ASYNC STDERR.
# -eof empty: If at end of file with no available data, result is empty string.
# -eof throw: If at end of file with no available data, throw ASYNC EOF.
# id: The asynchronous channel identifier returned by [async::exec].
proc ::async::get {args} {
    variable Channels

    # Parse arguments.
    set stderrMode throw
    set eofMode throw
    while {$args ne {}} {
        set args [lassign $args arg]
        if {[info exists id]} {
            return -code error "wrong # args: should be \"async::get\
                    ?-noWait? ?-keepNewline? ?-variable varName?\
                    ?-stderr ignore|merge|tag|read|throw? id\""
        } elseif {$arg in {-noWait -keepNewline}} {
            set [string range $arg 1 end] {}
        } elseif {$arg eq "-variable"} {
            set args [lassign $args varName]
        } elseif {$arg eq "-stderr"} {
            if {$args eq {}
             || [lindex $args 0] ni {ignore merge tag read throw}} {
                return -code error "-stderr switch must be followed by\
                        ignore, merge, tag, read, or throw"
            }
            set args [lassign $args stderrMode]
        } elseif {$arg eq "-eof"} {
            if {$args eq {} || [lindex $args 0] ni {empty throw}} {
                return -code error "-eof switch must be followed by\
                        empty or throw"
            }
            set args [lassign $args eofMode]
        } elseif {![dict exists $Channels $arg]} {
            return -code error "not an open asynchronous channel: $arg"
        } else {
            set id $arg
        }
    }

    # Confirm the requested directions are open.
    if {![dict exists $Channels $id open stdout]
     && ![dict exists $Channels $id open stderr]} {
        return -code error "both stdout and stderr closed for asynchronous\
                channel: $id"
    } elseif {$stderrMode eq "ignore"
           && ![dict exists $Channels $id open stdout]} {
        return -code error "stdout closed for asynchronous channel: $id"
    } elseif {$stderrMode eq "read"
           && ![dict exists $Channels $id open stderr]} {
        return -code error "stderr closed for asynchronous channel: $id"
    }

    # Loop until a complete line is available.  Return if -noWait was given and
    # both stdout and stderr are empty, incomplete, or unwanted.  At EOF, the
    # remainder of the buffer is considered to be complete even if it does not
    # end in newline.
    set rest {}
    while {1} {
        # First try reading from stderr, then stdout.
        foreach {dir skipMode} {stderr ignore stdout read} {
            # Check if this direction is open, not being ignored, and has a
            # complete line or has an incomplete line at end of file.
            set line [dict get $Channels $id buf $dir]
            if {$stderrMode ne $skipMode
             && [dict exists $Channels $id open $dir]
             && ([regexp {([^\n]*\n)(.*)} $line _ line rest]
              || ([dict exists $Channels $id eof $dir] && $line ne {}))} {
                # Remove the data from the buffer.
                dict set Channels $id buf $dir $rest

                # Unless -keepNewline, strip trailing newline.
                if {![info exists keepNewline]} {
                    regsub {\n$} $line {} line
                }

                # If -stderr throw (the default) and a complete line was read
                # from stderr, throw it rather than return it.
                if {$dir eq "stderr" && $stderrMode eq "throw"} {
                    throw {ASYNC STDERR} $line
                }

                # Build the result.  With -stderr tag, the result is a list
                # consisting of "stdout" or "stderr" followed by the received
                # data.  Otherwise, the result is the data.  -stderr merge is
                # the only case where the caller won't be able to tell where the
                # data came from.
                if {$stderrMode eq "tag"} {
                    set result [list $dir $line]
                } else {
                    set result $line
                }

                # Give the result to the caller.
                if {[info exists varName]} {
                    uplevel 1 [list ::set $varName $result]
                    return [string length $line]
                } else {
                    return $result
                }
            }
        }

        # Check if at EOF, and in which direction.
        set eofDir [switch $stderrMode {
        ignore  {if {[dict exists $Channels $id eof stdout]} {list stdout}}
        read    {if {[dict exists $Channels $id eof stderr]} {list stderr}}
        default {
            if {(![dict exists $Channels $id open stdout]
              || [dict exists $Channels $id eof stdout])
             && (![dict exists $Channels $id open stderr]
              || [dict exists $Channels $id eof stderr])} {
                if {[dict exists $Channels $id open stdout]} {
                    list stdout
                } else {
                    list stderr
                }
            }
        }}]

        # At this point, stderr and stdout are both unwanted or unavailable.
        if {$eofDir eq {} && ![info exists noWait]} {
            # If not at EOF and not -noWait, fill the buffers and retry.
            fill $id
        } elseif {$eofDir ne {} && $eofMode eq "throw"} {
            # At end of file with -eof throw, throw an exception.
            throw {ASYNC EOF} [dict get $Channels $id eof $eofDir]
        } elseif {[info exists varName]} {
            # If -eof empty is used at end of file, or if -noWait is used
            # when no data is available, and if -variable is used, set the
            # caller's variable to empty string and return -1.  The caller
            # must use [chan eof] to disambiguate.
            uplevel 1 [list ::set $varName {}]
            return -1
        } else {
            # As above, but without -variable.  Return empty.
            return
        }
    }
}

# ::async::read --
# Gets a block of data from the stdout and/or stderr of an asynchronous child
# process, according to the value of the -stderr switch, which defaults to
# -stderr throw.
#
# Yields until the requested data is available, unless -noWait is used.
#
# When called at end of file after all data has already been read and returned,
# the -eof switch determines whether empty string is returned or (the default
# behavior) ASYNC EOF is thrown.  By default, returns all data that is
# immediately available, yielding until at least one byte is available, but the
# -noWait, -all, -size, -min, and -max switches may be used to control the
# minimum and maximum amount of data that is returned.
#
# If EOF is reached on the channel (or both the channels) being read from, and
# -eof throw is used, ASYNC EOF is thrown.  With -stderr ignore, EOF on stdout
# causes ASYNC EOF to be thrown.  With -stderr read, EOF on stderr causes ASYNC
# EOF to be thrown.  In all other cases, stdout and stderr must both be at EOF
# (or be closed) for ASYNC EOF to be thrown.
#
# If -eof empty and -noWait are both used, [chan eof] must be used to determine
# if an empty result signifies end of file or the data not being available.
# [chan eof] can only be used to check the EOF status of stdout, not stderr.
#
# async::read ?-noWait? ?-dropNewline? ?-all? ?-size size? ?-min size?
#       ?-max size? ?-stderr mode? ?-eof mode? id
# -noWait: Returns immediately rather than yield if data is not available.
# -dropNewline: Strips the trailing newline at end of file, requires -all.
# -all: Reads until end of file, conflicts with -size, -min, and -max.
# -size size: Sets -min and -max to the same value.
# -min size: Minimum number of characters to read and return, default 1.
# -max size: Maximum number of characters to read and return, default unlimited.
# -stderr ignore: Ignores stderr and reads only stdout.
# -stderr merge: Reads from either stderr or stdout, whichever is available.
# -stderr tag: Like merge, except the return value is a tagged list.
# -stderr read: Ignores stdout and reads only stderr.
# -stderr throw: Returns stdout normally and throws stderr with ASYNC STDERR.
# -eof empty: If at end of file with no available data, result is empty string.
# -eof throw: If at end of file with no available data, throw ASYNC EOF.
# id: The asynchronous channel identifier returned by [async::exec].
proc ::async::read {args} {
    variable Channels

    # Parse arguments.
    set stderrMode throw
    set eofMode throw
    while {$args ne {}} {
        set args [lassign $args arg]
        if {[info exists id]} {
            return -code error "wrong # args: should be \"async::read\
                    ?-noWait? ?-dropNewline? ?-all? ?-size size? ?-min size?\
                    ?-max size? ?-stderr ignore|merge|tag|read|throw? id\""
        } elseif {$arg in {-noWait -dropNewline -all}} {
            set [string range $arg 1 end] {}
        } elseif {$arg in {-size -min -max}} {
            if {$args eq {}
             || ![string is entier [lindex $args 0]] || [lindex $args 0] <= 0} {
                return -code error "$arg switch must be followed by\
                        a positive integer"
            }
            set args [lassign $args [string range $arg 1 end]]
        } elseif {$arg eq "-stderr"} {
            if {![llength $args]
             || [lindex $args 0] ni {ignore merge tag read throw}} {
                return -code error "-stderr switch must be followed by\
                        ignore, merge, tag, read, or throw"
            }
            set args [lassign $args stderrMode]
        } elseif {$arg eq "-eof"} {
            if {$args eq {} || [lindex $args 0] ni {empty throw}} {
                return -code error "-eof switch must be followed by\
                        empty or throw"
            }
            set args [lassign $args eofMode]
        } elseif {![dict exists $Channels $arg]} {
            return -code error "not an open asynchronous channel: $arg"
        } else {
            set id $arg
        }
    }

    # Process -all, -dropNewline, -size, -min, and -max.
    if {[info exists all]} {
        foreach var {size min max} {
            if {[info exists $var]} {
                return -code error "-all conflicts with -$var"
            }
        }
    } elseif {[info exists dropNewline]} {
        return -code error "-dropNewline requires -all"
    } elseif {[info exists size]} {
        set min $size
        set max $size
    } elseif {[info exists min] && [info exists max] && $max < $min} {
        return -code error "-max cannot be less than -min"
    } elseif {![info exists min]} {
        set min 1
    }

    # Process -all is specified when -dropNewline is used.
    if {[info exists dropNewline] && ![info exists all]} {
    }

    # Confirm the requested directions are open.
    if {![dict exists $Channels $id open stdout]
     && ![dict exists $Channels $id open stderr]} {
        return -code error "both stdout and stderr closed for asynchronous\
                channel: $id"
    } elseif {$stderrMode eq "ignore"
           && ![dict exists $Channels $id open stdout]} {
        return -code error "stdout closed for asynchronous channel: $id"
    } elseif {$stderrMode eq "read"
           && ![dict exists $Channels $id open stderr]} {
        return -code error "stderr closed for asynchronous channel: $id"
    }

    # Loop until a complete block is available.  Return if -noWait was given and
    # both stdout and stderr are empty, incomplete, or unwanted.  At end of
    # file, the remainder of the buffer is considered to be complete even if its
    # size is less than -min value.
    while {1} {
        # First try reading from stderr, then stdout.
        foreach {dir skipMode} {stderr ignore stdout read} {
            # Skip this direction if it is closed or being ignored.
            if {$stderrMode eq $skipMode
             || ![dict exists $Channels $id open $dir]} {
                continue
            }

            # Skip if no data is available.
            if {[set block [dict get $Channels $id buf $dir]] eq {}} {
                continue
            }

            # When not at EOF, skip if -all is used or the block is incomplete.
            if {![dict exists $Channels $id eof $dir]
             && ([info exists all] || [string length $block] < $min)} {
                continue
            }

            # Remove the data from the buffer, limiting to -max if given.
            if {[info exists max]} {
                dict set Channels $id buf $dir [string range $block $max end]
                set block [string replace $block $max end]
            } else {
                dict set Channels $id buf $dir {}
            }

            # If -dropNewline, strip trailing newline.
            if {[info exists dropNewline]} {
                regsub {\n$} $block {} block
            }

            # If -stderr throw (the default) and a complete block was read from
            # stderr, throw it rather than return it.
            if {$dir eq "stderr" && $stderrMode eq "throw"} {
                throw {ASYNC STDERR} $block
            }

            # Return the result.  With -stderr tag, this is a list consisting of
            # "stdout" or "stderr" followed by the received data.  Otherwise,
            # the result is the data.  -stderr merge is the only case where the
            # caller won't be able to tell where the data came from.
            if {$stderrMode eq "tag"} {
                return [list $dir $block]
            } else {
                return $block
            }
        }

        # Check if at EOF, and in which direction.
        set eofDir [switch $stderrMode {
        ignore  {if {[dict exists $Channels $id eof stdout]} {list stdout}}
        read    {if {[dict exists $Channels $id eof stderr]} {list stderr}}
        default {
            if {(![dict exists $Channels $id open stdout]
              || [dict exists $Channels $id eof stdout])
             && (![dict exists $Channels $id open stderr]
              || [dict exists $Channels $id eof stderr])} {
                if {[dict exists $Channels $id open stdout]} {
                    list stdout
                } else {
                    list stderr
                }
            }
        }}]

        # At this point, stderr and stdout are both unwanted or unavailable.
        if {$eofDir eq {} && ![info exists noWait]} {
            # If not at EOF and not -noWait, fill the buffers and retry.
            fill $id
        } elseif {$eofDir ne {} && $eofMode eq "throw"} {
            # At end of file with -eof throw, throw an exception.
            throw {ASYNC EOF} [dict get $Channels $id eof $eofDir]
        } else {
            # Return empty if -eof empty is used at end of file, or if -noWait
            # is used when no data is available.  The caller must use [chan eof]
            # to disambiguate.
            return
        }
    }
}

# ::async::fill --
# Reads all available data from stdout and stderr into internal buffers.  The
# current coroutine yields until at least one character is successfully read or
# end of file is reached.  This procedure need not be called by user code.
proc ::async::fill {id} {
    variable Channels

    # Loop until data could be read.  "&" is used instead of "&&" to disable
    # short-circuit evaluation and force [flow] to be called on both stdout and
    # stderr, even when it returns nonzero for stdout.
    while {![flow $id stdout] & ![flow $id stderr]} {
        # Schedule this coroutine to be resumed when there is data.
        foreach dir {stdout stderr} {
            if {![dict exists $Channels $id eof $dir]
             && [dict exists $Channels $id open $dir]} {
                chan event [dict get $Channels $id chan $dir] readable\
                        [info coroutine]
            }
        }

        # Wait until at least one channel is readable.
        yield

        # Remove the scheduled channel event handlers.
        foreach dir {stdout stderr} {
            if {![dict exists $Channels $id eof $dir]
             && [dict exists $Channels $id open $dir]} {
                chan event [dict get $Channels $id chan $dir] readable {}
            }
        }
    }
}

# ::async::flow --
# Reads all immediately available data from stdout or stderr into internal
# buffers.  This procedure need not be called by user code.  Returns the number
# of characters that were read, 0 if no data was available, or -1 if at end of
# file or the requested direction was locally closed.
proc ::async::flow {id dir} {
    variable Channels

    if {[dict exists $Channels $id eof $dir]
     || ![dict exists $Channels $id open $dir]} {
        # Already at end of file, or the requested direction was closed.
        return -1
    } elseif {[catch {chan read [dict get $Channels $id chan $dir]} data]
           || ($data eq {} && [chan eof [dict get $Channels $id chan $dir]])} {
        # At end of file, or an error occurred.
        if {$data eq {}} {
            dict set Channels $id eof $dir "end of file"
        } else {
            dict set Channels $id eof $dir $data
        }
        return -1
    } else {
        # Not at end of file.  Possibly got some data.  Append to buffer.
        dict with Channels $id buf {append $dir $data}
        return [string length $data]
    }
}

# vim: set sts=4 sw=4 tw=80 et ft=tcl: