//  The main task is a load-balancer with heartbeating on workers so we
//  can detect crashed or blocked worker tasks:
773190
func main() {
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	backend, _ := zmq.NewSocket(zmq.ROUTER)
	defer frontend.Close()
	defer backend.Close()
	frontend.Bind("tcp://*:5555") //  For clients
	backend.Bind("tcp://*:5556")  //  For workers

	//  List of available workers
	workers := make([]worker_t, 0)

	//  Send out heartbeats at regular intervals
	heartbeat_at := time.Tick(HEARTBEAT_INTERVAL)

	poller1 := zmq.NewPoller()
	poller1.Add(backend, zmq.POLLIN)
	poller2 := zmq.NewPoller()
	poller2.Add(backend, zmq.POLLIN)
	poller2.Add(frontend, zmq.POLLIN)

	for {
		//  Poll frontend only if we have available workers
		var sockets []zmq.Polled
		var err error
		if len(workers) > 0 {
			sockets, err = poller2.Poll(HEARTBEAT_INTERVAL)
		} else {
			sockets, err = poller1.Poll(HEARTBEAT_INTERVAL)
		}
		if err != nil {
			break //  Interrupted
		}

		for _, socket := range sockets {
			switch socket.Socket {
			case backend:
				//  Handle worker activity on backend
				//  Use worker identity for load-balancing
				msg, err := backend.RecvMessage(0)
				if err != nil {
					break //  Interrupted
				}

				//  Any sign of life from worker means it's ready
				identity, msg := unwrap(msg)
				workers = s_worker_ready(s_worker_new(identity), workers)

				//  Validate control message, or return reply to client
				if len(msg) == 1 {
					if msg[0] != PPP_READY && msg[0] != PPP_HEARTBEAT {
						fmt.Println("E: invalid message from worker", msg)
					}
				} else {
					frontend.SendMessage(msg)
				}
			case frontend:
				//  Now get next client request, route to next worker
				msg, err := frontend.RecvMessage(0)
				if err != nil {
					break //  Interrupted
				}
				backend.SendMessage(workers[0].identity, msg)
				workers = workers[1:]
			}
		}

		//  We handle heartbeating after any socket activity. First we send
		//  heartbeats to any idle workers if it's time. Then we purge any
		//  dead workers:

		select {
		case <-heartbeat_at:
			for _, worker := range workers {
				backend.SendMessage(worker.identity, PPP_HEARTBEAT)
			}
		default:
		}
		workers = s_workers_purge(workers)
	}
}
963683
func worker_task() {
	worker, _ := zmq.NewSocket(zmq.REQ)
	defer worker.Close()
	worker.Connect("ipc://backend.ipc")

	//  Tell broker we're ready for work
	worker.SendMessage(WORKER_READY)

	//  Process messages as they arrive
	for {
		msg, e := worker.RecvMessage(0)
		if e != nil {
			break //  Interrupted
		}
		msg[len(msg)-1] = "OK"
		worker.SendMessage(msg)
	}
}
989332
func main() {
	verbose := false
	if len(os.Args) > 1 && os.Args[1] == "-v" {
		verbose = true
	}

	broker, _ := NewBroker(verbose)
	broker.Bind("tcp://*:5555")

	poller := zmq.NewPoller()
	poller.Add(broker.socket, zmq.POLLIN)

	//  Get and process messages forever or until interrupted
	for {
		polled, err := poller.Poll(HEARTBEAT_INTERVAL)
		if err != nil {
			break //  Interrupted
		}

		//  Process next input message, if any
		if len(polled) > 0 {
			msg, err := broker.socket.RecvMessage(0)
			if err != nil {
				break //  Interrupted
			}
			if broker.verbose {
				log.Printf("I: received message: %q\n", msg)
			}
			sender, msg := popStr(msg)
			_, msg = popStr(msg)
			header, msg := popStr(msg)

			switch header {
			case mdapi.MDPC_CLIENT:
				broker.ClientMsg(sender, msg)
			case mdapi.MDPW_WORKER:
				broker.WorkerMsg(sender, msg)
			default:
				log.Printf("E: invalid message: %q\n", msg)
			}
		}
		//  Disconnect and delete any expired workers
		//  Send heartbeats to idle workers if needed
		if time.Now().After(broker.heartbeat_at) {
			broker.Purge()
			for _, worker := range broker.waiting {
				worker.Send(mdapi.MDPW_HEARTBEAT, "", []string{})
			}
			broker.heartbeat_at = time.Now().Add(HEARTBEAT_INTERVAL)
		}
	}
	log.Println("W: interrupt received, shutting down...")
}
1137633
func (d *bucketDataSource) workerStart(server string, workerCh chan []uint16, tlsConfig *tls.Config) {
	backoffFactor := d.options.DataManagerBackoffFactor
	if backoffFactor <= 0.0 {
		backoffFactor = DefaultBucketDataSourceOptions.DataManagerBackoffFactor
	}
	sleepInitMS := d.options.DataManagerSleepInitMS
	if sleepInitMS <= 0 {
		sleepInitMS = DefaultBucketDataSourceOptions.DataManagerSleepInitMS
	}
	sleepMaxMS := d.options.DataManagerSleepMaxMS
	if sleepMaxMS <= 0 {
		sleepMaxMS = DefaultBucketDataSourceOptions.DataManagerSleepMaxMS
	}

	// Use exponential backoff loop to handle connect retries to the server.
	go func() {
		atomic.AddUint64(&d.stats.TotWorkerStart, 1)

		ExponentialBackoffLoop("cbdatasource.worker-"+server,
			func() int { return d.worker(server, workerCh, tlsConfig) },
			sleepInitMS, backoffFactor, sleepMaxMS)

		atomic.AddUint64(&d.stats.TotWorkerDone, 1)
	}()
}
360523
func (p *workerPool) Run() ([]interface{}, []error) {
	for i := 0; i < p.concurrency; i++ {
		go p.work()
	}

	p.wg.Add(len(p.Tasks))
	for _, task := range p.Tasks {
		p.tasksChan <- task
	}

	// all workers return
	close(p.tasksChan)

	p.wg.Wait()
	return p.workResults()
}
994391
func Main(port int, peers []string, diskAllocated int) {
	s, err := newServer(port, peers, diskAllocated)
	if err != nil {
		log.Fatal(err)
	}

	bitcoin.NewClient()

	s.Fatal(s.network.Listen())
}
1193680
func (tw *SpanWorker) Work() {
	const Timeout = 9 * time.Second
	capcmp := cap(tw.SpanChan) - 1
	for m := range tw.SpanChan {
		// If we are at or one below cap, increment the counter.
		if len(tw.SpanChan) >= capcmp {
			atomic.AddInt64(&tw.capCount, 1)
		}

		if m.Tags == nil && len(tw.commonTags) != 0 {
			m.Tags = make(map[string]string, len(tw.commonTags))
		}

		for k, v := range tw.commonTags {
			if _, has := m.Tags[k]; !has {
				m.Tags[k] = v
			}
		}

		// An SSF packet may contain a valid span, one or more valid metrics,
		// or both (a valid span *and* one or more valid metrics).
		// If it contains neither, it is the result of a client error, and the
		// span does not need to be passed to any sink.
		// If the span is empty but one or more metrics exist, the span still needs
		// to be passed to the sinks for potential metric extraction.
		if err := protocol.ValidateTrace(m); err != nil {
			if len(m.Metrics) == 0 {
				atomic.AddInt64(&tw.emptySSFCount, 1)
				log.WithError(err).Debug("Invalid SSF packet: packet contains neither valid metrics nor a valid span")
				continue
			}
		}

		var wg sync.WaitGroup
		for i, s := range tw.sinks {
			tags := tw.sinkTags[i]
			wg.Add(1)
			go func(i int, sink sinks.SpanSink, span *ssf.SSFSpan, wg *sync.WaitGroup) {
				defer wg.Done()

				done := make(chan struct{})
				start := time.Now()

				go func() {
					// Give each sink a change to ingest.
					err := sink.Ingest(span)
					if err != nil {
						if _, isNoTrace := err.(*protocol.InvalidTrace); !isNoTrace {
							// If a sink goes wacko and errors a lot, we stand to emit a
							// loooot of metrics towards all span workers here since
							// span ingest rates can be very high. C'est la vie.
							t := make([]string, 0, len(tags)+1)
							for k, v := range tags {
								t = append(t, k+":"+v)
							}

							t = append(t, "sink:"+sink.Name())
							tw.statsd.Incr("worker.span.ingest_error_total", t, 1.0)
						}
					}
					done <- struct{}{}
				}()

				select {
				case _ = <-done:
				case <-time.After(Timeout):
					log.WithFields(logrus.Fields{
						"sink":  sink.Name(),
						"index": i,
					}).Error("Timed out on sink ingestion")

					t := make([]string, 0, len(tags)+1)
					for k, v := range tags {
						t = append(t, k+":"+v)
					}

					t = append(t, "sink:"+sink.Name())
					tw.statsd.Incr("worker.span.ingest_timeout_total", t, 1.0)
				}
				atomic.AddInt64(&tw.cumulativeTimes[i], int64(time.Since(start)/time.Nanosecond))
			}(i, s, m, &wg)
		}
		wg.Wait()
	}
}
1298772
func main() {
	//  Arguments can be either of:
	//      -p  primary server, at tcp://localhost:5001
	//      -b  backup server, at tcp://localhost:5002
	statepub, _ := zmq.NewSocket(zmq.PUB)
	statesub, _ := zmq.NewSocket(zmq.SUB)
	statesub.SetSubscribe("")
	frontend, _ := zmq.NewSocket(zmq.ROUTER)
	fsm := &bstar_t{peer_expiry: time.Now().Add(2 * HEARTBEAT)}

	if len(os.Args) == 2 && os.Args[1] == "-p" {
		fmt.Println("I: Primary active, waiting for backup (passive)")
		frontend.Bind("tcp://*:5001")
		statepub.Bind("tcp://*:5003")
		statesub.Connect("tcp://localhost:5004")
		fsm.state = STATE_PRIMARY
	} else if len(os.Args) == 2 && os.Args[1] == "-b" {
		fmt.Println("I: Backup passive, waiting for primary (active)")
		frontend.Bind("tcp://*:5002")
		statepub.Bind("tcp://*:5004")
		statesub.Connect("tcp://localhost:5003")
		fsm.state = STATE_BACKUP
	} else {
		fmt.Println("Usage: bstarsrv { -p | -b }")
		return
	}
	//  We now process events on our two input sockets, and process these
	//  events one at a time via our finite-state machine. Our "work" for
	//  a client request is simply to echo it back:

	//  Set timer for next outgoing state message
	send_state_at := time.Now().Add(HEARTBEAT)

	poller := zmq.NewPoller()
	poller.Add(frontend, zmq.POLLIN)
	poller.Add(statesub, zmq.POLLIN)

LOOP:
	for {
		time_left := send_state_at.Sub(time.Now())
		if time_left < 0 {
			time_left = 0
		}
		polled, err := poller.Poll(time_left)
		if err != nil {
			break //  Context has been shut down
		}
		for _, socket := range polled {
			switch socket.Socket {
			case frontend:
				//  Have a client request
				msg, _ := frontend.RecvMessage(0)
				fsm.event = CLIENT_REQUEST
				if !StateMachine(fsm) {
					//  Answer client by echoing request back
					frontend.SendMessage(msg)
				}
			case statesub:
				//  Have state from our peer, execute as event
				message, _ := statesub.RecvMessage(0)
				i, _ := strconv.Atoi(message[0])
				fsm.event = event_t(i)
				if StateMachine(fsm) {
					break LOOP //  Error, so exit
				}
				fsm.peer_expiry = time.Now().Add(2 * HEARTBEAT)
			}
		}
		//  If we timed-out, send state to peer
		if time.Now().After(send_state_at) {
			statepub.SendMessage(int(fsm.state))
			send_state_at = time.Now().Add(HEARTBEAT)
		}
	}
	fmt.Println("W: interrupted")
}
1231443
func (w *worker) handleReorgTasks(reorgInfo *reorgInfo, totalAddedCount *int64, workers []*addIndexWorker, batchTasks []*reorgIndexTask) error {
	for i, task := range batchTasks {
		workers[i].taskCh <- task
	}

	startHandle := batchTasks[0].startHandle
	taskCnt := len(batchTasks)
	startTime := time.Now()
	nextHandle, taskAddedCount, err := w.waitTaskResults(workers, taskCnt, totalAddedCount, startHandle)
	elapsedTime := time.Since(startTime)
	if err == nil {
		err = w.isReorgRunnable(reorgInfo.d)
	}

	if err != nil {
		// update the reorg handle that has been processed.
		err1 := kv.RunInNewTxn(reorgInfo.d.store, true, func(txn kv.Transaction) error {
			return errors.Trace(reorgInfo.UpdateReorgMeta(txn, nextHandle, reorgInfo.EndHandle, reorgInfo.PhysicalTableID))
		})
		metrics.BatchAddIdxHistogram.WithLabelValues(metrics.LblError).Observe(elapsedTime.Seconds())
		logutil.Logger(ddlLogCtx).Warn("[ddl] add index worker handle batch tasks failed", zap.Int64("totalAddedCount", *totalAddedCount), zap.Int64("startHandle", startHandle), zap.Int64("nextHandle", nextHandle),
			zap.Int64("batchAddedCount", taskAddedCount), zap.String("taskFailedError", err.Error()), zap.String("takeTime", elapsedTime.String()), zap.NamedError("updateHandleError", err1))
		return errors.Trace(err)
	}

	// nextHandle will be updated periodically in runReorgJob, so no need to update it here.
	w.reorgCtx.setNextHandle(nextHandle)
	metrics.BatchAddIdxHistogram.WithLabelValues(metrics.LblOK).Observe(elapsedTime.Seconds())
	logutil.Logger(ddlLogCtx).Info("[ddl] add index worker handle batch tasks successful", zap.Int64("totalAddedCount", *totalAddedCount), zap.Int64("startHandle", startHandle),
		zap.Int64("nextHandle", nextHandle), zap.Int64("batchAddedCount", taskAddedCount), zap.String("takeTime", elapsedTime.String()))
	return nil
}
103062
func (p *ParallelManager) addWorker() {
	if atomic.LoadUint32(&p.workersNum) >= maxParallelWorkers {
		// Number of maximum workers is reached, no need to
		// to create a new one.
		return
	}

	// Update number of threads
	atomic.AddUint32(&p.workersNum, 1)

	// Start a new worker
	p.wg.Add(1)
	go func() {
		for {
			// Wait for jobs
			fn, ok := <-p.queueCh
			if !ok {
				// No more tasks, quit
				p.wg.Done()
				return
			}
			// Execute the task and send the result
			// to result channel.
			p.resultCh <- fn()
		}
	}()
}
803046
func main() {
	// DEBUG
	runtime.GOMAXPROCS(2)

	log.SetPrefix("[go-redis|bench] ")
	flag.Parse()

	fmt.Printf("\n\n== Bench synchclient == %d goroutines 1 AsyncClient  -- %d opts each --- \n\n", *workers, *opcnt)

	for _, task := range tasks {
		// REVU: creates a new client for each task run
		// wondering if using the same conn throughout will be more realistic?
		// regardless flushdb per task is OK
		benchTask(task, *opcnt, *workers, true)
	}

}
702631
func (w *worker) mainLoop() {
	defer w.txsSub.Unsubscribe()
	defer w.chainHeadSub.Unsubscribe()
	defer w.chainSideSub.Unsubscribe()

	for {
		select {
		case req := <-w.newWorkCh:
			w.commitNewWork(req.interrupt, req.noempty, req.timestamp)

		case ev := <-w.chainSideCh:
			// Short circuit for duplicate side blocks
			if _, exist := w.localUncles[ev.Block.Hash()]; exist {
				continue
			}
			if _, exist := w.remoteUncles[ev.Block.Hash()]; exist {
				continue
			}
			// Add side block to possible uncle block set depending on the author.
			if w.isLocalBlock != nil && w.isLocalBlock(ev.Block) {
				w.localUncles[ev.Block.Hash()] = ev.Block
			} else {
				w.remoteUncles[ev.Block.Hash()] = ev.Block
			}
			// If our mining block contains less than 2 uncle blocks,
			// add the new uncle block if valid and regenerate a mining block.
			if w.isRunning() && w.current != nil && w.current.uncles.Cardinality() < 2 {
				start := time.Now()
				if err := w.commitUncle(w.current, ev.Block.Header()); err == nil {
					var uncles []*types.Header
					w.current.uncles.Each(func(item interface{}) bool {
						hash, ok := item.(common.Hash)
						if !ok {
							return false
						}
						uncle, exist := w.localUncles[hash]
						if !exist {
							uncle, exist = w.remoteUncles[hash]
						}
						if !exist {
							return false
						}
						uncles = append(uncles, uncle.Header())
						return false
					})
					w.commit(uncles, nil, true, start)
				}
			}

		case ev := <-w.txsCh:
			// Apply transactions to the pending state if we're not mining.
			//
			// Note all transactions received may not be continuous with transactions
			// already included in the current mining block. These transactions will
			// be automatically eliminated.
			if !w.isRunning() && w.current != nil {
				w.mu.RLock()
				coinbase := w.coinbase
				w.mu.RUnlock()

				txs := make(map[common.Address]types.Transactions)
				for _, tx := range ev.Txs {
					acc, _ := types.Sender(w.current.signer, tx)
					txs[acc] = append(txs[acc], tx)
				}
				txset := types.NewTransactionsByPriceAndNonce(w.current.signer, txs)
				w.commitTransactions(txset, coinbase, nil)
				w.updateSnapshot()
			} else {
				// If clique is running in dev mode(period is 0), disable
				// advance sealing here.
				if w.chainConfig.Clique != nil && w.chainConfig.Clique.Period == 0 {
					w.commitNewWork(nil, true, time.Now().Unix())
				}
			}
			atomic.AddInt32(&w.newTxs, int32(len(ev.Txs)))

		// System stopped
		case <-w.exitCh:
			return
		case <-w.txsSub.Err():
			return
		case <-w.chainHeadSub.Err():
			return
		case <-w.chainSideSub.Err():
			return
		}
	}
}
1004275
func (c oauthClient) Task(taskID uint) (wl.Task, error) {
	url := fmt.Sprintf(
		"%s/tasks/%d",
		c.apiURL,
		taskID,
	)

	req, err := c.newGetRequest(url)
	if err != nil {
		return wl.Task{}, err
	}

	resp, err := c.do(req)
	if err != nil {
		return wl.Task{}, err
	}

	if resp.StatusCode != http.StatusOK {
		return wl.Task{}, fmt.Errorf("Unexpected response code %d - expected %d", resp.StatusCode, http.StatusOK)
	}

	task := transportTask{}
	err = json.NewDecoder(resp.Body).Decode(&task)
	if err != nil {
		return wl.Task{}, err
	}
	return taskFromTransport(task)
}
624899
func (brq *blockRetrievalQueue) notifyWorker(priority int) {
	// On-demand workers and prefetch workers share the priority queue. This
	// allows maximum time for requests to jump the queue, at least until the
	// worker actually begins working on it.
	//
	// Note that the worker being notified won't necessarily work on the exact
	// request that caused the notification. It's just a counter. That means
	// that sometimes on-demand workers will work on prefetch requests, and
	// vice versa. But the numbers should match.
	//
	// However, there are some pathological scenarios where if all the workers
	// of one type are making progress but the other type are not (which is
	// highly improbable), requests of one type could starve the other. By
	// design, on-demand requests _should_ starve prefetch requests, so this is
	// a problem only if prefetch requests can starve on-demand workers. But
	// because there are far more on-demand workers than prefetch workers, this
	// should never actually happen.
	workerCh := brq.workerCh
	if priority <= throttleRequestPriority && brq.throttledWorkCh != nil {
		workerCh = brq.throttledWorkCh
	} else if priority < defaultOnDemandRequestPriority {
		workerCh = brq.prefetchWorkerCh
	}
	brq.sendWork(workerCh)
}
534674
func managementTask(c *asyncConnHdl, ctl workerCtl) (sig *interrupt_code, te *taskStatus) {
	/* TODO:
	connected:
		clearChannels();
		startWokers();
	disconnected:
		?
	on_fault:
		disconnect();
		goto disconnected;
	on_exit:
		?
	*/
	//	log.Println("MGR: do task ...")
	select {
	case stat := <-c.feedback:
		// do the shutdown for now -- TODO: try reconnect
		if stat.event == faulted || stat.event == quit_processed {
			if stat.event == faulted {
				// REVU - pretty please TODO do the customized log
				log.Printf("<INFO> - %s (manager task) FAULT EVENT ", c)
			}
			// REVU - pretty please TODO do the customized log
			//			log.Printf("<INFO> %s - (manager task) SHUTTING DOWN ...", c)
			c.shutdown <- true

			// REVU - pretty please TODO do the customized log
			//			log.Printf("<INFO> %s - (manager task) RAISING SIGNAL STOP ...", c)
			go func() { c.reqProcCtl <- stop }()
			go func() { c.rspProcCtl <- stop }()
			go func() { c.heartbeatCtl <- stop }()
			go func() { c.managerCtl <- stop }()
		}
	case s := <-ctl:
		return &s, &ok_status
	}

	return nil, &ok_status
}
472193
func reqProcessingTask(c *asyncConnHdl, ctl workerCtl) (ic *interrupt_code, ts *taskStatus) {

	var err error
	var errmsg string

	bytecnt := 0
	blen := 0
	bufsize := c.spec().wBufSize
	var sig interrupt_code

	select {
	case req := <-c.pendingReqs:
		blen, err := c.processAsyncRequest(req)
		if err != nil {
			errmsg = fmt.Sprintf("processAsyncRequest error in initial phase")
			goto proc_error
		}
		bytecnt += blen
	case sig := <-ctl:
		return &sig, &ok_status
	}

	for bytecnt < bufsize {
		select {
		case req := <-c.pendingReqs:
			blen, err = c.processAsyncRequest(req)
			if err != nil {
				errmsg = fmt.Sprintf("processAsyncRequest error in batch phase")
				goto proc_error
			}
			if blen > 0 {
				bytecnt += blen
			} else {
				bytecnt = 0
			}

		case sig = <-ctl:
			ic = &sig
			goto done

		default:
			goto done
		}
	}

done:
	c.writer.Flush()
	return ic, &ok_status

proc_error:
	log.Println(errmsg, err)
	return nil, &taskStatus{snderr, err}
}
387489
func LongRunningTask() error {
	log.INFO.Print("Long running task started")
	for i := 0; i < 10; i++ {
		log.INFO.Print(10 - i)
		time.Sleep(1 * time.Second)
	}
	log.INFO.Print("Long running task finished")
	return nil
}