vendor/github.com/syndtr/goleveldb/leveldb/storage/file_storage.go

   1 // Copyright (c) 2012, Suryandaru Triandana <syndtr@gmail.com>
   2 // All rights reservefs.
   3 //
   4 // Use of this source code is governed by a BSD-style license that can be
   5 // found in the LICENSE file.
   6
   7 package storage
   8
   9 import (
  10         "errors"
  11         "fmt"
  12         "io/ioutil"
  13         "os"
  14         "path/filepath"
  15         "runtime"
  16         "strconv"
  17         "strings"
  18         "sync"
  19         "time"
  20 )
  21
  22 var (
  23         errFileOpen = errors.New("leveldb/storage: file still open")
  24         errReadOnly = errors.New("leveldb/storage: storage is read-only")
  25 )
  26
  27 type fileLock interface {
  28         release() error
  29 }
  30
  31 type fileStorageLock struct {
  32         fs *fileStorage
  33 }
  34
  35 func (lock *fileStorageLock) Unlock() {
  36         if lock.fs != nil {
  37                 lock.fs.mu.Lock()
  38                 defer lock.fs.mu.Unlock()
  39                 if lock.fs.slock == lock {
  40                         lock.fs.slock = nil
  41                 }
  42         }
  43 }
  44
  45 const logSizeThreshold = 1024 * 1024 // 1 MiB
  46
  47 // fileStorage is a file-system backed storage.
  48 type fileStorage struct {
  49         path     string
  50         readOnly bool
  51
  52         mu      sync.Mutex
  53         flock   fileLock
  54         slock   *fileStorageLock
  55         logw    *os.File
  56         logSize int64
  57         buf     []byte
  58         // Opened file counter; if open < 0 means closed.
  59         open int
  60         day  int
  61 }
  62
  63 // OpenFile returns a new filesytem-backed storage implementation with the given
  64 // path. This also acquire a file lock, so any subsequent attempt to open the
  65 // same path will fail.
  66 //
  67 // The storage must be closed after use, by calling Close method.
  68 func OpenFile(path string, readOnly bool) (Storage, error) {
  69         if fi, err := os.Stat(path); err == nil {
  70                 if !fi.IsDir() {
  71                         return nil, fmt.Errorf("leveldb/storage: open %s: not a directory", path)
  72                 }
  73         } else if os.IsNotExist(err) && !readOnly {
  74                 if err := os.MkdirAll(path, 0755); err != nil {
  75                         return nil, err
  76                 }
  77         } else {
  78                 return nil, err
  79         }
  80
  81         flock, err := newFileLock(filepath.Join(path, "LOCK"), readOnly)
  82         if err != nil {
  83                 return nil, err
  84         }
  85
  86         defer func() {
  87                 if err != nil {
  88                         flock.release()
  89                 }
  90         }()
  91
  92         var (
  93                 logw    *os.File
  94                 logSize int64
  95         )
  96         if !readOnly {
  97                 logw, err = os.OpenFile(filepath.Join(path, "LOG"), os.O_WRONLY|os.O_CREATE, 0644)
  98                 if err != nil {
  99                         return nil, err
 100                 }
 101                 logSize, err = logw.Seek(0, os.SEEK_END)
 102                 if err != nil {
 103                         logw.Close()
 104                         return nil, err
 105                 }
 106         }
 107
 108         fs := &fileStorage{
 109                 path:     path,
 110                 readOnly: readOnly,
 111                 flock:    flock,
 112                 logw:     logw,
 113                 logSize:  logSize,
 114         }
 115         runtime.SetFinalizer(fs, (*fileStorage).Close)
 116         return fs, nil
 117 }
 118
 119 func (fs *fileStorage) Lock() (Locker, error) {
 120         fs.mu.Lock()
 121         defer fs.mu.Unlock()
 122         if fs.open < 0 {
 123                 return nil, ErrClosed
 124         }
 125         if fs.readOnly {
 126                 return &fileStorageLock{}, nil
 127         }
 128         if fs.slock != nil {
 129                 return nil, ErrLocked
 130         }
 131         fs.slock = &fileStorageLock{fs: fs}
 132         return fs.slock, nil
 133 }
 134
 135 func itoa(buf []byte, i int, wid int) []byte {
 136         u := uint(i)
 137         if u == 0 && wid <= 1 {
 138                 return append(buf, '0')
 139         }
 140
 141         // Assemble decimal in reverse order.
 142         var b [32]byte
 143         bp := len(b)
 144         for ; u > 0 || wid > 0; u /= 10 {
 145                 bp--
 146                 wid--
 147                 b[bp] = byte(u%10) + '0'
 148         }
 149         return append(buf, b[bp:]...)
 150 }
 151
 152 func (fs *fileStorage) printDay(t time.Time) {
 153         if fs.day == t.Day() {
 154                 return
 155         }
 156         fs.day = t.Day()
 157         fs.logw.Write([]byte("=============== " + t.Format("Jan 2, 2006 (MST)") + " ===============\n"))
 158 }
 159
 160 func (fs *fileStorage) doLog(t time.Time, str string) {
 161         if fs.logSize > logSizeThreshold {
 162                 // Rotate log file.
 163                 fs.logw.Close()
 164                 fs.logw = nil
 165                 fs.logSize = 0
 166                 rename(filepath.Join(fs.path, "LOG"), filepath.Join(fs.path, "LOG.old"))
 167         }
 168         if fs.logw == nil {
 169                 var err error
 170                 fs.logw, err = os.OpenFile(filepath.Join(fs.path, "LOG"), os.O_WRONLY|os.O_CREATE, 0644)
 171                 if err != nil {
 172                         return
 173                 }
 174                 // Force printDay on new log file.
 175                 fs.day = 0
 176         }
 177         fs.printDay(t)
 178         hour, min, sec := t.Clock()
 179         msec := t.Nanosecond() / 1e3
 180         // time
 181         fs.buf = itoa(fs.buf[:0], hour, 2)
 182         fs.buf = append(fs.buf, ':')
 183         fs.buf = itoa(fs.buf, min, 2)
 184         fs.buf = append(fs.buf, ':')
 185         fs.buf = itoa(fs.buf, sec, 2)
 186         fs.buf = append(fs.buf, '.')
 187         fs.buf = itoa(fs.buf, msec, 6)
 188         fs.buf = append(fs.buf, ' ')
 189         // write
 190         fs.buf = append(fs.buf, []byte(str)...)
 191         fs.buf = append(fs.buf, '\n')
 192         fs.logw.Write(fs.buf)
 193 }
 194
 195 func (fs *fileStorage) Log(str string) {
 196         if !fs.readOnly {
 197                 t := time.Now()
 198                 fs.mu.Lock()
 199                 defer fs.mu.Unlock()
 200                 if fs.open < 0 {
 201                         return
 202                 }
 203                 fs.doLog(t, str)
 204         }
 205 }
 206
 207 func (fs *fileStorage) log(str string) {
 208         if !fs.readOnly {
 209                 fs.doLog(time.Now(), str)
 210         }
 211 }
 212
 213 func (fs *fileStorage) SetMeta(fd FileDesc) (err error) {
 214         if !FileDescOk(fd) {
 215                 return ErrInvalidFile
 216         }
 217         if fs.readOnly {
 218                 return errReadOnly
 219         }
 220
 221         fs.mu.Lock()
 222         defer fs.mu.Unlock()
 223         if fs.open < 0 {
 224                 return ErrClosed
 225         }
 226         defer func() {
 227                 if err != nil {
 228                         fs.log(fmt.Sprintf("CURRENT: %v", err))
 229                 }
 230         }()
 231         path := fmt.Sprintf("%s.%d", filepath.Join(fs.path, "CURRENT"), fd.Num)
 232         w, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 233         if err != nil {
 234                 return
 235         }
 236         _, err = fmt.Fprintln(w, fsGenName(fd))
 237         if err != nil {
 238                 fs.log(fmt.Sprintf("write CURRENT.%d: %v", fd.Num, err))
 239                 return
 240         }
 241         if err = w.Sync(); err != nil {
 242                 fs.log(fmt.Sprintf("flush CURRENT.%d: %v", fd.Num, err))
 243                 return
 244         }
 245         if err = w.Close(); err != nil {
 246                 fs.log(fmt.Sprintf("close CURRENT.%d: %v", fd.Num, err))
 247                 return
 248         }
 249         if err != nil {
 250                 return
 251         }
 252         if err = rename(path, filepath.Join(fs.path, "CURRENT")); err != nil {
 253                 fs.log(fmt.Sprintf("rename CURRENT.%d: %v", fd.Num, err))
 254                 return
 255         }
 256         // Sync root directory.
 257         if err = syncDir(fs.path); err != nil {
 258                 fs.log(fmt.Sprintf("syncDir: %v", err))
 259         }
 260         return
 261 }
 262
 263 func (fs *fileStorage) GetMeta() (fd FileDesc, err error) {
 264         fs.mu.Lock()
 265         defer fs.mu.Unlock()
 266         if fs.open < 0 {
 267                 return FileDesc{}, ErrClosed
 268         }
 269         dir, err := os.Open(fs.path)
 270         if err != nil {
 271                 return
 272         }
 273         names, err := dir.Readdirnames(0)
 274         // Close the dir first before checking for Readdirnames error.
 275         if ce := dir.Close(); ce != nil {
 276                 fs.log(fmt.Sprintf("close dir: %v", ce))
 277         }
 278         if err != nil {
 279                 return
 280         }
 281         // Find latest CURRENT file.
 282         var rem []string
 283         var pend bool
 284         var cerr error
 285         for _, name := range names {
 286                 if strings.HasPrefix(name, "CURRENT") {
 287                         pend1 := len(name) > 7
 288                         var pendNum int64
 289                         // Make sure it is valid name for a CURRENT file, otherwise skip it.
 290                         if pend1 {
 291                                 if name[7] != '.' || len(name) < 9 {
 292                                         fs.log(fmt.Sprintf("skipping %s: invalid file name", name))
 293                                         continue
 294                                 }
 295                                 var e1 error
 296                                 if pendNum, e1 = strconv.ParseInt(name[8:], 10, 0); e1 != nil {
 297                                         fs.log(fmt.Sprintf("skipping %s: invalid file num: %v", name, e1))
 298                                         continue
 299                                 }
 300                         }
 301                         path := filepath.Join(fs.path, name)
 302                         r, e1 := os.OpenFile(path, os.O_RDONLY, 0)
 303                         if e1 != nil {
 304                                 return FileDesc{}, e1
 305                         }
 306                         b, e1 := ioutil.ReadAll(r)
 307                         if e1 != nil {
 308                                 r.Close()
 309                                 return FileDesc{}, e1
 310                         }
 311                         var fd1 FileDesc
 312                         if len(b) < 1 || b[len(b)-1] != '\n' || !fsParseNamePtr(string(b[:len(b)-1]), &fd1) {
 313                                 fs.log(fmt.Sprintf("skipping %s: corrupted or incomplete", name))
 314                                 if pend1 {
 315                                         rem = append(rem, name)
 316                                 }
 317                                 if !pend1 || cerr == nil {
 318                                         metaFd, _ := fsParseName(name)
 319                                         cerr = &ErrCorrupted{
 320                                                 Fd:  metaFd,
 321                                                 Err: errors.New("leveldb/storage: corrupted or incomplete meta file"),
 322                                         }
 323                                 }
 324                         } else if pend1 && pendNum != fd1.Num {
 325                                 fs.log(fmt.Sprintf("skipping %s: inconsistent pending-file num: %d vs %d", name, pendNum, fd1.Num))
 326                                 rem = append(rem, name)
 327                         } else if fd1.Num < fd.Num {
 328                                 fs.log(fmt.Sprintf("skipping %s: obsolete", name))
 329                                 if pend1 {
 330                                         rem = append(rem, name)
 331                                 }
 332                         } else {
 333                                 fd = fd1
 334                                 pend = pend1
 335                         }
 336                         if err := r.Close(); err != nil {
 337                                 fs.log(fmt.Sprintf("close %s: %v", name, err))
 338                         }
 339                 }
 340         }
 341         // Don't remove any files if there is no valid CURRENT file.
 342         if fd.Zero() {
 343                 if cerr != nil {
 344                         err = cerr
 345                 } else {
 346                         err = os.ErrNotExist
 347                 }
 348                 return
 349         }
 350         if !fs.readOnly {
 351                 // Rename pending CURRENT file to an effective CURRENT.
 352                 if pend {
 353                         path := fmt.Sprintf("%s.%d", filepath.Join(fs.path, "CURRENT"), fd.Num)
 354                         if err := rename(path, filepath.Join(fs.path, "CURRENT")); err != nil {
 355                                 fs.log(fmt.Sprintf("CURRENT.%d -> CURRENT: %v", fd.Num, err))
 356                         }
 357                 }
 358                 // Remove obsolete or incomplete pending CURRENT files.
 359                 for _, name := range rem {
 360                         path := filepath.Join(fs.path, name)
 361                         if err := os.Remove(path); err != nil {
 362                                 fs.log(fmt.Sprintf("remove %s: %v", name, err))
 363                         }
 364                 }
 365         }
 366         return
 367 }
 368
 369 func (fs *fileStorage) List(ft FileType) (fds []FileDesc, err error) {
 370         fs.mu.Lock()
 371         defer fs.mu.Unlock()
 372         if fs.open < 0 {
 373                 return nil, ErrClosed
 374         }
 375         dir, err := os.Open(fs.path)
 376         if err != nil {
 377                 return
 378         }
 379         names, err := dir.Readdirnames(0)
 380         // Close the dir first before checking for Readdirnames error.
 381         if cerr := dir.Close(); cerr != nil {
 382                 fs.log(fmt.Sprintf("close dir: %v", cerr))
 383         }
 384         if err == nil {
 385                 for _, name := range names {
 386                         if fd, ok := fsParseName(name); ok && fd.Type&ft != 0 {
 387                                 fds = append(fds, fd)
 388                         }
 389                 }
 390         }
 391         return
 392 }
 393
 394 func (fs *fileStorage) Open(fd FileDesc) (Reader, error) {
 395         if !FileDescOk(fd) {
 396                 return nil, ErrInvalidFile
 397         }
 398
 399         fs.mu.Lock()
 400         defer fs.mu.Unlock()
 401         if fs.open < 0 {
 402                 return nil, ErrClosed
 403         }
 404         of, err := os.OpenFile(filepath.Join(fs.path, fsGenName(fd)), os.O_RDONLY, 0)
 405         if err != nil {
 406                 if fsHasOldName(fd) && os.IsNotExist(err) {
 407                         of, err = os.OpenFile(filepath.Join(fs.path, fsGenOldName(fd)), os.O_RDONLY, 0)
 408                         if err == nil {
 409                                 goto ok
 410                         }
 411                 }
 412                 return nil, err
 413         }
 414 ok:
 415         fs.open++
 416         return &fileWrap{File: of, fs: fs, fd: fd}, nil
 417 }
 418
 419 func (fs *fileStorage) Create(fd FileDesc) (Writer, error) {
 420         if !FileDescOk(fd) {
 421                 return nil, ErrInvalidFile
 422         }
 423         if fs.readOnly {
 424                 return nil, errReadOnly
 425         }
 426
 427         fs.mu.Lock()
 428         defer fs.mu.Unlock()
 429         if fs.open < 0 {
 430                 return nil, ErrClosed
 431         }
 432         of, err := os.OpenFile(filepath.Join(fs.path, fsGenName(fd)), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
 433         if err != nil {
 434                 return nil, err
 435         }
 436         fs.open++
 437         return &fileWrap{File: of, fs: fs, fd: fd}, nil
 438 }
 439
 440 func (fs *fileStorage) Remove(fd FileDesc) error {
 441         if !FileDescOk(fd) {
 442                 return ErrInvalidFile
 443         }
 444         if fs.readOnly {
 445                 return errReadOnly
 446         }
 447
 448         fs.mu.Lock()
 449         defer fs.mu.Unlock()
 450         if fs.open < 0 {
 451                 return ErrClosed
 452         }
 453         err := os.Remove(filepath.Join(fs.path, fsGenName(fd)))
 454         if err != nil {
 455                 if fsHasOldName(fd) && os.IsNotExist(err) {
 456                         if e1 := os.Remove(filepath.Join(fs.path, fsGenOldName(fd))); !os.IsNotExist(e1) {
 457                                 fs.log(fmt.Sprintf("remove %s: %v (old name)", fd, err))
 458                                 err = e1
 459                         }
 460                 } else {
 461                         fs.log(fmt.Sprintf("remove %s: %v", fd, err))
 462                 }
 463         }
 464         return err
 465 }
 466
 467 func (fs *fileStorage) Rename(oldfd, newfd FileDesc) error {
 468         if !FileDescOk(oldfd) || !FileDescOk(newfd) {
 469                 return ErrInvalidFile
 470         }
 471         if oldfd == newfd {
 472                 return nil
 473         }
 474         if fs.readOnly {
 475                 return errReadOnly
 476         }
 477
 478         fs.mu.Lock()
 479         defer fs.mu.Unlock()
 480         if fs.open < 0 {
 481                 return ErrClosed
 482         }
 483         return rename(filepath.Join(fs.path, fsGenName(oldfd)), filepath.Join(fs.path, fsGenName(newfd)))
 484 }
 485
 486 func (fs *fileStorage) Close() error {
 487         fs.mu.Lock()
 488         defer fs.mu.Unlock()
 489         if fs.open < 0 {
 490                 return ErrClosed
 491         }
 492         // Clear the finalizer.
 493         runtime.SetFinalizer(fs, nil)
 494
 495         if fs.open > 0 {
 496                 fs.log(fmt.Sprintf("close: warning, %d files still open", fs.open))
 497         }
 498         fs.open = -1
 499         if fs.logw != nil {
 500                 fs.logw.Close()
 501         }
 502         return fs.flock.release()
 503 }
 504
 505 type fileWrap struct {
 506         *os.File
 507         fs     *fileStorage
 508         fd     FileDesc
 509         closed bool
 510 }
 511
 512 func (fw *fileWrap) Sync() error {
 513         if err := fw.File.Sync(); err != nil {
 514                 return err
 515         }
 516         if fw.fd.Type == TypeManifest {
 517                 // Also sync parent directory if file type is manifest.
 518                 // See: https://code.google.com/p/leveldb/issues/detail?id=190.
 519                 if err := syncDir(fw.fs.path); err != nil {
 520                         fw.fs.log(fmt.Sprintf("syncDir: %v", err))
 521                         return err
 522                 }
 523         }
 524         return nil
 525 }
 526
 527 func (fw *fileWrap) Close() error {
 528         fw.fs.mu.Lock()
 529         defer fw.fs.mu.Unlock()
 530         if fw.closed {
 531                 return ErrClosed
 532         }
 533         fw.closed = true
 534         fw.fs.open--
 535         err := fw.File.Close()
 536         if err != nil {
 537                 fw.fs.log(fmt.Sprintf("close %s: %v", fw.fd, err))
 538         }
 539         return err
 540 }
 541
 542 func fsGenName(fd FileDesc) string {
 543         switch fd.Type {
 544         case TypeManifest:
 545                 return fmt.Sprintf("MANIFEST-%06d", fd.Num)
 546         case TypeJournal:
 547                 return fmt.Sprintf("%06d.log", fd.Num)
 548         case TypeTable:
 549                 return fmt.Sprintf("%06d.ldb", fd.Num)
 550         case TypeTemp:
 551                 return fmt.Sprintf("%06d.tmp", fd.Num)
 552         default:
 553                 panic("invalid file type")
 554         }
 555 }
 556
 557 func fsHasOldName(fd FileDesc) bool {
 558         return fd.Type == TypeTable
 559 }
 560
 561 func fsGenOldName(fd FileDesc) string {
 562         switch fd.Type {
 563         case TypeTable:
 564                 return fmt.Sprintf("%06d.sst", fd.Num)
 565         }
 566         return fsGenName(fd)
 567 }
 568
 569 func fsParseName(name string) (fd FileDesc, ok bool) {
 570         var tail string
 571         _, err := fmt.Sscanf(name, "%d.%s", &fd.Num, &tail)
 572         if err == nil {
 573                 switch tail {
 574                 case "log":
 575                         fd.Type = TypeJournal
 576                 case "ldb", "sst":
 577                         fd.Type = TypeTable
 578                 case "tmp":
 579                         fd.Type = TypeTemp
 580                 default:
 581                         return
 582                 }
 583                 return fd, true
 584         }
 585         n, _ := fmt.Sscanf(name, "MANIFEST-%d%s", &fd.Num, &tail)
 586         if n == 1 {
 587                 fd.Type = TypeManifest
 588                 return fd, true
 589         }
 590         return
 591 }
 592
 593 func fsParseNamePtr(name string, fd *FileDesc) bool {
 594         _fd, ok := fsParseName(name)
 595         if fd != nil {
 596                 *fd = _fd
 597         }
 598         return ok
 599 }