list ::coroutine main ::apply [list {work args} {

		variable queue
		thread::send -async $queue [list [namespace current] sourceon [list [thread::id] [info coroutine]]]
		::apply $work {*}$args [namespace current]
		#puts stderr [list [thread::id] sending source off to $queue]
		thread::send -async $queue [list [namespace current] sourceoff [list [thread::id] [info coroutine]]]
		thread::release
	} [namespace current]] $work {*}$args]]
	
	return $producer
}


proc station args {

	variable setup1 {
	}


	test basic {} -body {
		variable res

		set tnames [thread::names]
		#puts stderr [list {existing threads} $tnames]

		[daerth queue .spawn queue1] init
		[daerth queue .spawn queue2] init
		[daerth queue .spawn queue3] init


		set station1 [daerth transport source [queue1 thread] dest [
			queue2 thread] work {{daerth number} {

			while 1 {
				# puts stderr [list [thread::id] station1 working on $number]
				lassign [$daerth deliver [
					expr {$number + $number}]] number
			}
		}}]
		#puts stderr [list station1 $station1]


		set station2 [daerth transport source  [queue2 thread] dest [
			queue3 thread] work {{daerth number} {

			while 1 {
				# puts [list [thread::id] station2 working on $number]
				lassign [$daerth deliver [
					list $number [expr {$number + 1}]]] number
			}
		}}]


		set producer [daerth producer queue [queue1 thread] work {daerth {
			for {set i 0} {$i < 10} {incr i} {
				$daerth deliver $i
			}
			# puts stderr [list [thread::id] leaving work]
		}}]
		#puts stderr [list producer $producer]

		collect queue3

		thread send [queue1 thread] {thread::release}
		thread send [queue2 thread] {thread::release}
		thread send [queue3 thread] {thread::release}

		set tnames2 [thread::names]
		lappend res [expr {[llength $tnames] == [llength $tnames2]}]
		return $res
	} -cleanup [cleanup1] -result [sl {
		{0 1} {2 3} {4 5} {6 7} {8 9} {10 11} {12 13} {14 15} {16 17} {18 19} 1
	}]


	test daerth {} -setup $setup1 -body {
		variable res
		set tnames [thread names]
		set res2 {}
		foreach {consumercount filtercount producercount} [sl {
			1 1 1
			1 1 2
			1 1 3
			1 1 4
			1 2 1

					lappend res [list {res not double list1} $i [lindex $list1 $i] [
						lindex $res $i]]
					return $res2
				}
			}
			lappend res2 done
		}
		set tnames2 [thread names]
		lappend res2 [expr {[llength $tnames] == [llength $tnames2]}]
		return $res2
	} -cleanup [cleanup1] -result [sl {
		done done done done done done done done 1
	}]


	test daerth_primes {} -setup $setup1 -body {
		variable res
		set tnames [thread names]
		foreach {consumercount filtercount producercount} [sl {
			1 1 1
		}] {
			factory1 [list station_produce_primes 500 $producercount] $consumercount $filtercount
			set list1 [lsort -integer [::tsv::get tres list]]
			set res [lsort -integer $res[set res {}]]
			for {set i 0} {$i < [llength $list1]} {incr i} {
				if {[lindex $res $i] != [lindex $list1 $i] * 2} {
					lappend res2 [list {res not double list1} $i [lindex $list1 $i] [
						lindex $res $i]]
					return $res2
				}
			}
			lappend res2 done
		}
		set tnames2 [thread names]
		lappend res2 [expr {[llength $tnames] == [llength $tnames2]}]
		return $res2
	} -cleanup [cleanup1] -result [sl {
		done 1
	}]

	# Warning, this can take several minutes to run
	test daerth_primes_loadbalanced {} -body {
		set res2  {}
		set tnames [thread names]
		foreach {consumercount filtercount producercount} [sl {
			1 1 1
			# this runs about twice as fast as the previous run
			1 2 1 
		}] {
			factory_primes_loadbalanced [
				list station_produce_odd_numbers 1000 $producercount] \
					$consumercount $filtercount
			set list1 [lsort -integer [::tsv::get tres list]]
			lappend res2 [llength $list1]
		}
		set tnames2 [thread names]
		lappend res2 [expr {[llength $tnames] == [llength $tnames2]}]
		set res2
	} -cleanup [cleanup1] -result [sl {
		167 167
	}]
	
	cleanupTests
}

	# These filters are fed by the producers, and produce to  the leaf consumers.
	set filters [station_push_tres [queue1 thread] [queue2 thread] $filtercount]

	## leaf consumers
	set consumers [station_push_tres2 [queue2 thread] [queue3 thread] $consumercount]

	collect queue3

	thread send [queue1 thread] {thread::release}
	thread send [queue2 thread] {thread::release}
	thread send [queue3 thread] {thread::release}
}


proc factory_primes_loadbalanced {producers consumercount filtercount} {

	tsv::set tres list {} 

	set producers [{*}$producers [queue1 thread]]
	# puts stderr [list producers are $producers]
	set filters [station_isprime [queue1 thread] [queue2 thread] $filtercount]
	# puts stderr [list filters are $filters]
	set consumers [station_push_tres [queue2 thread] [queue3 thread] $consumercount] 

	collect queue3

	queue1 exit
	queue2 exit
	queue3 exit
}


proc station_isprime {source dest count} {
	set transports [daerth station count $count transport [
		list source $source dest $dest init {
			proc isprime x {

#! /bin/env tclsh

namespace import [yclprefix]::util 
package require {ycl daerth util}
namespace import [yclprefix]::daerth::util

proc exit _ {
	$_ .vars tid
	thread::send $tid thread::release
}
.my .method exit

proc init {_ args} {
	$_ .vars tid
	variable setup
	set opts [list limit 10 queuesize 10 tid {}]

	util opts opts args

#! /bin/env tclsh


package ifneeded [list ycl daerth] 0.2 [list apply {{dir} {
	package require {ycl package}
	package require {ycl shelf shelf}
	set name [yclprefix]::daerth::queue
	[[yclprefix] shelf shelf .spawn $name] init
	$name .eval [list source $dir/lib/queue.tcl]

	[yclprefix] package source daerth $dir/lib/daerth.tcl
	package provide [list ycl daerth] 0.2
}} $dir]


package ifneeded [list ycl daerth test] 0.2 [list apply {{dir} {
	package require {ycl package}
	[yclprefix] package source daerth::test $dir/lib/daerth.test.tcl
	package provide [list ycl daerth test] 0.2
}} $dir]


package ifneeded [list ycl daerth util] 0.2 [list apply {{dir} {
	package require {ycl package}
	[yclprefix] package source daerth::util $dir/lib/util.tcl
	package provide [list ycl daerth util] 0.2
}} $dir]