// Persist writes the snapshot to the given sink.
7805
func (m *nodeFSM) Persist(sink raft.SnapshotSink) error {
	if err := (*Node)(m).handler.Snapshot(sink); err != nil {
		sink.Cancel()
		return err
	}
	sink.Close()
	return nil
}
1052613
func (s *Snapshotter) Commit(ctx context.Context, name, key string, opts ...snapshots.Opt) error {
	log.G(ctx).WithFields(logrus.Fields{"name": name, "key": key}).Debug("commit")

	return s.withTransaction(ctx, true, func(ctx context.Context) error {
		id, _, _, err := storage.GetInfo(ctx, key)
		if err != nil {
			return err
		}

		deviceName := s.getDeviceName(id)
		size, err := s.pool.GetUsage(deviceName)
		if err != nil {
			return err
		}

		usage := snapshots.Usage{
			Size: size,
		}

		_, err = storage.CommitActive(ctx, key, name, usage, opts...)
		return err
	})
}
705587
func (c *Cache) Snapshot() (*Cache, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if c.snapshotting {
		return nil, ErrSnapshotInProgress
	}

	c.snapshotting = true
	c.tracker.IncSnapshotsActive() // increment the number of times we tried to do this

	// If no snapshot exists, create a new one, otherwise update the existing snapshot
	if c.snapshot == nil {
		c.snapshot = &Cache{
			store:   newRing(),
			tracker: newCacheTracker(c.tracker.metrics, c.tracker.labels),
		}
	}

	// Did a prior snapshot exist that failed?  If so, return the existing
	// snapshot to retry.
	if c.snapshot.Size() > 0 {
		return c.snapshot, nil
	}

	c.snapshot.store, c.store = c.store, c.snapshot.store
	snapshotSize := c.Size()

	c.snapshot.tracker.SetSnapshotSize(snapshotSize) // Save the size of the snapshot on the snapshot cache
	c.tracker.SetSnapshotSize(snapshotSize)          // Save the size of the snapshot on the live cache

	// Reset the cache's store.
	c.store.reset()
	c.tracker.SetCacheSize(0)
	c.lastSnapshot = time.Now()

	c.tracker.AddSnapshottedBytes(snapshotSize) // increment the number of bytes added to the snapshot
	c.tracker.SetDiskBytes(0)
	c.tracker.SetSnapshotsActive(0)

	return c.snapshot, nil
}
290044
func (s *MemoryStore) Save(tx ReadTx) (*pb.StoreSnapshot, error) {
	var snapshot pb.StoreSnapshot
	for _, os := range objectStorers {
		if err := os.Save(tx, &snapshot); err != nil {
			return nil, err
		}
	}

	return &snapshot, nil
}
1373347
func (_m *Store) Persist(txid string, blockHeight uint64, privateSimulationResults *rwset.TxPvtReadWriteSet) error {
	ret := _m.Called(txid, blockHeight, privateSimulationResults)

	var r0 error
	if rf, ok := ret.Get(0).(func(string, uint64, *rwset.TxPvtReadWriteSet) error); ok {
		r0 = rf(txid, blockHeight, privateSimulationResults)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}
649147
func (ds *DelayedSink) Store(loc BlockLocation, data []byte) error {
	time.Sleep(ds.Latency)
	return ds.Sink.Store(loc, data)
}
1198369
func snapshot(m ConcurrentMap) (chans []chan Tuple) {
	chans = make([]chan Tuple, SHARD_COUNT)
	wg := sync.WaitGroup{}
	wg.Add(SHARD_COUNT)
	// Foreach shard.
	for index, shard := range m {
		go func(index int, shard *ConcurrentMapShared) {
			// Foreach key, value pair.
			shard.RLock()
			chans[index] = make(chan Tuple, len(shard.items))
			wg.Done()
			for key, val := range shard.items {
				chans[index] <- Tuple{key, val}
			}
			shard.RUnlock()
			close(chans[index])
		}(index, shard)
	}
	wg.Wait()
	return chans
}
771722
func (c *asyncClient) Save() (stat FutureBool, err Error) {
	resp, err := c.conn.QueueRequest(&SAVE, [][]byte{})
	if err == nil {
		stat = resp.future.(FutureBool)
	}

	return

}
808034
func (sp *StandardProvider) Persistent() bool {
	return sp.Paths.KeyFile != "" && sp.Paths.CertFile != ""
}
298280
func (c *Core) persistAudit(ctx context.Context, table *MountTable, localOnly bool) error {
	if table.Type != auditTableType {
		c.logger.Error("given table to persist has wrong type", "actual_type", table.Type, "expected_type", auditTableType)
		return fmt.Errorf("invalid table type given, not persisting")
	}

	for _, entry := range table.Entries {
		if entry.Table != table.Type {
			c.logger.Error("given entry to persist in audit table has wrong table value", "path", entry.Path, "entry_table_type", entry.Table, "actual_type", table.Type)
			return fmt.Errorf("invalid audit entry found, not persisting")
		}
	}

	nonLocalAudit := &MountTable{
		Type: auditTableType,
	}

	localAudit := &MountTable{
		Type: auditTableType,
	}

	for _, entry := range table.Entries {
		if entry.Local {
			localAudit.Entries = append(localAudit.Entries, entry)
		} else {
			nonLocalAudit.Entries = append(nonLocalAudit.Entries, entry)
		}
	}

	if !localOnly {
		// Marshal the table
		compressedBytes, err := jsonutil.EncodeJSONAndCompress(nonLocalAudit, nil)
		if err != nil {
			c.logger.Error("failed to encode and/or compress audit table", "error", err)
			return err
		}

		// Create an entry
		entry := &logical.StorageEntry{
			Key:   coreAuditConfigPath,
			Value: compressedBytes,
		}

		// Write to the physical backend
		if err := c.barrier.Put(ctx, entry); err != nil {
			c.logger.Error("failed to persist audit table", "error", err)
			return err
		}
	}

	// Repeat with local audit
	compressedBytes, err := jsonutil.EncodeJSONAndCompress(localAudit, nil)
	if err != nil {
		c.logger.Error("failed to encode and/or compress local audit table", "error", err)
		return err
	}

	entry := &logical.StorageEntry{
		Key:   coreLocalAuditConfigPath,
		Value: compressedBytes,
	}

	if err := c.barrier.Put(ctx, entry); err != nil {
		c.logger.Error("failed to persist local audit table", "error", err)
		return err
	}

	return nil
}
968824
func (s *flotillaSnapshot) Persist(sink raft.SnapshotSink) error {
	defer sink.Close()
	defer s.pipeR.Close()
	_, e1 := io.Copy(sink, s.pipeR)
	e2 := <-s.copyErr

	if e2 != nil {
		sink.Cancel()
		return fmt.Errorf("Error copying snapshot to pipe: %s", e2)
	}
	if e1 != nil {
		sink.Cancel()
	} else {
		sink.Close()
	}
	return e1
}
56758
func (w *Wrapper) UpsertNode(s services.Server) (*services.KeepAlive, error) {
	return w.Write.UpsertNode(s)
}
843243
func (s *SlaveHealthChecker) Continue(slaveUPID *upid.UPID) {
	s.Lock()
	defer s.Unlock()
	s.paused = false
	s.slaveUPID = slaveUPID
}
404543
func (s *Wrapper) KeepAlive(ctx context.Context, lease Lease, expires time.Time) error {
	return s.backend.KeepAlive(ctx, lease, expires)
}
248162
func (kvm *Clone) Snapshot(wr io.Writer) error {
	kvm.mu.RLock()
	defer kvm.mu.RUnlock()
	data, err := json.Marshal(kvm.keys)
	if err != nil {
		return err
	}
	if _, err := wr.Write(data); err != nil {
		return err
	}
	return nil
}
1194267
func (e *Engine) writeSnapshotAndCommit(log *zap.Logger, closedFiles []string, snapshot *Cache) (err error) {
	defer func() {
		if err != nil {
			e.Cache.ClearSnapshot(false)
		}
	}()

	// write the new snapshot files
	newFiles, err := e.Compactor.WriteSnapshot(snapshot)
	if err != nil {
		log.Info("Error writing snapshot from compactor", zap.Error(err))
		return err
	}

	e.mu.RLock()
	defer e.mu.RUnlock()

	// update the file store with these new files
	if err := e.FileStore.Replace(nil, newFiles); err != nil {
		log.Info("Error adding new TSM files from snapshot", zap.Error(err))
		return err
	}

	// clear the snapshot from the in-memory cache, then the old WAL files
	e.Cache.ClearSnapshot(true)

	if err := e.WAL.Remove(closedFiles); err != nil {
		log.Info("Error removing closed WAL segments", zap.Error(err))
	}

	return nil
}
533838
func (dfs *DistributedFilesystem) snapshotSavePipe(vol volume.Volume, label string, excludes []string) (*io.PipeReader, <-chan error) {
	return savePipe(func(w io.Writer) error {
		return vol.Export(label, "", w, excludes)
	})
}