// resyncMonitors starts or stops resource monitors as needed to ensure that all
// (and only) those resources present in the map are monitored.
965501
func (gc *GarbageCollector) resyncMonitors(deletableResources map[schema.GroupVersionResource]struct{}) error {
	if err := gc.dependencyGraphBuilder.syncMonitors(deletableResources); err != nil {
		return err
	}
	gc.dependencyGraphBuilder.startMonitors()
	return nil
}
285250
func (_class ClusterClass) PoolResync(sessionID SessionRef, self ClusterRef) (_err error) {
	_method := "Cluster.pool_resync"
	_sessionIDArg, _err := convertSessionRefToXen(fmt.Sprintf("%s(%s)", _method, "session_id"), sessionID)
	if _err != nil {
		return
	}
	_selfArg, _err := convertClusterRefToXen(fmt.Sprintf("%s(%s)", _method, "self"), self)
	if _err != nil {
		return
	}
	_, _err =  _class.client.APICall(_method, _sessionIDArg, _selfArg)
	return
}
1235028
func (a *APIDefinition) IterateResources(it ResourceIterator) error {
	res := make([]*ResourceDefinition, len(a.Resources))
	i := 0
	for _, r := range a.Resources {
		res[i] = r
		i++
	}
	// Iterate parent resources first so that action parameters are
	// finalized prior to child actions needing them.
	isParent := func(p, c *ResourceDefinition) bool {
		par := c.Parent()
		for par != nil {
			if par == p {
				return true
			}
			par = par.Parent()
		}
		return false
	}
	sort.Slice(res, func(i, j int) bool {
		if isParent(res[i], res[j]) {
			return true
		}
		if isParent(res[j], res[i]) {
			return false
		}
		return res[i].Name < res[j].Name
	})
	for _, r := range res {
		if err := it(r); err != nil {
			return err
		}
	}
	return nil
}
1047307
func (s *StateSyncer) retrySyncFullEventFn() event {
	select {
	// trigger a full sync immediately.
	// this is usually called when a consul server was added to the cluster.
	// stagger the delay to avoid a thundering herd.
	case <-s.SyncFull.Notif():
		select {
		case <-time.After(s.stagger(s.serverUpInterval)):
			return syncFullNotifEvent
		case <-s.ShutdownCh:
			return shutdownEvent
		}

	// retry full sync after some time
	// todo(fs): why don't we use s.Interval here?
	case <-time.After(s.retryFailInterval + s.stagger(s.retryFailInterval)):
		return syncFullTimerEvent

	case <-s.ShutdownCh:
		return shutdownEvent
	}
}
166026
func (o *RsyncOptions) RunRsync() error {
	if err := o.Strategy.Copy(o.Source, o.Destination, o.Out, o.ErrOut); err != nil {
		return err
	}

	if !o.Watch {
		return nil
	}
	return o.WatchAndSync()
}
745966
func (p *processorListener) shouldResync(now time.Time) bool {
	p.resyncLock.Lock()
	defer p.resyncLock.Unlock()

	if p.resyncPeriod == 0 {
		return false
	}

	return now.After(p.nextResync) || now.Equal(p.nextResync)
}
1222099
func (chs *ChartChangeSync) SyncMirrors() {
	chs.logger.Log("info", "starting mirror sync")
	for _, err := range chs.mirrors.RefreshAll(chs.config.GitTimeout) {
		chs.logger.Log("error", fmt.Sprintf("failure while syncing mirror: %s", err))
	}
	chs.logger.Log("info", "finished syncing mirrors")
}
769209
func (m *SyncManager) GossipSyncers() map[route.Vertex]*GossipSyncer {
	m.syncersMu.Lock()
	defer m.syncersMu.Unlock()
	return m.gossipSyncers()
}
1150414
func (gc *Cfg) UpdateResources(res string) error {
	log.Infof("Received update resource for res %s", res)
	tempRm, err := resources.GetStateResourceManager()
	if err != nil {
		return err
	}

	ra := core.ResourceManager(tempRm)

	err = gc.checkErrors(res)
	if err != nil {
		return core.Errorf("process failed on error checks %s", err)
	}

	if res == "vlan" {
		var vlanRsrcCfg *bitset.BitSet
		vlanRsrcCfg, err = gc.initVLANBitset(gc.Auto.VLANs)
		if err != nil {
			return err
		}
		err = ra.RedefineResource("global", resources.AutoVLANResource, vlanRsrcCfg)
		if err != nil {
			log.Errorf("Error deleting vlan resource. Err: %v", err)
		}
	} else if res == "vxlan" {
		var vxlanRsrcCfg *resources.AutoVXLANCfgResource
		vxlanRsrcCfg, err = gc.initVXLANBitset(gc.Auto.VXLANs)
		if err != nil {
			return err
		}
		err = ra.RedefineResource("global", resources.AutoVXLANResource, vxlanRsrcCfg)
		if err != nil {
			log.Errorf("Error deleting vxlan resource. Err: %v", err)
		}
		g := &Oper{FreeVXLANsStart: vxlanRsrcCfg.FreeVXLANsStart}

		g.StateDriver = gc.StateDriver
		err = g.Write()
		if err != nil {
			log.Errorf("error '%s' updating global oper state %v \n", err, g)
			return err
		}
	}
	return err
}
962285
func ResyncPeriod(c *config.CompletedConfig) func() time.Duration {
	return func() time.Duration {
		factor := rand.Float64() + 1
		return time.Duration(float64(c.ComponentConfig.Generic.MinResyncPeriod.Nanoseconds()) * factor)
	}
}
23387
func (sc *syncContext) startedSyncPhase() bool {
	for _, res := range sc.syncRes.Resources {
		if !res.IsHook() {
			return true
		}
		if res.HookType == appv1.HookTypeSync || res.HookType == appv1.HookTypePostSync {
			return true
		}
	}
	return false
}
542954
func (s *ClusterMetadata) SetResources(v *JobResource) *ClusterMetadata {
	s.Resources = v
	return s
}
234302
func (d *Daemon) initMaps() error {
	if option.Config.DryMode {
		return nil
	}

	if _, err := lxcmap.LXCMap.OpenOrCreate(); err != nil {
		return err
	}

	// The ipcache is shared between endpoints. Parallel mode needs to be
	// used to allow existing endpoints that have not been regenerated yet
	// to continue using the existing ipcache until the endpoint is
	// regenerated for the first time. Existing endpoints are using a
	// policy map which is potentially out of sync as local identities are
	// re-allocated on startup. Parallel mode allows to continue using the
	// old version until regeneration. Note that the old version is not
	// updated with new identities. This is fine as any new identity
	// appearing would require a regeneration of the endpoint anyway in
	// order for the endpoint to gain the privilege of communication.
	if _, err := ipcachemap.IPCache.OpenParallel(); err != nil {
		return err
	}

	if _, err := metricsmap.Metrics.OpenOrCreate(); err != nil {
		return err
	}

	if _, err := tunnel.TunnelMap.OpenOrCreate(); err != nil {
		return err
	}

	if err := openServiceMaps(); err != nil {
		log.WithError(err).Fatal("Unable to open service maps")
	}

	// Set up the list of IPCache listeners in the daemon, to be
	// used by syncLXCMap().
	ipcache.IPIdentityCache.SetListeners([]ipcache.IPIdentityMappingListener{
		&envoy.NetworkPolicyHostsCache,
		bpfIPCache.NewListener(d),
	})

	// Insert local host entries to bpf maps
	if err := d.syncLXCMap(); err != nil {
		return err
	}

	// Start the controller for periodic sync
	// The purpose of the controller is to ensure that the host entries are
	// reinserted to the bpf maps if they are ever removed from them.
	// TODO: Determine if we can get rid of this when we have more rigorous
	//       desired/realized state implementation for the bpf maps.
	controller.NewManager().UpdateController("lxcmap-bpf-host-sync",
		controller.ControllerParams{
			DoFunc: func(ctx context.Context) error {
				return d.syncLXCMap()
			},
			RunInterval: 5 * time.Second,
		})

	// Start the controller for periodic sync of the metrics map with
	// the prometheus server.
	controller.NewManager().UpdateController("metricsmap-bpf-prom-sync",
		controller.ControllerParams{
			DoFunc:      metricsmap.SyncMetricsMap,
			RunInterval: 5 * time.Second,
		})

	// Clean all lb entries
	if !option.Config.RestoreState {
		log.Debug("cleaning up all BPF LB maps")

		d.loadBalancer.BPFMapMU.Lock()
		defer d.loadBalancer.BPFMapMU.Unlock()

		if option.Config.EnableIPv6 {
			if err := lbmap.Service6Map.DeleteAll(); err != nil {
				return err
			}
			if err := lbmap.RRSeq6Map.DeleteAll(); err != nil {
				return err
			}
		}
		if err := d.RevNATDeleteAll(); err != nil {
			return err
		}

		if option.Config.EnableIPv4 {
			if err := lbmap.Service4Map.DeleteAll(); err != nil {
				return err
			}
			if err := lbmap.RRSeq4Map.DeleteAll(); err != nil {
				return err
			}
		}

		// If we are not restoring state, all endpoints can be
		// deleted. Entries will be re-populated.
		lxcmap.LXCMap.DeleteAll()
	}

	return nil
}
206290
func (wfc *WorkflowController) ResyncConfig() error {
	cmClient := wfc.kubeclientset.CoreV1().ConfigMaps(wfc.namespace)
	cm, err := cmClient.Get(wfc.configMap, metav1.GetOptions{})
	if err != nil {
		return errors.InternalWrapError(err)
	}
	return wfc.updateConfig(cm)
}
526852
func (helper *CacheHelper) DoResync(resyncEv datasync.ResyncEvent) error {
	var wasError error
	//idx.Put()
	ifaces, found := resyncEv.GetValues()[helper.Prefix]
	if found {
		// Step 1: fill the existing things
		resyncNames := map[string]interface{}{}
		for {
			item, stop := ifaces.GetNext()
			if stop {
				break
			}
			ifaceName, err := helper.ParseName(item.GetKey())
			if err != nil {
				wasError = err
			} else {
				current := proto.Clone(helper.DataPrototype)
				item.GetValue(current)
				helper.IDX.Put(ifaceName, current)
				resyncNames[ifaceName] = nil
			}
		}

		// Step 2:
		existingNames := []string{} //TODO
		for _, existingName := range existingNames {
			if _, found := resyncNames[existingName]; !found {
				helper.IDX.Delete(existingName)
			}
		}
	}
	return wasError
}
1272464
func (s *DataMsgServiceServer) DataResyncs(ctx context.Context, req *msg.DataResyncRequests) (
	*msg.DataResyncReplies, error) {
	resyncs := req.GetDataResyncs()
	if resyncs != nil {
		//TODO propagate event like in Kafka transport

		/*localtxn := syncbase.NewLocalBytesTxn(s.adapter.localtransp.PropagateBytesResync)
		if len(resyncs) == 0 {
			log.Debug("received empty resync => DELETE ALL")
		}
		for _, chReq := range resyncs {
			//TODO chReq.ContentType
			localtxn.Put(chReq.Key, chReq.Content)
		}
		err := localtxn.Commit()*/
		var err error
		if err != nil {
			return &msg.DataResyncReplies{MsgId: replySeq(), Error: &msg.Error{Message: err.Error()} /*TODO all other fields*/}, err
		}

		return &msg.DataResyncReplies{MsgId: replySeq() /*TODO all other fields*/}, nil
	}

	err := errors.New("unexpected place - nil resyncs")
	return &msg.DataResyncReplies{MsgId: replySeq(), Error: &msg.Error{Message: err.Error()} /*TODO all other fields*/}, err
}
502247
func (m *SyncManager) gossipSyncers() map[route.Vertex]*GossipSyncer {
	numSyncers := len(m.inactiveSyncers) + len(m.activeSyncers)
	syncers := make(map[route.Vertex]*GossipSyncer, numSyncers)

	for _, syncer := range m.inactiveSyncers {
		syncers[syncer.cfg.peerPub] = syncer
	}
	for _, syncer := range m.activeSyncers {
		syncers[syncer.cfg.peerPub] = syncer
	}

	return syncers
}