remote: option to skip delta compression while encoding packfiles

One use of go-git is to transfer git data from a non-standard git repo
(not stored in a file system, for example) to a "remote" backed by a
standard, local .git repo

In this scenario, delta compression is not needed to reduce transfer
time over the "network", because there is no network.  The underlying
storage layer has already taken care of the data tranfer, and sending
the objects to local .git storage doesn't require compression.  So
this PR gives the user the option to turn off compression when it
isn't needed.

Of course, this results in a larger, uncompressed local .git repo, but
the user can then run `git gc` or `git repack` on that repo if they
care about the storage costs.

Turning this on reduces total push time of a 36K repo by 50 seconds
(out of a pre-PR total of 3m26s).

Author: strib

options.go

@@ -175,6 +175,13 @@ type PushOptions struct {
 	// Progress is where the human readable information sent by the server is
 	// stored, if nil nothing is stored.
 	Progress sideband.Progress
+	// SkipCompression can be true if the caller doesn't need
+	// delta-compressed objects pushed to the remote.  This can be useful to
+	// avoid CPU overhead when the objects don't need to be
+	// transferred across a network.  If the remote wants a compressed
+	// repo after the transfer, it can run `git gc --aggressive` or
+	// `git repack -a -d -f --depth=250 --window=250` as needed.
+	SkipCompression bool
 }
 
 // Validate validates the fields and sets the default values.

plumbing/format/packfile/delta_selector.go

@@ -33,12 +33,20 @@ func newDeltaSelector(s storer.EncodedObjectStorer) *deltaSelector {
 
 // ObjectsToPack creates a list of ObjectToPack from the hashes provided,
 // creating deltas if it's suitable, using an specific internal logic
-func (dw *deltaSelector) ObjectsToPack(hashes []plumbing.Hash) ([]*ObjectToPack, error) {
-	otp, err := dw.objectsToPack(hashes)
+func (dw *deltaSelector) ObjectsToPack(
+	hashes []plumbing.Hash,
+	skipCompression bool,
+) ([]*ObjectToPack, error) {
+
+	otp, err := dw.objectsToPack(hashes, skipCompression)
 	if err != nil {
 		return nil, err
 	}
 
+	if skipCompression {
+		return otp, nil
+	}
+
 	dw.sort(otp)
 
 	var objectGroups [][]*ObjectToPack
@@ -77,10 +85,19 @@ func (dw *deltaSelector) ObjectsToPack(hashes []plumbing.Hash) ([]*ObjectToPack,
 	return otp, nil
 }
 
-func (dw *deltaSelector) objectsToPack(hashes []plumbing.Hash) ([]*ObjectToPack, error) {
+func (dw *deltaSelector) objectsToPack(
+	hashes []plumbing.Hash,
+	skipCompression bool,
+) ([]*ObjectToPack, error) {
 	var objectsToPack []*ObjectToPack
 	for _, h := range hashes {
-		o, err := dw.encodedDeltaObject(h)
+		var o plumbing.EncodedObject
+		var err error
+		if skipCompression {
+			o, err = dw.encodedObject(h)
+		} else {
+			o, err = dw.encodedDeltaObject(h)
+		}
 		if err != nil {
 			return nil, err
 		}
@@ -93,6 +110,10 @@ func (dw *deltaSelector) objectsToPack(hashes []plumbing.Hash) ([]*ObjectToPack,
 		objectsToPack = append(objectsToPack, otp)
 	}
 
+	if skipCompression {
+		return objectsToPack, nil
+	}
+
 	if err := dw.fixAndBreakChains(objectsToPack); err != nil {
 		return nil, err
 	}

plumbing/format/packfile/delta_selector_test.go

@@ -146,31 +146,31 @@ func (s *DeltaSelectorSuite) createTestObjects() {
 func (s *DeltaSelectorSuite) TestObjectsToPack(c *C) {
 	// Different type
 	hashes := []plumbing.Hash{s.hashes["base"], s.hashes["treeType"]}
-	otp, err := s.ds.ObjectsToPack(hashes)
+	otp, err := s.ds.ObjectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	c.Assert(len(otp), Equals, 2)
 	c.Assert(otp[0].Object, Equals, s.store.Objects[s.hashes["base"]])
 	c.Assert(otp[1].Object, Equals, s.store.Objects[s.hashes["treeType"]])
 
 	// Size radically different
 	hashes = []plumbing.Hash{s.hashes["bigBase"], s.hashes["target"]}
-	otp, err = s.ds.ObjectsToPack(hashes)
+	otp, err = s.ds.ObjectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	c.Assert(len(otp), Equals, 2)
 	c.Assert(otp[0].Object, Equals, s.store.Objects[s.hashes["bigBase"]])
 	c.Assert(otp[1].Object, Equals, s.store.Objects[s.hashes["target"]])
 
 	// Delta Size Limit with no best delta yet
 	hashes = []plumbing.Hash{s.hashes["smallBase"], s.hashes["smallTarget"]}
-	otp, err = s.ds.ObjectsToPack(hashes)
+	otp, err = s.ds.ObjectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	c.Assert(len(otp), Equals, 2)
 	c.Assert(otp[0].Object, Equals, s.store.Objects[s.hashes["smallBase"]])
 	c.Assert(otp[1].Object, Equals, s.store.Objects[s.hashes["smallTarget"]])
 
 	// It will create the delta
 	hashes = []plumbing.Hash{s.hashes["base"], s.hashes["target"]}
-	otp, err = s.ds.ObjectsToPack(hashes)
+	otp, err = s.ds.ObjectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	c.Assert(len(otp), Equals, 2)
 	c.Assert(otp[0].Object, Equals, s.store.Objects[s.hashes["target"]])
@@ -185,7 +185,7 @@ func (s *DeltaSelectorSuite) TestObjectsToPack(c *C) {
 		s.hashes["o2"],
 		s.hashes["o3"],
 	}
-	otp, err = s.ds.ObjectsToPack(hashes)
+	otp, err = s.ds.ObjectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	c.Assert(len(otp), Equals, 3)
 	c.Assert(otp[0].Object, Equals, s.store.Objects[s.hashes["o1"]])
@@ -208,7 +208,7 @@ func (s *DeltaSelectorSuite) TestObjectsToPack(c *C) {
 
 	// Don't sort so we can easily check the sliding window without
 	// creating a bunch of new objects.
-	otp, err = s.ds.objectsToPack(hashes)
+	otp, err = s.ds.objectsToPack(hashes, false)
 	c.Assert(err, IsNil)
 	err = s.ds.walk(otp)
 	c.Assert(err, IsNil)

plumbing/format/packfile/encoder.go

@@ -14,10 +14,10 @@ import (
 // Encoder gets the data from the storage and write it into the writer in PACK
 // format
 type Encoder struct {
-	selector     *deltaSelector
-	w            *offsetWriter
-	zw           *zlib.Writer
-	hasher       plumbing.Hasher
+	selector *deltaSelector
+	w        *offsetWriter
+	zw       *zlib.Writer
+	hasher   plumbing.Hasher
 	// offsets is a map of object hashes to corresponding offsets in the packfile.
 	// It is used to determine offset of the base of a delta when a OFS_DELTA is
 	// used.
@@ -47,8 +47,11 @@ func NewEncoder(w io.Writer, s storer.EncodedObjectStorer, useRefDeltas bool) *E
 
 // Encode creates a packfile containing all the objects referenced in hashes
 // and writes it to the writer in the Encoder.
-func (e *Encoder) Encode(hashes []plumbing.Hash) (plumbing.Hash, error) {
-	objects, err := e.selector.ObjectsToPack(hashes)
+func (e *Encoder) Encode(
+	hashes []plumbing.Hash,
+	skipCompression bool,
+) (plumbing.Hash, error) {
+	objects, err := e.selector.ObjectsToPack(hashes, skipCompression)
 	if err != nil {
 		return plumbing.ZeroHash, err
 	}
@@ -137,7 +140,7 @@ func (e *Encoder) writeOfsDeltaHeader(deltaOffset int64, base plumbing.Hash) err
 
 	// for OFS_DELTA, offset of the base is interpreted as negative offset
 	// relative to the type-byte of the header of the ofs-delta entry.
-	relativeOffset := deltaOffset-baseOffset
+	relativeOffset := deltaOffset - baseOffset
 	if relativeOffset <= 0 {
 		return fmt.Errorf("bad offset for OFS_DELTA entry: %d", relativeOffset)
 	}

plumbing/format/packfile/encoder_advanced_test.go

@@ -27,12 +27,24 @@ func (s *EncoderAdvancedSuite) TestEncodeDecode(c *C) {
 	fixs.Test(c, func(f *fixtures.Fixture) {
 		storage, err := filesystem.NewStorage(f.DotGit())
 		c.Assert(err, IsNil)
-		s.testEncodeDecode(c, storage)
+		s.testEncodeDecode(c, storage, false)
 	})
 
 }
 
-func (s *EncoderAdvancedSuite) testEncodeDecode(c *C, storage storer.Storer) {
+func (s *EncoderAdvancedSuite) TestEncodeDecodeSkipCompression(c *C) {
+	fixs := fixtures.Basic().ByTag("packfile").ByTag(".git")
+	fixs = append(fixs, fixtures.ByURL("https://github.com/src-d/go-git.git").
+		ByTag("packfile").ByTag(".git").One())
+	fixs.Test(c, func(f *fixtures.Fixture) {
+		storage, err := filesystem.NewStorage(f.DotGit())
+		c.Assert(err, IsNil)
+		s.testEncodeDecode(c, storage, true)
+	})
+
+}
+
+func (s *EncoderAdvancedSuite) testEncodeDecode(c *C, storage storer.Storer, skipCompression bool) {
 
 	objIter, err := storage.IterEncodedObjects(plumbing.AnyObject)
 	c.Assert(err, IsNil)
@@ -57,7 +69,7 @@ func (s *EncoderAdvancedSuite) testEncodeDecode(c *C, storage storer.Storer) {
 
 	buf := bytes.NewBuffer(nil)
 	enc := NewEncoder(buf, storage, false)
-	_, err = enc.Encode(hashes)
+	_, err = enc.Encode(hashes, skipCompression)
 	c.Assert(err, IsNil)
 
 	scanner := NewScanner(buf)

plumbing/format/packfile/encoder_test.go

@@ -26,7 +26,7 @@ func (s *EncoderSuite) SetUpTest(c *C) {
 }
 
 func (s *EncoderSuite) TestCorrectPackHeader(c *C) {
-	hash, err := s.enc.Encode([]plumbing.Hash{})
+	hash, err := s.enc.Encode([]plumbing.Hash{}, false)
 	c.Assert(err, IsNil)
 
 	hb := [20]byte(hash)
@@ -47,7 +47,7 @@ func (s *EncoderSuite) TestCorrectPackWithOneEmptyObject(c *C) {
 	_, err := s.store.SetEncodedObject(o)
 	c.Assert(err, IsNil)
 
-	hash, err := s.enc.Encode([]plumbing.Hash{o.Hash()})
+	hash, err := s.enc.Encode([]plumbing.Hash{o.Hash()}, false)
 	c.Assert(err, IsNil)
 
 	// PACK + VERSION(2) + OBJECT NUMBER(1)
@@ -74,13 +74,13 @@ func (s *EncoderSuite) TestMaxObjectSize(c *C) {
 	o.SetType(plumbing.CommitObject)
 	_, err := s.store.SetEncodedObject(o)
 	c.Assert(err, IsNil)
-	hash, err := s.enc.Encode([]plumbing.Hash{o.Hash()})
+	hash, err := s.enc.Encode([]plumbing.Hash{o.Hash()}, false)
 	c.Assert(err, IsNil)
 	c.Assert(hash.IsZero(), Not(Equals), true)
 }
 
 func (s *EncoderSuite) TestHashNotFound(c *C) {
-	h, err := s.enc.Encode([]plumbing.Hash{plumbing.NewHash("BAD")})
+	h, err := s.enc.Encode([]plumbing.Hash{plumbing.NewHash("BAD")}, false)
 	c.Assert(h, Equals, plumbing.ZeroHash)
 	c.Assert(err, NotNil)
 	c.Assert(err, Equals, plumbing.ErrObjectNotFound)

plumbing/transport/server/server.go

@@ -165,7 +165,7 @@ func (s *upSession) UploadPack(ctx context.Context, req *packp.UploadPackRequest
 	pr, pw := io.Pipe()
 	e := packfile.NewEncoder(pw, s.storer, false)
 	go func() {
-		_, err := e.Encode(objs)
+		_, err := e.Encode(objs, false)
 		pw.CloseWithError(err)
 	}()
 

plumbing/transport/test/receive_pack.go

@@ -348,7 +348,7 @@ func (s *ReceivePackSuite) testSendPackDeleteReference(c *C) {
 func (s *ReceivePackSuite) emptyPackfile() io.ReadCloser {
 	var buf bytes.Buffer
 	e := packfile.NewEncoder(&buf, memory.NewStorage(), false)
-	_, err := e.Encode(nil)
+	_, err := e.Encode(nil, false)
 	if err != nil {
 		panic(err)
 	}

remote.go

@@ -149,7 +149,8 @@ func (r *Remote) PushContext(ctx context.Context, o *PushOptions) error {
 		}
 	}
 
-	rs, err := pushHashes(ctx, s, r.s, req, hashesToPush)
+	rs, err := pushHashes(
+		ctx, s, r.s, req, hashesToPush, o.SkipCompression)
 	if err != nil {
 		return err
 	}
@@ -800,14 +801,15 @@ func pushHashes(
 	sto storer.EncodedObjectStorer,
 	req *packp.ReferenceUpdateRequest,
 	hs []plumbing.Hash,
+	skipCompression bool,
 ) (*packp.ReportStatus, error) {
 
 	rd, wr := io.Pipe()
 	req.Packfile = rd
 	done := make(chan error)
 	go func() {
 		e := packfile.NewEncoder(wr, sto, false)
-		if _, err := e.Encode(hs); err != nil {
+		if _, err := e.Encode(hs, skipCompression); err != nil {
 			done <- wr.CloseWithError(err)
 			return
 		}

remote_test.go

@@ -380,6 +380,30 @@ func (s *RemoteSuite) TestPushContext(c *C) {
 	c.Assert(err, NotNil)
 }
 
+func (s *RemoteSuite) TestPushContextSkipCompression(c *C) {
+	url := c.MkDir()
+	_, err := PlainInit(url, true)
+	c.Assert(err, IsNil)
+
+	fs := fixtures.ByURL("https://github.com/git-fixtures/tags.git").One().DotGit()
+	sto, err := filesystem.NewStorage(fs)
+	c.Assert(err, IsNil)
+
+	r := newRemote(sto, &config.RemoteConfig{
+		Name: DefaultRemoteName,
+		URLs: []string{url},
+	})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err = r.PushContext(ctx, &PushOptions{
+		RefSpecs:        []config.RefSpec{"refs/tags/*:refs/tags/*"},
+		SkipCompression: true,
+	})
+	c.Assert(err, NotNil)
+}
+
 func (s *RemoteSuite) TestPushTags(c *C) {
 	url := c.MkDir()
 	server, err := PlainInit(url, true)