// MultiRead reads multiple entities at one time
188376
func (c *Connector) MultiRead(ctx context.Context, ei *dosa.EntityInfo, keys []map[string]dosa.FieldValue, minimumFields []string) ([]*dosa.FieldValuesOrError, error) {
	// Convert the fields from the client's map to a set of fields to read
	rpcMinimumFields := makeRPCminimumFields(minimumFields)

	// convert the keys to RPC's Value
	rpcFields := make([]dosarpc.FieldValueMap, len(keys))
	for i, kmap := range keys {
		rpcFields[i] = make(dosarpc.FieldValueMap)
		for key, value := range kmap {
			rv, err := RawValueFromInterface(value)
			if err != nil {
				return nil, err
			}
			if rv == nil {
				continue
			}
			rpcValue := &dosarpc.Value{ElemValue: rv}
			rpcFields[i][key] = rpcValue
		}
	}

	// perform the multi read request
	request := &dosarpc.MultiReadRequest{
		Ref:          entityInfoToSchemaRef(ei),
		KeyValues:    rpcFields,
		FieldsToRead: rpcMinimumFields,
	}

	response, err := c.client.MultiRead(ctx, request, getHeaders(c.headers)...)
	if err != nil {
		if !dosarpc.Dosa_MultiRead_Helper.IsException(err) {
			return nil, errors.Wrap(err, "failed to MultiRead due to network issue")
		}

		return nil, errors.Wrap(err, "failed to MultiRead")
	}

	rpcResults := response.Results
	results := make([]*dosa.FieldValuesOrError, len(rpcResults))
	for i, rpcResult := range rpcResults {
		results[i] = &dosa.FieldValuesOrError{Values: make(map[string]dosa.FieldValue), Error: nil}
		for name, value := range rpcResult.EntityValues {
			for _, col := range ei.Def.Columns {
				if col.Name == name {
					results[i].Values[name] = RawValueAsInterface(*value.ElemValue, col.Type)
					break
				}
			}
		}
		if rpcResult.Error != nil {
			results[i].Error = wrapIDLError(rpcResult.Error)
		}
	}

	return results, nil
}
241393
func NewBatchRead(key *Key, binNames []string) *BatchRead {
	res := &BatchRead{
		Key:      key,
		BinNames: binNames,
	}

	if len(binNames) == 0 {
		res.ReadAllBins = true
	}

	return res
}
345547
func (r *ByteReader) Unread(n int) {
	if r.off >= n {
		r.off -= n
	} else {
		r.off = 0
	}
}
817546
func globalRandRead(b []byte) {
	pos := 0
	val := 0
	for n := 0; n < len(b); n++ {
		if pos == 0 {
			val = rand.Int()
			pos = 7
		}
		b[n] = byte(val)
		val >>= 8
		pos--
	}
}
761596
func (c *Connector) MultiUpsert(ctx context.Context, ei *dosa.EntityInfo, multiValues []map[string]dosa.FieldValue) ([]error, error) {
	values, err := fieldValueMapsFromClientMaps(multiValues)
	if err != nil {
		return nil, err
	}

	// ttl := dosa.NoTTL().Nanoseconds()
	// if ei.TTL != nil {
	// 	ttl = ei.TTL.Nanoseconds()
	// }

	// perform the multi upsert request
	request := &dosarpc.MultiUpsertRequest{
		Ref:      entityInfoToSchemaRef(ei),
		Entities: values,
		// TTL:      &ttl, mgode@ has not yet committed origin/ttl-for-multi-upsert
	}

	response, err := c.client.MultiUpsert(ctx, request, getHeaders(c.headers)...)
	if err != nil {
		if !dosarpc.Dosa_MultiUpsert_Helper.IsException(err) {
			return nil, errors.Wrap(err, "failed to MultiUpsert due to network issue")
		}

		return nil, errors.Wrap(err, "failed to MultiUpsert")
	}

	rpcErrors := response.Errors
	results := make([]error, len(rpcErrors))
	for i, rpcError := range rpcErrors {
		if rpcError != nil {
			results[i] = wrapIDLError(rpcError)
		}
	}

	return results, nil
}
93236
func (zip GobMeddler) PreRead(fieldAddr interface{}) (scanTarget interface{}, err error) {
	// give a pointer to a byte buffer to grab the raw data
	return new([]byte), nil
}
389614
func (m *MockClient) MultiRead(arg0 context.Context, arg1 []string, arg2 ...dosa.DomainObject) (dosa.MultiResult, error) {
	varargs := []interface{}{arg0, arg1}
	for _, a := range arg2 {
		varargs = append(varargs, a)
	}
	ret := m.ctrl.Call(m, "MultiRead", varargs...)
	ret0, _ := ret[0].(dosa.MultiResult)
	ret1, _ := ret[1].(error)
	return ret0, ret1
}
1248172
func (r *Reader) NumUnread() int {
	return int(uint(r.Size()) - r.position)
}
606475
func (b *CumulativeBucket) MultiAdd(res *Result) {
	if res.Count == 0 {
		return
	}
	atomic.AddInt64(&b.count, res.Count)
	atomic.AddInt64(&b.sum, res.Sum)
	b.sumOfSquares.Add(res.SumOfSquares)
}
535611
func (s *CfgNetworkState) ReadAll() ([]core.State, error) {
	return s.StateDriver.ReadAllState(networkConfigPathPrefix, s, json.Unmarshal)
}
1106611
func (c *connection) readRespLoop() {
	defer func() {
		c.mu.Lock()
		for _, cc := range c.respc {
			close(cc)
		}
		c.respc = make(map[int32]chan []byte)
		c.mu.Unlock()
	}()

	rd := bufio.NewReader(c.rw)
	for {
		if c.readTimeout > 0 {
			err := c.rw.SetReadDeadline(time.Now().Add(c.readTimeout))
			if err != nil {
				c.logger.Error("msg", "SetReadDeadline failed",
					"error", err)
			}
		}
		correlationID, b, err := proto.ReadResp(rd)
		if err != nil {
			c.mu.Lock()
			if c.stopErr == nil {
				c.stopErr = err
				close(c.stop)
			}
			c.mu.Unlock()
			return
		}

		c.mu.Lock()
		rc, ok := c.respc[correlationID]
		delete(c.respc, correlationID)
		c.mu.Unlock()
		if !ok {
			c.logger.Warn(
				"msg", "response to unknown request",
				"correlationID", correlationID)
			continue
		}

		select {
		case <-c.stop:
			c.mu.Lock()
			if c.stopErr == nil {
				c.stopErr = ErrClosed
			}
			c.mu.Unlock()
		case rc <- b:
		}
		close(rc)
	}
}
302408
func (entries *DirEntries) ReadAll() (result []*DirEntry, err error) {
	defer entries.Close()

	for entries.Next() {
		result = append(result, entries.Entry())
	}
	err = entries.Err()

	return
}
862236
func (c *Connector) MultiUpsert(ctx context.Context, ei *dosa.EntityInfo, multiValues []map[string]dosa.FieldValue) (result []error, err error) {
	return nil, new(ErrNotImplemented)
}
16378
func (c *Client) MultipleBatch(operations []BatchOperationIndexed, opts ...interface{}) (res MultipleBatchRes, err error) {
	body := map[string][]BatchOperationIndexed{"requests": operations}
	path := c.path("/indexes/*/batch")
	err = c.transport.Request(&res, http.MethodPost, path, body, call.Write, opts...)
	res.wait = c.waitTask
	return
}
1031218
func NewMultistream(c io.ReadWriteCloser, proto string) Multistream {
	return &lazyClientConn{
		protos: []string{proto},
		con:    c,
	}
}
1252087
func BlockingRead(fd int, b []byte) (int, *tcpip.Error) {
	for {
		n, _, e := syscall.RawSyscall(syscall.SYS_READ, uintptr(fd), uintptr(unsafe.Pointer(&b[0])), uintptr(len(b)))
		if e == 0 {
			return int(n), nil
		}

		event := PollEvent{
			FD:     int32(fd),
			Events: 1, // POLLIN
		}

		_, e = BlockingPoll(&event, 1, -1)
		if e != 0 && e != syscall.EINTR {
			return 0, TranslateErrno(e)
		}
	}
}
10226
func (server *Server) readAll(request *http.Request) []byte {
	defer request.Body.Close()
	body, _ := ioutil.ReadAll(request.Body)
	return body
}