//  Each worker task works on one request at a time and sends a random number
//  of replies back, with random delays between replies:
342644
func server_worker() {

	worker, _ := zmq.NewSocket(zmq.DEALER)
	defer worker.Close()
	worker.Connect("inproc://backend")

	for {
		//  The DEALER socket gives us the reply envelope and message
		msg, _ := worker.RecvMessage(0)
		identity, content := pop(msg)

		//  Send 0..4 replies back
		replies := rand.Intn(5)
		for reply := 0; reply < replies; reply++ {
			//  Sleep for some fraction of a second
			time.Sleep(time.Duration(rand.Intn(1000)+1) * time.Millisecond)
			worker.SendMessage(identity, content)
		}
	}
}
776622
func discardTask(id string, responsec responder) task {
	return func(c *cache) error {
		bucket, ok := c.buckets[id]
		if !ok {
			// Not found, so nothing to discard.
			responsec <- func() (Cacheable, error) {
				return nil, nil
			}
			return nil
		}
		// Discard Cacheable, notify possible waiters,
		// delete bucket, and notify caller.
		var err error
		if bucket.cacheable != nil {
			err = bucket.cacheable.Discard()
		}
		for _, waiter := range bucket.waiters {
			waiter <- func() (Cacheable, error) {
				return nil, errors.New(ErrDiscardedWhileLoading, errorMessages, id)
			}
		}
		delete(c.buckets, id)
		if err != nil {
			err = errors.Annotate(err, ErrDiscard, errorMessages, id)
		}
		responsec <- func() (Cacheable, error) {
			return nil, err
		}
		return nil
	}
}
1226218
func (ListServiceOfferings) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListServiceOfferingsResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListServiceOfferingsResponse was expected, got %T", resp))
		return
	}

	for i := range items.ServiceOffering {
		if !callback(&items.ServiceOffering[i], nil) {
			break
		}
	}
}
525752
func (client *Client) asyncRequest(ctx context.Context, asyncCommand AsyncCommand) (interface{}, error) {
	var err error

	resp := asyncCommand.AsyncResponse()
	client.AsyncRequestWithContext(
		ctx,
		asyncCommand,
		func(j *AsyncJobResult, e error) bool {
			if e != nil {
				err = e
				return false
			}
			if j.JobStatus != Pending {
				if r := j.Result(resp); r != nil {
					err = r
				}
				return false
			}
			return true
		},
	)
	return resp, err
}
602809
func (w *Worker) Work() {
	for {
		select {
		case <-w.ch:
			log.Println("worker done")
			return
		case <-time.After(w.Interval):
			for {
				if didWork := w.WorkOne(); !didWork {
					break // didn't do any work, go back to sleep
				}
			}
		}
	}
}
618355
func NewTaskQueue(syncFn func(string) error) *taskQueue {
	return &taskQueue{
		queue:      workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
		sync:       syncFn,
		workerDone: make(chan struct{}),
	}
}
874924
func clearTask(responsec responder) task {
	return func(c *cache) error {
		var errs []error
		for id, bucket := range c.buckets {
			switch bucket.status {
			case statusLoading:
				for _, waiter := range bucket.waiters {
					waiter <- func() (Cacheable, error) {
						return nil, errors.New(ErrDiscardedWhileLoading, errorMessages, id)
					}
				}
			case statusLoaded:
				if err := bucket.cacheable.Discard(); err != nil {
					errs = append(errs, err)
				}
			}
		}
		c.buckets = make(map[string]*bucket)
		var err error
		if len(errs) > 0 {
			err = errors.Collect(errs...)
		}
		responsec <- func() (Cacheable, error) {
			return nil, err
		}
		return nil
	}
}
722240
func (c *asyncConnHdl) worker(id int, name string, task workerTask, ctl workerCtl, fb chan workerStatus) {
	// REVU - pretty please TODO do the customized log
	//	log.Printf("<INFO> %s - %s STARTED.", c, name)
	var signal interrupt_code
	var tstat *taskStatus

	// todo: add startup hook for worker

await_signal:
	//	log.Println(name, "_worker: await_signal.")
	signal = <-ctl

on_interrupt:
	//		log.Println(name, "_worker: on_interrupt: ", signal);
	switch signal {
	case stop:
		goto before_stop
	case pause:
		goto await_signal
	case start:
		goto work
	}

work:
	//		fmt.Println(name, "_worker: work!");
	select {
	case signal = <-ctl:
		goto on_interrupt
	default:
		is, stat := task(c, ctl) // todo is a task context type
		if stat == nil {
			log.Println("<BUG> nil stat from worker ", name)
		}
		if stat.code != ok {
			//			fmt.Println(name, "_worker: task error!")
			tstat = stat
			goto on_error
		} else if is != nil {
			signal = *is
			goto on_interrupt
		}
		goto work
	}

on_error:
	//log.Println(name, "_worker: on_error!")
	// TODO: log it, send it, and go back to wait_start:
	log.Println(name, "_worker task raised error: ", tstat)
	fb <- workerStatus{id, faulted, tstat, &ctl}
	goto await_signal

before_stop:
	//	fmt.Println(name, "_worker: before_stop!")
	// TODO: add shutdown hook for worker

	// REVU - pretty please TODO do the customized log
	//	log.Printf("<INFO> %s - %s STOPPED.", c, name)
}
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))
	}
}
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)
	}
}
70981
func (s *Scheduler) Every(interval uint64) *Job {
	job := NewJob(interval)
	s.jobs[s.size] = job
	s.size++
	return job
}
692579
func (worker *Worker) retryTaskIn(signature *tasks.Signature, retryIn time.Duration) error {
	// Update task state to RETRY
	if err := worker.server.GetBackend().SetStateRetry(signature); err != nil {
		return fmt.Errorf("Set state to 'retry' for task %s returned error: %s", signature.UUID, err)
	}

	// Delay task by retryIn duration
	eta := time.Now().UTC().Add(retryIn)
	signature.ETA = &eta

	log.WARNING.Printf("Task %s failed. Going to retry in %.0f seconds.", signature.UUID, retryIn.Seconds())

	// Send the task back to the queue
	_, err := worker.server.SendTask(signature)
	return err
}
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.
}
389201
func (p *Pool) sendNextJobs(n int) error {
	jobs, err := p.getNextJobs(p.config.BatchSize)
	if err != nil {
		return err
	}
	// Send the jobs across the channel, where they will be picked up
	// by exactly one worker
	for _, job := range jobs {
		p.jobs <- job
	}
	return nil
}
991547
func (w *Worker) TaskProgress(taskId string, progress int, msg string) (err error) {
	payload := map[string]interface{}{
		"msg":     msg,
		"percent": progress,
	}

	err = w.tasks(taskId, "progress").Req("POST", payload, nil)
	return
}
662659
func (f *asyncFlusher) waitAll() (int, error) {
	close(f.chunks)
	f.wg.Wait()
	if len(f.errs) == 0 {
		return f.rowsSent, nil
	}
	return f.rowsSent, f.errs
}
104468
func (ListUsers) Each(resp interface{}, callback IterateItemFunc) {
	items, ok := resp.(*ListUsersResponse)
	if !ok {
		callback(nil, fmt.Errorf("wrong type, ListUsersResponse was expected, got %T", resp))
		return
	}

	for i := range items.User {
		if !callback(&items.User[i], nil) {
			break
		}
	}
}