Skip to content

Commit

Permalink
Merge pull request #244 from ktock/decompressjs2
Browse files Browse the repository at this point in the history 
imagemounter: Decompress layers in JS wrapper
  • Loading branch information
@ktock
ktock authored Mar 5, 2024
2 parents 58847de + 446875c commit 43e793f
Show file tree
Hide file tree
Showing 6 changed files with 468 additions and 42 deletions.
8 changes: 5 additions & 3 deletions extras/imagemounter/go.mod
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ require (
github.com/opencontainers/runc v1.1.5
github.com/opencontainers/runtime-spec v1.1.0
github.com/sirupsen/logrus v1.9.3
golang.org/x/sync v0.4.0
golang.org/x/sync v0.6.0
)

require (
Expand All @@ -37,7 +37,7 @@ require (
github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/hashicorp/go-retryablehttp v0.7.4 // indirect
github.com/insomniacslk/dhcp v0.0.0-20220504074936-1ca156eafb9f // indirect
github.com/klauspost/compress v1.17.2 // indirect
github.com/klauspost/compress v1.17.7 // indirect
github.com/miekg/dns v1.1.55 // indirect
github.com/moby/locker v1.0.1 // indirect
github.com/moby/sys/mountinfo v0.6.2 // indirect
Expand Down Expand Up @@ -70,6 +70,8 @@ replace github.com/u-root/uio => github.com/ktock/u-root-uio v0.0.0-202309111429

replace github.com/hugelgupf/p9 => github.com/ktock/p9 v0.0.0-20240109162600-dac0f97fe886

replace github.com/containerd/stargz-snapshotter => github.com/ktock/stargz-snapshotter v0.0.1-0.20240109162903-e5206b85e4cc
replace github.com/containerd/stargz-snapshotter => github.com/ktock/stargz-snapshotter v0.0.1-0.20240304123548-c3fcce188d68

replace github.com/containerd/containerd => github.com/ktock/containerd v1.2.1-0.20240109162750-28ec64e033db

replace github.com/containerd/stargz-snapshotter/estargz => github.com/ktock/stargz-snapshotter/estargz v0.0.0-20240304123548-c3fcce188d68
16 changes: 8 additions & 8 deletions extras/imagemounter/go.sum
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,6 @@ github.com/containerd/continuity v0.4.3 h1:6HVkalIp+2u1ZLH1J/pYX2oBVXlJZvh1X1A7b
github.com/containerd/continuity v0.4.3/go.mod h1:F6PTNCKepoxEaXLQp3wDAjygEnImnZ/7o4JzpodfroQ=
github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
github.com/containerd/stargz-snapshotter/estargz v0.15.1 h1:eXJjw9RbkLFgioVaTG+G/ZW/0kEe2oEKCdS/ZxIyoCU=
github.com/containerd/stargz-snapshotter/estargz v0.15.1/go.mod h1:gr2RNwukQ/S9Nv33Lt6UC7xEx58C+LHRdoqbEKjz1Kk=
github.com/containerd/ttrpc v1.2.2 h1:9vqZr0pxwOF5koz6N0N3kJ0zDHokrcPxIR/ZR2YFtOs=
github.com/containerd/ttrpc v1.2.2/go.mod h1:sIT6l32Ph/H9cvnJsfXM5drIVzTr5A2flTf1G5tYZak=
github.com/containerd/typeurl/v2 v2.1.1 h1:3Q4Pt7i8nYwy2KmQWIw2+1hTvwTE/6w9FqcttATPO/4=
Expand Down Expand Up @@ -146,8 +144,8 @@ github.com/kaey/framebuffer v0.0.0-20140402104929-7b385489a1ff/go.mod h1:tS4qtlc
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.10.6/go.mod h1:aoV0uJVorq1K+umq18yTdKaF57EivdYsUV+/s2qKfXs=
github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
github.com/klauspost/compress v1.17.7 h1:ehO88t2UGzQK66LMdE8tibEd1ErmzZjNEqWkjLAKQQg=
github.com/klauspost/compress v1.17.7/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/klauspost/pgzip v1.2.4/go.mod h1:Ch1tH69qFZu15pkjo5kYi6mth2Zzwzt50oCQKQE9RUs=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
Expand All @@ -160,8 +158,10 @@ github.com/ktock/insomniacslk-dhcp v0.0.0-20230911142651-b86573a014b1 h1:KD92SLh
github.com/ktock/insomniacslk-dhcp v0.0.0-20230911142651-b86573a014b1/go.mod h1:h+MxyHxRg9NH3terB1nfRIUaQEcI0XOVkdR9LNBlp8E=
github.com/ktock/p9 v0.0.0-20240109162600-dac0f97fe886 h1:w26G030q4XUqgRPaGfQcyFi1a5R4Nt9VC+6P07c7CrU=
github.com/ktock/p9 v0.0.0-20240109162600-dac0f97fe886/go.mod h1:d5H7Bc5VS6NgKyPmAGLB6qMj5rvRpP/u0HaVI5pp7kY=
github.com/ktock/stargz-snapshotter v0.0.1-0.20240109162903-e5206b85e4cc h1:EXrPMV/2hLjwbJMTVSM1y6a/AnPxLLD5sXEY7R832zE=
github.com/ktock/stargz-snapshotter v0.0.1-0.20240109162903-e5206b85e4cc/go.mod h1:74D+J1m1RMXytLmWxegXWhtOSRHPWZKpKc2NdK3S+us=
github.com/ktock/stargz-snapshotter v0.0.1-0.20240304123548-c3fcce188d68 h1:qzxytQYHEjdoqEMcAvZGAZCiy78zhGT5K+5afuX8/Uo=
github.com/ktock/stargz-snapshotter v0.0.1-0.20240304123548-c3fcce188d68/go.mod h1:74D+J1m1RMXytLmWxegXWhtOSRHPWZKpKc2NdK3S+us=
github.com/ktock/stargz-snapshotter/estargz v0.0.0-20240304123548-c3fcce188d68 h1:WwJnVlL+6g3VF+NNluCrAy2BRiMLwEJH4SMqDrbKxMc=
github.com/ktock/stargz-snapshotter/estargz v0.0.0-20240304123548-c3fcce188d68/go.mod h1:gr2RNwukQ/S9Nv33Lt6UC7xEx58C+LHRdoqbEKjz1Kk=
github.com/ktock/u-root-uio v0.0.0-20230911142931-5cf720bc8a29 h1:BNd7VYl9yNjxsA4Bt9YKAyKJKIymo1v9f7M2cWhh9iU=
github.com/ktock/u-root-uio v0.0.0-20230911142931-5cf720bc8a29/go.mod h1:LpEX5FO/cB+WF4TYGY1V5qktpaZLkKkSegbr0V4eYXA=
github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
Expand Down Expand Up @@ -332,8 +332,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ=
golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
Expand Down
163 changes: 141 additions & 22 deletions extras/imagemounter/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@ package main
import (
"archive/tar"
"bytes"
"compress/gzip"
"context"
"crypto"
"crypto/ecdsa"
Expand Down Expand Up @@ -280,8 +279,8 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
return nil, fmt.Errorf("failed to send request")
}

var buf []byte = make([]byte, 1048576)
var isEOF uint32
var reqBodyD []byte = make([]byte, 1048576)
var nwritten uint32 = 0
idx := 0
chunksize := 0
Expand All @@ -292,7 +291,7 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
for {
if idx >= chunksize {
// chunk is fully written. full another one.
chunksize, err = bodyR.Read(reqBodyD)
chunksize, err = bodyR.Read(buf)
if err != nil && err != io.EOF {
return nil, err
}
Expand All @@ -303,7 +302,7 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
}
res := http_writebody(
id,
uint32(uintptr(unsafe.Pointer(&[]byte(reqBodyD[idx:])[0]))),
uint32(uintptr(unsafe.Pointer(&[]byte(buf[idx:])[0]))),
uint32(chunksize),
uint32(uintptr(unsafe.Pointer(&nwritten))),
isEOF,
Expand Down Expand Up @@ -333,22 +332,21 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
time.Sleep(5 * time.Millisecond)
}

var respD []byte = make([]byte, 1048576)
var respsize uint32
var respFull []byte
isEOF = 0
for {
res := http_recv(
id,
uint32(uintptr(unsafe.Pointer(&[]byte(respD)[0]))),
1048576,
uint32(uintptr(unsafe.Pointer(&[]byte(buf)[0]))),
uint32(len(buf)),
uint32(uintptr(unsafe.Pointer(&respsize))),
uint32(uintptr(unsafe.Pointer(&isEOF))),
)
if res != 0 {
return nil, fmt.Errorf("failed to receive response")
}
respFull = append(respFull, respD[:int(respsize)]...)
respFull = append(respFull, buf[:int(respsize)]...)
if isEOF == 1 {
break
}
Expand All @@ -357,17 +355,17 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
if err := json.Unmarshal(respFull, &resp); err != nil {
return nil, err
}
respFull = nil

isEOF = 0
pr, pw := io.Pipe()
go func() {
var body []byte = make([]byte, 1048576)
var bodysize uint32
for {
res := http_readbody(
id,
uint32(uintptr(unsafe.Pointer(&[]byte(body)[0]))),
1048576,
uint32(uintptr(unsafe.Pointer(&[]byte(buf)[0]))),
uint32(len(buf)),
uint32(uintptr(unsafe.Pointer(&bodysize))),
uint32(uintptr(unsafe.Pointer(&isEOF))),
)
Expand All @@ -376,7 +374,7 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
return
}
if bodysize > 0 {
if _, err := pw.Write(body[:int(bodysize)]); err != nil {
if _, err := pw.Write(buf[:int(bodysize)]); err != nil {
pw.CloseWithError(err)
return
}
Expand All @@ -385,6 +383,7 @@ func doHttpRoundTrip(req *http.Request) (*http.Response, error) {
break
}
}
buf = nil
pw.Close()
}()
r := fetchResponseToHTTPResponse(req, &resp)
Expand Down Expand Up @@ -732,7 +731,6 @@ func fsFromImage(ctx context.Context, addr string, platform imagespec.Platform,
}, nil, map[string]esgzremote.Handler{
"url-reader": &layerOCILayoutURLHandler{addr},
}, reference.Spec{}, newLayerOCILayoutExternalReaderAt(addr))
// }, reference.Spec{}, newLayerOCILayoutURLReader(addr))
if err != nil {
return nil, nil, nil, nil, err
}
Expand Down Expand Up @@ -760,16 +758,44 @@ func fsFromImage(ctx context.Context, addr string, platform imagespec.Platform,
defer r.Close()
if withDecompression {
raw := r
zr, err := gzip.NewReader(raw)
zr, err := newWasmDecompressor(raw)
if err != nil {
return nil, err
}
defer zr.Close()
r = zr
r = &readerWithCloser{zr, func() error {
return raw.Close()
}}
}
data, err := io.ReadAll(r)
if err != nil {
return nil, err

var rs [][]byte
var curoff int
var chunkSize = 10 * 1024 * 1024
b := make([]byte, chunkSize)
for {
n, err := r.Read(b[curoff:])
curoff += n
if err != nil {
if err == io.EOF {
rs = append(rs, b[:curoff])
break
}
return nil, err
}
if curoff == len(b) {
rs = append(rs, b)
curoff = 0
b = make([]byte, chunkSize)
}
}
r.Close()
var totalLen int
for _, s := range rs {
totalLen += len(s)
}
data := make([]byte, totalLen)
var i int
for _, s := range rs {
i += copy(data[i:], s)
}
return io.NewSectionReader(bytes.NewReader(data), 0, int64(len(data))), nil
})
Expand Down Expand Up @@ -887,8 +913,8 @@ func fetchManifestAndConfigOCILayout(ctx context.Context, addr string, platform
return manifest, config, configD, nil
}

func fetchLayers(ctx context.Context, manifest imagespec.Manifest, config EStargzLayerConfig, hosts esgzsource.RegistryHosts, handlers map[string]esgzremote.Handler, refspec reference.Spec, unlazyReader func(imagespec.Descriptor, bool) (io.ReaderAt, error)) ([]NodeLayer, func(), error) {
layers, waitInit, err := getEStargzLayers(ctx, manifest, config, hosts, handlers, refspec)
func fetchLayers(ctx context.Context, manifest imagespec.Manifest, config EStargzLayerConfig, hosts esgzsource.RegistryHosts, handlers map[string]esgzremote.Handler, refspec reference.Spec, unlazyReader func(imagespec.Descriptor, bool) (io.ReaderAt, error)) (layers []NodeLayer, waitInit func(), err error) {
layers, waitInit, err = getEStargzLayers(ctx, manifest, config, hosts, handlers, refspec)
if err == nil {
return layers, waitInit, nil
}
Expand Down Expand Up @@ -1094,7 +1120,9 @@ func getEStargzLayers(ctx context.Context, manifest imagespec.Manifest, config E
tm.DoPrioritizedTask()
defer tm.DonePrioritizedTask()
return b.ReadAt(p, offset)
}), 0, b.Size()))
}), 0, b.Size()),
esgzmetadata.WithDecompressors(newGzipDecompressor()),
)
if err != nil {
return err
}
Expand Down Expand Up @@ -2301,3 +2329,94 @@ func headerToAttr(h *tar.Header) (p9.Attr, p9.QID) {
}
return out, q
}

//go:wasmimport env decompress_init
func decompress_init(idP uint32) uint32

//go:wasmimport env decompress_write
func decompress_write(id uint32, bufP uint32, buflen uint32, isEOF uint32) uint32

//go:wasmimport env decompress_read
func decompress_read(id uint32, bufP uint32, buflen uint32, recvLenP uint32, isEOFP uint32) uint32

type wasmDecompressor struct {
id uint32
raw io.Reader
rawEOF bool
}

func newWasmDecompressor(r io.Reader) (io.Reader, error) {
var id uint32
res := decompress_init(uint32(uintptr(unsafe.Pointer(&id))))
if res != 0 {
return nil, fmt.Errorf("failed to init decompressor")
}
return &wasmDecompressor{id: id, raw: r}, nil
}

func (d *wasmDecompressor) Read(p []byte) (n int, _ error) {
if !d.rawEOF {
n, err := d.raw.Read(p)
if err != nil && err != io.EOF {
return n, err
}
var isEOF = uint32(0)
if err == io.EOF {
isEOF = 1
d.rawEOF = true
}
res := decompress_write(d.id,
uint32(uintptr(unsafe.Pointer(&[]byte(p)[0]))),
uint32(n),
isEOF,
)
if res != 0 {
return 0, fmt.Errorf("failed to write compressed data")
}
}
var recvLen uint32
var isEOF = uint32(0)
res := decompress_read(d.id,
uint32(uintptr(unsafe.Pointer(&[]byte(p)[0]))),
uint32(len(p)),
uint32(uintptr(unsafe.Pointer(&recvLen))),
uint32(uintptr(unsafe.Pointer(&isEOF))),
)
if res != 0 {
return 0, fmt.Errorf("failed to read compressed data")
}
var err error
if isEOF == 1 {
err = io.EOF
}
return int(recvLen), err
}

type gzipDecompressor struct {
*estargz.GzipDecompressor
}

func newGzipDecompressor() *gzipDecompressor {
return &gzipDecompressor{&estargz.GzipDecompressor{}}
}

func (gz *gzipDecompressor) Reader(r io.Reader) (io.ReadCloser, error) {
zr, err := newWasmDecompressor(r)
if err != nil {
return nil, err
}
return io.NopCloser(zr), nil
}

type readerWithCloser struct {
r io.Reader
closeFunc func() error
}

func (r *readerWithCloser) Read(p []byte) (int, error) {
return r.r.Read(p)
}

func (r *readerWithCloser) Close() error {
return r.closeFunc()
}
Loading

0 comments on commit 43e793f

Please sign in to comment.