// As the name says,
// blocks until the stream is empty, until we meet EOF
329597
func ensureReaderClosed(stream io.ReadCloser) {
	if stream == nil {
		return
	}
	io.Copy(ioutil.Discard, stream)
	stream.Close()
}
54199
func streamEventCallback(st C.virStreamPtr, events int, callbackID int) {
	callbackFunc := getCallbackId(callbackID)

	callback, ok := callbackFunc.(StreamEventCallback)
	if !ok {
		panic("Incorrect stream event func callback")
	}

	callback(&Stream{ptr: st}, StreamEventType(events))
}
138432
func (fc *DocumentsWriterFlushControl) waitIfStalled() {
	if fc.infoStream.IsEnabled("DWFC") {
		fc.infoStream.Message(
			"DWFC", "waitIfStalled: numFlushesPending: %v netBytes: %v flushingBytes: %v fullFlush: %v",
			fc.flushQueue.Len(), fc.netBytes(), fc.flushBytes(), fc.fullFlush)
	}
	fc.DocumentsWriterStallControl.waitIfStalled()
}
439959
func (p *StatusTable) Stop(name string) {
	p.Lock()
	defer p.Unlock()

	p.pool[name] = false
}
152885
func (t *http2Server) finishStream(s *Stream, rst bool, rstCode http2.ErrCode, hdr *headerFrame, eosReceived bool) {
	oldState := t.deleteStream(s, eosReceived)
	// If the stream is already closed, then don't put trailing header to controlbuf.
	if oldState == streamDone {
		return
	}

	hdr.cleanup = &cleanupStream{
		streamID: s.id,
		rst:      rst,
		rstCode:  rstCode,
		onWrite:  func() {},
	}
	t.controlBuf.put(hdr)
}
1264282
func (stream *Stream) operateWithTimeout(invoke StreamOperation) (*orderer.StepResponse, error) {
	timer := time.NewTimer(stream.Timeout)
	defer timer.Stop()

	var operationEnded sync.WaitGroup
	operationEnded.Add(1)

	responseChan := make(chan struct {
		res *orderer.StepResponse
		err error
	}, 1)

	go func() {
		defer operationEnded.Done()
		res, err := invoke()
		responseChan <- struct {
			res *orderer.StepResponse
			err error
		}{res: res, err: err}
	}()

	select {
	case r := <-responseChan:
		if r.err != nil {
			stream.Cancel(r.err)
		}
		return r.res, r.err
	case <-timer.C:
		stream.Logger.Warningf("Stream %d to %s(%s) was forcibly terminated because timeout (%v) expired",
			stream.ID, stream.NodeName, stream.Endpoint, stream.Timeout)
		stream.Cancel(errTimeout)
		// Wait for the operation goroutine to end
		operationEnded.Wait()
		return nil, errTimeout
	}
}
446616
func (es *EventStream) Stop() {
	if es.stream != nil {
		stop(es.stream, es.rlref)
		es.stream = nil
	}

	// Remove eventstream from the registry
	registry.Delete(es.registryID)
	es.registryID = 0
}
420016
func (r *marathonClient) wait(name string, timeout time.Duration, fn func(string) bool) error {
	timer := time.NewTimer(timeout)
	defer timer.Stop()

	ticker := time.NewTicker(r.config.PollingWaitTime)
	defer ticker.Stop()
	for {
		if fn(name) {
			return nil
		}

		select {
		case <-timer.C:
			return ErrTimeoutError
		case <-ticker.C:
			continue
		}
	}
}
958910
func (s *streamState) removeStream(name string) bundlerStream {
	if si, idx := s.streamIndex(name); si != nil {
		heap.Remove(s, idx)
		return si
	}
	return nil
}
332941
func (s *Server) waitFor(search string) (err error) {
	var line string

	scanner := bufio.NewScanner(s.stdout)
	for scanner.Scan() {
		line = scanner.Text()
		fmt.Fprintf(&s.stdoutBuf, "%s\n", line)
		if strings.Contains(line, search) {
			return nil
		}
	}
	err = scanner.Err()
	if err == nil {
		err = io.EOF
	}
	return err
}
380086
func (m *Streamer) Stop(streamID StreamID) {
	strm, ok := m.loadStream(streamID)
	if !ok {
		return
	}

	close(strm.done)

	go func() {
		// wait some time to ensure clients have connected, once they've
		// retrieved the stream from the map it's safe to delete the key
		time.Sleep(m.graceTime)

		m.streams.Delete(streamID)
	}()
}
1193873
func (p *BlockPuller) Close() {
	if p.cancelStream != nil {
		p.cancelStream()
	}
	p.cancelStream = nil

	if p.conn != nil {
		p.conn.Close()
	}
	p.conn = nil
	p.endpoint = ""
	p.latestSeq = 0
	p.blockBuff = nil
}
628192
func (bw *BufferedFileWriter) Flush() {
	bw.mu.Lock()
	// could add check if bw.done already
	//  should never happen
	bw.bufWriter.Flush()
	bw.origWriter.f.Sync()
	bw.mu.Unlock()
}
486941
func (s *Scanner) EOF() int {
	if s.Err != nil {
		return ScanError
	}
	if s.endTop {
		return ScanEnd
	}
	s.Step(s, ' ')
	if s.endTop {
		return ScanEnd
	}
	if s.Err == nil {
		s.Err = &SyntaxError{"unexpected end of JSON input", s.bytes}
	}
	return ScanError
}
26563
func (w *huffmanBitWriter) writeBlockHuff(eof bool, input []byte) {
	if w.err != nil {
		return
	}

	// Clear histogram
	for i := range w.literalFreq {
		w.literalFreq[i] = 0
	}

	// Add everything as literals
	histogram(input, w.literalFreq)

	w.literalFreq[endBlockMarker] = 1

	const numLiterals = endBlockMarker + 1
	const numOffsets = 1

	w.literalEncoding.generate(w.literalFreq, 15)

	// Figure out smallest code.
	// Always use dynamic Huffman or Store
	var numCodegens int

	// Generate codegen and codegenFrequencies, which indicates how to encode
	// the literalEncoding and the offsetEncoding.
	w.generateCodegen(numLiterals, numOffsets, w.literalEncoding, huffOffset)
	w.codegenEncoding.generate(w.codegenFreq[:], 7)
	size, numCodegens := w.dynamicSize(w.literalEncoding, huffOffset, 0)

	// Store bytes, if we don't get a reasonable improvement.
	if ssize, storable := w.storedSize(input); storable && ssize < (size+size>>4) {
		w.writeStoredHeader(len(input), eof)
		w.writeBytes(input)
		return
	}

	// Huffman.
	w.writeDynamicHeader(numLiterals, numOffsets, numCodegens, eof)
	encoding := w.literalEncoding.codes[:257]
	n := w.nbytes
	for _, t := range input {
		// Bitwriting inlined, ~30% speedup
		c := encoding[t]
		w.bits |= uint64(c.code) << w.nbits
		w.nbits += uint(c.len)
		if w.nbits < 48 {
			continue
		}
		// Store 6 bytes
		bits := w.bits
		w.bits >>= 48
		w.nbits -= 48
		bytes := w.bytes[n : n+6]
		bytes[0] = byte(bits)
		bytes[1] = byte(bits >> 8)
		bytes[2] = byte(bits >> 16)
		bytes[3] = byte(bits >> 24)
		bytes[4] = byte(bits >> 32)
		bytes[5] = byte(bits >> 40)
		n += 6
		if n < bufferFlushSize {
			continue
		}
		w.write(w.bytes[:n])
		if w.err != nil {
			return // Return early in the event of write failures
		}
		n = 0
	}
	w.nbytes = n
	w.writeCode(encoding[endBlockMarker])
}
331852
func (b *bufWriter) finish() *os.File {
	b.flush()
	f := b.file
	f.Seek(0, 0)
	return f
}
282743
func WriteWait(writeWait time.Duration) func(*wsConnection) {
	return func(wsc *wsConnection) {
		wsc.writeWait = writeWait
	}
}