simdjson_amd64.go

//go:build !appengine && !noasm && gc
// +build !appengine,!noasm,gc

/*
 * MinIO Cloud Storage, (C) 2020 MinIO, Inc.
 *
 * 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 simdjson

import (
	"bufio"
	"bytes"
	"errors"
	"fmt"
	"io"
	"runtime"
	"sync"

	"github.com/klauspost/cpuid/v2"
)

var wantFeatures = cpuid.CombineFeatures(cpuid.AVX2, cpuid.CLMUL)

// SupportedCPU will return whether the CPU is supported.
func SupportedCPU() bool {
	return cpuid.CPU.HasAll(wantFeatures)
}

func newInternalParsedJson(reuse *ParsedJson, opts []ParserOption) (*internalParsedJson, error) {
	if !SupportedCPU() {
		return nil, errors.New("Host CPU does not meet target specs")
	}
	var pj *internalParsedJson
	if reuse != nil && reuse.internal != nil {
		pj = reuse.internal
		pj.ParsedJson = *reuse
		pj.ParsedJson.internal = nil
		reuse = &ParsedJson{}
	}
	if pj == nil {
		pj = &internalParsedJson{}
	}
	pj.copyStrings = true
	for _, opt := range opts {
		if err := opt(pj); err != nil {
			return nil, err
		}
	}
	return pj, nil
}

// Parse an object or array from a block of data and return the parsed JSON.
// An optional block of previously parsed json can be supplied to reduce allocations.
func Parse(b []byte, reuse *ParsedJson, opts ...ParserOption) (*ParsedJson, error) {
	pj, err := newInternalParsedJson(reuse, opts)
	if err != nil {
		return nil, err
	}
	err = pj.parseMessage(b, false)
	if err != nil {
		return nil, err
	}
	parsed := &pj.ParsedJson
	parsed.internal = pj
	return parsed, nil
}

// ParseND will parse newline delimited JSON objects or arrays.
// An optional block of previously parsed json can be supplied to reduce allocations.
func ParseND(b []byte, reuse *ParsedJson, opts ...ParserOption) (*ParsedJson, error) {
	pj, err := newInternalParsedJson(reuse, opts)
	if err != nil {
		return nil, err
	}
	err = pj.parseMessage(bytes.TrimSpace(b), true)
	if err != nil {
		return nil, err
	}
	return &pj.ParsedJson, nil
}

// A Stream is used to stream back results.
// Either Error or Value will be set on returned results.
type Stream struct {
	Value *ParsedJson
	Error error
}

// ParseNDStream will parse a stream and return parsed JSON to the supplied result channel.
// The method will return immediately.
// Each element is contained within a root tag.
//
//	<root>Element 1</root><root>Element 2</root>...
//
// Each result will contain an unspecified number of full elements,
// so it can be assumed that each result starts and ends with a root tag.
// The parser will keep parsing until writes to the result stream blocks.
// A stream is finished when a non-nil Error is returned.
// If the stream was parsed until the end the Error value will be io.EOF
// The channel will be closed after an error has been returned.
// An optional channel for returning consumed results can be provided.
// There is no guarantee that elements will be consumed, so always use
// non-blocking writes to the reuse channel.
func ParseNDStream(r io.Reader, res chan<- Stream, reuse <-chan *ParsedJson) {
	if !SupportedCPU() {
		go func() {
			res <- Stream{
				Value: nil,
				Error: fmt.Errorf("Host CPU does not meet target specs"),
			}
			close(res)
		}()
		return
	}
	const tmpSize = 10 << 20
	buf := bufio.NewReaderSize(r, tmpSize)
	tmpPool := sync.Pool{New: func() interface{} {
		return make([]byte, tmpSize+1024)
	}}
	conc := (runtime.GOMAXPROCS(0) + 1) / 2
	queue := make(chan chan Stream, conc)
	go func() {
		// Forward finished items in order.
		defer close(res)
		end := false
		for items := range queue {
			i := <-items
			select {
			case res <- i:
			default:
				if !end {
					// Block if we haven't returned an error
					res <- i
				}
			}
			if i.Error != nil {
				end = true
			}
		}
	}()
	go func() {
		defer close(queue)
		for {
			tmp := tmpPool.Get().([]byte)
			tmp = tmp[:tmpSize]
			n, err := buf.Read(tmp)
			if err != nil && err != io.EOF {
				queueError(queue, err)
				return
			}
			tmp = tmp[:n]
			// Read until Newline
			if err != io.EOF {
				b, err2 := buf.ReadBytes('\n')
				if err2 != nil && err2 != io.EOF {
					queueError(queue, err2)
					return
				}
				tmp = append(tmp, b...)
				// Forward io.EOF
				err = err2
			}

			if len(tmp) > 0 {
				result := make(chan Stream, 0)
				queue <- result
				go func() {
					var pj internalParsedJson
					pj.copyStrings = true
					select {
					case v := <-reuse:
						if cap(v.Message) >= tmpSize+1024 {
							tmpPool.Put(v.Message)
							v.Message = nil
						}
						pj.ParsedJson = *v

					default:
					}
					parseErr := pj.parseMessage(tmp, true)
					if parseErr != nil {
						result <- Stream{
							Value: nil,
							Error: fmt.Errorf("parsing input: %w", parseErr),
						}
						return
					}
					parsed := pj.ParsedJson
					result <- Stream{
						Value: &parsed,
						Error: nil,
					}
				}()
			} else {
				tmpPool.Put(tmp)
			}
			if err != nil {
				// Should only really be io.EOF
				queueError(queue, err)
				return
			}
		}
	}()
}

func queueError(queue chan chan Stream, err error) {
	result := make(chan Stream, 0)
	queue <- result
	result <- Stream{
		Value: nil,
		Error: err,
	}
}