func NewRandomUnreliableTransport(f trace.Frame, nok, ndrop int, expa, expb time.Duration) *Transport {
	return NewTransport(f, func(f0, f1 trace.Frame, a0, a1 net.Addr) (net.Conn, net.Conn) {
		nok, ndrop := rand.Intn(nok+1), rand.Intn(ndrop+1)
		nok = max(nok, 1)
		f.Printf("TRANSPORT PROFILE NOK=%d, NDROP=%d", nok, ndrop)
		return NewSievePipe(f0, f1, a0, a1, nok, ndrop, expa, expb)
	})
}

func NewTransport(frame trace.Frame, sub *codec.Transport) *Transport {
	t := &Transport{
		frame:  frame,
		sub:    sub,
		Dialer: NewDialer(frame.Refine("dialer"), sub),
	}
	frame.Bind(t)
	return t
}

func NewListener(frame trace.Frame, sub *codec.Listener) *Listener {
	l := &Listener{frame: frame, sub: sub, ach: make(chan *Conn)}
	frame.Bind(l)
	go func() {
		for {
			NewAcceptBridge(l.frame.Refine("accept"), l, l.sub.Accept())
		}
	}()
	return l
}

// NewBuffer creates a new buffer with limit m.
func NewBuffer(frame trace.Frame, m int) *Buffer {
	b := &Buffer{
		wch: make(chan struct{}, m+1), // +1 for the final EOF, so it does not block
		// The capacity of rch is chosen so that writers to rch will never block.
		rch: make(chan struct{}, 4*m+2),
	}
	frame.Bind(b)
	b.Frame = frame
	for i := 0; i < m; i++ {
		b.wch <- struct{}{}
	}
	return b
}

func testWrite(fr trace.Frame, t *testing.T, c *Conn, ready chan<- int) {
	defer func() {
		ready <- 1
	}()
	for i := 0; i < N; i++ {
		if err := c.Write([]byte{byte(i), byte(i + 1), byte(i + 2)}); err != nil {
			t.Errorf("write (%s)", err)
			failNow()
		}
		fr.Printf("WROTE %d/%d", i+1, N)
	}
	if err := c.Close(); err != nil {
		t.Fatalf("write-side close (%s)", err)
		failNow()
	}
	fr.Printf("CLOSED WRITE")
}

func (a *Conn) Start(frame trace.Frame, id chainID, addr net.Addr, linker linker, scrb func()) {
	frame.Bind(a)
	a.frame = frame
	a.scrb = scrb
	a.id = id
	a.addr = addr
	a.linker = linker
	a.cascade = MakeCascade(frame)
	// A buffer size 1 on rch, helps remove a deadlock in the TestConn.
	// Essentially it ensures that Read and Write (on two ends of a
	// connection) cannot deadlock each other when a successful Write also
	// requires a stitch. We throw in a couple of extra buffer spaces to
	// prevent any potential deadlock between Read and Kill.
	a.rch = make(chan interface{}, 3)
	a.kch = make(chan struct{})
	go a.readLoop()
}

func NewConn(frame trace.Frame, under *chain.Conn) *Conn {
	c := &Conn{
		frame: frame,
		// Only 1 needed in readLoop; get 3 just to be safe
		rch: make(chan interface{}, 3),
		// Capacity 1 (below) unblocks writeSync, invoked in readLoop right after a connection stitch
		// stitch is received, racing with a user write waiting on waitForLink.
		// In particular, if execUserWrite is waiting on waitForLink, it would prevent readLoop from
		// moving on to adopt the new connection.
		sch: make(chan *control, 1),
		ach: make(chan struct{}),
		sub: under,
		bfr: NewBuffer(frame.Refine("buffer"), MemoryCap),
	}
	c.frame.Bind(c)
	go c.readLoop()
	go c.writeLoop() // write loop for user and sync messages
	return c
}

func testRead(fr trace.Frame, t *testing.T, c *Conn, ready chan<- int) {
	defer func() {
		ready <- 1
	}()
	for i := 0; i < N; i++ {
		q, err := c.Read()
		if err != nil {
			t.Fatalf("read (%s)", err)
			failNow()
		}
		z := []byte{byte(i), byte(i + 1), byte(i + 2)}
		if !reflect.DeepEqual(q, z) {
			t.Fatalf("expecting %#v, got %#v", z, q)
			failNow()
		}
		fr.Printf("READ %d/%d", i+1, N)
	}
	if err := c.Close(); err != nil {
		t.Fatalf("read-side close (%s)", err)
		failNow()
	}
	fr.Printf("CLOSED READ")
}

func NewUnreliableTransport(f trace.Frame, nok, ndrop int, expa, expb time.Duration) *Transport {
	return NewTransport(f, func(f0, f1 trace.Frame, a0, a1 net.Addr) (net.Conn, net.Conn) {
		f.Printf("TRANSPORT PROFILE NOK=%d, NDROP=%d", nok, ndrop)
		return NewSievePipe(f0, f1, a0, a1, nok, ndrop, expa, expb)
	})
}