// WithWaitGroup adds all non long-running requests to wait group, which is used for graceful shutdown.
778687
func WithWaitGroup(handler http.Handler, longRunning apirequest.LongRunningRequestCheck, wg *utilwaitgroup.SafeWaitGroup) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		ctx := req.Context()
		requestInfo, ok := apirequest.RequestInfoFrom(ctx)
		if !ok {
			// if this happens, the handler chain isn't setup correctly because there is no request info
			responsewriters.InternalError(w, req, errors.New("no RequestInfo found in the context"))
			return
		}

		if !longRunning(req, requestInfo) {
			if err := wg.Add(1); err != nil {
				http.Error(w, "apiserver is shutting down.", http.StatusInternalServerError)
				return
			}
			defer wg.Done()
		}

		handler.ServeHTTP(w, req)
	})
}
890915
func (c *Client) QueryGroup(id int32) (*GroupNode, error) {
	if err := c.oscConn.Send(osc.Message{
		Address: groupQueryTreeAddress,
		Arguments: osc.Arguments{
			osc.Int(id),
			osc.Int(1),
		},
	}); err != nil {
		return nil, err
	}
	// wait for response
	var resp osc.Message
	select {
	case resp = <-c.gqueryTreeChan:
	case <-time.After(2 * time.Second):
		return nil, errors.New("timeout waiting for response")
	}
	if numArgs := len(resp.Arguments); numArgs < 3 {
		return nil, fmt.Errorf("expected 3 arguments for message, got %d", numArgs)
	}
	// Throw away the flag that tells us we want to include synth controls in the reply.
	// We already know we requested that!
	resp.Arguments = resp.Arguments[1:]
	return c.parseGroup(resp)
}
108168
func (rwg *RepeatedWaitGroup) WaitUnlessPaused(ctx context.Context) (
	bool, error) {
	paused, isIdleCh, pauseCh := func() (bool, chan struct{}, chan struct{}) {
		rwg.lock.Lock()
		defer rwg.lock.Unlock()
		if !rwg.paused && rwg.pauseCh == nil {
			rwg.pauseCh = make(chan struct{})
		}
		return rwg.paused, rwg.isIdleCh, rwg.pauseCh
	}()

	if isIdleCh == nil {
		return false, nil
	}

	if paused {
		return true, nil
	}

	select {
	case <-isIdleCh:
		return false, nil
	case <-pauseCh:
		return true, nil
	case <-ctx.Done():
		return false, ctx.Err()
	}
}
1210606
func (rwg *RepeatedWaitGroup) Resume() {
	rwg.lock.Lock()
	defer rwg.lock.Unlock()
	if rwg.pauseCh != nil {
		panic("Non-nil pauseCh on resume!")
	}
	rwg.paused = false
}
87473
func (c *AutoScaling) WaitUntilGroupInService(input *DescribeAutoScalingGroupsInput) error {
	return c.WaitUntilGroupInServiceWithContext(aws.BackgroundContext(), input)
}
1032231
func (c *conn) LeaveGroup(group uint32) error {
	return os.NewSyscallError("setsockopt", c.s.SetSockoptInt(
		unix.SOL_NETLINK,
		unix.NETLINK_DROP_MEMBERSHIP,
		int(group),
	))
}
17705
func (dg *deliverGroup) ClientWait(dc *deliverClient) {
	defer dg.wg.Done()
	for {
		resp, err := dc.Connection.Recv()
		if err != nil {
			err = errors.WithMessage(err, fmt.Sprintf("error receiving from deliver filtered at %s", dc.Address))
			dg.setError(err)
			return
		}
		switch r := resp.Type.(type) {
		case *pb.DeliverResponse_FilteredBlock:
			filteredTransactions := r.FilteredBlock.FilteredTransactions
			for _, tx := range filteredTransactions {
				if tx.Txid == dg.TxID {
					logger.Infof("txid [%s] committed with status (%s) at %s", dg.TxID, tx.TxValidationCode, dc.Address)
					return
				}
			}
		case *pb.DeliverResponse_Status:
			err = errors.Errorf("deliver completed with status (%s) before txid received", r.Status)
			dg.setError(err)
			return
		default:
			err = errors.Errorf("received unexpected response type (%T) from %s", r, dc.Address)
			dg.setError(err)
			return
		}
	}
}
748201
func (b *Backend) InitGroup(groupUUID string, taskUUIDs []string) error {
	groupMeta := &tasks.GroupMeta{
		GroupUUID: groupUUID,
		TaskUUIDs: taskUUIDs,
		CreatedAt: time.Now().UTC(),
	}

	encoded, err := json.Marshal(groupMeta)
	if err != nil {
		return err
	}

	conn := b.open()
	defer conn.Close()

	_, err = conn.Do("SET", groupUUID, encoded)
	if err != nil {
		return err
	}

	return b.setExpirationTime(groupUUID)
}
23503
func (client *Client) LeaveGroup(groupID string) *LeaveGroupCall {
	return &LeaveGroupCall{
		c:       client,
		groupID: groupID,
	}
}
75661
func (wg *WaitGroup) Go(cb func()) {
	wg.wg.Add(1)
	go func() {
		cb()
		wg.wg.Done()
	}()
}
464204
func HaltGroup(tag int) {
	_tag := (C.int)(tag)
	C.Mix_HaltGroup(_tag)
}
206748
func (s *sessionQueueSet) ApplyAndWait(getApply func(*sessionQueue) func()) {
	var wg sync.WaitGroup
	for _, sessionq := range *s {
		wg.Add(1)
		go func(sq *sessionQueue) {
			defer wg.Done()
			getApply(sq)()
		}(sessionq)
	}
	wg.Wait()
}
781281
func (wg *WaitGroup) WaitFor(timeout time.Duration) bool {
	if timeout == time.Duration(0) {
		wg.Wait()
		return true // ### return, always true ###
	}

	start := time.Now()
	spin := NewSpinner(SpinPriorityHigh)
	for wg.Active() {
		if time.Since(start) > timeout {
			return false // ### return, timed out ###
		}
		spin.Yield()
	}
	return true
}
1032568
func (b *Broker) SyncGroup(request *SyncGroupRequest) (*SyncGroupResponse, error) {
	response := new(SyncGroupResponse)

	err := b.sendAndReceive(request, response)
	if err != nil {
		return nil, err
	}

	return response, nil
}
785651
func (s *Scheduler) ExecuteTaskGroup(ctx context.Context,
	fc *flow.Flow,
	taskGroupStatus *pb.FlowExecutionStatus_TaskGroup,
	wg *sync.WaitGroup,
	taskGroup *plan.TaskGroup,
	bid float64, relatedFiles []resource.FileResource) {

	defer wg.Done()

	tasks := taskGroup.Tasks
	lastTask := tasks[len(tasks)-1]
	if tasks[0].Step.IsOnDriverSide {
		// these should be only one task on the driver side
		if err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.localExecute(ctx, fc, exeStatus, lastTask, wg)
		}); err != nil {
			log.Fatalf("Failed to execute on driver side: %v", err)
		}
		return
	}
	if !needsInputFromDriver(tasks[0]) {
		// wait until inputs are registed
		s.shardLocator.waitForInputDatasetShardLocations(tasks[0])
	}
	if isInputOnDisk(tasks[0]) && !isRestartableTasks(tasks) {
		// for non-restartable taskGroup, wait until on disk inputs are completed
		for _, stepGroup := range taskGroup.ParentStepGroup.Parents {
			stepGroup.WaitForAllTasksToComplete()
		}
	}

	// fmt.Printf("inputs of %s is %s\n", tasks[0].Name(), s.allInputLocations(tasks[0]))

	pickedServerChan := make(chan market.Supply, 1)
	s.Market.AddDemand(market.Requirement(taskGroup), bid, pickedServerChan)

	// get assigned executor location
	supply := <-pickedServerChan
	allocation := supply.Object.(*pb.Allocation)
	defer s.Market.ReturnSupply(supply)

	if needsInputFromDriver(tasks[0]) {
		// tell the driver to write to me
		for _, shard := range tasks[0].InputShards {
			// println("registering", shard.Name(), "at", allocation.Location.URL())
			s.setShardLocation(shard, pb.DataLocation{
				Name:     shard.Name(),
				Location: allocation.Location,
				OnDisk:   shard.Dataset.GetIsOnDiskIO(),
			})
		}
	}

	for _, shard := range lastTask.OutputShards {
		// println("registering", shard.Name(), "at", allocation.Location.URL(), "onDisk", shard.Dataset.GetIsOnDiskIO())
		s.setShardLocation(shard, pb.DataLocation{
			Name:     shard.Name(),
			Location: allocation.Location,
			OnDisk:   shard.Dataset.GetIsOnDiskIO(),
		})
	}

	// send driver code only when using go mapper reducer
	var hasGoCode bool
	for _, t := range tasks {
		hasGoCode = hasGoCode || t.Step.IsGoCode
	}
	if hasGoCode {
		relatedFiles = append(relatedFiles, resource.FileResource{os.Args[0], "."})
	}

	if len(relatedFiles) > 0 {
		err := withClient(allocation.Location.URL(), func(client pb.GleamAgentClient) error {
			for _, relatedFile := range relatedFiles {
				err := sendRelatedFile(ctx, client, fc.HashCode, relatedFile)
				if err != nil {
					taskGroup.MarkStop(err)
					return err
				}
			}
			return nil
		})
		if err != nil {
			log.Fatalf("Failed to send related files: %v", err)
		}
	}

	fn := func() error {
		err := taskGroupStatus.Track(func(exeStatus *pb.FlowExecutionStatus_TaskGroup_Execution) error {
			return s.remoteExecuteOnLocation(ctx, fc, taskGroupStatus, exeStatus, taskGroup, allocation, wg)
		})
		if err != nil {
			log.Printf("Failed to remoteExecuteOnLocation %v: %v", allocation, err)
		}
		taskGroup.MarkStop(err)
		return err
	}

	util.ExecuteWithCleanup(
		ctx,
		func() error {
			if isRestartableTasks(tasks) {
				return util.TimeDelayedRetry(fn, time.Minute, 3*time.Minute)
			} else {
				return fn()
			}
		},
		func() {
			var w sync.WaitGroup
			for _, shard := range lastTask.OutputShards {
				w.Add(1)
				// println("deleting", shard.Name(), "from", allocation.Location.URL())
				go func(shard *flow.DatasetShard) {
					defer w.Done()
					if err := sendDeleteRequest(allocation.Location.URL(), &pb.DeleteDatasetShardRequest{
						Name: shard.Name(),
					}); err != nil {
						println("Purging dataset error:", err.Error())
					}
				}(shard)
			}
			w.Wait()
		},
	)

}
1101008
func (e *Engine) VirtualizationWait(ctx context.Context, ID, state string) (*enginetypes.VirtualizationWaitResult, error) {
	waitBody, errorCh := e.client.ContainerWait(ctx, ID, dockercontainer.WaitConditionNotRunning)
	r := &enginetypes.VirtualizationWaitResult{}
	select {
	case b := <-waitBody:
		if b.Error != nil {
			r.Message = b.Error.Message
		}
		r.Code = b.StatusCode
		return r, nil
	case err := <-errorCh:
		r.Message = err.Error()
		r.Code = -1
		return r, err
	}
}
837554
func getReady(name string, wg *sync.WaitGroup) {
	defer wg.Done()
	fmt.Printf("%s started getting ready\n", name)
	// the time it takes to get ready is at least 60 seconds, plus a random number up to 90 seconds
	sec := rand.Intn(30) + 60
	// this is where they get ready
	time.Sleep(time.Duration(sec) * time.Second)
	fmt.Printf("%s spent %d seconds getting ready\n", name, sec)
}