// Use a single redis.AsyncClient with specified number
// of workers to bench concurrent load on the async client
773210
func benchTask(taskspec taskSpec, iterations int, workers int, printReport bool) (delta time.Duration, err error) {
	// channel to signal completion
	signal := make(chan int, workers)

	// spec and initialize an AsyncClient
	// will flush your db13 as noted in README ..
	spec := redis.DefaultSpec().Db(13).Password("go-redis")
	client, e := redis.NewAsynchClientWithSpec(spec)
	if e != nil {
		log.Println("Error creating client for worker: ", e)
		return -1, e
	}

	defer client.Quit()

	// panics
	setup(client)

	t0 := time.Now()
	for i := 0; i < workers; i++ {
		id := fmt.Sprintf("%d", i)
		go taskspec.task(id, signal, client, iterations)
	}

	// wait for completion
	for i := 0; i < workers; i++ {
		<-signal
	}
	delta = time.Now().Sub(t0)

	if printReport {
		report(taskspec.name, workers, delta, iterations*workers)
	}

	return
}
99679
func UseNumber() {
	json = jsoniter.Config{
		UseNumber:              true,
		EscapeHTML:             true,
		SortMapKeys:            true,
		ValidateJsonRawMessage: true,
	}.Froze()
}
632536
func NewClient(namespace string, pool *redis.Pool) *Client {
	return &Client{
		namespace: namespace,
		pool:      pool,
	}
}
267968
func main5() {
	api, _ := bandwidth.New(os.Getenv("CATAPULT_USER_ID"), os.Getenv("CATAPULT_API_TOKEN"), os.Getenv("CATAPULT_API_SECRET"))
	fromNumber := os.Args[1]
	toNumber := os.Args[2]
	text := os.Args[3]
	id, err := api.CreateMessage(&bandwidth.CreateMessageData{From: fromNumber, To: toNumber, Text: text})
	if err != nil {
		fmt.Printf("Error: %s", err.Error())
		return
	}
	fmt.Printf("Message ID is %s", id)
}
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)
	}
}
959715
func (b *Broker) nextTask(queue string) (result []byte, err error) {
	conn := b.open()
	defer conn.Close()

	items, err := redis.ByteSlices(conn.Do("BLPOP", queue, 1000))
	if err != nil {
		return []byte{}, err
	}

	// items[0] - the name of the key where an element was popped
	// items[1] - the value of the popped element
	if len(items) != 2 {
		return []byte{}, redis.ErrNil
	}

	result = items[1]

	return result, nil
}
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
}
1304198
func (w *Worker) Serve(addr string, h http.Handler) {
	// Create server
	s := &http.Server{Addr: addr, Handler: h}

	// Create task
	t := w.NewTask()

	// Execute the rest in a goroutine
	astilog.Infof("astiworker: serving on %s", addr)
	go func() {
		// Serve
		var chanDone = make(chan error)
		go func() {
			if err := s.ListenAndServe(); err != nil {
				chanDone <- err
			}
		}()

		// Wait for context or chanDone to be done
		select {
		case <-w.ctx.Done():
			if w.ctx.Err() != context.Canceled {
				astilog.Error(errors.Wrap(w.ctx.Err(), "astiworker: context error"))
			}
		case err := <-chanDone:
			if err != nil {
				astilog.Error(errors.Wrap(err, "astiworker: serving failed"))
			}
		}

		// Shutdown
		astilog.Infof("astiworker: shutting down server on %s", addr)
		if err := s.Shutdown(context.Background()); err != nil {
			astilog.Error(errors.Wrapf(err, "astiworker: shutting down server on %s failed", addr))
		}

		// Task is done
		t.Done()
	}()
}
176477
func (r *Redis) Gather(acc telegraf.Accumulator) error {
	if !r.initialized {
		err := r.init(acc)
		if err != nil {
			return err
		}
	}

	var wg sync.WaitGroup

	for _, client := range r.clients {
		wg.Add(1)
		go func(client Client) {
			defer wg.Done()
			acc.AddError(r.gatherServer(client, acc))
		}(client)
	}

	wg.Wait()
	return nil
}
279928
func (wsh *WebsocketHub) NumClients() int {
	// Swallow any type assertion error since the default int of 0 is OK.
	n, _ := wsh.numClients.Load().(int)
	return n
}
385333
func (c *RedisClient) GetMulti(count int, timeout time.Duration, queues ...string) ([]Job, error) {

	if len(queues) == 0 {
		return nil, errors.New("disque: no queues specified")
	}
	if count < 0 {
		return nil, fmt.Errorf("disque: invalid count %d", count)
	}

	args := redis.Args{}
	if timeout > 0 {
		args = args.Add("TIMEOUT", int64(timeout/time.Millisecond))
	}
	if count > 0 {
		args = args.Add("COUNT", count)
	}
	args = args.Add("FROM")
	args = args.AddFlat(queues)

	vals, err := redis.Values(c.conn.Do("GETJOB", args...))

	if err != nil {
		return nil, fmt.Errorf("disque: could not get jobs: %s", err)
	}

	ret := make([]Job, 0, len(vals))

	for _, v := range vals {
		if arr, ok := v.([]interface{}); ok {

			ret = append(ret, Job{
				Queue: string(arr[0].([]byte)),
				id:    string(arr[1].([]byte)),
				Data:  arr[2].([]byte),
			})

		}

	}

	return ret, nil

}
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.
}
849365
func (dc *DispatcherClient) CallAsync(funcName string, request interface{}) (*AsyncResult, error) {
	req := dc.getRequest(funcName, request)

	innerAr, err := dc.c.CallAsync(req)
	if err != nil {
		return nil, err
	}

	ch := make(chan struct{})
	ar := &AsyncResult{
		Done: ch,
	}

	go func() {
		<-innerAr.Done
		ar.Response, ar.Error = getResponse(innerAr.Response, innerAr.Error)
		close(ch)
	}()

	return ar, nil
}
1368669
func (r *Runner) queueHandler(ctx context.Context) {
consumeQueue:
	for {
		select {
		case task := <-r.taskQueue:
			r.handleTask(task)
		case <-ctx.Done():
			break consumeQueue
		}
	}

	// consume remainders
	for len(r.taskQueue) > 0 {
		r.handleTask(<-r.taskQueue)
	}
}
1061187
func (client *AMIClient) AsyncAction(action string, params Params) (<-chan *AMIResponse, error) {
	var output string
	client.mutexAsyncAction.Lock()
	defer client.mutexAsyncAction.Unlock()

	output = fmt.Sprintf("Action: %s\r\n", strings.TrimSpace(action))
	if params == nil {
		params = Params{}
	}
	if _, ok := params["ActionID"]; !ok {
		params["ActionID"] = fmt.Sprintf("%d", time.Now().UnixNano())
	}

	if _, ok := client.response[params["ActionID"]]; !ok {
		client.response[params["ActionID"]] = make(chan *AMIResponse, 1)
	}
	for k, v := range params {
		output = output + fmt.Sprintf("%s: %s\r\n", k, strings.TrimSpace(v))
	}
	if err := client.conn.PrintfLine("%s", output); err != nil {
		return nil, err
	}

	return client.response[params["ActionID"]], nil
}
131417
func (c *Client) allocSync() {
	staggered := false
	syncTicker := time.NewTicker(allocSyncIntv)
	updates := make(map[string]*structs.Allocation)
	for {
		select {
		case <-c.shutdownCh:
			syncTicker.Stop()
			return
		case alloc := <-c.allocUpdates:
			// Batch the allocation updates until the timer triggers.
			updates[alloc.ID] = alloc
		case <-syncTicker.C:
			// Fast path if there are no updates
			if len(updates) == 0 {
				continue
			}

			sync := make([]*structs.Allocation, 0, len(updates))
			for _, alloc := range updates {
				sync = append(sync, alloc)
			}

			// Send to server.
			args := structs.AllocUpdateRequest{
				Alloc:        sync,
				WriteRequest: structs.WriteRequest{Region: c.Region()},
			}

			var resp structs.GenericResponse
			if err := c.RPC("Node.UpdateAlloc", &args, &resp); err != nil {
				c.logger.Error("error updating allocations", "error", err)
				syncTicker.Stop()
				syncTicker = time.NewTicker(c.retryIntv(allocSyncRetryIntv))
				staggered = true
			} else {
				updates = make(map[string]*structs.Allocation)
				if staggered {
					syncTicker.Stop()
					syncTicker = time.NewTicker(allocSyncIntv)
					staggered = false
				}
			}
		}
	}
}
1265439
func worker_task(i int) {
	worker, _ := zmq.NewSocket(zmq.REQ)
	defer worker.Close()
	worker.Connect("ipc://" + self + "-localbe.ipc")

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

	//  Process messages as they arrive
	for {
		msg, err := worker.RecvMessage(0)
		if err != nil {
			break //  Interrupted
		}

		//  Workers are busy for 0/1 seconds
		time.Sleep(time.Duration(rand.Intn(2)) * time.Second)
		n := len(msg) - 1
		worker.SendMessage(msg[:n], fmt.Sprintf("%s %s-%d", msg[n], self, i))
	}
}