// worker does all the parsing and validation of the certificate(s)
// contained in a single file. It first reads all the data in the
// file, then begins parsing certificates in the file. Those
// certificates are then checked for revocation.
198769
func worker(paths chan string, bundler chan *x509.Certificate, pool *sync.WaitGroup) {
	defer (*pool).Done()
	for {
		path, ok := <-paths
		if !ok {
			return
		}

		log.Infof("Loading %s", path)

		fileData, err := ioutil.ReadFile(path)
		if err != nil {
			log.Warningf("%v", err)
			continue
		}

		for {
			var block *pem.Block
			if len(fileData) == 0 {
				break
			}
			block, fileData = pem.Decode(fileData)
			if block == nil {
				log.Warningf("%s: no PEM data found", path)
				break
			} else if block.Type != "CERTIFICATE" {
				log.Info("Skipping non-certificate")
				continue
			}

			cert, err := x509.ParseCertificate(block.Bytes)
			if err != nil {
				log.Warningf("Invalid certificate: %v", err)
				continue
			}

			log.Infof("Validating %+v", cert.Subject)
			revoked, ok := revoke.VerifyCertificate(cert)
			if !ok {
				log.Warning("Failed to verify certificate.")
			} else if !revoked {
				bundler <- cert
			} else {
				log.Info("Skipping revoked certificate")
			}
		}
	}
}
1369033
func runWorkers(
	ctx context.Context,
	o *gcs.Object,
	bucket gcs.Bucket) (results []result, err error) {
	b := syncutil.NewBundle(ctx)

	// Set up a channel that is closed upon SIGINT.
	stop := make(chan struct{})
	go func() {
		c := make(chan os.Signal, 1)
		signal.Notify(c, os.Interrupt)

		<-c
		log.Printf("SIGINT received. Stopping soon...")
		close(stop)
	}()

	// Start several workers making random reads.
	var wg sync.WaitGroup
	resultChan := make(chan result)
	for i := 0; i < *fWorkers; i++ {
		wg.Add(1)
		b.Add(func(ctx context.Context) (err error) {
			defer wg.Done()
			err = makeReads(ctx, o, bucket, resultChan, stop)
			if err != nil {
				err = fmt.Errorf("makeReads: %v", err)
				return
			}

			return
		})
	}

	go func() {
		wg.Wait()
		close(resultChan)
	}()

	// Accumulate results.
	b.Add(func(ctx context.Context) (err error) {
		for r := range resultChan {
			results = append(results, r)
		}

		return
	})

	err = b.Join()
	return
}
455183
func (ew *EventWorker) Work() {
	for {
		select {
		case s := <-ew.sampleChan:
			ew.mutex.Lock()
			ew.samples = append(ew.samples, s)
			ew.mutex.Unlock()
		}
	}
}
514673
func serverWorker(sock mangos.Socket, id int) {
	var err error

	delay := rand.Intn(int(time.Second))

	for {
		var m *mangos.Message

		if m, err = sock.RecvMsg(); err != nil {
			return
		}

		m.Body = make([]byte, 4)

		time.Sleep(time.Duration(delay))

		binary.BigEndian.PutUint32(m.Body[0:], uint32(id))

		if err = sock.SendMsg(m); err != nil {
			return
		}
	}
}
360502
func (e eval) Run(in worker.Input, resChan chan worker.Result) {
	result := worker.Result{
		WorkerName: workerName,
	}
	crlRes := Result{}

	// Grab first CRL response we can from our certificate
	crlResponses, err := getCRLResponses(in.Certificate)
	if err != nil {
		resChan <- worker.Result{
			Success:    false,
			WorkerName: workerName,
			Errors:     []string{fmt.Sprintf("error getting CRL response, err=%v", err)},
			Result:     nil,
		}
		return
	}
	if len(crlResponses) == 0 {
		resChan <- worker.Result{
			Success:    false,
			WorkerName: workerName,
			Errors:     []string{fmt.Sprintf("no CRL responses received, err=%v", err)},
			Result:     nil,
		}
		return
	}

	// TODO(adam): store all CRL responses
	// TODO(adam): Process all responses? Or just one that's signed?

	certList, err := x509.ParseCRL(crlResponses[0])
	if err != nil {
		resChan <- worker.Result{
			Success:    false,
			WorkerName: workerName,
			Errors:     []string{fmt.Sprintf("error reading CRL response for %d", in.Certificate.ID)},
			Result:     nil,
		}
		return
	}

	// Verify the CRL
	if err = verifyCRL(certList, in.Certificate, in.CertificateChain); err != nil {
		resChan <- worker.Result{
			Success:    false,
			WorkerName: workerName,
			Errors:     []string{fmt.Sprintf("error verifying CRL, err=%v", err)},
			Result:     nil,
		}
		return
	}

	x509Cert, err := in.Certificate.ToX509()
	if err != nil {
		resChan <- worker.Result{
			Success:    false,
			WorkerName: workerName,
			Errors:     []string{fmt.Sprintf("error converting Certificate %d to x509.Certificate, err=%v", in.Certificate.ID, err)},
			Result:     nil,
		}
		return
	}

	// Check if our certificate is in the revoked list
	revoked := certList.TBSCertList.RevokedCertificates
	for i := range revoked {
		if x509Cert.SerialNumber.Cmp(revoked[i].SerialNumber) == 0 {
			// certificate is in revoked list, serials match
			crlRes.RevocationTime = revoked[i].RevocationTime
			crlRes.Revoked = true

			if err := in.DBHandle.UpdateCertMarkAsRevoked(in.Certificate.ID, certList.TBSCertList.ThisUpdate); err != nil {
				resChan <- worker.Result{
					Success:    false,
					WorkerName: workerName,
					Errors:     []string{fmt.Sprintf("error update Certificate %d revocation in database, err=%v", in.Certificate.ID, err)},
					Result:     nil,
				}
				return
			}

			// also mark is_valid=false on scans table
			if _, err = in.DBHandle.Exec(`UPDATE scans SET is_valid=false WHERE id=$1`, in.Scanid); err != nil {
				resChan <- worker.Result{
					Success:    false,
					WorkerName: workerName,
					Errors:     []string{fmt.Sprintf("error update Scan %d is_valid in database, err=%v", in.Scanid, err)},
					Result:     nil,
				}
				return
			}

			break
		}
	}

	// Marshal the response
	bs, err := json.Marshal(&crlRes)
	if err != nil {
		result.Success = false
		result.Errors = append(result.Errors, err.Error())
	} else {
		result.Success = true
		result.Result = bs
	}
	resChan <- result
}
1347123
func (s *SigPool) poolWorker() {
	defer s.wg.Done()

	for {
		select {

		// We've just received a new signature job. Given the items
		// contained within the message, we'll craft a signature and
		// send the result along with a possible error back to the
		// caller.
		case sigMsg := <-s.signJobs:
			rawSig, err := s.signer.SignOutputRaw(
				sigMsg.Tx, &sigMsg.SignDesc,
			)
			if err != nil {
				select {
				case sigMsg.Resp <- SignJobResp{
					Sig: lnwire.Sig{},
					Err: err,
				}:
					continue
				case <-sigMsg.Cancel:
					continue
				case <-s.quit:
					return
				}
			}

			sig, err := lnwire.NewSigFromRawSignature(rawSig)
			select {
			case sigMsg.Resp <- SignJobResp{
				Sig: sig,
				Err: err,
			}:
			case <-sigMsg.Cancel:
				continue
			case <-s.quit:
				return
			}

		// We've just received a new verification job from the outside
		// world. We'll attempt to construct the sighash, parse the
		// signature, and finally verify the signature.
		case verifyMsg := <-s.verifyJobs:
			sigHash, err := verifyMsg.SigHash()
			if err != nil {
				select {
				case verifyMsg.ErrResp <- &HtlcIndexErr{
					error:     err,
					VerifyJob: &verifyMsg,
				}:
					continue
				case <-verifyMsg.Cancel:
					continue
				}
			}

			rawSig := verifyMsg.Sig

			if !rawSig.Verify(sigHash, verifyMsg.PubKey) {
				err := fmt.Errorf("invalid signature "+
					"sighash: %x, sig: %x", sigHash,
					rawSig.Serialize())

				select {
				case verifyMsg.ErrResp <- &HtlcIndexErr{
					error:     err,
					VerifyJob: &verifyMsg,
				}:
				case <-verifyMsg.Cancel:
				case <-s.quit:
					return
				}
			} else {
				select {
				case verifyMsg.ErrResp <- nil:
				case <-verifyMsg.Cancel:
				case <-s.quit:
					return
				}
			}

		// The sigPool sig pool is exiting, so we will as well.
		case <-s.quit:
			return
		}
	}
}
31148
func (worker *Worker) err(e error) {
	if worker.ErrorHandler != nil {
		worker.ErrorHandler(e)
	}
}
162811
func (m *Manager) load() error {
	m.rw.Lock()
	defer m.rw.Unlock()

	domain := strings.TrimSuffix(m.Domain, ".")

	certKey, err := m.cachedKey(domain + keySuffix)
	if err != nil {
		return err
	}

	buf, err := ioutil.ReadFile(filepath.Join(m.CacheDir, domain+certSuffix))
	if err != nil {
		return err
	}

	var b *pem.Block
	var der [][]byte
	for {
		b, buf = pem.Decode(buf)
		if b == nil {
			break
		}
		if b.Type != "CERTIFICATE" {
			return ErrInvalidCertificate
		}
		der = append(der, b.Bytes)
		if buf == nil {
			break
		}
	}
	if len(der) == 0 {
		return ErrInvalidCertificate
	}

	leaf, err := parseCert(domain, der, certKey)
	if err != nil {
		return err
	}

	if time.Now().After(leaf.NotAfter) {
		return ErrCertificateExpired
	}

	m.cert = &tls.Certificate{
		Certificate: der,
		Leaf:        leaf,
		PrivateKey:  certKey,
	}

	return nil
}
754904
func (w *Worker) Work() {
	for {
		select {
		case m := <-w.PacketChan:
			w.ProcessMetric(&m)
		case m := <-w.ImportChan:
			for _, j := range m {
				w.ImportMetric(j)
			}
		case ms := <-w.ImportMetricChan:
			for _, m := range ms {
				w.ImportMetricGRPC(m)
			}
		case <-w.QuitChan:
			// We have been asked to stop.
			log.WithField("worker", w.id).Error("Stopping")
			return
		}
	}
}
446918
func ParseCertificateFile(certFile string) (*Certificate, error) {
	certPEM, err := ioutil.ReadFile(certFile)
	if err != nil {
		return nil, err
	}

	return ParseCertificatePEM(certPEM)
}
610158
func (lbc *loadBalancerController) worker() {
	for {
		key, _ := lbc.queue.Get()
		glog.Infof("Sync triggered by service %v", key)
		if err := lbc.sync(false); err != nil {
			glog.Warningf("Requeuing %v because of error: %v", key, err)
			lbc.queue.Add(key)
		}
		lbc.queue.Done(key)
	}
}
1180821
func (s *packetManager) workerChan(runWorker func(chan orderedRequest),
) chan orderedRequest {

	// multiple workers for faster read/writes
	rwChan := make(chan orderedRequest, SftpServerWorkerCount)
	for i := 0; i < SftpServerWorkerCount; i++ {
		runWorker(rwChan)
	}

	// single worker to enforce sequential processing of everything else
	cmdChan := make(chan orderedRequest)
	runWorker(cmdChan)

	pktChan := make(chan orderedRequest, SftpServerWorkerCount)
	go func() {
		for pkt := range pktChan {
			switch pkt.requestPacket.(type) {
			case *sshFxpReadPacket, *sshFxpWritePacket:
				s.incomingPacket(pkt)
				rwChan <- pkt
				continue
			case *sshFxpClosePacket:
				// wait for reads/writes to finish when file is closed
				// incomingPacket() call must occur after this
				s.working.Wait()
			}
			s.incomingPacket(pkt)
			// all non-RW use sequential cmdChan
			cmdChan <- pkt
		}
		close(rwChan)
		close(cmdChan)
		s.close()
	}()

	return pktChan
}
641620
func (d *bucketDataSource) worker(server string, workerCh chan []uint16, tlsConfig *tls.Config) int {
	atomic.AddUint64(&d.stats.TotWorkerBody, 1)

	if !d.isRunning() {
		return -1
	}

	atomic.AddUint64(&d.stats.TotWorkerConnect, 1)
	emptyWorkerCh := func() int {
		for {
			select {
			case _, ok := <-workerCh:
				if !ok {
					return -1 // workerCh was closed
				}

				// Else, keep looping to consume workerCh.

			default:
				return 0 // Stop loop when workerCh is empty.
			}
		}
	}

	var client *memcached.Client
	var err error
	if tlsConfig == nil {
		connect := d.options.Connect
		if connect == nil {
			connect = memcached.Connect
		}
		client, err = connect("tcp", server)
	} else {
		connectTLS := d.options.ConnectTLS
		if connectTLS == nil {
			connectTLS = memcached.ConnectTLS
		}
		client, err = connectTLS("tcp", server, tlsConfig)
	}

	if err != nil {
		atomic.AddUint64(&d.stats.TotWorkerConnectErr, 1)
		d.receiver.OnError(fmt.Errorf("worker connect, server: %s, err: %v",
			server, err))

		// If we can't connect, then maybe a node was rebalanced out
		// or failed-over, so consume the workerCh so that the
		// refresh-cluster goroutine will be unblocked and can receive
		// our kick.
		ret := emptyWorkerCh()

		d.Kick("worker-connect-err")

		return ret
	}
	atomic.AddUint64(&d.stats.TotWorkerConnectOk, 1)

	defer func() {
		atomic.AddUint64(&d.stats.TotWorkerClientClose, 1)
		client.Close()
		atomic.AddUint64(&d.stats.TotWorkerClientCloseDone, 1)
	}()

	var user, pswd string

	if d.auth != nil {
		if auth, ok := d.auth.(couchbase.GenericMcdAuthHandler); ok {
			atomic.AddUint64(&d.stats.TotWorkerAuthenticateMemcachedConn, 1)
			err = auth.AuthenticateMemcachedConn(server, client)
			if err != nil {
				atomic.AddUint64(&d.stats.TotWorkerAuthenticateMemcachedConnErr, 1)
				d.receiver.OnError(fmt.Errorf("worker auth,"+
					" AuthenticateMemcachedConn, server: %s, err: %v",
					server, err))

				// If we can't authenticate, then maybe a node was
				// rebalanced out, so consume the workerCh so that the
				// refresh-cluster goroutine will be unblocked and can
				// receive our kick.
				ret := emptyWorkerCh()

				d.Kick("worker-auth-AuthenticateMemcachedConn")

				return ret
			}
			atomic.AddUint64(&d.stats.TotWorkerAuthenticateMemcachedConnOk, 1)
		} else if auth, ok := d.auth.(couchbase.AuthWithSaslHandler); ok {
			user, pswd = auth.GetSaslCredentials()
		} else {
			user, pswd, _ = d.auth.GetCredentials()
		}

		if user != "" {
			atomic.AddUint64(&d.stats.TotWorkerAuth, 1)
			res, err := client.Auth(user, pswd)
			if err != nil {
				atomic.AddUint64(&d.stats.TotWorkerAuthErr, 1)
				d.receiver.OnError(fmt.Errorf("worker auth, server: %s,"+
					" user: %s, err: %v", server, user, err))

				// If we can't authenticate, then maybe a node was
				// rebalanced out, so consume the workerCh so that the
				// refresh-cluster goroutine will be unblocked and can
				// receive our kick.
				ret := emptyWorkerCh()

				d.Kick("worker-auth-client")

				return ret
			}

			if res.Status != gomemcached.SUCCESS {
				atomic.AddUint64(&d.stats.TotWorkerAuthFail, 1)
				d.receiver.OnError(&AuthFailError{ServerURL: server, User: user})
				return 0
			}
		}
	}

	uprOpenName := d.options.Name
	if uprOpenName == "" {
		uprOpenName = fmt.Sprintf("cbdatasource-%x", rand.Int63())
	}

	// Server handshakes.
	var openUprFlags uint32
	openUprFlags, err = serverHandShake(client, d)
	if err != nil {
		// Just bump the stats and continue with errors logged
		// since even at feature mismatch the openConn can proceed.
		atomic.AddUint64(&d.stats.TotHandShakeErr, 1)
		if d.options.Logf != nil {
			d.options.Logf("cbdatasource: serverHandShake, err: %v", err)
		}
	}

	// Select the bucket for this connection.
	if user != d.bucketName {
		_, err = client.SelectBucket(d.bucketName)
		if err != nil {
			atomic.AddUint64(&d.stats.TotSelectBucketErr, 1)
			d.receiver.OnError(err)
			return 0
		}
	}

	err = UPROpen(client, uprOpenName, d.options, openUprFlags)
	if err != nil {
		atomic.AddUint64(&d.stats.TotWorkerUPROpenErr, 1)
		d.receiver.OnError(err)
		return 0
	}
	atomic.AddUint64(&d.stats.TotWorkerUPROpenOk, 1)

	ackBytes :=
		uint32(d.options.FeedBufferAckThreshold * float32(d.options.FeedBufferSizeBytes))

	sendCh := make(chan interface{}, 1)
	sendEndCh := make(chan struct{})
	recvEndCh := make(chan struct{})

	cleanup := func(progress int, err error) int {
		atomic.AddUint64(&d.stats.TotWorkerCleanup, 1)

		if err != nil {
			d.receiver.OnError(err)
		}

		go func() {
			<-recvEndCh
			close(sendCh)
		}()

		atomic.AddUint64(&d.stats.TotWorkerCleanupDone, 1)
		return progress
	}

	currVBuckets := make(map[uint16]*VBucketState)
	currVBucketsMutex := sync.Mutex{} // Protects currVBuckets.

	go func() { // Sender goroutine.
		defer func() {
			close(sendEndCh)
			atomic.AddUint64(&d.stats.TotWorkerTransmitDone, 1)
		}()

		atomic.AddUint64(&d.stats.TotWorkerTransmitStart, 1)
		for msg := range sendCh {
			atomic.AddUint64(&d.stats.TotWorkerTransmit, 1)
			mcPkt, ok := msg.(*gomemcached.MCRequest)
			if ok {
				// Transmit a request.
				err := client.Transmit(mcPkt)
				if err != nil {
					atomic.AddUint64(&d.stats.TotWorkerTransmitErr, 1)
					d.receiver.OnError(fmt.Errorf("client.Transmit, err: %v", err))
					return
				}
			} else {
				mcPkt, ok := msg.(*gomemcached.MCResponse)
				if ok {
					// Transmit a response.
					err := client.TransmitResponse(mcPkt)
					if err != nil {
						atomic.AddUint64(&d.stats.TotWorkerTransmitErr, 1)
						d.receiver.OnError(fmt.Errorf("client.TransmitResponse, err: %v", err))
						return
					}
				} else {
					// Unknown packet.
					d.receiver.OnError(fmt.Errorf("Unidentified packet to transmit!"))
					return
				}
			}
			atomic.AddUint64(&d.stats.TotWorkerTransmitOk, 1)
		}
	}()

	go func() { // Receiver goroutine.
		defer func() {
			close(recvEndCh)
			atomic.AddUint64(&d.stats.TotWorkerReceiveDone, 1)
		}()

		traceCapacity := d.options.TraceCapacity
		if traceCapacity == 0 {
			traceCapacity = DefaultBucketDataSourceOptions.TraceCapacity
		}
		if traceCapacity < 0 {
			traceCapacity = 0
		}

		trs := map[uint16]*trace.RingBuffer{} // Keyed by vbucket.

		if d.options.Logf != nil {
			d.options.Logf("cbdatasource: receiver tracing,"+
				" server: %s, name: %s, capacity: %d",
				server, uprOpenName, traceCapacity)

			defer func() {
				vbids := make(sort.IntSlice, 0, len(trs))
				for vbid, tr := range trs {
					if tr != nil {
						vbids = append(vbids, int(vbid))
					}
				}
				sort.Sort(vbids)

				var buf bytes.Buffer
				for _, vbid := range vbids {
					tr := trs[uint16(vbid)]
					if tr != nil {
						fmt.Fprintf(&buf, " vb: %d => %s;", vbid,
							trace.MsgsToString(tr.Msgs(), ",", " "))
					}
				}

				d.options.Logf("cbdatasource: receiver closed,"+
					" server: %s, name: %s, traces:%s",
					server, uprOpenName, buf.String())
			}()
		}

		atomic.AddUint64(&d.stats.TotWorkerReceiveStart, 1)

		var hdr [gomemcached.HDR_LEN]byte
		var pkt gomemcached.MCRequest
		var res gomemcached.MCResponse

		// Track received bytes in case we need to buffer-ack.
		recvBytesTotal := uint32(0)

		conn := client.Hijack()

		for {
			// TODO: memory allocation here.
			atomic.AddUint64(&d.stats.TotWorkerReceive, 1)
			_, err := pkt.Receive(conn, hdr[:])
			if err != nil {
				atomic.AddUint64(&d.stats.TotWorkerReceiveErr, 1)
				d.receiver.OnError(fmt.Errorf("pkt.Receive, err: %v", err))
				return
			}
			atomic.AddUint64(&d.stats.TotWorkerReceiveOk, 1)

			tr := trs[pkt.VBucket]
			if tr == nil {
				tr = trace.NewRingBuffer(traceCapacity, trace.ConsolidateByTitle)
				trs[pkt.VBucket] = tr
			}

			if pkt.Opcode == gomemcached.UPR_MUTATION ||
				pkt.Opcode == gomemcached.UPR_DELETION ||
				pkt.Opcode == gomemcached.UPR_EXPIRATION {
				tr.Add("md", nil)

				atomic.AddUint64(&d.stats.TotUPRDataChange, 1)

				vbucketID := pkt.VBucket

				currVBucketsMutex.Lock()

				vbucketState := currVBuckets[vbucketID]
				if vbucketState == nil || vbucketState.State != "running" {
					currVBucketsMutex.Unlock()
					atomic.AddUint64(&d.stats.TotUPRDataChangeStateErr, 1)
					d.receiver.OnError(fmt.Errorf("error: DataChange,"+
						" wrong vbucketState: %#v, err: %v", vbucketState, err))
					return
				}

				if !vbucketState.SnapSaved {
					// NOTE: Following the ep-engine's approach, we
					// wait to persist SnapStart/SnapEnd until we see
					// the first mutation/deletion in the new snapshot
					// range.  That reduces a race window where if we
					// kill and restart this process right now after a
					// setVBucketMetaData() and before the next,
					// first-mutation-in-snapshot, then a restarted
					// stream-req using this just-saved
					// SnapStart/SnapEnd might have a lastSeq number <
					// SnapStart, where Couchbase Server will respond
					// to the stream-req with an ERANGE error code.
					v, _, err := d.getVBucketMetaData(vbucketID)
					if err != nil || v == nil {
						currVBucketsMutex.Unlock()
						d.receiver.OnError(fmt.Errorf("error: DataChange,"+
							" getVBucketMetaData, vbucketID: %d, err: %v",
							vbucketID, err))
						return
					}

					v.SnapStart = vbucketState.SnapStart
					v.SnapEnd = vbucketState.SnapEnd

					err = d.setVBucketMetaData(vbucketID, v)
					if err != nil {
						currVBucketsMutex.Unlock()
						d.receiver.OnError(fmt.Errorf("error: DataChange,"+
							" getVBucketMetaData, vbucketID: %d, err: %v",
							vbucketID, err))
						return
					}

					vbucketState.SnapSaved = true
				}

				currVBucketsMutex.Unlock()

				seq := binary.BigEndian.Uint64(pkt.Extras[:8])

				if pkt.Opcode == gomemcached.UPR_MUTATION {
					atomic.AddUint64(&d.stats.TotUPRDataChangeMutation, 1)
					err = d.receiver.DataUpdate(vbucketID, pkt.Key, seq, &pkt)
				} else {
					if pkt.Opcode == gomemcached.UPR_DELETION {
						atomic.AddUint64(&d.stats.TotUPRDataChangeDeletion, 1)
					} else {
						atomic.AddUint64(&d.stats.TotUPRDataChangeExpiration, 1)
					}
					err = d.receiver.DataDelete(vbucketID, pkt.Key, seq, &pkt)
				}

				if err != nil {
					atomic.AddUint64(&d.stats.TotUPRDataChangeErr, 1)
					d.receiver.OnError(fmt.Errorf("error: DataChange, err: %v", err))
					return
				}

				atomic.AddUint64(&d.stats.TotUPRDataChangeOk, 1)

				// Update received bytes for mutations, deletions, expirations
				recvBytesTotal += uint32(gomemcached.HDR_LEN) +
					uint32(len(pkt.Key)+len(pkt.Extras)+len(pkt.Body))
			} else {
				tr.Add(fmt.Sprintf("%d", pkt.Opcode), nil)

				res.Opcode = pkt.Opcode
				res.Opaque = pkt.Opaque
				res.Status = gomemcached.Status(pkt.VBucket)
				res.Extras = pkt.Extras
				res.Cas = pkt.Cas
				res.Key = pkt.Key
				res.Body = pkt.Body

				atomic.AddUint64(&d.stats.TotWorkerHandleRecv, 1)
				currVBucketsMutex.Lock()
				err := d.handleRecv(sendCh, currVBuckets, &res)
				currVBucketsMutex.Unlock()
				if err != nil {
					atomic.AddUint64(&d.stats.TotWorkerHandleRecvErr, 1)
					d.receiver.OnError(fmt.Errorf("error: HandleRecv, err: %v", err))
					return
				}
				atomic.AddUint64(&d.stats.TotWorkerHandleRecvOk, 1)

				if pkt.Opcode == gomemcached.UPR_SNAPSHOT ||
					pkt.Opcode == gomemcached.SET_VBUCKET ||
					pkt.Opcode == gomemcached.UPR_STREAMEND {
					// Update bytes received only for select packets
					// See: https://github.com/couchbase/kv_engine/blob/master/docs..
					//      ../dcp/documentation/flow-control.md#buffering-messages
					recvBytesTotal += uint32(gomemcached.HDR_LEN) +
						uint32(len(pkt.Key)+len(pkt.Extras)+len(pkt.Body))
				}
			}

			if ackBytes > 0 && recvBytesTotal > ackBytes {
				atomic.AddUint64(&d.stats.TotUPRBufferAck, 1)
				ack := &gomemcached.MCRequest{Opcode: gomemcached.UPR_BUFFERACK}
				ack.Extras = make([]byte, 4) // TODO: Memory mgmt.
				binary.BigEndian.PutUint32(ack.Extras, uint32(recvBytesTotal))
				sendCh <- ack
				recvBytesTotal = 0
			}
		}
	}()

	atomic.AddUint64(&d.stats.TotWorkerBodyKick, 1)
	d.Kick("new-worker")

	pingTimeoutMS := d.options.PingTimeoutMS
	if pingTimeoutMS <= 0 {
		pingTimeoutMS = DefaultBucketDataSourceOptions.PingTimeoutMS
	}

	var prevActivity uint64

	for {
		currVBucketsMutex.Lock()
		d.logVBucketStates(server, uprOpenName, "worker, looping beg",
			currVBuckets, nil)
		currVBucketsMutex.Unlock()

		select {
		case <-sendEndCh:
			atomic.AddUint64(&d.stats.TotWorkerSendEndCh, 1)
			return cleanup(0, nil)

		case <-recvEndCh:
			// If we lost a connection, then maybe a node was rebalanced out,
			// or failed over, so ask for a cluster refresh just in case.
			d.Kick("recvEndCh")

			atomic.AddUint64(&d.stats.TotWorkerRecvEndCh, 1)
			return cleanup(0, nil)

		case wantVBucketIDs, alive := <-workerCh:
			atomic.AddUint64(&d.stats.TotRefreshWorker, 1)

			if !alive {
				atomic.AddUint64(&d.stats.TotRefreshWorkerDone, 1)
				return cleanup(-1, nil) // We've been asked to shutdown.
			}

			currVBucketsMutex.Lock()
			d.logVBucketStates(server, uprOpenName, "refreshWorker-prior",
				currVBuckets, nil)
			err := d.refreshWorker(sendCh, currVBuckets, wantVBucketIDs)
			d.logVBucketStates(server, uprOpenName, "refreshWorker-after",
				currVBuckets, err)
			currVBucketsMutex.Unlock()
			if err != nil {
				return cleanup(0, err)
			}

			atomic.AddUint64(&d.stats.TotRefreshWorkerOk, 1)

		case <-time.After(time.Duration(pingTimeoutMS) * time.Millisecond):
			atomic.AddUint64(&d.stats.TotPingTimeout, 1)

			currActivity := atomic.LoadUint64(&d.stats.TotWorkerTransmit) +
				atomic.LoadUint64(&d.stats.TotWorkerReceive)
			if currActivity == prevActivity { // If no activity, then ping.
				atomic.AddUint64(&d.stats.TotPingReq, 1)
				sendCh <- &gomemcached.MCRequest{Opcode: gomemcached.NOOP}
				atomic.AddUint64(&d.stats.TotPingReqDone, 1)
			}

			prevActivity = currActivity
		}
	}

	return cleanup(-1, nil) // Unreached.
}
400186
func (w Worker) Run() {
	w.Ready <- w

	for incoming := range w.DataChan {
		data := incoming.Data

		switch data[0] {
		case BundleTag[0]:
			bundle, err := ParseBundle(data, incoming.Sender)
			if err != nil {
				w.ErrChan <- err
			}
			if err := w.Dispatcher.Dispatch(bundle, w.ExactMatch); err != nil {
				w.ErrChan <- errors.Wrap(err, "dispatch bundle")
			}
		case MessageChar:
			msg, err := ParseMessage(data, incoming.Sender)
			if err != nil {
				w.ErrChan <- err
			}
			if err := w.Dispatcher.Invoke(msg, w.ExactMatch); err != nil {
				w.ErrChan <- errors.Wrap(err, "dispatch message")
			}
		default:
			w.ErrChan <- ErrParse
		}
		// Announce the worker is ready again.
		w.Ready <- w
	}
}
572371
func (op *CaasOperatorAgent) Workers() (worker.Worker, error) {
	updateAgentConfLogging := func(loggingConfig string) error {
		return op.AgentConf.ChangeConfig(func(setter agent.ConfigSetter) error {
			setter.SetLoggingConfig(loggingConfig)
			return nil
		})
	}

	agentConfig := op.AgentConf.CurrentConfig()
	manifolds := CaasOperatorManifolds(caasoperator.ManifoldsConfig{
		Agent:                agent.APIHostPortsSetter{op},
		AgentConfigChanged:   op.configChangedVal,
		Clock:                clock.WallClock,
		LogSource:            op.bufferedLogger.Logs(),
		UpdateLoggerConfig:   updateAgentConfLogging,
		PrometheusRegisterer: op.prometheusRegistry,
		LeadershipGuarantee:  15 * time.Second,
		PreUpgradeSteps:      op.preUpgradeSteps,
		UpgradeStepsLock:     op.upgradeComplete,
		ValidateMigration:    op.validateMigration,
		MachineLock:          op.machineLock,
		PreviousAgentVersion: agentConfig.UpgradedToVersion(),
	})

	engine, err := dependency.NewEngine(dependencyEngineConfig())
	if err != nil {
		return nil, err
	}
	if err := dependency.Install(engine, manifolds); err != nil {
		if err := worker.Stop(engine); err != nil {
			logger.Errorf("while stopping engine with bad manifolds: %v", err)
		}
		return nil, err
	}
	if err := startIntrospection(introspectionConfig{
		Agent:              op,
		Engine:             engine,
		MachineLock:        op.machineLock,
		NewSocketName:      DefaultIntrospectionSocketName,
		PrometheusGatherer: op.prometheusRegistry,
		WorkerFunc:         introspection.NewWorker,
	}); err != nil {
		// If the introspection worker failed to start, we just log error
		// but continue. It is very unlikely to happen in the real world
		// as the only issue is connecting to the abstract domain socket
		// and the agent is controlled by by the OS to only have one.
		logger.Errorf("failed to start introspection worker: %v", err)
	}
	return engine, nil
}
1041042
func (t *taskQueue) worker() {
	for {
		key, quit := t.queue.Get()
		if quit {
			close(t.workerDone)
			return
		}
		glog.V(3).Infof("syncing %v", key)
		if err := t.sync(key.(string)); err != nil {
			glog.V(3).Infof("requeuing %v, err %v", key, err)
			t.requeue(key.(string))
		} else {
			t.queue.Forget(key)
		}

		t.queue.Done(key)
	}
}
482062
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Annotate(err, "validating config")
	}
	updater, err := config.NewUpdater()
	if err != nil {
		return nil, errors.Annotate(err, "getting new updater")
	}
	w := &updaterWorker{
		config:  config,
		updater: updater,
	}
	w.tomb.Go(w.loop)
	return w, nil
}