//Sync returns unbuffered channel which will get resolved with the
//value of the action when fullfilled or when the supplied value of
//time has passed it will eject
485041
func (a *Action) Sync(ms int) <-chan interface{} {
	m := make(chan interface{})

	if a.cache != nil {
		go func() {
			m <- a.cache
			close(m)
		}()
	} else {
		if ms <= 0 {
			ms = 1
		}

		md := time.Duration(ms) * time.Millisecond

		a.When(func(b interface{}, _ ActionInterface) {
			go func() {
				m <- b
				close(m)
			}()
		})

		go func() {
			<-time.After(md)
			<-m
		}()
	}

	return m
}
1360040
func (cache *Cache) Sync(st Writer) error {
	if cache.readonly {
		// Sync is (should be) a no-op for read-only - any modifications should have been caught in respective methods
		return nil
	}
	cache.Lock()
	defer cache.Unlock()
	var addresses crypto.Addresses
	for address := range cache.accounts {
		addresses = append(addresses, address)
	}

	sort.Sort(addresses)
	for _, address := range addresses {
		accInfo := cache.accounts[address]
		accInfo.RLock()
		if accInfo.removed {
			err := st.RemoveAccount(address)
			if err != nil {
				return err
			}
		} else if accInfo.updated {
			// First update account in case it needs to be created
			err := st.UpdateAccount(accInfo.account)
			if err != nil {
				return err
			}
			// Sort keys
			var keys binary.Words256
			for key := range accInfo.storage {
				keys = append(keys, key)
			}
			sort.Sort(keys)
			// Update account's storage
			for _, key := range keys {
				value := accInfo.storage[key]
				err := st.SetStorage(address, key, value)
				if err != nil {
					return err
				}
			}

		}
		accInfo.RUnlock()
	}
	return nil
}
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
}
658960
func (env *Env) Sync(force bool) error {
	ret := C.mdb_env_sync(env._env, cbool(force))
	return operrno("mdb_env_sync", ret)
}
615316
func (a *Agent) PauseSync() {
	// Do this outside of lock as it has it's own locking
	a.sync.Pause()

	// Coordinate local state watchers
	a.syncMu.Lock()
	defer a.syncMu.Unlock()
	if a.syncCh == nil {
		a.syncCh = make(chan struct{})
	}
}
858933
func flush(stream FSEventStreamRef, sync bool) {
	if sync {
		C.FSEventStreamFlushSync(stream)
	} else {
		C.FSEventStreamFlushAsync(stream)
	}
}
1166669
func (agent *ActionAgent) lock(ctx context.Context) error {
	agent.actionMutex.Lock()
	agent.actionMutexLocked = true

	// After we take the lock (which could take a long time), we
	// check the client is still here.
	select {
	case <-ctx.Done():
		agent.actionMutexLocked = false
		agent.actionMutex.Unlock()
		return ctx.Err()
	default:
		return nil
	}
}
56110
func (pm *ProtocolManager) synchronise(peer *peer) {
	// Short circuit if no peers are available
	if peer == nil {
		return
	}

	// Make sure the peer's TD is higher than our own.
	if !pm.needToSync(peer.headBlockInfo()) {
		return
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()
	pm.blockchain.(*light.LightChain).SyncCht(ctx)
	pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), downloader.LightSync)
}
1145728
func (logger *Logger) Sync(eventId uint64) {
	logger.mutex.RLock()
	for _, eventHandlerSpec := range logger.eventHandlerMap {
		if atomic.LoadUint64(&eventHandlerSpec.lastSentEventId) < eventId {
			// lastSentEventId wasn't incremented, meaning it was dropped. no point waiting for it
			continue
		}

		// wait for the lastProcessedEventId to become >= eventId
		eventHandlerSpec.lastProcessedEventIdCond.L.Lock()
		for eventHandlerSpec.lastProcessedEventId < eventId {
			eventHandlerSpec.lastProcessedEventIdCond.Wait()
		}
		eventHandlerSpec.lastProcessedEventIdCond.L.Unlock()
	}
	logger.mutex.RUnlock()
}
988229
func (l *WAL) scheduleSync() {
	// If we're not the first to sync, then another goroutine is fsyncing the wal for us.
	if !atomic.CompareAndSwapUint64(&l.syncCount, 0, 1) {
		return
	}

	// Fsync the wal and notify all pending waiters
	go func() {
		var timerCh <-chan time.Time

		// time.NewTicker requires a > 0 delay, since 0 indicates no delay, use a closed
		// channel which will always be ready to read from.
		if l.syncDelay == 0 {
			// Create a RW chan and close it
			timerChrw := make(chan time.Time)
			close(timerChrw)
			// Convert it to a read-only
			timerCh = timerChrw
		} else {
			t := time.NewTicker(l.syncDelay)
			defer t.Stop()
			timerCh = t.C
		}
		for {
			select {
			case <-timerCh:
				l.mu.Lock()
				if len(l.syncWaiters) == 0 {
					atomic.StoreUint64(&l.syncCount, 0)
					l.mu.Unlock()
					return
				}

				l.sync()
				l.mu.Unlock()
			case <-l.closing:
				atomic.StoreUint64(&l.syncCount, 0)
				return
			}
		}
	}()
}
603978
func (s *schedule) Sync(timestamp uint32) {
	atomic.StoreInt64(&s.offset, time.Now().UnixNano()-int64(timestamp)*1000)
}
124058
func (w *Watcher) sync() error {
	w.needSync = 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) {
		if len(entry) == 0 {
			logger.Tracef("got empty changelog document")
		}
		id := entry[0]
		if id.Name != "_id" {
			panic("watcher: _id field isn't first entry")
		}
		if first {
			w.lastId = id.Value
			first = false
		}
		if id.Value == lastId {
			break
		}
		logger.Tracef("got changelog document: %#v", 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) {
				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 {
					logger.Warningf("changelog has revno with type %T: %#v", r[i], r[i])
					continue
				}
				if revno < 0 {
					revno = -1
				}
				// Queue notifications for per-collection watches.
				for _, info := range w.watches[watchKey{c.Name, nil}] {
					if info.filter != nil && !info.filter(d[i]) {
						continue
					}
					evt := event{
						ch:        info.ch,
						key:       key,
						isDeleted: revno == -1,
						revno:     revno,
					}
					w.syncEvents = append(w.syncEvents, evt)
				}
				// Queue notifications for per-document watches.
				infos := w.watches[key]
				for i, info := range infos {
					if revno > info.revno || revno < 0 && info.revno >= 0 {
						infos[i].revno = revno
						evt := event{
							ch:        info.ch,
							key:       key,
							isDeleted: revno == -1,
							revno:     revno,
						}
						w.syncEvents = append(w.syncEvents, evt)
					}
				}
			}
		}
	}
	if err := iter.Close(); err != nil {
		if qerr, ok := err.(*mgo.QueryError); ok {
			// CappedPositionLost is code 136.
			// Just in case that changes for some reason, we'll also check the error message.
			if qerr.Code == 136 || strings.Contains(qerr.Message, "CappedPositionLost") {
				logger.Warningf("watcher iterator failed due to txn log collection overflow")
				err = cappedPositionLostError
			}
		}
		return errors.Annotate(err, "watcher iteration error")
	}
	return nil
}
996390
func (w *SegmentWAL) Sync() error {
	var head *segmentFile
	var err error

	// Flush the writer and retrieve the reference to the head segment under mutex lock.
	func() {
		w.mtx.Lock()
		defer w.mtx.Unlock()
		if err = w.flush(); err != nil {
			return
		}
		head = w.head()
	}()
	if err != nil {
		return errors.Wrap(err, "flush buffer")
	}
	if head != nil {
		// But only fsync the head segment after releasing the mutex as it will block on disk I/O.
		start := time.Now()
		err := fileutil.Fdatasync(head.File)
		w.metrics.fsyncDuration.Observe(time.Since(start).Seconds())
		return err
	}
	return nil
}
387880
func (c *config) Sync() error {
	if err := c.opts.Loader.Sync(); err != nil {
		return err
	}

	snap, err := c.opts.Loader.Snapshot()
	if err != nil {
		return err
	}

	c.Lock()
	defer c.Unlock()

	c.snap = snap
	vals, err := c.opts.Reader.Values(snap.ChangeSet)
	if err != nil {
		return err
	}
	c.vals = vals

	return nil
}
439924
func (c *OpenChannel) FullSync() error {
	c.Lock()
	defer c.Unlock()

	return c.Db.Update(c.fullSync)
}
1350522
func (o *os) sync() {
	if len(o.opts.Sources) == 0 {
		log.Printf("Zero sources available to sync")
		return
	}

	var sets []*ChangeSet

	for _, source := range o.opts.Sources {
		ch, err := source.Read()
		// should we actually skip failing sources?
		// best effort merging right? but what if we
		// already have good config? that would be screwed
		if err != nil {
			o.RLock()
			vals := o.vals
			o.RUnlock()

			// if we have no config, we're going to try
			// load something
			if vals == nil {
				continue
			} else {
				return
			}
		}
		sets = append(sets, ch)
	}

	set, err := o.opts.Reader.Parse(sets...)
	if err != nil {
		log.Printf("Failed to parse ChangeSets %v", err)
		return
	}

	o.Lock()
	o.vals, _ = o.opts.Reader.Values(set)
	o.cset = set
	o.Unlock()

	o.update()
}
246635
func (api *PublicDownloaderAPI) Syncing(ctx context.Context) (*rpc.Subscription, error) {
	notifier, supported := rpc.NotifierFromContext(ctx)
	if !supported {
		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
	}

	rpcSub := notifier.CreateSubscription()

	go func() {
		statuses := make(chan interface{})
		sub := api.SubscribeSyncStatus(statuses)

		for {
			select {
			case status := <-statuses:
				notifier.Notify(rpcSub.ID, status)
			case <-rpcSub.Err():
				sub.Unsubscribe()
				return
			case <-notifier.Closed():
				sub.Unsubscribe()
				return
			}
		}
	}()

	return rpcSub, nil
}