//  Worker using REQ socket to do load-balancing
//
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)
	}
}
1098867
func Worker(a *App) {
	for subreddit := range subredditsToCrawl {
		if audience, subscribers, articles, err := readDOMData(subreddit); err == nil {
			// store the audience+subscribers and update the last crawl time
			// ----------------------
			if err := a.DB().InsertAudienceValue(subreddit, audience, subscribers); err != nil {
				log.Println("err:", err.Error())
			} else {
				log.Printf("info: subreddit %s has %d active users (%d subscribers)\n", subreddit, audience, subscribers)
			}

			// store each articles if not already present at this rank
			// ----------------------
			if err := storeArticles(a, articles); err != nil {
				log.Printf("err: %s", err.Error())
			}
		} else if err != nil {
			log.Println("err:", err.Error())
		}

		r := time.Duration(((rand.Int() % 2) + MIN_SECONDS_BETWEEN_EACH_CRAWL)) * time.Second

		time.Sleep(r) // wait a bit before the next crawl
	}
}
915135
func (a *adapterBase) worker(workerNum int, ctx interface{}) {
	a.Trace("xfer: adapter %q worker %d starting", a.Name(), workerNum)
	waitForAuth := workerNum > 0
	signalAuthOnResponse := workerNum == 0

	// First worker is the only one allowed to start immediately
	// The rest wait until successful response from 1st worker to
	// make sure only 1 login prompt is presented if necessary
	// Deliberately outside jobChan processing so we know worker 0 will process 1st item
	if waitForAuth {
		a.Trace("xfer: adapter %q worker %d waiting for Auth", a.Name(), workerNum)
		a.authWait.Wait()
		a.Trace("xfer: adapter %q worker %d auth signal received", a.Name(), workerNum)
	}

	for job := range a.jobChan {
		t := job.T

		var authCallback func()
		if signalAuthOnResponse {
			authCallback = func() {
				a.authWait.Done()
				signalAuthOnResponse = false
			}
		}
		a.Trace("xfer: adapter %q worker %d processing job for %q", a.Name(), workerNum, t.Oid)

		// Actual transfer happens here
		var err error
		if t.Size < 0 {
			err = fmt.Errorf("Git LFS: object %q has invalid size (got: %d)", t.Oid, t.Size)
		} else {
			err = a.transferImpl.DoTransfer(ctx, t, a.cb, authCallback)
		}

		// Mark the job as completed, and alter all listeners
		job.Done(err)

		a.Trace("xfer: adapter %q worker %d finished job for %q", a.Name(), workerNum, t.Oid)
	}
	// This will only happen if no jobs were submitted; just wake up all workers to finish
	if signalAuthOnResponse {
		a.authWait.Done()
	}
	a.Trace("xfer: adapter %q worker %d stopping", a.Name(), workerNum)
	a.transferImpl.WorkerEnding(workerNum, ctx)
	a.workerWait.Done()
}
696234
func (svr *Server) Serve() error {
	var wg sync.WaitGroup
	runWorker := func(ch chan orderedRequest) {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if err := svr.sftpServerWorker(ch); err != nil {
				svr.conn.Close() // shuts down recvPacket
			}
		}()
	}
	pktChan := svr.pktMgr.workerChan(runWorker)

	var err error
	var pkt requestPacket
	var pktType uint8
	var pktBytes []byte
	for {
		pktType, pktBytes, err = svr.recvPacket()
		if err != nil {
			break
		}

		pkt, err = makePacket(rxPacket{fxp(pktType), pktBytes})
		if err != nil {
			switch errors.Cause(err) {
			case errUnknownExtendedPacket:
				if err := svr.serverConn.sendError(pkt, ErrSshFxOpUnsupported); err != nil {
					debug("failed to send err packet: %v", err)
					svr.conn.Close() // shuts down recvPacket
					break
				}
			default:
				debug("makePacket err: %v", err)
				svr.conn.Close() // shuts down recvPacket
				break
			}
		}

		pktChan <- svr.pktMgr.newOrderedRequest(pkt)
	}

	close(pktChan) // shuts down sftpServerWorkers
	wg.Wait()      // wait for all workers to exit

	// close any still-open files
	for handle, file := range svr.openFiles {
		fmt.Fprintf(svr.debugStream, "sftp server file with handle %q left open: %v\n", handle, file.Name())
		file.Close()
	}
	return err // error from recvPacket
}
109979
func (h configReloadHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	if err := config.ReloadGlobalConfig(); err != nil {
		writeError(w, err)
	} else {
		writeData(w, "success!")
	}
}
1250817
func (w *workManager) sendToWorker(j queuedJob) {
	switch j.Job().Type() {
	case collectJobType:
		w.collectchan <- j
	case publishJobType:
		w.publishchan <- j
	case processJobType:
		w.processchan <- j
	}
}
717835
func (rq *ResourceQuotaController) worker(queue workqueue.RateLimitingInterface) func() {
	workFunc := func() bool {
		key, quit := queue.Get()
		if quit {
			return true
		}
		defer queue.Done(key)
		rq.workerLock.RLock()
		defer rq.workerLock.RUnlock()
		err := rq.syncHandler(key.(string))
		if err == nil {
			queue.Forget(key)
			return false
		}
		utilruntime.HandleError(err)
		queue.AddRateLimited(key)
		return false
	}

	return func() {
		for {
			if quit := workFunc(); quit {
				klog.Infof("resource quota controller worker shutting down")
				return
			}
		}
	}
}
136490
func (worker *Worker) AddServer(net, addr string) (err error) {
	// Create a new job server's client as a agent of server
	a, err := newAgent(net, addr, worker)
	if err != nil {
		return err
	}
	worker.agents = append(worker.agents, a)
	return
}
1352328
func (cli *socketClient) FlushSync() error {
	reqRes := cli.queueRequest(types.ToRequestFlush())
	if err := cli.Error(); err != nil {
		return err
	}
	reqRes.Wait() // NOTE: if we don't flush the queue, its possible to get stuck here
	return cli.Error()
}
10086
func subHandler(sock mangos.Socket) {
	count := 0
	for {
		msg := fmt.Sprintf("PUB #%d %s", count, time.Now().String())
		if e := sock.Send([]byte(msg)); e != nil {
			die("Cannot send pub: %v", e)
		}
		time.Sleep(5 * time.Second)
		count++
	}
}
1093946
func (s *WebsocketServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	logger.Println("WebsocketServer.ServeHTTP", r.Method, r.RequestURI)
	// TODO: subprotocol?
	conn, err := s.Upgrader.Upgrade(w, r, nil)
	if err != nil {
		logger.Println("Error upgrading to websocket connection:", err)
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	s.handleWebsocket(conn)
}
946373
func Ack() error {
	client, err := NewClientForPath(os.Getenv("EINHORN_SOCK_PATH"))
	if err != nil {
		return err
	}

	defer client.Close()

	return client.SendRequest(&ClientAckRequest{
		Command: "worker:ack",
		Pid:     os.Getpid(),
	})
}
180067
func (wp *WorkerPool) Drain() {
	wg := sync.WaitGroup{}
	for _, w := range wp.workers {
		wg.Add(1)
		go func(w *worker) {
			w.drain()
			wg.Done()
		}(w)
	}
	wg.Wait()
}
800997
func main() {
	rand.Seed(time.Now().UnixNano())

	go client_task()
	go client_task()
	go client_task()
	go server_task()

	//  Run for 5 seconds then quit
	time.Sleep(5 * time.Second)
}
836784
func newBulkWorker(p *BulkProcessor, i int) *bulkWorker {
	return &bulkWorker{
		p:           p,
		i:           i,
		bulkActions: p.bulkActions,
		bulkSize:    p.bulkSize,
		service:     NewBulkService(p.c),
		flushC:      make(chan struct{}),
		flushAckC:   make(chan struct{}),
	}
}
792583
func NewSocketWorker(wo *Websocket) *SocketWorker {
	data := make(chan interface{})

	sw := SocketWorker{
		wo:     wo,
		closer: make(chan bool),
		data:   data,
		queue:  flux.NewQueue(data),
	}

	go sw.manage()
	return &sw
}
1012091
func (r *readPump) Work() {
	cnx := redis.PubSubConn{Conn: r.cnx}
	defer close(r.closer)

	for {
		msg := cnx.Receive()

		if err, isErr := msg.(error); isErr && shouldNotifyUser(err) {
			select {
			case r.errs <- err:
			case <-r.closer:
				return
			}
		} else if !isErr {
			select {
			case r.data <- msg:
			case <-r.closer:
				return
			}
		}
	}
}