// newWorkerState initializes a new readWorkerState, which will be called
// whenever a new goroutine is allocated to begin processing read tasks.
347699
func (r *Read) newWorkerState() WorkerState {
	return &readWorkerState{
		bufferPool: r.bufferPool,
		readBuf:    r.bufferPool.Take(),
	}
}
407037
func NewWorker(opt Opt) (*Worker, error) {
	sm, err := source.NewManager()
	if err != nil {
		return nil, err
	}

	cm := opt.CacheManager
	sm.Register(opt.ImageSource)

	gs, err := git.NewSource(git.Opt{
		CacheAccessor: cm,
		MetadataStore: opt.MetadataStore,
	})
	if err == nil {
		sm.Register(gs)
	} else {
		logrus.Warnf("Could not register builder git source: %s", err)
	}

	hs, err := http.NewSource(http.Opt{
		CacheAccessor: cm,
		MetadataStore: opt.MetadataStore,
		Transport:     opt.Transport,
	})
	if err == nil {
		sm.Register(hs)
	} else {
		logrus.Warnf("Could not register builder http source: %s", err)
	}

	ss, err := local.NewSource(local.Opt{
		CacheAccessor: cm,
		MetadataStore: opt.MetadataStore,
	})
	if err == nil {
		sm.Register(ss)
	} else {
		logrus.Warnf("Could not register builder local source: %s", err)
	}

	return &Worker{
		Opt:           opt,
		SourceManager: sm,
	}, nil
}
928531
func NewState(db dbm.DB) *State {
	forest, err := storage.NewMutableForest(storage.NewPrefixDB(db, forestPrefix), defaultCacheCapacity)
	if err != nil {
		// This should only happen if we have negative cache capacity, which for us is a positive compile-time constant
		panic(fmt.Errorf("could not create new state because error creating MutableForest"))
	}
	ring := validator.NewRing(nil, DefaultValidatorsWindowSize)
	rs := ReadState{Forest: forest, History: ring}
	ws := writeState{forest: forest, ring: ring}
	return &State{
		db:         db,
		ReadState:  rs,
		writeState: ws,
		logger:     logging.NewNoopLogger(),
	}
}
145534
func NewWorker(srv *Server) (*Worker, error) {
	w := &Worker{
		srv:    srv,
		logger: srv.logger.ResetNamed("worker"),
		start:  time.Now(),
	}
	w.pauseCond = sync.NewCond(&w.pauseLock)
	go w.run()
	return w, nil
}
71490
func NewWorker(config Config) (RestoreStatusWorker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	restoreStatus, err := config.RestoreInfoWatcher.RestoreStatus()
	if err != nil {
		return nil, errors.Trace(err)
	}
	w := &restoreWorker{
		config: config,
		status: restoreStatus,
	}
	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	}); err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
200979
func NewWorker(
	algorithm pruneAlgorithm,
	registryClientFactory RegistryClientFactoryFunc,
	registryURL *url.URL,
	imagePrunerFactory ImagePrunerFactoryFunc,
	streamPruner ImageStreamDeleter,
	layerLinkPruner LayerLinkDeleter,
	blobPruner BlobDeleter,
	manifestPruner ManifestDeleter,
) (Worker, error) {
	client, err := registryClientFactory()
	if err != nil {
		return nil, err
	}

	imagePruner, err := imagePrunerFactory()
	if err != nil {
		return nil, err
	}

	return &worker{
		algorithm:       algorithm,
		registryClient:  client,
		registryURL:     registryURL,
		imagePruner:     imagePruner,
		streamPruner:    streamPruner,
		layerLinkPruner: layerLinkPruner,
		blobPruner:      blobPruner,
		manifestPruner:  manifestPruner,
	}, nil
}
197203
func NewWorker(config Config) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	apiPorts := config.APIInfo.Ports()
	if n := len(apiPorts); n != 1 {
		return nil, errors.Errorf("api.Info has %d unique ports, expected 1", n)
	}

	w := &Worker{
		config:       config,
		connections:  make(chan net.Conn),
		dialRequests: make(chan dialRequest),
		apiPort:      apiPorts[0],
	}

	const logPrefix = "[transport] "
	logWriter := &raftutil.LoggoWriter{logger, loggo.DEBUG}
	logLogger := log.New(logWriter, logPrefix, 0)
	stream, err := newStreamLayer(config.LocalID, config.Hub, w.connections, config.Clock, &Dialer{
		APIInfo: config.APIInfo,
		DialRaw: w.dialRaw,
		Path:    config.Path,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	transport := raft.NewNetworkTransportWithConfig(&raft.NetworkTransportConfig{
		Logger:  logLogger,
		MaxPool: maxPoolSize,
		Stream:  stream,
		Timeout: config.Timeout,
	})
	w.Transport = transport

	var h http.Handler = NewHandler(w.connections, w.catacomb.Dying())
	h = &httpcontext.BasicAuthHandler{
		Handler:       h,
		Authenticator: w.config.Authenticator,
		Authorizer:    httpcontext.AuthorizerFunc(controllerAuthorizer),
	}
	h = &httpcontext.ImpliedModelHandler{
		Handler:   h,
		ModelUUID: w.config.APIInfo.ModelTag.Id(),
	}

	w.config.Mux.AddHandler("GET", w.config.Path, h)

	if err := catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: func() error {
			defer transport.Close()
			defer w.config.Mux.RemoveHandler("GET", w.config.Path)
			return w.loop()
		},
		Init: []worker.Worker{stream},
	}); err != nil {
		transport.Close()
		w.config.Mux.RemoveHandler("GET", w.config.Path)
		return nil, errors.Trace(err)
	}
	return w, nil
}
651625
func New(config Config) (*Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	// Read it before the worker starts, so that we have a value
	// guaranteed before we return the worker. Because we read this
	// before we start the internal watcher, we'll need an additional
	// read triggered by the first change event; this will *probably*
	// be the same value, but we can't assume it.
	life, err := config.Facade.Life(config.Entity)
	if err != nil {
		return nil, filter(errors.Trace(err))
	}

	w := &Worker{
		config: config,
		life:   life,
	}
	err = catacomb.Invoke(catacomb.Plan{
		Site: &w.catacomb,
		Work: w.loop,
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return w, nil
}
7155
func NewStartWorkerArgs(scopeURL string) *StartWorkerArgs {
	args := new(StartWorkerArgs)
	args.ScopeURL = scopeURL
	return args
}
717521
func (s *Filesystem) WriteState(state *states.State) error {
	// TODO: this should use a more robust method of writing state, by first
	// writing to a temp file on the same filesystem, and renaming the file over
	// the original.

	defer s.mutex()()

	if s.readFile == nil {
		err := s.refreshState()
		if err != nil {
			return err
		}
	}

	return s.writeState(state, nil)
}
419708
func NewWorker() worker.Worker {
	var w terminationWorker
	c := make(chan os.Signal, 1)
	signal.Notify(c, TerminationSignal)
	w.tomb.Go(func() error {
		defer signal.Stop(c)
		return w.loop(c)
	})
	return &w
}
1069967
func newWorker(a agent.Agent, apiCaller base.APICaller) (worker.Worker, error) {
	currentConfig := a.CurrentConfig()

	// TODO(fwereade): this functionality should be on the
	// machiner facade instead -- or, better yet, separate
	// the networking concerns from the lifecycle ones and
	// have completely separate workers.
	//
	// (With their own facades.)
	agentFacade, err := apiagent.NewState(apiCaller)
	if err != nil {
		return nil, errors.Trace(err)
	}
	modelConfig, err := agentFacade.ModelConfig()
	if err != nil {
		return nil, errors.Errorf("cannot read environment config: %v", err)
	}

	ignoreMachineAddresses, _ := modelConfig.IgnoreMachineAddresses()
	// Containers only have machine addresses, so we can't ignore them.
	tag := currentConfig.Tag()
	if names.IsContainerMachine(tag.Id()) {
		ignoreMachineAddresses = false
	}
	if ignoreMachineAddresses {
		logger.Infof("machine addresses not used, only addresses from provider")
	}
	accessor := APIMachineAccessor{apimachiner.NewState(apiCaller)}
	w, err := NewMachiner(Config{
		MachineAccessor:              accessor,
		Tag:                          tag.(names.MachineTag),
		ClearMachineAddressesOnStart: ignoreMachineAddresses,
		NotifyMachineDead: func() error {
			return agent.SetCanUninstall(a)
		},
	})
	if err != nil {
		return nil, errors.Annotate(err, "cannot start machiner worker")
	}
	return w, err
}
1117834
func NewWorker(config WorkerConfig) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	started := make(chan struct{})
	w := &agentConfigUpdater{
		config:       config,
		mongoProfile: config.MongoProfile,
	}
	w.tomb.Go(func() error {
		return w.loop(started)
	})
	select {
	case <-started:
	case <-time.After(10 * time.Second):
		return nil, errors.New("worker failed to start properly")
	}
	return w, nil
}
825864
func newWorker(a agent.Agent, apiCaller base.APICaller, machineLock machinelock.Lock, clock clock.Clock) (worker.Worker, error) {
	apiConn, ok := apiCaller.(api.Connection)
	if !ok {
		return nil, errors.New("unable to obtain api.Connection")
	}
	rebootState, err := apiConn.Reboot()
	if err != nil {
		return nil, errors.Trace(err)
	}
	w, err := NewReboot(rebootState, a.CurrentConfig(), machineLock, clock)
	if err != nil {
		return nil, errors.Annotate(err, "cannot start reboot worker")
	}
	return w, nil
}
1255068
func NewRetryStrategyWorker(config WorkerConfig) (worker.Worker, error) {
	if err := config.Validate(); err != nil {
		return nil, errors.Trace(err)
	}
	w, err := watcher.NewNotifyWorker(watcher.NotifyConfig{
		Handler: retryStrategyHandler{config},
	})
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &RetryStrategyWorker{w, config.RetryStrategy}, nil
}
1011293
func newWorker(
	m *manager,
	probeType probeType,
	pod *v1.Pod,
	container v1.Container) *worker {

	w := &worker{
		stopCh:       make(chan struct{}, 1), // Buffer so stop() can be non-blocking.
		pod:          pod,
		container:    container,
		probeType:    probeType,
		probeManager: m,
	}

	switch probeType {
	case readiness:
		w.spec = container.ReadinessProbe
		w.resultsManager = m.readinessManager
		w.initialValue = results.Failure
	case liveness:
		w.spec = container.LivenessProbe
		w.resultsManager = m.livenessManager
		w.initialValue = results.Success
	}

	basicMetricLabels := prometheus.Labels{
		"probe_type": w.probeType.String(),
		"container":  w.container.Name,
		"pod":        w.pod.Name,
		"namespace":  w.pod.Namespace,
		"pod_uid":    string(w.pod.UID),
	}

	w.proberResultsSuccessfulMetricLabels = deepCopyPrometheusLabels(basicMetricLabels)
	w.proberResultsSuccessfulMetricLabels["result"] = probeResultSuccessful

	w.proberResultsFailedMetricLabels = deepCopyPrometheusLabels(basicMetricLabels)
	w.proberResultsFailedMetricLabels["result"] = probeResultFailed

	w.proberResultsUnknownMetricLabels = deepCopyPrometheusLabels(basicMetricLabels)
	w.proberResultsUnknownMetricLabels["result"] = probeResultUnknown

	return w
}
477219
func NewBlockWorker(wr *wrangler.Wrangler) (Worker, error) {
	return &BlockWorker{
		StatusWorker: NewStatusWorker(),
		wr:           wr,
	}, nil
}