mirror of
https://github.com/octoleo/syncthing.git
synced 2024-09-19 13:19:01 +00:00
This adds metadata updates to the same write batch as the underlying file change. The odds of a metadata update going missing is greatly reduced. Bonus change: actually commit the transaction in recalcMeta.
This commit is contained in:
parent
c3637f2191
commit
6a840a040b
@ -48,10 +48,15 @@ type ReadTransaction interface {
|
|||||||
// purposes of saving memory when transactions are in-RAM. Note that
|
// purposes of saving memory when transactions are in-RAM. Note that
|
||||||
// transactions may be checkpointed *anyway* even if this is not called, due to
|
// transactions may be checkpointed *anyway* even if this is not called, due to
|
||||||
// resource constraints, but this gives you a chance to decide when.
|
// resource constraints, but this gives you a chance to decide when.
|
||||||
|
//
|
||||||
|
// Functions can be passed to Checkpoint. These are run if and only if the
|
||||||
|
// checkpoint will result in a flush, and will run before the flush. The
|
||||||
|
// transaction can be accessed via a closure. If an error is returned from
|
||||||
|
// these functions the flush will be aborted and the error bubbled.
|
||||||
type WriteTransaction interface {
|
type WriteTransaction interface {
|
||||||
ReadTransaction
|
ReadTransaction
|
||||||
Writer
|
Writer
|
||||||
Checkpoint() error
|
Checkpoint(...func() error) error
|
||||||
Commit() error
|
Commit() error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -150,8 +150,8 @@ func (t *leveldbTransaction) Put(key, val []byte) error {
|
|||||||
return t.checkFlush(dbFlushBatchMax)
|
return t.checkFlush(dbFlushBatchMax)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *leveldbTransaction) Checkpoint() error {
|
func (t *leveldbTransaction) Checkpoint(preFlush ...func() error) error {
|
||||||
return t.checkFlush(dbFlushBatchMin)
|
return t.checkFlush(dbFlushBatchMin, preFlush...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (t *leveldbTransaction) Commit() error {
|
func (t *leveldbTransaction) Commit() error {
|
||||||
@ -167,10 +167,15 @@ func (t *leveldbTransaction) Release() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// checkFlush flushes and resets the batch if its size exceeds the given size.
|
// checkFlush flushes and resets the batch if its size exceeds the given size.
|
||||||
func (t *leveldbTransaction) checkFlush(size int) error {
|
func (t *leveldbTransaction) checkFlush(size int, preFlush ...func() error) error {
|
||||||
if len(t.batch.Dump()) < size {
|
if len(t.batch.Dump()) < size {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
for _, hook := range preFlush {
|
||||||
|
if err := hook(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
return t.flush()
|
return t.flush()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -124,11 +124,17 @@ func (db *Lowlevel) updateRemoteFiles(folder, device []byte, fs []protocol.FileI
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := t.Checkpoint(); err != nil {
|
if err := t.Checkpoint(func() error {
|
||||||
|
return meta.toDB(t, folder)
|
||||||
|
}); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := meta.toDB(t, folder); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
return t.Commit()
|
return t.Commit()
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -227,11 +233,17 @@ func (db *Lowlevel) updateLocalFiles(folder []byte, fs []protocol.FileInfo, meta
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := t.Checkpoint(); err != nil {
|
if err := t.Checkpoint(func() error {
|
||||||
|
return meta.toDB(t, folder)
|
||||||
|
}); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := meta.toDB(t, folder); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
return t.Commit()
|
return t.Commit()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -62,8 +62,8 @@ func (m *metadataTracker) Marshal() ([]byte, error) {
|
|||||||
|
|
||||||
// toDB saves the marshalled metadataTracker to the given db, under the key
|
// toDB saves the marshalled metadataTracker to the given db, under the key
|
||||||
// corresponding to the given folder
|
// corresponding to the given folder
|
||||||
func (m *metadataTracker) toDB(db *Lowlevel, folder []byte) error {
|
func (m *metadataTracker) toDB(t readWriteTransaction, folder []byte) error {
|
||||||
key, err := db.keyer.GenerateFolderMetaKey(nil, folder)
|
key, err := t.keyer.GenerateFolderMetaKey(nil, folder)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@ -79,7 +79,7 @@ func (m *metadataTracker) toDB(db *Lowlevel, folder []byte) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
err = db.Put(key, bs)
|
err = t.Put(key, bs)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
m.dirty = false
|
m.dirty = false
|
||||||
}
|
}
|
||||||
|
@ -117,6 +117,8 @@ func (s *FileSet) recalcMeta() error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
defer t.close()
|
||||||
|
|
||||||
var deviceID protocol.DeviceID
|
var deviceID protocol.DeviceID
|
||||||
err = t.withAllFolderTruncated([]byte(s.folder), func(device []byte, f FileInfoTruncated) bool {
|
err = t.withAllFolderTruncated([]byte(s.folder), func(device []byte, f FileInfoTruncated) bool {
|
||||||
copy(deviceID[:], device)
|
copy(deviceID[:], device)
|
||||||
@ -128,7 +130,10 @@ func (s *FileSet) recalcMeta() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
s.meta.SetCreated()
|
s.meta.SetCreated()
|
||||||
return s.meta.toDB(s.db, []byte(s.folder))
|
if err := s.meta.toDB(t, []byte(s.folder)); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return t.Commit()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Verify the local sequence number from actual sequence entries. Returns
|
// Verify the local sequence number from actual sequence entries. Returns
|
||||||
@ -184,7 +189,20 @@ func (s *FileSet) Drop(device protocol.DeviceID) {
|
|||||||
s.meta.resetAll(device)
|
s.meta.resetAll(device)
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := s.meta.toDB(s.db, []byte(s.folder)); backend.IsClosed(err) {
|
t, err := s.db.newReadWriteTransaction()
|
||||||
|
if backend.IsClosed(err) {
|
||||||
|
return
|
||||||
|
} else if err != nil {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
|
defer t.close()
|
||||||
|
|
||||||
|
if err := s.meta.toDB(t, []byte(s.folder)); backend.IsClosed(err) {
|
||||||
|
return
|
||||||
|
} else if err != nil {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
|
if err := t.Commit(); backend.IsClosed(err) {
|
||||||
return
|
return
|
||||||
} else if err != nil {
|
} else if err != nil {
|
||||||
panic(err)
|
panic(err)
|
||||||
@ -202,12 +220,6 @@ func (s *FileSet) Update(device protocol.DeviceID, fs []protocol.FileInfo) {
|
|||||||
s.updateMutex.Lock()
|
s.updateMutex.Lock()
|
||||||
defer s.updateMutex.Unlock()
|
defer s.updateMutex.Unlock()
|
||||||
|
|
||||||
defer func() {
|
|
||||||
if err := s.meta.toDB(s.db, []byte(s.folder)); err != nil && !backend.IsClosed(err) {
|
|
||||||
panic(err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
|
|
||||||
if device == protocol.LocalDeviceID {
|
if device == protocol.LocalDeviceID {
|
||||||
// For the local device we have a bunch of metadata to track.
|
// For the local device we have a bunch of metadata to track.
|
||||||
if err := s.db.updateLocalFiles([]byte(s.folder), fs, s.meta); err != nil && !backend.IsClosed(err) {
|
if err := s.db.updateLocalFiles([]byte(s.folder), fs, s.meta); err != nil && !backend.IsClosed(err) {
|
||||||
|
Loading…
Reference in New Issue
Block a user