// runDecoder will create a sync decoder that will decodeAsync a block of data.
772397
func (d *frameDec) runDecoder(dst []byte, dec *blockDec) ([]byte, error) {
	// TODO: Init to dictionary
	d.history.reset()
	saved := d.history.b

	// We use the history for output to avoid copying it.
	d.history.b = dst
	// Store input length, so we only check new data.
	crcStart := len(dst)
	var err error
	for {
		err = dec.reset(d.rawInput, d.WindowSize)
		if err != nil {
			break
		}
		if debug {
			println("next block:", dec)
		}
		err = dec.decodeBuf(&d.history)
		if err != nil || dec.Last {
			break
		}
		if uint64(len(d.history.b)) > d.o.maxDecodedSize {
			err = ErrDecoderSizeExceeded
			break
		}
	}
	dst = d.history.b
	if err == nil {
		if d.HasCheckSum {
			var n int
			n, err = d.crc.Write(dst[crcStart:])
			if err == nil {
				if n != len(dst)-crcStart {
					err = io.ErrShortWrite
				}
			}
			err = d.checkCRC()
		}
	}
	d.history.b = saved
	return dst, err
}
831729
func (runner *gleamRunner) runMapperReducer() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	if runner.Option.IsProfiling {
		cpuProfFile := fmt.Sprintf("mr_cpu_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)
		pwd, _ := os.Getwd()
		// println("saving pprof to", pwd+"/"+cpuProfFile)

		cf, err := os.Create(cpuProfFile)
		if err != nil {
			log.Fatalf("failed to create cpu profile file %s: %v", pwd+"/"+cpuProfFile, err)
		}
		pprof.StartCPUProfile(cf)
		defer pprof.StopCPUProfile()

		memProfFile := fmt.Sprintf("mr_mem_%d-s%d-t%d.pprof", runner.Option.HashCode,
			runner.Option.StepId, runner.Option.TaskId)

		mf, err := os.Create(memProfFile)
		// println("saving pprof to", pwd+"/"+memProfFile)
		if err != nil {
			log.Fatalf("failed to create memory profile file %s: %v", pwd+"/"+memProfFile, err)
		}

		defer func() {
			runtime.GC()
			pprof.Lookup("heap").WriteTo(mf, 0)

		}()
	}

	stat.FlowHashCode = uint32(runner.Option.HashCode)
	stat.Stats = []*pb.InstructionStat{
		{
			StepId: int32(runner.Option.StepId),
			TaskId: int32(runner.Option.TaskId),
		},
	}

	if runner.Option.Mapper != "" {
		if fn, ok := mappers[MapperId(runner.Option.Mapper)]; ok {
			if err := runner.processMapper(ctx, fn.Mapper); err != nil {
				log.Fatalf("Failed to execute mapper %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing mapper function %v. Args: %v", runner.Option.Mapper, os.Args)
	}

	if runner.Option.Reducer != "" {
		if runner.Option.KeyFields == "" {
			log.Fatalf("Also expecting values for -gleam.keyFields! Actual arguments: %v", os.Args)
		}
		if fn, ok := reducers[ReducerId(runner.Option.Reducer)]; ok {
			keyPositions := strings.Split(runner.Option.KeyFields, ",")
			var keyIndexes []int
			for _, keyPosition := range keyPositions {
				keyIndex, keyIndexError := strconv.Atoi(keyPosition)
				if keyIndexError != nil {
					log.Fatalf("Failed to parse key index positions %v: %v", runner.Option.KeyFields, keyIndexError)
				}
				keyIndexes = append(keyIndexes, keyIndex)
			}

			if err := runner.processReducer(ctx, fn.Reducer, keyIndexes); err != nil {
				log.Fatalf("Failed to execute reducer %v: %v", os.Args, err)
			}
			return
		}
		log.Fatalf("Missing reducer function %v. Args: %v", runner.Option.Reducer, os.Args)

	}

	log.Fatalf("Failed to find function to execute. Args: %v", os.Args)
}
962794
func (j *JSONPb) NewDecoder(r io.Reader) runtime.Decoder {
	d := json.NewDecoder(r)
	return runtime.DecoderFunc(func(v interface{}) error { return decodeJSONPb(d, v) })
}
809993
func NewDecoder(enc *Encoding, r io.Reader) io.Reader {
	return newDecoder(enc, r)
}
1075389
func (p *Proto) NewDecoder(reader io.Reader) runtime.Decoder {
	return runtime.DecoderFunc(func(value interface{}) error {
		buffer, err := ioutil.ReadAll(reader)
		if err != nil {
			return err
		}
		return p.Unmarshal(buffer, value)
	})
}
230569
func (marshaller *ProtoMarshaller) NewDecoder(reader io.Reader) Decoder {
	return DecoderFunc(func(value interface{}) error {
		buffer, err := ioutil.ReadAll(reader)
		if err != nil {
			return err
		}
		return marshaller.Unmarshal(buffer, value)
	})
}
659920
func NewDecoder(r io.Reader, format Format) Decoder {
	switch format {
	case FmtProtoDelim:
		return &protoDecoder{r: r}
	}
	return &textDecoder{r: r}
}
1019678
func jsonDecoder(body io.Reader, v interface{}) error {
	d := json.NewDecoder(body)
	return d.Decode(v)
}
386243
func (m *RWMutex) RUnlock() {
	m.mutex.RUnlock()
	atomic.AddInt64(&m.readers, -1)
}
366668
func unmarshalerDecoder(d *decodeState, v reflect.Value, opts decOpts) int {
	um, ok := v.Interface().(Unmarshaler)
	if !ok {
		panic(fmt.Errorf("Non-Unmarshaler passed to unmarshalerEncoder"))
	}

	read, err := um.UnmarshalTLS(d.Bytes())
	if err != nil {
		panic(err)
	}

	if read > d.Len() {
		panic(fmt.Errorf("Invalid return value from UnmarshalTLS"))
	}

	d.Next(read)
	return read
}
857238
func (j *JSONPb) NewDecoder(r io.Reader) Decoder {
	d := json.NewDecoder(r)
	return DecoderWrapper{Decoder: d}
}
918797
func NewDecoder(rd io.Reader) Decoder {
	dec := gob.NewDecoder(rd)
	return func(r *Result) error { return dec.Decode(r) }
}
665629
func NewDecoder(r io.Reader) *decoder {
	d := &decoder{
		scanner:  bufio.NewScanner(r),
		nextFn:   decodeInit,
		line:     0,
		buffered: "",
	}
	return d
}
789183
func NewDecoder(r io.ReadCloser, d runtime.Decoder) Decoder {
	return &decoder{
		reader:   r,
		decoder:  d,
		buf:      make([]byte, 1024),
		maxBytes: 16 * 1024 * 1024,
	}
}
1037288
func NewDecoder(r io.Reader) *Decoder {
	return &Decoder{r, defaultDelimiter, defaultEscape, false, false}
}
1011248
func (p *Process) runLinker() {
	processSock := path.Join(p.containerPath, "processes", fmt.Sprintf("%s.sock", p.ID()))

	link, err := link.Create(processSock, p.stdout, p.stderr)
	if err != nil {
		p.completed(-1, err)
		return
	}

	p.stdin.AddSink(link)

	p.link = link
	close(p.linked)

	p.completed(p.link.Wait())

	// don't leak stdin pipe
	p.stdin.Close()
}
744124
func NewDecoder(r io.Reader, opts ...Option) *Decoder {
	return &Decoder{
		r:   r,
		opt: applyOptions(opts),
	}
}