gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(rpc2): request and response with checksum

Browse Source 
. #22495324

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-08-13 15:39:10 +08:00
parent d3e06e797a
commit 27b2588bde
12 changed files with 1033 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
blobstore/common/rpc2/body.go
View File

@ -15,9 +15,6 @@
package rpc2
import (
"bytes"
"fmt"
"hash/crc32"
"io"
"sync"
@ -25,19 +22,13 @@ import (
)
type bodyAndTrailer struct {
sr *transport.SizedReader
br io.Reader
// body remain, read trailer if remain == 0
remain int
closeOnce sync.Once
req *Request
sr *transport.SizedReader // smux reader
br Body // body reader
remain int // body remain, read trailer if remain == 0
req *Request
trailerOnce sync.Once
trailer *FixedHeader
trailerWriter io.Writer
trailerChecker func(*FixedHeader) error
closeOnce sync.Once
}
func (r *bodyAndTrailer) tryReadTrailer() error {
@ -47,19 +38,18 @@ func (r *bodyAndTrailer) tryReadTrailer() error {
var err error
if r.remain == 0 { // try to read trailer
r.trailerOnce.Do(func() {
_, err = r.trailer.ReadFrom(r.sr)
_, err = r.req.Trailer.ReadFrom(r.sr)
})
if err == nil || err == io.EOF {
err = r.trailerChecker(r.trailer)
}
}
return err
}
func (r *bodyAndTrailer) Read(p []byte) (int, error) {
if len(p) > r.remain {
p = p[:r.remain]
}
n, err := r.br.Read(p)
r.remain -= n
r.trailerWriter.Write(p[:n])
if err == nil {
err = r.tryReadTrailer()
}
@ -67,10 +57,12 @@ func (r *bodyAndTrailer) Read(p []byte) (int, error) {
}
func (r *bodyAndTrailer) WriteTo(w io.Writer) (int64, error) {
if _, ok := w.(*LimitedWriter); !ok {
if lw, ok := w.(*LimitedWriter); !ok {
return 0, ErrLimitedWriter
} else if lw.a > int64(r.remain) {
return 0, io.ErrShortWrite
}
n, err := r.sr.WriteTo(io.MultiWriter(w, r.trailerWriter))
n, err := r.br.WriteTo(w)
if err == errLimitedWrite {
err = nil
}
@ -85,7 +77,7 @@ func (r *bodyAndTrailer) Close() (err error) {
errx := r.tryReadTrailer()
r.closeOnce.Do(func() {
err = r.sr.Close()
if r.req != nil {
if r.req.client != nil {
if err == nil && r.sr.Finished() {
err = r.req.client.Connector.Put(r.req.Context(), r.req.conn)
} else {
@ -100,34 +92,18 @@ func (r *bodyAndTrailer) Close() (err error) {
}
// makeBodyWithTrailer body and trailer remain.
func makeBodyWithTrailer(sr *transport.SizedReader, l int64, req *Request, trailer *FixedHeader) Body {
writer := io.MultiWriter()
checker := func(trailer *FixedHeader) error { return nil }
if trailer.Has(headerInternalCrc) {
crc := crc32.NewIEEE()
writer = io.MultiWriter(crc)
checker = func(trailer *FixedHeader) error {
exp := []byte(trailer.Get(headerInternalCrc))
act := crc.Sum(nil)
if !bytes.Equal(exp, act) {
return &Error{
Status: 400,
Reason: "Checksum",
Detail: fmt.Sprintf("rpc2: internal crc exp(%v) act(%v)", exp, act),
}
}
return nil
}
func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, l int64) Body {
var br Body
if req.checksum != nil {
br = newEdBody(*req.checksum, sr, int(l), false)
} else {
br = sr
}
r := &bodyAndTrailer{
sr: sr,
br: io.LimitReader(sr, l),
remain: int(l),
req: req,
trailer: trailer,
trailerWriter: writer,
trailerChecker: checker,
sr: sr,
br: br,
remain: int(l),
req: req,
}
r.tryReadTrailer()
return r

295
blobstore/common/rpc2/checksum.go Normal file
View File

@ -0,0 +1,295 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
// original | body |
// encoded | payload1+cell | payload2+cell | ... |
// len(payload1) = len(payload2) = blockSize, cell = hash.Hash.Size()
import (
"bytes"
"fmt"
"hash"
"hash/crc32"
"io"
)
const DefaultBlockSize = 64 << 10
var algorithms = map[ChecksumAlgorithm]func() hash.Hash{
ChecksumAlgorithm_Crc_IEEE: func() hash.Hash { return crc32.NewIEEE() },
}
func (cb *ChecksumBlock) EncodeSize(originalSize int64) int64 {
if cb == nil {
return originalSize
}
hasher := algorithms[cb.Algorithm]()
payload := int64(cb.BlockSize)
blocks := (originalSize + (payload - 1)) / payload
return originalSize + int64(hasher.Size())*blocks
}
func (cb *ChecksumBlock) Hasher() hash.Hash {
return algorithms[cb.Algorithm]()
}
func (cb *ChecksumBlock) Readable(b []byte) any {
switch cb.Algorithm {
case ChecksumAlgorithm_Crc_IEEE:
return uint32(b[0])<<24 | uint32(b[1])<<16 | uint32(b[2])<<8 | uint32(b[3])
default:
return nil
}
}
func unmarshalBlock(b []byte) (*ChecksumBlock, error) {
var block ChecksumBlock
if err := block.Unmarshal(b); err != nil {
return nil, fmt.Errorf("rpc2: internal checksum %s", err.Error())
}
if _, exist := algorithms[block.Algorithm]; !exist || block.BlockSize == 0 {
return nil, fmt.Errorf("rpc2: checksum(%s) not implements", block.String())
}
return &block, nil
}
func checksumError(block ChecksumBlock, exp, act []byte) *Error {
return &Error{
Status: 400,
Reason: "Checksum",
Detail: fmt.Sprintf("rpc2: internal checksum exp(%v) act(%v)",
block.Readable(exp), block.Readable(act)),
}
}
// body encoder and decoder
type edBody struct {
block ChecksumBlock
encode bool
hasher hash.Hash
remain int
nx int // block index
cx int // cell index, -1 means no sum cached
cell []byte
err error
Body
}
func newEdBody(block ChecksumBlock, body Body, remain int, encode bool) *edBody {
hasher := block.Hasher()
return &edBody{
block: block,
encode: encode,
hasher: hasher,
remain: remain,
cx: -1,
cell: make([]byte, hasher.Size()),
Body: body,
}
}
// encodeRead the parameter p is sumx.FreamWrite.data
func (r *edBody) encodeRead(p []byte) (nn int, err error) {
var n int
if r.cx >= 0 { // has remaining checksum
n = copy(p, r.cell[r.cx:])
nn += n
r.cx += n
if r.cx < r.hasher.Size() {
return
}
r.cx = -1
p = p[n:]
}
if r.remain <= 0 {
if nn == 0 {
err = io.EOF
}
return
}
blockSize := int(r.block.BlockSize)
tryRead := blockSize - r.nx
if tryRead > r.remain {
tryRead = r.remain
}
if len(p) < tryRead {
n, err = r.Body.Read(p)
} else {
n, err = r.Body.Read(p[:tryRead])
}
r.hasher.Write(p[:n])
nn += n
r.nx += n
r.remain -= n
if r.nx == blockSize || r.remain == 0 {
copy(r.cell, r.hasher.Sum(nil))
r.hasher = r.block.Hasher()
r.cx = 0
r.nx = 0
}
return
}
// decodeRead the parameter p is handler's memory location
func (r *edBody) decodeRead(p []byte) (nn int, err error) {
if r.err != nil {
return 0, r.err
}
if r.remain <= 0 {
return 0, io.EOF
}
if r.cx >= 0 {
if _, err = io.ReadFull(r.Body, r.cell[r.cx:]); err != nil {
r.err = err
return 0, err
}
act := r.hasher.Sum(nil)
if !bytes.Equal(r.cell, act) {
r.err = checksumError(r.block, r.cell, act)
return 0, r.err
}
r.cx = -1
r.hasher = r.block.Hasher()
}
blockSize := int(r.block.BlockSize)
tryRead := blockSize - r.nx
if tryRead > r.remain {
tryRead = r.remain
}
var n int
if len(p) < tryRead {
n, err = r.Body.Read(p)
} else {
n, err = r.Body.Read(p[:tryRead])
}
r.hasher.Write(p[:n])
nn += n
r.nx += n
r.remain -= n
if r.nx == blockSize || r.remain == 0 {
_, err = io.ReadFull(r.Body, r.cell)
if err != nil {
r.err = err
return 0, err
}
act := r.hasher.Sum(nil)
if !bytes.Equal(r.cell, act) {
r.err = checksumError(r.block, r.cell, act)
return 0, r.err
}
r.hasher = r.block.Hasher()
r.nx = 0
}
return
}
func (r *edBody) Read(p []byte) (nn int, err error) {
var n int
for len(p) > 0 {
if r.encode {
n, err = r.encodeRead(p)
} else {
n, err = r.decodeRead(p)
}
nn += n
p = p[n:]
if n == 0 || err != nil {
break
}
}
return
}
func (r *edBody) WriteTo(w io.Writer) (int64, error) {
if r.encode {
return r.Body.WriteTo(w)
}
return r.Body.WriteTo(&edBodyWriter{edBody: r, w: w})
}
type edBodyWriter struct {
*edBody
w io.Writer
}
// Write the parameter p is sumx.FreamRead.data
func (r *edBodyWriter) Write(p []byte) (nn int, err error) {
if r.err != nil {
return 0, r.err
}
var n int
if r.cx >= 0 {
n = copy(r.cell[r.cx:], p)
nn += n
r.cx += n
if r.cx < r.hasher.Size() {
return
}
act := r.hasher.Sum(nil)
if !bytes.Equal(r.cell, act) {
r.err = checksumError(r.block, r.cell, act)
return 0, r.err
}
r.cx = -1
r.hasher = r.block.Hasher()
p = p[n:]
}
if r.remain <= 0 {
if nn == 0 {
err = io.EOF
}
return
}
blockSize := int(r.block.BlockSize)
tryRead := blockSize - r.nx
if tryRead > r.remain {
tryRead = r.remain
}
if len(p) > tryRead {
p = p[:tryRead]
}
n, err = r.w.Write(p)
r.hasher.Write(p[:n])
nn += n
r.nx += n
r.remain -= n
if r.nx == blockSize || r.remain == 0 {
r.cx = 0
r.nx = 0
}
return
}

345
blobstore/common/rpc2/checksum_test.go Normal file
View File

@ -0,0 +1,345 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
crand "crypto/rand"
"fmt"
"hash"
"hash/crc32"
"io"
mrand "math/rand"
"testing"
"github.com/stretchr/testify/require"
)
var defBlock = ChecksumBlock{
Algorithm: ChecksumAlgorithm_Crc_IEEE,
BlockSize: DefaultBlockSize,
}
type randReadWriter struct {
rhasher hash.Hash32
whasher hash.Hash32
}
func (r *randReadWriter) Read(p []byte) (int, error) {
crand.Read(p)
r.rhasher.Write(p)
return len(p), nil
}
func (r *randReadWriter) Write(p []byte) (int, error) {
r.whasher.Write(p)
return len(p), nil
}
type transReadWriter struct {
step int
off int
data []byte
}
func (t *transReadWriter) Read(p []byte) (int, error) {
if t.off >= len(t.data) {
return 0, io.EOF
}
n := copy(p, t.data[t.off:])
t.off += n
return n, nil
}
func (t *transReadWriter) ReadFrom(r io.Reader) (nn int64, err error) {
var n int
for t.off < len(t.data) {
p := t.data[t.off:]
if len(p) >= t.step {
p = p[:t.step]
}
n, err = r.Read(p)
nn += int64(n)
t.off += n
if err != nil {
break
}
}
return
}
func (t *transReadWriter) WriteTo(w io.Writer) (nn int64, err error) {
if t.off >= len(t.data) {
return 0, io.EOF
}
var n int
for {
p := t.data[t.off:]
if len(p) >= t.step {
p = p[:t.step]
}
n, err = w.Write(p)
nn += int64(n)
t.off += n
if t.off == len(t.data) {
return nn, nil
}
if n == 0 || err != nil {
return nn, err
}
}
}
func (t *transReadWriter) Close() error { return nil }
func TestChecksumUnmarshal(t *testing.T) {
b, err := defBlock.Marshal()
require.NoError(t, err)
_, err = unmarshalBlock(b[:len(b)-1])
require.Error(t, err)
block, err := unmarshalBlock(b)
require.NoError(t, err)
require.Equal(t, defBlock.Algorithm, block.Algorithm)
block.BlockSize = 0
b, err = block.Marshal()
require.NoError(t, err)
_, err = unmarshalBlock(b)
require.Error(t, err)
}
func TestEncodeDecodeBodyBase(t *testing.T) {
for _, step := range []int{1, 2, 3, 7, 16, 1 << 10, 64 << 10} {
for _, blockSize := range []int{128, 513, 1 << 10, 64 << 10} {
for idx := range [10]struct{}{} {
size := mrand.Intn(9<<14) + mrand.Intn(11) + 1
if idx == 0 {
size = blockSize - 1
}
logName := fmt.Sprintf("step: %d block:%d size:%d", step, blockSize, size)
block := ChecksumBlock{
Algorithm: ChecksumAlgorithm_Crc_IEEE,
BlockSize: uint32(blockSize),
}
clientBody := &randReadWriter{rhasher: crc32.NewIEEE()}
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: step, data: make([]byte, block.EncodeSize(int64(size)))}
nn, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err, logName)
require.Equal(t, int64(len(transBody.data)), nn)
_, err = encodeBody.Read(make([]byte, 1))
require.Equal(t, io.EOF, err, logName)
transBody.off = 0
decodeBody := newEdBody(block, transBody, size, false)
serverBody := &randReadWriter{whasher: crc32.NewIEEE()}
if idx%2 == 0 {
nn, err = decodeBody.WriteTo(serverBody)
require.NoError(t, err, logName)
require.Equal(t, block.EncodeSize(int64(size)), nn, logName)
require.Equal(t, clientBody.rhasher.Sum32(), serverBody.whasher.Sum32(), logName)
_, err = decodeBody.WriteTo(serverBody)
require.ErrorIs(t, io.EOF, err, logName)
} else {
b := make([]byte, size)
n, err := io.ReadFull(decodeBody, b)
require.NoError(t, err, logName)
require.Equal(t, size, n, logName)
serverBody.Write(b)
require.Equal(t, clientBody.rhasher.Sum32(), serverBody.whasher.Sum32(), logName)
_, err = decodeBody.Read(make([]byte, 1))
require.ErrorIs(t, io.EOF, err, logName)
}
}
}
}
}
func TestEncodeDecodeBodyMissmatch(t *testing.T) {
{
size := 12
clientBody := &randReadWriter{rhasher: crc32.NewIEEE()}
encodeBody := newEdBody(defBlock, clientNopBody(io.NopCloser(clientBody)), size, true)
require.Panics(t, func() { encodeBody.WriteTo(nil) })
transBody := &transReadWriter{step: 32, data: make([]byte, defBlock.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
transBody.data[10] += 1
decodeBody := newEdBody(defBlock, transBody, size, false)
b := make([]byte, size)
for range [3]struct{}{} {
_, err = io.ReadFull(decodeBody, b)
require.Error(t, err)
status, code, detail := DetectError(err)
require.Equal(t, 400, status, detail)
require.Equal(t, "Checksum", code, detail)
}
}
{
block := ChecksumBlock{Algorithm: ChecksumAlgorithm_Crc_IEEE, BlockSize: 1}
size := 12
clientBody := &randReadWriter{rhasher: crc32.NewIEEE()}
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: 1, data: make([]byte, block.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
transBody.data[10] += 1
decodeBody := newEdBody(block, transBody, size, false)
serverBody := &randReadWriter{whasher: crc32.NewIEEE()}
for range [3]struct{}{} {
_, err = decodeBody.WriteTo(serverBody)
require.Error(t, err)
status, code, detail := DetectError(err)
require.Equal(t, 400, status, detail)
require.Equal(t, "Checksum", code, detail)
}
}
}
func TestEncodeDecodeBodyNodata(t *testing.T) {
size := 12
clientBody := &randReadWriter{rhasher: crc32.NewIEEE()}
encodeBody := newEdBody(defBlock, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: 32, data: make([]byte, defBlock.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
transBody.data = transBody.data[1:]
decodeBody := newEdBody(defBlock, transBody, size, false)
b := make([]byte, size)
for range [3]struct{}{} {
_, err = io.ReadFull(decodeBody, b)
require.Error(t, err)
}
var block *ChecksumBlock
require.Equal(t, int64(10), block.EncodeSize(10))
block = &ChecksumBlock{}
require.Nil(t, block.Readable(nil))
}
type oneByte struct {
done bool
}
func (b *oneByte) Write(p []byte) (int, error) {
if b.done {
return 0, io.ErrShortWrite
}
b.done = true
return 1, io.ErrShortWrite
}
func TestEncodeDecodeBodyReadWriteto(t *testing.T) {
block := ChecksumBlock{Algorithm: ChecksumAlgorithm_Crc_IEEE, BlockSize: 3}
size := 12
clientBody := &randReadWriter{rhasher: crc32.NewIEEE()}
{
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: 1, data: make([]byte, block.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
decodeBody := newEdBody(block, transBody, size, false)
for range [12 / 2]struct{}{} {
decodeBody.WriteTo(&oneByte{})
_, err = io.ReadFull(decodeBody, make([]byte, 1))
require.NoError(t, err)
}
}
{
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: 1, data: make([]byte, block.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
transBody.data = transBody.data[:3]
decodeBody := newEdBody(block, transBody, size, false)
decodeBody.WriteTo(&oneByte{})
decodeBody.WriteTo(&oneByte{})
decodeBody.WriteTo(&oneByte{})
_, err = io.ReadFull(decodeBody, make([]byte, 1))
require.Error(t, err)
}
{
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), size, true)
transBody := &transReadWriter{step: 1, data: make([]byte, block.EncodeSize(int64(size)))}
_, err := transBody.ReadFrom(encodeBody)
require.NoError(t, err)
transBody.off = 0
transBody.data[1] += 1
decodeBody := newEdBody(block, transBody, size, false)
decodeBody.WriteTo(&oneByte{})
decodeBody.WriteTo(&oneByte{})
decodeBody.WriteTo(&oneByte{})
_, err = io.ReadFull(decodeBody, make([]byte, 1))
require.Error(t, err)
}
}
type noneReadWriter struct{}
func (r *noneReadWriter) Read(p []byte) (int, error) { return len(p), nil }
func (r *noneReadWriter) Write(p []byte) (int, error) { return len(p), nil }
func BenchmarkEncodeDecodeBody(b *testing.B) {
cases := []struct {
step, blockSize, size int
}{
{32 << 10, 4 << 10, 1 << 20},
{32 << 10, 32 << 10, 1 << 20},
{32 << 10, 64 << 10, 1 << 20},
{1 << 20, 64 << 10, 8 << 20},
{1 << 20, 512 << 10, 8 << 20},
{4 << 20, 1 << 20, 16 << 20},
}
for _, cs := range cases {
b.Run(fmt.Sprintf("step(%d)-block(%d)-size(%d)", cs.step, cs.blockSize, cs.size),
func(b *testing.B) {
block := ChecksumBlock{
Algorithm: ChecksumAlgorithm_Crc_IEEE,
BlockSize: uint32(cs.blockSize),
}
clientBody := &noneReadWriter{}
encodeBody := newEdBody(block, clientNopBody(io.NopCloser(clientBody)), cs.size, true)
buff := make([]byte, block.EncodeSize(int64(cs.size)))
b.ResetTimer()
for ii := 0; ii <= b.N; ii++ {
transBody := &transReadWriter{step: cs.step, data: buff}
transBody.ReadFrom(encodeBody)
transBody.off = 0
decodeBody := newEdBody(block, transBody, cs.size, false)
serverBody := &noneReadWriter{}
decodeBody.WriteTo(serverBody)
}
},
)
}
}

12
blobstore/common/rpc2/example/client.go
View File

@ -4,7 +4,6 @@ import (
"bytes"
"context"
"encoding/json"
"hash/crc32"
"io"
"net"
"time"
@ -53,18 +52,13 @@ func runClient() {
buff := []byte("ping")
req, _ := rpc2.NewRequest(context.Background(), "", "/ping", &para, bytes.NewReader(buff))
req.Trailer.SetLen("trailer-1", 1)
var crcString string
req.AfterBody = func() error {
req.Trailer.Set("trailer-1", "xX")
hash := crc32.NewIEEE()
hash.Write(buff)
crcString = string(hash.Sum(nil))
log.Info("internal crc", hash.Sum(nil))
return nil
}
req.OptionCrc()
req.Option(func(r *rpc2.Request) {
log.Info("request option")
log.Info("run request option")
})
var ret pingPara
@ -82,10 +76,6 @@ func runClient() {
log.Infof("after request header : %+v", req.Header.M)
log.Infof("after request trailer: %+v", req.Trailer.M)
if crcString != req.Trailer.Get("internal-crc") {
panic("crc")
}
got, err := io.ReadAll(resp.Body)
if err != nil {
panic(rpc2.ErrorString(err))

14
blobstore/common/rpc2/example/server.go
View File

@ -32,19 +32,19 @@ func handlePing(w rpc2.ResponseWriter, req *rpc2.Request) error {
log.Info(req.RequestHeader.ToString())
var para pingPara
para.Unmarshal(req.GetParameter())
w.SetContentLength(req.ContentLength)
w.SetContentLength(int64(len("response -> ")) + req.ContentLength)
buff := make([]byte, req.ContentLength)
if _, err := io.ReadFull(req.Body, buff); err != nil {
return err
}
if err := req.Body.Close(); err != nil {
return err
}
req.Body.Close()
log.Info("body :", string(buff))
log.Info("trailer:", req.Trailer.M)
w.WriteOK(&para)
w.Header().Set("ignored", "x") // ignore
log.Info(req.Trailer.M)
w.Write(buff)
return nil
_, err := w.Write(append([]byte("response -> "), buff...))
return err
}
func handleStream(_ rpc2.ResponseWriter, req *rpc2.Request) error {

4
blobstore/common/rpc2/header.go
View File

@ -25,8 +25,8 @@ const (
MaxHeaders = 1 << 10 // 1024
MaxHeaderLength = 4 << 10 // 4K
HeaderInternalPrefix = "internal-"
headerInternalCrc = HeaderInternalPrefix + "crc"
HeaderInternalPrefix = "internal-"
headerInternalChecksum = HeaderInternalPrefix + "stream-checksum"
)
func withinLen(s string) bool { return len(s) <= MaxHeaderLength }

50
blobstore/common/rpc2/request.go
View File

@ -17,7 +17,6 @@ package rpc2
import (
"context"
"fmt"
"hash/crc32"
"io"
"sync"
"time"
@ -66,6 +65,8 @@ type Request struct {
opts []OptionRequest
checksum *ChecksumBlock
Body Body
GetBody func() (io.ReadCloser, error) // client side
@ -93,12 +94,13 @@ func (req *Request) write(deadline time.Time) error {
var cell headerCell
cell.Set(reqHeaderSize)
size := _headerCell + reqHeaderSize + int(req.ContentLength) + req.Trailer.AllSize()
encodeLen := req.checksum.EncodeSize(req.ContentLength)
size := _headerCell + reqHeaderSize + int(encodeLen) + req.Trailer.AllSize()
req.conn.SetDeadline(deadline)
_, err := req.conn.SizedWrite(io.MultiReader(cell.Reader(),
req.RequestHeader.MarshalToReader(),
io.LimitReader(req.Body, req.ContentLength),
io.LimitReader(req.Body, encodeLen), // the body was encoded
req.trailerReader(),
), size)
return err
@ -122,9 +124,9 @@ func (req *Request) request(deadline time.Time) (*Response, error) {
}
}
resp.Body = makeBodyWithTrailer(
req.conn.NewSizedReader(int(resp.ContentLength)+resp.Trailer.AllSize(), frame),
resp.ContentLength, req, &resp.Trailer)
resp.Body = makeBodyWithTrailer(req.conn.NewSizedReader(
int(req.checksum.EncodeSize(resp.ContentLength))+resp.Trailer.AllSize(), frame),
req, resp.ContentLength)
return resp, nil
}
@ -141,24 +143,32 @@ func (req *Request) Option(opt OptionRequest) *Request {
}
func (req *Request) OptionCrc() *Request {
req.Header.Set(headerInternalCrc, "1")
return req.OptionChecksum(ChecksumBlock{
Algorithm: ChecksumAlgorithm_Crc_IEEE,
BlockSize: DefaultBlockSize,
})
}
func (req *Request) OptionChecksum(block ChecksumBlock) *Request {
if _, exist := algorithms[block.Algorithm]; !exist || block.BlockSize == 0 {
panic(fmt.Sprintf("rpc2: checksum(%s) not implements", block.String()))
}
if req.checksum != nil {
return req
}
cb, err := block.Marshal()
if err != nil {
return req
}
req.checksum = &block
req.Header.Set(headerInternalChecksum, string(cb))
if req.ContentLength == 0 {
return req
}
req.Trailer.SetLen(headerInternalCrc, 4)
req.opts = append(req.opts, func(r *Request) {
crc := crc32.NewIEEE()
r.Body = crcBody{
Reader: io.TeeReader(r.Body, crc),
WriterTo: r.Body,
Closer: r.Body,
}
afterBody := r.AfterBody
r.AfterBody = func() error {
req.Trailer.Set(headerInternalCrc, string(crc.Sum(nil)))
return afterBody()
}
req.opts = append(req.opts, func(r *Request) {
r.Body = newEdBody(block, r.Body, int(req.ContentLength), true)
})
return req
}

60
blobstore/common/rpc2/response.go
View File

@ -17,7 +17,6 @@ package rpc2
import (
"bytes"
"fmt"
"hash/crc32"
"io"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
@ -68,7 +67,7 @@ type response struct {
hasWroteHeader bool
midWriter io.Writer
bodyEncoder *edBody
remain int // body remain
toWrite int
@ -79,8 +78,8 @@ type response struct {
func (resp *response) SetContentLength(l int64) {
resp.hdr.ContentLength = l
resp.remain = int(l)
if l <= 0 {
resp.hdr.Trailer.Del(headerInternalCrc)
if resp.bodyEncoder != nil {
resp.bodyEncoder.remain = int(l)
}
}
@ -130,17 +129,17 @@ func (resp *response) Write(p []byte) (int, error) {
if resp.remain < len(p) {
p = p[:resp.remain]
}
resp.remain -= len(p)
resp.toWrite += len(p)
resp.toList = append(resp.toList, bytes.NewReader(p))
if resp.remain == 0 {
resp.toWrite += resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList, &trailerReader{
Fn: resp.afterBody,
Trailer: &resp.hdr.Trailer,
})
if resp.remain != len(p) {
return 0, io.ErrShortWrite
}
resp.midWriter.Write(p)
r, toWrite := resp.encodeBody(bytes.NewReader(p))
resp.toWrite += toWrite + resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList, r, &trailerReader{
Fn: resp.afterBody,
Trailer: &resp.hdr.Trailer,
})
resp.remain = 0
if err := resp.Flush(); err != nil {
return 0, err
}
@ -154,13 +153,12 @@ func (resp *response) ReadFrom(r io.Reader) (n int64, err error) {
}
}
remain := resp.remain
resp.toWrite += remain + resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList,
io.TeeReader(io.LimitReader(r, int64(remain)), resp.midWriter),
&trailerReader{
Fn: resp.afterBody,
Trailer: &resp.hdr.Trailer,
})
r, toWrite := resp.encodeBody(io.LimitReader(r, int64(remain)))
resp.toWrite += toWrite + resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList, r, &trailerReader{
Fn: resp.afterBody,
Trailer: &resp.hdr.Trailer,
})
resp.remain = 0
if err := resp.Flush(); err != nil {
return 0, err
@ -193,18 +191,20 @@ func (resp *response) AfterBody(fn func() error) {
}
func (resp *response) options(req *Request) {
resp.midWriter = io.MultiWriter()
if req.Header.Get(headerInternalCrc) != "" {
resp.hdr.Trailer.SetLen(headerInternalCrc, 4)
crc := crc32.NewIEEE()
resp.midWriter = crc
resp.afterBody = func() error {
resp.hdr.Trailer.Set(headerInternalCrc, string(crc.Sum(nil)))
return nil
}
if sum := req.Header.Get(headerInternalChecksum); sum != "" {
resp.hdr.Header.Set(headerInternalChecksum, sum)
resp.bodyEncoder = newEdBody(*req.checksum, nil, 0, true)
}
}
func (resp *response) encodeBody(r io.Reader) (io.Reader, int) {
if resp.bodyEncoder == nil {
return r, resp.remain
}
resp.bodyEncoder.Body = clientNopBody(io.NopCloser(r))
return resp.bodyEncoder, int(resp.bodyEncoder.block.EncodeSize(int64(resp.remain)))
}
func baseResponse() *Response {
return &Response{
ResponseHeader: ResponseHeader{

9
blobstore/common/rpc2/rpc2.go
View File

@ -87,12 +87,6 @@ func clientNopBody(rc io.ReadCloser) Body {
return nopBody{rc}
}
type crcBody struct {
io.Reader
io.WriterTo
io.Closer
}
var NoParameter Codec = noParameter{}
type noParameter struct{}
@ -103,12 +97,13 @@ func (noParameter) Unmarshal([]byte) error { return nil }
// LimitedWriter wrap Body with WriteTo
type LimitedWriter struct {
w io.Writer
a int64
n int64
e error
}
func LimitWriter(w io.Writer, limit int64) io.Writer {
return &LimitedWriter{w, limit, nil}
return &LimitedWriter{w, limit, limit, nil}
}
func (lw *LimitedWriter) Write(p []byte) (int, error) {

310
blobstore/common/rpc2/rpc2.pb.go
View File

@ -54,6 +54,34 @@ func (StreamCmd) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_af0916bb5e6806d0, []int{0}
}
type ChecksumAlgorithm int32
const (
ChecksumAlgorithm_NONE ChecksumAlgorithm = 0
ChecksumAlgorithm_Crc_IEEE ChecksumAlgorithm = 1
ChecksumAlgorithm_Crc_FAST ChecksumAlgorithm = 2
)
var ChecksumAlgorithm_name = map[int32]string{
0: "NONE",
1: "Crc_IEEE",
2: "Crc_FAST",
}
var ChecksumAlgorithm_value = map[string]int32{
"NONE": 0,
"Crc_IEEE": 1,
"Crc_FAST": 2,
}
func (x ChecksumAlgorithm) String() string {
return proto.EnumName(ChecksumAlgorithm_name, int32(x))
}
func (ChecksumAlgorithm) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_af0916bb5e6806d0, []int{1}
}
type Header struct {
M map[string]string `protobuf:"bytes,1,rep,name=m,proto3" json:"m,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"`
@ -512,8 +540,64 @@ func (m *Error) GetDetail() string {
return ""
}
type ChecksumBlock struct {
Algorithm ChecksumAlgorithm `protobuf:"varint,1,opt,name=algorithm,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumAlgorithm" json:"algorithm,omitempty"`
BlockSize uint32 `protobuf:"varint,2,opt,name=blockSize,proto3" json:"blockSize,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *ChecksumBlock) Reset() { *m = ChecksumBlock{} }
func (m *ChecksumBlock) String() string { return proto.CompactTextString(m) }
func (*ChecksumBlock) ProtoMessage() {}
func (*ChecksumBlock) Descriptor() ([]byte, []int) {
return fileDescriptor_af0916bb5e6806d0, []int{6}
}
func (m *ChecksumBlock) XXX_Unmarshal(b []byte) error {
return m.Unmarshal(b)
}
func (m *ChecksumBlock) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
if deterministic {
return xxx_messageInfo_ChecksumBlock.Marshal(b, m, deterministic)
} else {
b = b[:cap(b)]
n, err := m.MarshalToSizedBuffer(b)
if err != nil {
return nil, err
}
return b[:n], nil
}
}
func (m *ChecksumBlock) XXX_Merge(src proto.Message) {
xxx_messageInfo_ChecksumBlock.Merge(m, src)
}
func (m *ChecksumBlock) XXX_Size() int {
return m.Size()
}
func (m *ChecksumBlock) XXX_DiscardUnknown() {
xxx_messageInfo_ChecksumBlock.DiscardUnknown(m)
}
var xxx_messageInfo_ChecksumBlock proto.InternalMessageInfo
func (m *ChecksumBlock) GetAlgorithm() ChecksumAlgorithm {
if m != nil {
return m.Algorithm
}
return ChecksumAlgorithm_NONE
}
func (m *ChecksumBlock) GetBlockSize() uint32 {
if m != nil {
return m.BlockSize
}
return 0
}
func init() {
proto.RegisterEnum("cubefs.blobstore.common.rpc2.StreamCmd", StreamCmd_name, StreamCmd_value)
proto.RegisterEnum("cubefs.blobstore.common.rpc2.ChecksumAlgorithm", ChecksumAlgorithm_name, ChecksumAlgorithm_value)
proto.RegisterType((*Header)(nil), "cubefs.blobstore.common.rpc2.Header")
proto.RegisterMapType((map[string]string)(nil), "cubefs.blobstore.common.rpc2.Header.MEntry")
proto.RegisterType((*FixedValue)(nil), "cubefs.blobstore.common.rpc2.FixedValue")
@ -522,49 +606,55 @@ func init() {
proto.RegisterType((*RequestHeader)(nil), "cubefs.blobstore.common.rpc2.RequestHeader")
proto.RegisterType((*ResponseHeader)(nil), "cubefs.blobstore.common.rpc2.ResponseHeader")
proto.RegisterType((*Error)(nil), "cubefs.blobstore.common.rpc2.Error")
proto.RegisterType((*ChecksumBlock)(nil), "cubefs.blobstore.common.rpc2.ChecksumBlock")
}
func init() { proto.RegisterFile("rpc2.proto", fileDescriptor_af0916bb5e6806d0) }
var fileDescriptor_af0916bb5e6806d0 = []byte{
// 580 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x54, 0x5d, 0x6b, 0x13, 0x41,
0x14, 0xed, 0x64, 0xbb, 0x9b, 0xee, 0xad, 0xad, 0x61, 0xa8, 0x65, 0x29, 0x25, 0x5d, 0x42, 0xa1,
0xab, 0xe2, 0x56, 0x62, 0x1f, 0xfc, 0x00, 0xc1, 0xd8, 0x96, 0x16, 0xb4, 0x95, 0xa9, 0x08, 0xfa,
0x20, 0x6c, 0x76, 0xaf, 0x69, 0x70, 0x77, 0x27, 0xce, 0x4e, 0x8a, 0xfd, 0x3b, 0xfe, 0x0b, 0xff,
0x41, 0x1f, 0x7d, 0x15, 0x24, 0x48, 0x1e, 0xfd, 0x15, 0x32, 0xb3, 0x13, 0x93, 0x80, 0x86, 0xe8,
0xa3, 0x6f, 0xf7, 0x9c, 0xdc, 0x73, 0x73, 0xcf, 0x99, 0x9b, 0x00, 0x88, 0x5e, 0xdc, 0x0c, 0x7b,
0x82, 0x4b, 0x4e, 0x37, 0xe3, 0x7e, 0x1b, 0xdf, 0x15, 0x61, 0x3b, 0xe5, 0xed, 0x42, 0x72, 0x81,
0x61, 0xcc, 0xb3, 0x8c, 0xe7, 0xa1, 0xea, 0xd9, 0x58, 0xeb, 0xf0, 0x0e, 0xd7, 0x8d, 0xbb, 0xaa,
0x2a, 0x35, 0x8d, 0x4f, 0x04, 0x9c, 0x23, 0x8c, 0x12, 0x14, 0xf4, 0x01, 0x90, 0xcc, 0x23, 0xbe,
0x15, 0x2c, 0x37, 0x6f, 0x87, 0xb3, 0x46, 0x85, 0xa5, 0x20, 0x7c, 0x7e, 0x90, 0x4b, 0x71, 0xc9,
0x48, 0x46, 0x77, 0xc0, 0x29, 0x64, 0xd4, 0x4e, 0xd1, 0xab, 0xf8, 0x24, 0x58, 0x6a, 0x5d, 0x1f,
0x0e, 0xb6, 0x0c, 0xf3, 0x63, 0xb0, 0x45, 0xee, 0x30, 0x03, 0x36, 0xf6, 0xc0, 0x29, 0x55, 0xb4,
0x06, 0xd6, 0x7b, 0xbc, 0xf4, 0x88, 0x4f, 0x02, 0x97, 0xa9, 0x92, 0xae, 0x81, 0x7d, 0x11, 0xa5,
0xfd, 0x72, 0x86, 0xcb, 0x4a, 0xf0, 0xb0, 0x72, 0x9f, 0x34, 0xf6, 0x00, 0x0e, 0xbb, 0x1f, 0x31,
0x79, 0xa5, 0x18, 0xa5, 0x4c, 0x31, 0xd7, 0xca, 0x15, 0xa6, 0xca, 0xdf, 0x2b, 0x1b, 0xdf, 0x08,
0x2c, 0x6b, 0x99, 0xf1, 0xb7, 0x3f, 0xf6, 0x77, 0x77, 0xb6, 0xbf, 0x09, 0x95, 0x31, 0xd9, 0x5a,
0xbc, 0x1a, 0x6c, 0x2d, 0xfc, 0x95, 0xd5, 0xb7, 0x33, 0xac, 0x3e, 0x9e, 0x5c, 0x78, 0xb9, 0x19,
0xcc, 0xb1, 0x8e, 0xf6, 0x3e, 0x19, 0xca, 0x67, 0x0b, 0x56, 0x18, 0x7e, 0xe8, 0x63, 0x21, 0x8d,
0x41, 0x0f, 0xaa, 0x17, 0x28, 0x8a, 0x2e, 0x2f, 0xc3, 0xb1, 0xd9, 0x08, 0xaa, 0x80, 0xb2, 0xa8,
0xd3, 0x8d, 0xf5, 0xf7, 0xd9, 0xac, 0x04, 0xf4, 0x00, 0xdc, 0x42, 0x0a, 0x8c, 0xb2, 0xa7, 0x59,
0xe2, 0x59, 0x3e, 0x09, 0x56, 0x9b, 0x3b, 0xb3, 0x37, 0x39, 0x1b, 0xb5, 0xb3, 0xb1, 0x92, 0xd6,
0x01, 0x04, 0x66, 0x5c, 0xe2, 0x93, 0x24, 0x11, 0xde, 0xa2, 0x76, 0x39, 0xc1, 0xd0, 0x6d, 0x58,
0x29, 0xd1, 0x51, 0x94, 0x27, 0x29, 0x0a, 0xcf, 0xd6, 0x2d, 0xd3, 0xa4, 0x5a, 0x5e, 0x8a, 0x28,
0xc6, 0xe3, 0x7d, 0xcf, 0xd1, 0x9f, 0x8f, 0xa0, 0xd2, 0xc7, 0x3c, 0x97, 0x98, 0xcb, 0x67, 0x98,
0x77, 0xe4, 0xb9, 0x57, 0xf5, 0x49, 0x60, 0xb1, 0x69, 0x92, 0xb6, 0xc0, 0x39, 0xd7, 0x31, 0x78,
0x4b, 0x3a, 0xd3, 0xed, 0x79, 0x4e, 0xd8, 0x3c, 0xab, 0x51, 0xd2, 0x63, 0xbd, 0x43, 0x57, 0xed,
0xe8, 0xea, 0x21, 0x37, 0xe7, 0xbe, 0x13, 0x33, 0x69, 0xa4, 0xa7, 0x9b, 0xe0, 0xf6, 0x22, 0x11,
0x65, 0x28, 0x51, 0x78, 0xe0, 0x93, 0xe0, 0x1a, 0x1b, 0x13, 0x8d, 0xaf, 0x15, 0x58, 0x65, 0x58,
0xf4, 0x78, 0x5e, 0xe0, 0x3f, 0x3e, 0xde, 0xba, 0xbe, 0x43, 0xd9, 0x2f, 0x74, 0xe2, 0x36, 0x33,
0x48, 0xf1, 0x02, 0xa3, 0x82, 0xe7, 0x26, 0x66, 0x83, 0xd4, 0x14, 0x14, 0x82, 0x0b, 0x93, 0x6e,
0x09, 0xfe, 0xc7, 0x6c, 0x4f, 0xc1, 0x3e, 0xd0, 0xde, 0xc6, 0x09, 0x91, 0x3f, 0x24, 0x54, 0x99,
0x4a, 0x68, 0x1d, 0x9c, 0x04, 0x65, 0xd4, 0x4d, 0xf5, 0x6f, 0xc1, 0x65, 0x06, 0xdd, 0xda, 0x05,
0xf7, 0xd7, 0xdd, 0xd3, 0x2a, 0x58, 0x27, 0xa7, 0x2f, 0x6b, 0x0b, 0xaa, 0x38, 0x7b, 0x7d, 0x52,
0x23, 0xaa, 0x78, 0x71, 0x76, 0x54, 0xab, 0xa8, 0xe2, 0xf0, 0xf8, 0xa4, 0x66, 0xb5, 0x6e, 0x5c,
0x0d, 0xeb, 0xe4, 0xcb, 0xb0, 0x4e, 0xbe, 0x0f, 0xeb, 0xe4, 0x4d, 0x35, 0xdc, 0x7d, 0xa4, 0x1c,
0xb5, 0x1d, 0xfd, 0x8f, 0x7b, 0xef, 0x67, 0x00, 0x00, 0x00, 0xff, 0xff, 0xe2, 0x78, 0xe8, 0x45,
0xb3, 0x05, 0x00, 0x00,
// 670 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x55, 0xc1, 0x6a, 0xdb, 0x4a,
0x14, 0xcd, 0xd8, 0xb1, 0x6c, 0xdd, 0xc4, 0x79, 0x7a, 0x43, 0x5e, 0x10, 0x21, 0x38, 0xc6, 0x04,
0xa2, 0x97, 0xc7, 0x93, 0x8b, 0x9b, 0x45, 0xd3, 0x42, 0x21, 0x4e, 0x14, 0x62, 0x68, 0x9c, 0x32,
0x0e, 0x85, 0x76, 0xd1, 0x22, 0x4b, 0x53, 0x5b, 0x44, 0xd2, 0xb8, 0xa3, 0x71, 0x68, 0x4a, 0xbf,
0xa6, 0x7f, 0xd1, 0x3f, 0xc8, 0xb2, 0xdb, 0x42, 0x09, 0xc5, 0xcb, 0x7e, 0x45, 0x99, 0x91, 0x54,
0x3b, 0xb4, 0x35, 0x69, 0x97, 0xdd, 0xdd, 0x73, 0x7d, 0xcf, 0xd1, 0x3d, 0x47, 0x33, 0x16, 0x00,
0x1f, 0x79, 0x2d, 0x7b, 0xc4, 0x99, 0x60, 0x78, 0xc3, 0x1b, 0xf7, 0xe9, 0xcb, 0xc4, 0xee, 0x87,
0xac, 0x9f, 0x08, 0xc6, 0xa9, 0xed, 0xb1, 0x28, 0x62, 0xb1, 0x2d, 0x67, 0xd6, 0x57, 0x07, 0x6c,
0xc0, 0xd4, 0x60, 0x53, 0x56, 0x29, 0xa7, 0xf1, 0x0e, 0x81, 0x76, 0x4c, 0x5d, 0x9f, 0x72, 0xbc,
0x07, 0x28, 0x32, 0x51, 0xbd, 0x68, 0x2d, 0xb5, 0xfe, 0xb3, 0xe7, 0x49, 0xd9, 0x29, 0xc1, 0x3e,
0x71, 0x62, 0xc1, 0x2f, 0x09, 0x8a, 0xf0, 0x36, 0x68, 0x89, 0x70, 0xfb, 0x21, 0x35, 0x0b, 0x75,
0x64, 0x55, 0xda, 0x7f, 0x4d, 0xae, 0x37, 0xb3, 0xce, 0x97, 0xeb, 0x4d, 0xf4, 0x3f, 0xc9, 0xc0,
0xfa, 0x2e, 0x68, 0x29, 0x0b, 0x1b, 0x50, 0x3c, 0xa7, 0x97, 0x26, 0xaa, 0x23, 0x4b, 0x27, 0xb2,
0xc4, 0xab, 0x50, 0xba, 0x70, 0xc3, 0x71, 0xaa, 0xa1, 0x93, 0x14, 0xdc, 0x2f, 0xdc, 0x43, 0x8d,
0x5d, 0x80, 0xa3, 0xe0, 0x35, 0xf5, 0x9f, 0xc8, 0x8e, 0x64, 0x86, 0x34, 0x56, 0xcc, 0x2a, 0x91,
0xe5, 0x8f, 0x99, 0x8d, 0x4f, 0x08, 0x96, 0x14, 0x2d, 0xf3, 0x77, 0x38, 0xf5, 0x77, 0x67, 0xbe,
0xbf, 0x19, 0x56, 0x66, 0xb2, 0xbd, 0x78, 0x75, 0xbd, 0xb9, 0xf0, 0x4b, 0x56, 0x9f, 0xcf, 0xb1,
0xfa, 0x70, 0x76, 0xe1, 0xa5, 0x96, 0x75, 0x8b, 0x75, 0x94, 0xf7, 0xd9, 0x50, 0xde, 0x17, 0xa1,
0x4a, 0xe8, 0xab, 0x31, 0x4d, 0x44, 0x66, 0xd0, 0x84, 0xf2, 0x05, 0xe5, 0x49, 0xc0, 0xd2, 0x70,
0x4a, 0x24, 0x87, 0x32, 0xa0, 0xc8, 0x1d, 0x04, 0x9e, 0x7a, 0x5e, 0x89, 0xa4, 0x00, 0x3b, 0xa0,
0x27, 0x82, 0x53, 0x37, 0x3a, 0x88, 0x7c, 0xb3, 0x58, 0x47, 0xd6, 0x4a, 0x6b, 0x7b, 0xfe, 0x26,
0xbd, 0x7c, 0x9c, 0x4c, 0x99, 0xb8, 0x06, 0xc0, 0x69, 0xc4, 0x04, 0xdd, 0xf7, 0x7d, 0x6e, 0x2e,
0x2a, 0x97, 0x33, 0x1d, 0xbc, 0x05, 0xd5, 0x14, 0x1d, 0xbb, 0xb1, 0x1f, 0x52, 0x6e, 0x96, 0xd4,
0xc8, 0xcd, 0xa6, 0x5c, 0x5e, 0x70, 0xd7, 0xa3, 0x9d, 0x43, 0x53, 0x53, 0xbf, 0xe7, 0x50, 0xf2,
0x3d, 0x16, 0x0b, 0x1a, 0x8b, 0x47, 0x34, 0x1e, 0x88, 0xa1, 0x59, 0xae, 0x23, 0xab, 0x48, 0x6e,
0x36, 0x71, 0x1b, 0xb4, 0xa1, 0x8a, 0xc1, 0xac, 0xa8, 0x4c, 0xb7, 0x6e, 0x73, 0x84, 0xb3, 0xd7,
0x9a, 0x31, 0x71, 0x47, 0xed, 0x10, 0xc8, 0x1d, 0x75, 0x25, 0xf2, 0xef, 0xad, 0xcf, 0x49, 0xa6,
0x94, 0xf3, 0xf1, 0x06, 0xe8, 0x23, 0x97, 0xbb, 0x11, 0x15, 0x94, 0x9b, 0x50, 0x47, 0xd6, 0x32,
0x99, 0x36, 0x1a, 0x1f, 0x0b, 0xb0, 0x42, 0x68, 0x32, 0x62, 0x71, 0x42, 0x7f, 0xf3, 0xe5, 0xad,
0xa9, 0x73, 0x28, 0xc6, 0x89, 0x4a, 0xbc, 0x44, 0x32, 0x24, 0xfb, 0x9c, 0xba, 0x09, 0x8b, 0xb3,
0x98, 0x33, 0x24, 0x55, 0x28, 0xe7, 0x8c, 0x67, 0xe9, 0xa6, 0xe0, 0x4f, 0xcc, 0xf6, 0x14, 0x4a,
0x8e, 0xf2, 0x36, 0x4d, 0x08, 0xfd, 0x24, 0xa1, 0xc2, 0x8d, 0x84, 0xd6, 0x40, 0xf3, 0xa9, 0x70,
0x83, 0x50, 0xdd, 0x05, 0x9d, 0x64, 0xa8, 0xf1, 0x16, 0xaa, 0x07, 0x43, 0xea, 0x9d, 0x27, 0xe3,
0xa8, 0x1d, 0x32, 0xef, 0x1c, 0x9f, 0x80, 0xee, 0x86, 0x03, 0xc6, 0x03, 0x31, 0x8c, 0x94, 0xf6,
0x4a, 0xab, 0x39, 0xdf, 0x4c, 0xce, 0xdf, 0xcf, 0x69, 0x64, 0xaa, 0x20, 0xed, 0xf4, 0xa5, 0x6e,
0x2f, 0x78, 0x93, 0xfe, 0x21, 0x54, 0xc9, 0xb4, 0xb1, 0xd3, 0x04, 0xfd, 0xdb, 0xad, 0xc3, 0x65,
0x28, 0x76, 0x4f, 0xcf, 0x8c, 0x05, 0x59, 0xf4, 0x9e, 0x76, 0x0d, 0x24, 0x8b, 0xc7, 0xbd, 0x63,
0xa3, 0x20, 0x8b, 0xa3, 0x4e, 0xd7, 0x28, 0xee, 0xec, 0xc1, 0xdf, 0xdf, 0x3d, 0x0e, 0x57, 0x60,
0xb1, 0x7b, 0xda, 0x75, 0x8c, 0x05, 0xbc, 0x0c, 0x95, 0x03, 0xee, 0xbd, 0xe8, 0x38, 0x8e, 0x63,
0xa0, 0x1c, 0x1d, 0xed, 0xf7, 0xce, 0x8c, 0x42, 0xfb, 0x9f, 0xab, 0x49, 0x0d, 0x7d, 0x98, 0xd4,
0xd0, 0xe7, 0x49, 0x0d, 0x3d, 0x2b, 0xdb, 0xcd, 0x07, 0x72, 0xfb, 0xbe, 0xa6, 0x3e, 0x15, 0x77,
0xbf, 0x06, 0x00, 0x00, 0xff, 0xff, 0xfd, 0x28, 0xeb, 0x96, 0x6c, 0x06, 0x00, 0x00,
}
func (m *Header) Marshal() (dAtA []byte, err error) {
@ -952,6 +1042,43 @@ func (m *Error) MarshalToSizedBuffer(dAtA []byte) (int, error) {
return len(dAtA) - i, nil
}
func (m *ChecksumBlock) Marshal() (dAtA []byte, err error) {
size := m.Size()
dAtA = make([]byte, size)
n, err := m.MarshalToSizedBuffer(dAtA[:size])
if err != nil {
return nil, err
}
return dAtA[:n], nil
}
func (m *ChecksumBlock) MarshalTo(dAtA []byte) (int, error) {
size := m.Size()
return m.MarshalToSizedBuffer(dAtA[:size])
}
func (m *ChecksumBlock) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i := len(dAtA)
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.BlockSize != 0 {
i = encodeVarintRpc2(dAtA, i, uint64(m.BlockSize))
i--
dAtA[i] = 0x10
}
if m.Algorithm != 0 {
i = encodeVarintRpc2(dAtA, i, uint64(m.Algorithm))
i--
dAtA[i] = 0x8
}
return len(dAtA) - i, nil
}
func encodeVarintRpc2(dAtA []byte, offset int, v uint64) int {
offset -= sovRpc2(v)
base := offset
@ -1136,6 +1263,24 @@ func (m *Error) Size() (n int) {
return n
}
func (m *ChecksumBlock) Size() (n int) {
if m == nil {
return 0
}
var l int
_ = l
if m.Algorithm != 0 {
n += 1 + sovRpc2(uint64(m.Algorithm))
}
if m.BlockSize != 0 {
n += 1 + sovRpc2(uint64(m.BlockSize))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
func sovRpc2(x uint64) (n int) {
return (math_bits.Len64(x|1) + 6) / 7
}
@ -2390,6 +2535,95 @@ func (m *Error) Unmarshal(dAtA []byte) error {
}
return nil
}
func (m *ChecksumBlock) Unmarshal(dAtA []byte) error {
l := len(dAtA)
iNdEx := 0
for iNdEx < l {
preIndex := iNdEx
var wire uint64
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRpc2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
wire |= uint64(b&0x7F) << shift
if b < 0x80 {
break
}
}
fieldNum := int32(wire >> 3)
wireType := int(wire & 0x7)
if wireType == 4 {
return fmt.Errorf("proto: ChecksumBlock: wiretype end group for non-group")
}
if fieldNum <= 0 {
return fmt.Errorf("proto: ChecksumBlock: illegal tag %d (wire type %d)", fieldNum, wire)
}
switch fieldNum {
case 1:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field Algorithm", wireType)
}
m.Algorithm = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRpc2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.Algorithm |= ChecksumAlgorithm(b&0x7F) << shift
if b < 0x80 {
break
}
}
case 2:
if wireType != 0 {
return fmt.Errorf("proto: wrong wireType = %d for field BlockSize", wireType)
}
m.BlockSize = 0
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRpc2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
m.BlockSize |= uint32(b&0x7F) << shift
if b < 0x80 {
break
}
}
default:
iNdEx = preIndex
skippy, err := skipRpc2(dAtA[iNdEx:])
if err != nil {
return err
}
if (skippy < 0) || (iNdEx+skippy) < 0 {
return ErrInvalidLengthRpc2
}
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
if iNdEx > l {
return io.ErrUnexpectedEOF
}
return nil
}
func skipRpc2(dAtA []byte) (n int, err error) {
l := len(dAtA)
iNdEx := 0

11
blobstore/common/rpc2/rpc2.proto
View File

@ -81,3 +81,14 @@ message Error {
string reason = 2;
string detail = 3;
}
enum ChecksumAlgorithm {
NONE = 0;
Crc_IEEE = 1;
Crc_FAST = 2;
}
message ChecksumBlock {
ChecksumAlgorithm algorithm = 1;
uint32 blockSize = 2;
}

18
blobstore/common/rpc2/server.go
View File

@ -233,7 +233,9 @@ func (s *Server) handleStream(stream *transport.Stream) {
} else {
status, reason, detail := DetectError(err)
resp.hdr.Reason = reason
resp.hdr.Error = detail.Error()
if detail != nil {
resp.hdr.Error = detail.Error()
}
resp.WriteHeader(status, NoParameter)
}
}
@ -274,9 +276,17 @@ func (s *Server) readRequest(stream *transport.Stream) (*Request, error) {
_, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", traceID)
req := &Request{RequestHeader: hdr, ctx: ctx, conn: stream}
req.Body = makeBodyWithTrailer(
stream.NewSizedReader(int(req.ContentLength)+req.Trailer.AllSize(), frame),
req.ContentLength, nil, &req.Trailer)
if sum := hdr.Header.Get(headerInternalChecksum); sum != "" {
block, err := unmarshalBlock([]byte(sum))
if err != nil {
frame.Close()
return nil, err
}
req.checksum = block
}
req.Body = makeBodyWithTrailer(stream.NewSizedReader(
int(req.checksum.EncodeSize(req.ContentLength))+req.Trailer.AllSize(), frame),
req, req.ContentLength)
if hdr.StreamCmd == StreamCmd_SYN {
req.stream = &serverStream{req: req}