-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Integrate zstd compression into chain exchange (#842)
* Integrate zstd compression into chain exchange The GPBFT message exchange over pubsub already uses zstd compression on top of CBOR encoded messages. The work here integrates the same style of compression for chain exchange messages, with additional unification of the encoding mechanism across the two. The work refactors the root level encoding implementation into a generic encoder decoder that both chain exchange and gpbft used. Tests and benchmarks are updated to reflect this. The benchmarking of partial gmessage encoding is also adjusted to fix a few redundant statements and bugs in testing. Fixes #819 * Strictly Limit the size of decompressed values to 1 MiB The default message size limit in GossipSub is 1 MiB, which is unchanged in Lotus. This means when decompressing values, we can never have a valid compressed message that expands to larger than 1 MiB. Set this limit explicitly in the zstd decoder. * Massage the flaky test to submission
- Loading branch information
Showing
7 changed files
with
212 additions
and
112 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| package encoding | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "fmt" | ||
|
|
||
| "github.com/klauspost/compress/zstd" | ||
| cbg "github.com/whyrusleeping/cbor-gen" | ||
| ) | ||
|
|
||
| // maxDecompressedSize is the default maximum amount of memory allocated by the | ||
| // zstd decoder. The limit of 1MiB is chosen based on the default maximum message | ||
| // size in GossipSub. | ||
| const maxDecompressedSize = 1 << 20 | ||
|
|
||
| type CBORMarshalUnmarshaler interface { | ||
| cbg.CBORMarshaler | ||
| cbg.CBORUnmarshaler | ||
| } | ||
|
|
||
| type EncodeDecoder[T CBORMarshalUnmarshaler] interface { | ||
| Encode(v T) ([]byte, error) | ||
| Decode([]byte, T) error | ||
| } | ||
|
|
||
| type CBOR[T CBORMarshalUnmarshaler] struct{} | ||
|
|
||
| func NewCBOR[T CBORMarshalUnmarshaler]() *CBOR[T] { | ||
| return &CBOR[T]{} | ||
| } | ||
|
|
||
| func (c *CBOR[T]) Encode(m T) ([]byte, error) { | ||
| var buf bytes.Buffer | ||
| if err := m.MarshalCBOR(&buf); err != nil { | ||
| return nil, err | ||
| } | ||
| return buf.Bytes(), nil | ||
| } | ||
|
|
||
| func (c *CBOR[T]) Decode(v []byte, t T) error { | ||
| r := bytes.NewReader(v) | ||
| return t.UnmarshalCBOR(r) | ||
| } | ||
|
|
||
| type ZSTD[T CBORMarshalUnmarshaler] struct { | ||
| cborEncoding *CBOR[T] | ||
| compressor *zstd.Encoder | ||
| decompressor *zstd.Decoder | ||
| } | ||
|
|
||
| func NewZSTD[T CBORMarshalUnmarshaler]() (*ZSTD[T], error) { | ||
| writer, err := zstd.NewWriter(nil) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| reader, err := zstd.NewReader(nil, zstd.WithDecoderMaxMemory(maxDecompressedSize)) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &ZSTD[T]{ | ||
| cborEncoding: &CBOR[T]{}, | ||
| compressor: writer, | ||
| decompressor: reader, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *ZSTD[T]) Encode(m T) ([]byte, error) { | ||
| cborEncoded, err := c.cborEncoding.Encode(m) | ||
| if len(cborEncoded) > maxDecompressedSize { | ||
| // Error out early if the encoded value is too large to be decompressed. | ||
| return nil, fmt.Errorf("encoded value cannot exceed maximum size: %d > %d", len(cborEncoded), maxDecompressedSize) | ||
| } | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| compressed := c.compressor.EncodeAll(cborEncoded, make([]byte, 0, len(cborEncoded))) | ||
| return compressed, nil | ||
| } | ||
|
|
||
| func (c *ZSTD[T]) Decode(v []byte, t T) error { | ||
| cborEncoded, err := c.decompressor.DecodeAll(v, make([]byte, 0, len(v))) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return c.cborEncoding.Decode(cborEncoded, t) | ||
| } |
Oops, something went wrong.