1	/*
     2	Copyright 2011 The Perkeep Authors
     3	
     4	Licensed under the Apache License, Version 2.0 (the "License");
     5	you may not use this file except in compliance with the License.
     6	You may obtain a copy of the License at
     7	
     8	     http://www.apache.org/licenses/LICENSE-2.0
     9	
    10	Unless required by applicable law or agreed to in writing, software
    11	distributed under the License is distributed on an "AS IS" BASIS,
    12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    13	See the License for the specific language governing permissions and
    14	limitations under the License.
    15	*/
    16	
    17	package schema
    18	
    19	import (
    20		"bufio"
    21		"bytes"
    22		"context"
    23		"errors"
    24		"fmt"
    25		"io"
    26		"os"
    27		"strings"
    28		"time"
    29	
    30		"go4.org/rollsum"
    31	
    32		"perkeep.org/pkg/blob"
    33		"perkeep.org/pkg/blobserver"
    34	
    35		"go4.org/syncutil"
    36	)
    37	
    38	const (
    39		// maxBlobSize is the largest blob we ever make when cutting up
    40		// a file.
    41		maxBlobSize = 1 << 20
    42	
    43		// firstChunkSize is the ideal size of the first chunk of a
    44		// file.  It's kept smaller for the file(1) command, which
    45		// likes to read 96 kB on Linux and 256 kB on OS X.  Related
    46		// are tools which extract the EXIF metadata from JPEGs,
    47		// ID3 from mp3s, etc.  Nautilus, OS X Finder, etc.
    48		// The first chunk may be larger than this if cutting the file
    49		// here would create a small subsequent chunk (e.g. a file one
    50		// byte larger than firstChunkSize)
    51		firstChunkSize = 256 << 10
    52	
    53		// bufioReaderSize is an explicit size for our bufio.Reader,
    54		// so we don't rely on NewReader's implicit size.
    55		// We care about the buffer size because it affects how far
    56		// in advance we can detect EOF from an io.Reader that doesn't
    57		// know its size.  Detecting an EOF bufioReaderSize bytes early
    58		// means we can plan for the final chunk.
    59		bufioReaderSize = 32 << 10
    60	
    61		// tooSmallThreshold is the threshold at which rolling checksum
    62		// boundaries are ignored if the current chunk being built is
    63		// smaller than this.
    64		tooSmallThreshold = 64 << 10
    65	)
    66	
    67	// WriteFileFromReaderWithModTime creates and uploads a "file" JSON schema
    68	// composed of chunks of r, also uploading the chunks.  The returned
    69	// BlobRef is of the JSON file schema blob.
    70	// Both filename and modTime are optional.
    71	func WriteFileFromReaderWithModTime(ctx context.Context, bs blobserver.StatReceiver, filename string, modTime time.Time, r io.Reader) (blob.Ref, error) {
    72		if strings.Contains(filename, "/") {
    73			return blob.Ref{}, fmt.Errorf("schema.WriteFileFromReader: filename %q shouldn't contain a slash", filename)
    74		}
    75	
    76		m := NewFileMap(filename)
    77		if !modTime.IsZero() {
    78			m.SetModTime(modTime)
    79		}
    80		return WriteFileMap(ctx, bs, m, r)
    81	}
    82	
    83	// WriteFileFromReader creates and uploads a "file" JSON schema
    84	// composed of chunks of r, also uploading the chunks.  The returned
    85	// BlobRef is of the JSON file schema blob.
    86	// The filename is optional.
    87	func WriteFileFromReader(ctx context.Context, bs blobserver.StatReceiver, filename string, r io.Reader) (blob.Ref, error) {
    88		return WriteFileFromReaderWithModTime(ctx, bs, filename, time.Time{}, r)
    89	}
    90	
    91	// WriteFileMap uploads chunks of r to bs while populating file and
    92	// finally uploading file's Blob. The returned blobref is of file's
    93	// JSON blob.
    94	func WriteFileMap(ctx context.Context, bs blobserver.StatReceiver, file *Builder, r io.Reader) (blob.Ref, error) {
    95		return writeFileMapRolling(ctx, bs, file, r)
    96	}
    97	
    98	func serverHasBlob(ctx context.Context, bs blobserver.BlobStatter, br blob.Ref) (have bool, err error) {
    99		_, err = blobserver.StatBlob(ctx, bs, br)
   100		if err == nil {
   101			have = true
   102		} else if errors.Is(err, os.ErrNotExist) {
   103			err = nil
   104		}
   105		return
   106	}
   107	
   108	type span struct {
   109		from, to int64
   110		bits     int
   111		br       blob.Ref
   112		children []span
   113	}
   114	
   115	func (s *span) isSingleBlob() bool {
   116		return len(s.children) == 0
   117	}
   118	
   119	func (s *span) size() int64 {
   120		size := s.to - s.from
   121		for _, cs := range s.children {
   122			size += cs.size()
   123		}
   124		return size
   125	}
   126	
   127	// noteEOFReader keeps track of when it's seen EOF, but otherwise
   128	// delegates entirely to r.
   129	type noteEOFReader struct {
   130		r      io.Reader
   131		sawEOF bool
   132	}
   133	
   134	func (r *noteEOFReader) Read(p []byte) (n int, err error) {
   135		n, err = r.r.Read(p)
   136		if errors.Is(err, io.EOF) {
   137			r.sawEOF = true
   138		}
   139		return
   140	}
   141	
   142	func uploadString(ctx context.Context, bs blobserver.StatReceiver, br blob.Ref, s string) (blob.Ref, error) {
   143		if !br.Valid() {
   144			panic("invalid blobref")
   145		}
   146		hasIt, err := serverHasBlob(ctx, bs, br)
   147		if err != nil {
   148			return blob.Ref{}, err
   149		}
   150		if hasIt {
   151			return br, nil
   152		}
   153		_, err = blobserver.ReceiveNoHash(ctx, bs, br, strings.NewReader(s))
   154		if err != nil {
   155			return blob.Ref{}, err
   156		}
   157		return br, nil
   158	}
   159	
   160	// uploadBytes populates bb (a builder of either type "bytes" or
   161	// "file", which is a superset of "bytes"), sets it to the provided
   162	// size, and populates with provided spans.  The bytes or file schema
   163	// blob is uploaded and its blobref is returned.
   164	func uploadBytes(ctx context.Context, bs blobserver.StatReceiver, bb *Builder, size int64, s []span) *uploadBytesFuture {
   165		future := newUploadBytesFuture()
   166		parts := []BytesPart{}
   167		addBytesParts(ctx, bs, &parts, s, future)
   168	
   169		if err := bb.PopulateParts(size, parts); err != nil {
   170			future.errc <- err
   171			return future
   172		}
   173	
   174		// Hack until perkeep.org/issue/102 is fixed. If we happen to upload
   175		// the "file" schema before any of its parts arrive, then the indexer
   176		// can get confused.  So wait on the parts before, and then upload
   177		// the "file" blob afterwards.
   178		if bb.Type() == TypeFile {
   179			future.errc <- nil
   180			_, err := future.Get() // may not be nil, if children parts failed
   181			future = newUploadBytesFuture()
   182			if err != nil {
   183				future.errc <- err
   184				return future
   185			}
   186		}
   187	
   188		json := bb.Blob().JSON()
   189		br := blob.RefFromString(json)
   190		future.br = br
   191		go func() {
   192			_, err := uploadString(ctx, bs, br, json)
   193			future.errc <- err
   194		}()
   195		return future
   196	}
   197	
   198	func newUploadBytesFuture() *uploadBytesFuture {
   199		return &uploadBytesFuture{
   200			errc: make(chan error, 1),
   201		}
   202	}
   203	
   204	// An uploadBytesFuture is an eager result of a still-in-progress uploadBytes call.
   205	// Call Get to wait and get its final result.
   206	type uploadBytesFuture struct {
   207		br       blob.Ref
   208		errc     chan error
   209		children []*uploadBytesFuture
   210	}
   211	
   212	// BlobRef returns the optimistic blobref of this uploadBytes call without blocking.
   213	func (f *uploadBytesFuture) BlobRef() blob.Ref {
   214		return f.br
   215	}
   216	
   217	// Get blocks for all children and returns any final error.
   218	func (f *uploadBytesFuture) Get() (blob.Ref, error) {
   219		for _, f := range f.children {
   220			if _, err := f.Get(); err != nil {
   221				return blob.Ref{}, err
   222			}
   223		}
   224		return f.br, <-f.errc
   225	}
   226	
   227	// addBytesParts uploads the provided spans to bs, appending elements to *dst.
   228	func addBytesParts(ctx context.Context, bs blobserver.StatReceiver, dst *[]BytesPart, spans []span, parent *uploadBytesFuture) {
   229		for _, sp := range spans {
   230			if len(sp.children) == 1 && sp.children[0].isSingleBlob() {
   231				// Remove an occasional useless indirection of
   232				// what would become a bytes schema blob
   233				// pointing to a single blobref.  Just promote
   234				// the blobref child instead.
   235				child := sp.children[0]
   236				*dst = append(*dst, BytesPart{
   237					BlobRef: child.br,
   238					Size:    uint64(child.size()),
   239				})
   240				sp.children = nil
   241			}
   242			if len(sp.children) > 0 {
   243				childrenSize := int64(0)
   244				for _, cs := range sp.children {
   245					childrenSize += cs.size()
   246				}
   247				future := uploadBytes(ctx, bs, newBytes(), childrenSize, sp.children)
   248				parent.children = append(parent.children, future)
   249				*dst = append(*dst, BytesPart{
   250					BytesRef: future.BlobRef(),
   251					Size:     uint64(childrenSize),
   252				})
   253			}
   254			if sp.from == sp.to {
   255				panic("Shouldn't happen. " + fmt.Sprintf("weird span with same from & to: %#v", sp))
   256			}
   257			*dst = append(*dst, BytesPart{
   258				BlobRef: sp.br,
   259				Size:    uint64(sp.to - sp.from),
   260			})
   261		}
   262	}
   263	
   264	// writeFileMap uploads chunks of r to bs while populating fileMap and
   265	// finally uploading fileMap. The returned blobref is of fileMap's
   266	// JSON blob. It uses rolling checksum for the chunks sizes.
   267	func writeFileMapRolling(ctx context.Context, bs blobserver.StatReceiver, file *Builder, r io.Reader) (blob.Ref, error) {
   268		n, spans, err := writeFileChunks(ctx, bs, file, r)
   269		if err != nil {
   270			return blob.Ref{}, err
   271		}
   272		// The top-level content parts
   273		return uploadBytes(ctx, bs, file, n, spans).Get()
   274	}
   275	
   276	// WriteFileChunks uploads chunks of r to bs while populating file.
   277	// It does not upload file.
   278	func WriteFileChunks(ctx context.Context, bs blobserver.StatReceiver, file *Builder, r io.Reader) error {
   279		size, spans, err := writeFileChunks(ctx, bs, file, r)
   280		if err != nil {
   281			return err
   282		}
   283		parts := []BytesPart{}
   284		future := newUploadBytesFuture()
   285		addBytesParts(ctx, bs, &parts, spans, future)
   286		future.errc <- nil // Get will still block on addBytesParts' children
   287		if _, err := future.Get(); err != nil {
   288			return err
   289		}
   290		return file.PopulateParts(size, parts)
   291	}
   292	
   293	func writeFileChunks(ctx context.Context, bs blobserver.StatReceiver, file *Builder, r io.Reader) (n int64, spans []span, outerr error) {
   294		src := &noteEOFReader{r: r}
   295		bufr := bufio.NewReaderSize(src, bufioReaderSize)
   296		spans = []span{} // the tree of spans, cut on interesting rollsum boundaries
   297		rs := rollsum.New()
   298		var last int64
   299		var buf bytes.Buffer
   300		blobSize := 0 // of the next blob being built, should be same as buf.Len()
   301	
   302		const chunksInFlight = 32 // at ~64 KB chunks, this is ~2MB memory per file
   303		gatec := syncutil.NewGate(chunksInFlight)
   304		firsterrc := make(chan error, 1)
   305	
   306		// uploadLastSpan runs in the same goroutine as the loop below and is responsible for
   307		// starting uploading the contents of the buf.  It returns false if there's been
   308		// an error and the loop below should be stopped.
   309		uploadLastSpan := func() bool {
   310			chunk := buf.String()
   311			buf.Reset()
   312			br := blob.RefFromString(chunk)
   313			spans[len(spans)-1].br = br
   314			select {
   315			case outerr = <-firsterrc:
   316				return false
   317			default:
   318				// No error seen so far, continue.
   319			}
   320			gatec.Start()
   321			go func() {
   322				defer gatec.Done()
   323				if _, err := uploadString(ctx, bs, br, chunk); err != nil {
   324					select {
   325					case firsterrc <- err:
   326					default:
   327					}
   328				}
   329			}()
   330			return true
   331		}
   332	
   333		for {
   334			c, err := bufr.ReadByte()
   335			if errors.Is(err, io.EOF) {
   336				if n != last {
   337					spans = append(spans, span{from: last, to: n})
   338					if !uploadLastSpan() {
   339						return
   340					}
   341				}
   342				break
   343			}
   344			if err != nil {
   345				return 0, nil, err
   346			}
   347	
   348			buf.WriteByte(c)
   349			n++
   350			blobSize++
   351			rs.Roll(c)
   352	
   353			var bits int
   354			onRollSplit := rs.OnSplit()
   355			switch {
   356			case blobSize == maxBlobSize:
   357				bits = 20 // arbitrary node weight; 1<<20 == 1MB
   358			case src.sawEOF:
   359				// Don't split. End is coming soon enough.
   360				continue
   361			case onRollSplit && n > firstChunkSize && blobSize > tooSmallThreshold:
   362				bits = rs.Bits()
   363			case n == firstChunkSize:
   364				bits = 18 // 1 << 18 == 256KB
   365			default:
   366				// Don't split.
   367				continue
   368			}
   369			blobSize = 0
   370	
   371			// Take any spans from the end of the spans slice that
   372			// have a smaller 'bits' score and make them children
   373			// of this node.
   374			var children []span
   375			childrenFrom := len(spans)
   376			for childrenFrom > 0 && spans[childrenFrom-1].bits < bits {
   377				childrenFrom--
   378			}
   379			if nCopy := len(spans) - childrenFrom; nCopy > 0 {
   380				children = make([]span, nCopy)
   381				copy(children, spans[childrenFrom:])
   382				spans = spans[:childrenFrom]
   383			}
   384	
   385			spans = append(spans, span{from: last, to: n, bits: bits, children: children})
   386			last = n
   387			if !uploadLastSpan() {
   388				return
   389			}
   390		}
   391	
   392		// Loop was already hit earlier.
   393		if outerr != nil {
   394			return 0, nil, outerr
   395		}
   396	
   397		// Wait for all uploads to finish, one way or another, and then
   398		// see if any generated errors.
   399		// Once this loop is done, we own all the tokens in gatec, so nobody
   400		// else can have one outstanding.
   401		for i := 0; i < chunksInFlight; i++ {
   402			gatec.Start()
   403		}
   404		select {
   405		case err := <-firsterrc:
   406			return 0, nil, err
   407		default:
   408		}
   409	
   410		return n, spans, nil
   411	
   412	}