// Returns streaming bitrot writer implementation.
519584
func newStreamingBitrotWriter(disk StorageAPI, volume, filePath string, length int64, algo BitrotAlgorithm, shardSize int64) io.WriteCloser {
	r, w := io.Pipe()
	h := algo.New()
	bw := &streamingBitrotWriter{w, h, shardSize, make(chan struct{})}
	go func() {
		totalFileSize := int64(-1) // For compressed objects length will be unknown (represented by length=-1)
		if length != -1 {
			bitrotSumsTotalSize := ceilFrac(length, shardSize) * int64(h.Size()) // Size used for storing bitrot checksums.
			totalFileSize = bitrotSumsTotalSize + length
		}
		err := disk.CreateFile(volume, filePath, totalFileSize, r)
		r.CloseWithError(err)
		close(bw.canClose)
	}()
	return bw
}
895536
func (bv *BitVector) writeBits(output store.IndexOutput) error {
	return store.Stream(output).
		WriteInt(int32(bv.size)).
		WriteInt(int32(bv.Count())).
		WriteBytes(bv.bits).
		Close()
}
1220555
func MemoryStreamGetVirtual(context *gin.Context) {
	rateLimit, convertErr := util.ConvertStringToTimeDuration(context.Param("rateLimit"))
	context.Stream(func(w io.Writer) bool {
		if convertErr != nil {
			return false
		}
		time.Sleep(rateLimit * time.Second)
		body, err := memory.ServiceGetVirtualMemory()
		return util.StreamHandleResponse(context, body, err, "MemoryStreamGetVirtual")
	})
}
408469
func bitrotCheckFile(disk StorageAPI, volume string, filePath string, tillOffset int64, algo BitrotAlgorithm, sum []byte, shardSize int64) (err error) {
	if algo != HighwayHash256S {
		buf := []byte{}
		// For whole-file bitrot we don't need to read the entire file as the bitrot verify happens on the server side even if we read 0-bytes.
		_, err = disk.ReadFile(volume, filePath, 0, buf, NewBitrotVerifier(algo, sum))
		return err
	}
	buf := make([]byte, shardSize)
	r := newStreamingBitrotReader(disk, volume, filePath, tillOffset, algo, shardSize)
	defer closeBitrotReaders([]io.ReaderAt{r})
	var offset int64
	for {
		if offset == tillOffset {
			break
		}
		var n int
		tmpBuf := buf
		if int64(len(tmpBuf)) > (tillOffset - offset) {
			tmpBuf = tmpBuf[:(tillOffset - offset)]
		}
		n, err = r.ReadAt(tmpBuf, offset)
		if err != nil {
			return err
		}
		offset += int64(n)
	}
	return nil
}
424771
func (st smallBlobStreamer) StreamBlobs(ctx context.Context, dest chan<- blobserver.BlobAndToken, contToken string) (err error) {
	small := st.sto.small
	if bs, ok := small.(blobserver.BlobStreamer); ok {
		return bs.StreamBlobs(ctx, dest, contToken)
	}
	defer close(dest)
	donec := ctx.Done()
	return blobserver.EnumerateAllFrom(ctx, small, contToken, func(sb blob.SizedRef) error {
		select {
		case dest <- blobserver.BlobAndToken{
			Blob: blob.NewBlob(sb.Ref, sb.Size, func(ctx context.Context) ([]byte, error) {
				return slurpSizedRef(ctx, small, sb)
			}),
			Token: sb.Ref.StringMinusOne(), // streamer is >=, enumerate is >
		}:
			return nil
		case <-donec:
			return ctx.Err()
		}
	})
}
812490
func binaryWrite(w io.Writer, order binary.ByteOrder, data interface{}) (n int64, err error) {
	buf := bytes.Buffer{}
	if err = binary.Write(&buf, order, data); err != nil {
		return 0, err
	}

	written, err := w.Write(buf.Bytes())
	return int64(written), err
}
729434
func (s *Stream) Write(b []byte) (n int, err error) {
	s.sendLock.Lock()
	defer s.sendLock.Unlock()
	total := 0
	for total < len(b) {
		n, err := s.write(b[total:])
		total += n
		if err != nil {
			return total, err
		}
	}
	return total, nil
}
530413
func (s stream) NewDecoder(r io.Reader) *StreamDecoder {
	dec := NewDecoder(r)
	streamDec := &StreamDecoder{
		Decoder: dec,
		done:    make(chan struct{}, 1),
		mux:     sync.RWMutex{},
	}
	return streamDec
}
379904
func (c *Client) RemoteStreamingRpcHandler(method string) (structs.StreamingRpcHandler, error) {
	server := c.servers.FindServer()
	if server == nil {
		return nil, noServersErr
	}

	conn, err := c.streamingRpcConn(server, method)
	if err != nil {
		// Move off to another server
		c.rpcLogger.Error("error performing RPC to server", "error", err, "rpc", method, "server", server.Addr)
		c.servers.NotifyFailedServer(server)
		return nil, err
	}

	return bridgedStreamingRpcHandler(conn), nil
}
467919
func WriteOutputTarStream(fg storage.FileGetter, up storage.Unpacker, w io.Writer) error {
	// ... Since these are interfaces, this is possible, so let's not have a nil pointer
	if fg == nil || up == nil {
		return nil
	}
	var copyBuffer []byte
	var crcHash hash.Hash
	var crcSum []byte
	var multiWriter io.Writer
	for {
		entry, err := up.Next()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}
		switch entry.Type {
		case storage.SegmentType:
			if _, err := w.Write(entry.Payload); err != nil {
				return err
			}
		case storage.FileType:
			if entry.Size == 0 {
				continue
			}
			fh, err := fg.Get(entry.GetName())
			if err != nil {
				return err
			}
			if crcHash == nil {
				crcHash = crc64.New(storage.CRCTable)
				crcSum = make([]byte, 8)
				multiWriter = io.MultiWriter(w, crcHash)
				copyBuffer = byteBufferPool.Get().([]byte)
				defer byteBufferPool.Put(copyBuffer)
			} else {
				crcHash.Reset()
			}

			if _, err := copyWithBuffer(multiWriter, fh, copyBuffer); err != nil {
				fh.Close()
				return err
			}

			if !bytes.Equal(crcHash.Sum(crcSum[:0]), entry.Payload) {
				// I would rather this be a comparable ErrInvalidChecksum or such,
				// but since it's coming through the PipeReader, the context of
				// _which_ file would be lost...
				fh.Close()
				return fmt.Errorf("file integrity checksum failed for %q", entry.GetName())
			}
			fh.Close()
		}
	}
}
1259386
func (stream *Stream) WriteUint8(val uint8) {
	stream.buf = writeFirstBuf(stream.buf, digits[val])
}
791822
func (f *FileRotator) flushBuffer() error {
	f.bufLock.Lock()
	defer f.bufLock.Unlock()
	if f.bufw != nil {
		return f.bufw.Flush()
	}
	return nil
}
463291
func NewBitReader(data []byte) *BitReader {
	b := new(BitReader)
	b.Reset(data)
	return b
}
638526
func (rng *Pcg128x64) Stream(streamH, streamL uint64) {
  // stream must be odd for LCG, so shift left 1 and turn on the 1 bit
  rng.lcg.Stream = Uint128{streamH,streamL}.ShiftLeft(1)
  rng.lcg.Stream.L |= 1
}
827163
func WriteArbitrateX3(
	clientAddr chan<- protocol.Addr,
	clientData chan<- protocol.WriteData,
	clientResp <-chan protocol.WriteResp,
	serverAddr0 <-chan protocol.Addr,
	serverData0 <-chan protocol.WriteData,
	serverResp0 chan<- protocol.WriteResp,
	serverAddr1 <-chan protocol.Addr,
	serverData1 <-chan protocol.WriteData,
	serverResp1 chan<- protocol.WriteResp,
	serverAddr2 <-chan protocol.Addr,
	serverData2 <-chan protocol.WriteData,
	serverResp2 chan<- protocol.WriteResp) {

	// Specify the input selection channels.
	dataChanSelect := make(chan byte)
	respChanSelect := make(chan byte)

	// Run write data channel handler.
	go func() {
		for {
			var writeData protocol.WriteData
			chanSelect := <-dataChanSelect

			// Terminate transfers on write data channel 'last' flag.
			isLast := false
			for !isLast {
				switch chanSelect {
				case 0:
					writeData = <-serverData0
				case 1:
					writeData = <-serverData1
				default:
					writeData = <-serverData2
				}
				clientData <- writeData
				isLast = writeData.Last
			}
		}
	}()

	// Run response channel handler.
	go func() {
		for {
			chanSelect := <-respChanSelect
			writeResp := <-clientResp
			switch chanSelect {
			case 0:
				serverResp0 <- writeResp
			case 1:
				serverResp1 <- writeResp
			default:
				serverResp2 <- writeResp
			}
		}
	}()

	// Use intermediate variables for efficient implementation.
	var writeAddr protocol.Addr
	var dataChanId byte
	for {
		select {
		case writeAddr = <-serverAddr0:
			dataChanId = 0
		case writeAddr = <-serverAddr1:
			dataChanId = 1
		case writeAddr = <-serverAddr2:
			dataChanId = 2
		}
		clientAddr <- writeAddr
		dataChanSelect <- dataChanId
		respChanSelect <- dataChanId
	}
}
1335792
func BER2DER(ber []byte) ([]byte, error) {
	if len(ber) == 0 {
		return nil, ASN1Error{"ber2der: input ber is empty"}
	}
	//fmt.Printf("--> ber2der: Transcoding %d bytes\n", len(ber))
	out := new(bytes.Buffer)

	obj, _, err := readObject(ber, 0)
	if err != nil {
		return nil, err
	}
	obj.encodeTo(out)

	// if offset < len(ber) {
	//	return nil, fmt.Errorf("ber2der: Content longer than expected. Got %d, expected %d", offset, len(ber))
	//}

	return out.Bytes(), nil
}
633864
func DoByteBufStreamingRoundTrip(stream testpb.BenchmarkService_StreamingCallClient, reqSize, respSize int) error {
	out := make([]byte, reqSize)
	if err := stream.(grpc.ClientStream).SendMsg(&out); err != nil {
		return fmt.Errorf("/BenchmarkService/StreamingCall.(ClientStream).SendMsg(_) = %v, want <nil>", err)
	}
	var in []byte
	if err := stream.(grpc.ClientStream).RecvMsg(&in); err != nil {
		// EOF is a valid error here.
		if err == io.EOF {
			return nil
		}
		return fmt.Errorf("/BenchmarkService/StreamingCall.(ClientStream).RecvMsg(_) = %v, want <nil>", err)
	}
	return nil
}