// Sync the saved state of a chunk by its T0.
778557
func (a *AggMetric) SyncChunkSaveState(ts uint32) {
	a.Lock()
	defer a.Unlock()
	if ts > a.lastSaveFinish {
		a.lastSaveFinish = ts
	}
	if ts > a.lastSaveStart {
		a.lastSaveStart = ts
	}
	log.Debugf("AM: metric %s at chunk T0=%d has been saved.", a.key, ts)
}
455229
func (m *KubernetesMonitor) Resync(ctx context.Context) error {
	return m.dockerMonitor.Resync(ctx)
}
901748
func (cache *Cache) Sync(state Writer) error {
	cache.Lock()
	defer cache.Unlock()
	// Determine order for names
	// note names may be of any length less than some limit
	var hashes ProposalHashArray
	for hash := range cache.proposals {
		hashes = append(hashes, hash)
	}
	sort.Stable(hashes)

	// Update or delete names.
	for _, hash := range hashes {
		proposalInfo := cache.proposals[hash]
		proposalInfo.RLock()
		if proposalInfo.removed {
			err := state.RemoveProposal(hash[:])
			if err != nil {
				proposalInfo.RUnlock()
				return err
			}
		} else if proposalInfo.updated {
			err := state.UpdateProposal(hash[:], proposalInfo.ballot)
			if err != nil {
				proposalInfo.RUnlock()
				return err
			}
		}
		proposalInfo.RUnlock()
	}
	return nil
}
565291
func SaveState(fd uintptr) (*State, error) {
	mode, e := winterm.GetConsoleMode(fd)
	if e != nil {
		return nil, e
	}

	return &State{mode: mode}, nil
}
1181097
func (a *AggMetric) persist(pos int) {
	chunk := a.chunks[pos]
	pre := time.Now()

	if a.lastSaveStart >= chunk.Series.T0 {
		// this can happen if
		// a) there are 2 primary MT nodes both saving chunks to Cassandra
		// b) a primary failed and this node was promoted to be primary but metric consuming is lagging.
		// c) chunk was persisted by GC (stale) and then new data triggered another persist call
		// d) dropFirstChunk is enabled and this is the first chunk
		log.Debugf("AM: persist(): duplicate persist call for chunk.")
		return
	}

	// create an array of chunks that need to be sent to the writeQueue.
	pending := make([]*ChunkWriteRequest, 1)
	// add the current chunk to the list of chunks to send to the writeQueue
	cwr := NewChunkWriteRequest(a, a.key, chunk, a.ttl, a.chunkSpan, time.Now())
	pending[0] = &cwr

	// if we recently became the primary, there may be older chunks
	// that the old primary did not save.  We should check for those
	// and save them.
	previousPos := pos - 1
	if previousPos < 0 {
		previousPos += len(a.chunks)
	}
	previousChunk := a.chunks[previousPos]
	for (previousChunk.Series.T0 < chunk.Series.T0) && (a.lastSaveStart < previousChunk.Series.T0) {
		log.Debugf("AM: persist(): old chunk needs saving. Adding %s:%d to writeQueue", a.key, previousChunk.Series.T0)
		cwr := NewChunkWriteRequest(a, a.key, previousChunk, a.ttl, a.chunkSpan, time.Now())
		pending = append(pending, &cwr)
		previousPos--
		if previousPos < 0 {
			previousPos += len(a.chunks)
		}
		previousChunk = a.chunks[previousPos]
	}

	// Every chunk with a T0 <= this chunks' T0 is now either saved, or in the writeQueue.
	a.lastSaveStart = chunk.Series.T0

	log.Debugf("AM: persist(): sending %d chunks to write queue", len(pending))

	pendingChunk := len(pending) - 1

	// if the store blocks,
	// the calling function will block waiting for persist() to complete.
	// This is intended to put backpressure on our message handlers so
	// that they stop consuming messages, leaving them to buffer at
	// the message bus. The "pending" array of chunks are processed
	// last-to-first ensuring that older data is added to the store
	// before newer data.
	for pendingChunk >= 0 {
		log.Debugf("AM: persist(): sealing chunk %d/%d (%s:%d) and adding to write queue.", pendingChunk, len(pending), a.key, chunk.Series.T0)
		a.store.Add(pending[pendingChunk])
		pendingChunk--
	}
	persistDuration.Value(time.Now().Sub(pre))
	return
}
456907
func (c *syncClient) Save() (err Error) {
	_, err = c.conn.ServiceRequest(&SAVE, [][]byte{})
	return

}
128482
func chunk(s []interface{}, sz int) [][]interface{} {
	r := [][]interface{}{}
	j := len(s)
	for i := 0; i < j; i += sz {
		r = append(r, s[i:i+sz])
	}
	return r
}
965156
func (lbc *loadBalancerController) sync(dryRun bool) error {
	if !lbc.epController.HasSynced() || !lbc.svcController.HasSynced() {
		time.Sleep(100 * time.Millisecond)
		return errDeferredSync
	}
	httpSvc, httpsTermSvc, tcpSvc := lbc.getServices()
	if len(httpSvc) == 0 && len(httpsTermSvc) == 0 && len(tcpSvc) == 0 {
		return nil
	}
	if err := lbc.cfg.write(
		map[string][]service{
			"http":      httpSvc,
			"httpsTerm": httpsTermSvc,
			"tcp":       tcpSvc,
		}, dryRun); err != nil {
		return err
	}
	if dryRun {
		return nil
	}
	return lbc.cfg.reload()
}
515447
func (s *ZKVirtualIPSynchronizer) Sync(pool p.ResourcePool, assignments map[string]string) error {
	virtualIPMap := s.getVirtualIPMap(pool)

	errorList := SyncError{}
	for _, ip := range s.virtualIPsWithNoAssignment(virtualIPMap, assignments) {
		err := s.handler.Assign(ip.PoolID, ip.IP, ip.Netmask, ip.BindInterface)
		if err != nil {
			errorList = append(errorList, err)
		}
	}

	for _, ip := range s.assignmentsWithNoVirtualIP(virtualIPMap, assignments) {
		err := s.handler.Unassign(pool.ID, ip)
		if err != nil {
			errorList = append(errorList, err)
		}
	}

	if len(errorList) > 0 {
		return errorList
	}

	return nil
}
976781
func SaveState(fd uintptr) (*State, error) {
	var oldState State
	if err := tcget(fd, &oldState.termios); err != 0 {
		return nil, err
	}

	return &oldState, nil
}
171655
func (c *Client) Sync() (string, error) {
	var ret string

	err := c.command(
		&ret,
		[]byte("SYNC"),
	)

	return ret, err
}
3801
func (d *downloader) downloadChunk(chunk dlchunk) error {
	in := &s3.GetObjectInput{}
	awsutil.Copy(in, d.in)

	// Get the next byte range of data
	in.Range = aws.String(chunk.ByteRange())

	var n int64
	var err error
	for retry := 0; retry <= d.partBodyMaxRetries; retry++ {
		var resp *s3.GetObjectOutput
		resp, err = d.cfg.S3.GetObjectWithContext(d.ctx, in, d.cfg.RequestOptions...)
		if err != nil {
			return err
		}
		d.setTotalBytes(resp) // Set total if not yet set.

		n, err = io.Copy(&chunk, resp.Body)
		resp.Body.Close()
		if err == nil {
			break
		}

		chunk.cur = 0
		logMessage(d.cfg.S3, aws.LogDebugWithRequestRetries,
			fmt.Sprintf("DEBUG: object part body download interrupted %s, err, %v, retrying attempt %d",
				aws.StringValue(in.Key), err, retry))
	}

	d.incrWritten(n)

	return err
}
1136566
func (w *Watcher) Sync() {
	done := make(chan bool)
	w.sendReq(reqSync{done})
	select {
	case <-done:
	case <-w.tomb.Dying():
	}
}
261924
func (f *DeltaFIFO) Resync() error {
	f.lock.Lock()
	defer f.lock.Unlock()

	if f.knownObjects == nil {
		return nil
	}

	keys := f.knownObjects.ListKeys()
	for _, k := range keys {
		if err := f.syncKeyLocked(k); err != nil {
			return err
		}
	}
	return nil
}
943045
func NewSyncPool(minSize, maxSize, factor int) *SyncPool {
	n := 0
	for chunkSize := minSize; chunkSize <= maxSize; chunkSize *= factor {
		n++
	}
	pool := &SyncPool{
		make([]sync.Pool, n),
		make([]int, n),
		minSize, maxSize,
	}
	n = 0
	for chunkSize := minSize; chunkSize <= maxSize; chunkSize *= factor {
		pool.classesSize[n] = chunkSize
		pool.classes[n].New = func(size int) func() interface{} {
			return func() interface{} {
				buf := make([]byte, size)
				return &buf
			}
		}(chunkSize)
		n++
	}
	return pool
}
59363
func (w *TxnWatcher) sync() (bool, error) {
	w.logger.Tracef("txn watcher %p starting sync", w)
	added := false
	// Iterate through log events in reverse insertion order (newest first).
	iter := w.iteratorFunc()
	seen := make(map[watchKey]bool)
	first := true
	lastId := w.lastId
	var entry bson.D
	for iter.Next(&entry) {
		w.iteratorStepCount++
		if len(entry) == 0 {
			w.logger.Tracef("got empty changelog document")
		}
		id := entry[0]
		if id.Name != "_id" {
			w.logger.Warningf("watcher: _id field isn't first entry")
			continue
		}
		if first {
			w.lastId = id.Value
			first = false
		}
		if id.Value == lastId {
			break
		}
		w.logger.Tracef("%p step %d got changelog document: %#v", w, w.iteratorStepCount, entry)
		for _, c := range entry[1:] {
			// See txn's Runner.ChangeLog for the structure of log entries.
			var d, r []interface{}
			dr, _ := c.Value.(bson.D)
			for _, item := range dr {
				switch item.Name {
				case "d":
					d, _ = item.Value.([]interface{})
				case "r":
					r, _ = item.Value.([]interface{})
				}
			}
			if len(d) == 0 || len(d) != len(r) {
				w.logger.Warningf("changelog has invalid collection document: %#v", c)
				continue
			}
			for i := len(d) - 1; i >= 0; i-- {
				key := watchKey{c.Name, d[i]}
				if seen[key] {
					continue
				}
				seen[key] = true
				revno, ok := r[i].(int64)
				if !ok {
					w.logger.Warningf("changelog has revno with type %T: %#v", r[i], r[i])
					continue
				}
				if revno < 0 {
					revno = -1
				}
				w.syncEvents = append(w.syncEvents, Change{
					C:     c.Name,
					Id:    d[i],
					Revno: revno,
				})
				w.changesCount++
				added = true
			}
		}
	}
	if err := iter.Close(); err != nil {
		return false, errors.Annotate(err, "watcher iteration error")
	}
	return added, 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
				}
			}
		}
	}
}