// goroutine safe
557062
func (p *Processor) Marshal(msg interface{}) ([][]byte, error) {
	msgType := reflect.TypeOf(msg)

	// id
	_id, ok := p.msgID[msgType]
	if !ok {
		err := fmt.Errorf("message %s not registered", msgType)
		return nil, err
	}

	id := make([]byte, 2)
	if p.littleEndian {
		binary.LittleEndian.PutUint16(id, _id)
	} else {
		binary.BigEndian.PutUint16(id, _id)
	}

	// data
	data, err := proto.Marshal(msg.(proto.Message))
	return [][]byte{id, data}, err
}
836532
func (f *StringForwarder) loop(callback func(string)) {
	f.mu.Lock()
	defer f.mu.Unlock()
	for {
		for !f.stopped && f.current == nil {
			f.cond.Wait()
		}
		if f.current == nil {
			return
		}
		f.invokeCallback(callback, *f.current)
		f.current = nil
	}
}
157862
func (e *EventsInstance) GoExecute(eventName string, next func(bool), sessions ...interface{}) {
	if e.listeners == nil {
		next(true)
		return
	}
	c := make(chan int)
	n := 0
	e.lock.RLock()
	defer e.lock.RUnlock()
	if l, ok := e.listeners[eventName]; ok {
		if len(l) > 0 {
			for _, h := range l {
				n++
				//h 的原型为 func(interface{}, func(bool))
				go h(func(_ bool) {
					c <- 1
				}, sessions...)
			}
		}
	}
	for n > 0 {
		i := <-c
		if i == 1 {
			n--
		}
	}
	if next == nil {
		return
	}
	next(true)
}
1142766
func (e *Endpoint) Go(function string, args interface{}, reply interface{}, done chan *rpc.Call) *rpc.Call {
	call := &rpc.Call{}
	call.ServiceMethod = function
	call.Args = args
	call.Reply = reply
	if done == nil {
		done = make(chan *rpc.Call, 10)
	} else {
		if cap(done) == 0 {
			log.Panic("birpc: done channel is unbuffered")
		}
	}
	call.Done = done

	msg := &Message{
		Func: function,
		Args: args,
	}

	e.client.mutex.Lock()
	e.client.seq++
	msg.ID = e.client.seq
	e.client.pending[msg.ID] = call
	e.client.mutex.Unlock()

	// put sending in a goroutine so a malicious client that
	// refuses to read cannot ever make a .Go call block
	go e.send(msg)
	return call
}
882954
func (g *Group) Go(f func() error, index int) {
	g.wg.Add(1)

	go func() {
		defer g.wg.Done()

		if err := f(); err != nil {
			g.errs[index] = err
		}
	}()
}
185856
func goProofOfWork(trytes Trytes, mwm int, optRate chan int64, parallelism ...int) (Trytes, error) {
	if trytes == "" {
		return "", ErrInvalidTrytesForProofOfWork
	}

	// if any goroutine finds a nonce, then the cancel flag is set to true
	// and thereby all other ongoing Proof-of-Work tasks will halt.
	cancelled := false

	tr := MustTrytesToTrits(trytes)

	c := curl.NewCurlP81().(*curl.Curl)
	c.Absorb(tr[:(TransactionTrinarySize - HashTrinarySize)])
	copy(c.State, tr[TransactionTrinarySize-HashTrinarySize:])

	numGoroutines := proofOfWorkParallelism(parallelism...)
	var result Trytes
	var rate chan int64
	if optRate != nil {
		rate = make(chan int64, numGoroutines)
	}
	exit := make(chan struct{})
	nonceChan := make(chan Trytes)

	for i := 0; i < numGoroutines; i++ {
		go func(i int) {
			lmid, hmid := Para(c.State)
			lmid[nonceOffset] = PearlDiverMidStateLow0
			hmid[nonceOffset] = PearlDiverMidStateHigh0
			lmid[nonceOffset+1] = PearlDiverMidStateLow1
			hmid[nonceOffset+1] = PearlDiverMidStateHigh1
			lmid[nonceOffset+2] = PearlDiverMidStateLow2
			hmid[nonceOffset+2] = PearlDiverMidStateHigh2
			lmid[nonceOffset+3] = PearlDiverMidStateLow3
			hmid[nonceOffset+3] = PearlDiverMidStateHigh3

			incrN(i, lmid, hmid)
			nonce, r, _ := Loop(lmid, hmid, mwm, &cancelled, check, int(c.Rounds))

			if rate != nil {
				rate <- int64(math.Abs(float64(r)))
			}
			if r >= 0 && len(nonce) > 0 {
				select {
				case <-exit:
				case nonceChan <- MustTritsToTrytes(nonce):
					cancelled = true
				}
			}
		}(i)
	}

	if rate != nil {
		var rateSum int64
		for i := 0; i < numGoroutines; i++ {
			rateSum += <-rate
		}
		optRate <- rateSum
	}

	result = <-nonceChan
	close(exit)
	cancelled = true
	return result, nil
}
238617
func GoroutineScope(thread Thread) (*EvalScope, error) {
	locations, err := ThreadStacktrace(thread, 1)
	if err != nil {
		return nil, err
	}
	if len(locations) < 1 {
		return nil, errors.New("could not decode first frame")
	}
	g, err := GetG(thread)
	if err != nil {
		return nil, err
	}
	return FrameToScope(thread.BinInfo(), thread, g, locations...), nil
}
1003936
func (t *T) TryUntilf(
	f func() bool, timeout time.Duration, spec string, args ...interface{},
) {
	prefix := fmt.Sprintf(spec, args...) + ": "

	end := time.Now().Add(timeout)
	for time.Now().Before(end) {
		if f() {
			return
		}
		// Yield the processor so that other goroutines have a chance to work.
		// This is necessary since the function may not actually sleep at
		// all.
		runtime.Gosched()
	}

	t.Fatalf("%sTimeout after %s", prefix, timeout)
}
1121206
func (executor *UnboundedExecutor) Go(handler func(ctx context.Context)) {
	pc := reflect.ValueOf(handler).Pointer()
	f := runtime.FuncForPC(pc)
	funcName := f.Name()
	file, line := f.FileLine(pc)
	executor.activeGoroutinesMutex.Lock()
	defer executor.activeGoroutinesMutex.Unlock()
	startFrom := fmt.Sprintf("%s:%d", file, line)
	executor.activeGoroutines[startFrom] += 1
	go func() {
		defer func() {
			recovered := recover()
			// if you want to quit a goroutine without trigger HandlePanic
			// use runtime.Goexit() to quit
			if recovered != nil {
				if executor.HandlePanic == nil {
					HandlePanic(recovered, funcName)
				} else {
					executor.HandlePanic(recovered, funcName)
				}
			}
			executor.activeGoroutinesMutex.Lock()
			executor.activeGoroutines[startFrom] -= 1
			executor.activeGoroutinesMutex.Unlock()
		}()
		handler(executor.ctx)
	}()
}
410531
func LogStack() {
	buf := make([]byte, 1<<20)
	stacklen := runtime.Stack(buf, true)
	log.Infof("*** goroutine dump...\n%s\n*** end\n", buf[:stacklen])
}
1173735
func GoWithRecover(goroutine func(), customRecover func(err interface{})) {
	go func() {
		defer func() {
			if err := recover(); err != nil {
				customRecover(err)
			}
		}()
		goroutine()
	}()
}
843327
func (s *Simple) run() {
	for {
		select {
		case <-time.After(s.config.GCPeriodicity):
			s.GC()
		case <-s.halt:
			s.closeError <- nil
			// there is no cleanup required, so we just return
			return
		}
	}
}
652523
func (s *server) Stop() {
	if s.State() == Stopped {
		return
	}

	close(s.stopped)

	// make sure all goroutines have stopped before we close the log
	s.routineGroup.Wait()

	s.log.close()
	s.setState(Stopped)
}
242571
func GoroutineID() uint64 {
	b := make([]byte, 64)
	b = b[:runtime.Stack(b, false)]
	// Parse the 4707 out of "goroutine 4707 ["
	b = bytes.TrimPrefix(b, goroutineSpace)
	i := bytes.IndexByte(b, ' ')
	if i < 0 {
		panic(fmt.Sprintf("No space found in %q", b))
	}
	b = b[:i]
	n, err := strconv.ParseUint(string(b), 10, 64)
	if err != nil {
		panic(fmt.Sprintf("Failed to parse goroutine ID out of %q: %v", b, err))
	}
	return n
}
1136114
func (c *SafeCounter) Inc(key string) {
	c.mux.Lock()
	// Lock so only one goroutine at a time can access the map c.v.
	c.v[key]++
	c.mux.Unlock()
}
552765
func Any(asyncs ...Async) Async {
	return &asyncImpl{
		cancel: func() {
			for _, a := range asyncs {
				a.Cancel()
			}
		},
		promise: Promise(func() error {
			// TODO (m0sth8): rewrite this with select for every async
			for _, a := range asyncs {
				err := <-a.Result()
				if err != nil {
					return err
				}
			}
			return nil
		}),
	}
}
1240723
func (p *Pool) Go(goroutine func(stop chan bool)) {
	p.lock.Lock()
	newRoutine := routine{
		goroutine: goroutine,
		stop:      make(chan bool, 1),
	}
	p.routines = append(p.routines, newRoutine)
	p.waitGroup.Add(1)
	Go(func() {
		goroutine(newRoutine.stop)
		p.waitGroup.Done()
	})
	p.lock.Unlock()
}