sdk/go/arvados/collection_fs.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: Apache-2.0
   4
   5 package arvados
   6
   7 import (
   8         "crypto/md5"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "net/http"
  13         "os"
  14         "path"
  15         "regexp"
  16         "sort"
  17         "strconv"
  18         "strings"
  19         "sync"
  20         "time"
  21 )
  22
  23 var (
  24         ErrReadOnlyFile      = errors.New("read-only file")
  25         ErrNegativeOffset    = errors.New("cannot seek to negative offset")
  26         ErrFileExists        = errors.New("file exists")
  27         ErrInvalidOperation  = errors.New("invalid operation")
  28         ErrDirectoryNotEmpty = errors.New("directory not empty")
  29         ErrPermission        = os.ErrPermission
  30
  31         maxBlockSize = 1 << 26
  32 )
  33
  34 type File interface {
  35         io.Reader
  36         io.Writer
  37         io.Closer
  38         io.Seeker
  39         Size() int64
  40         Readdir(int) ([]os.FileInfo, error)
  41         Stat() (os.FileInfo, error)
  42         Truncate(int64) error
  43 }
  44
  45 type keepClient interface {
  46         ReadAt(locator string, p []byte, off int) (int, error)
  47 }
  48
  49 type fileinfo struct {
  50         name    string
  51         mode    os.FileMode
  52         size    int64
  53         modTime time.Time
  54 }
  55
  56 // Name implements os.FileInfo.
  57 func (fi fileinfo) Name() string {
  58         return fi.name
  59 }
  60
  61 // ModTime implements os.FileInfo.
  62 func (fi fileinfo) ModTime() time.Time {
  63         return fi.modTime
  64 }
  65
  66 // Mode implements os.FileInfo.
  67 func (fi fileinfo) Mode() os.FileMode {
  68         return fi.mode
  69 }
  70
  71 // IsDir implements os.FileInfo.
  72 func (fi fileinfo) IsDir() bool {
  73         return fi.mode&os.ModeDir != 0
  74 }
  75
  76 // Size implements os.FileInfo.
  77 func (fi fileinfo) Size() int64 {
  78         return fi.size
  79 }
  80
  81 // Sys implements os.FileInfo.
  82 func (fi fileinfo) Sys() interface{} {
  83         return nil
  84 }
  85
  86 func (fi fileinfo) Stat() os.FileInfo {
  87         return fi
  88 }
  89
  90 // A CollectionFileSystem is an http.Filesystem plus Stat() and
  91 // support for opening writable files.
  92 type CollectionFileSystem interface {
  93         http.FileSystem
  94         Stat(name string) (os.FileInfo, error)
  95         Create(name string) (File, error)
  96         OpenFile(name string, flag int, perm os.FileMode) (File, error)
  97         Mkdir(name string, perm os.FileMode) error
  98         Remove(name string) error
  99         MarshalManifest(string) (string, error)
 100 }
 101
 102 type fileSystem struct {
 103         dirnode
 104 }
 105
 106 func (fs *fileSystem) OpenFile(name string, flag int, perm os.FileMode) (File, error) {
 107         return fs.dirnode.OpenFile(path.Clean(name), flag, perm)
 108 }
 109
 110 func (fs *fileSystem) Open(name string) (http.File, error) {
 111         return fs.dirnode.OpenFile(path.Clean(name), os.O_RDONLY, 0)
 112 }
 113
 114 func (fs *fileSystem) Create(name string) (File, error) {
 115         return fs.dirnode.OpenFile(path.Clean(name), os.O_CREATE|os.O_RDWR|os.O_TRUNC, 0)
 116 }
 117
 118 func (fs *fileSystem) Stat(name string) (os.FileInfo, error) {
 119         f, err := fs.OpenFile(name, os.O_RDONLY, 0)
 120         if err != nil {
 121                 return nil, err
 122         }
 123         defer f.Close()
 124         return f.Stat()
 125 }
 126
 127 type inode interface {
 128         os.FileInfo
 129         Parent() inode
 130         Read([]byte, filenodePtr) (int, filenodePtr, error)
 131         Write([]byte, filenodePtr) (int, filenodePtr, error)
 132         Truncate(int64) error
 133         Readdir() []os.FileInfo
 134         Stat() os.FileInfo
 135         sync.Locker
 136         RLock()
 137         RUnlock()
 138 }
 139
 140 // filenode implements inode.
 141 type filenode struct {
 142         fileinfo
 143         parent   *dirnode
 144         extents  []extent
 145         repacked int64 // number of times anything in []extents has changed len
 146         sync.RWMutex
 147 }
 148
 149 // filenodePtr is an offset into a file that is (usually) efficient to
 150 // seek to. Specifically, if filenode.repacked==filenodePtr.repacked
 151 // then filenode.extents[filenodePtr.extentIdx][filenodePtr.extentOff]
 152 // corresponds to file offset filenodePtr.off. Otherwise, it is
 153 // necessary to reexamine len(filenode.extents[0]) etc. to find the
 154 // correct extent and offset.
 155 type filenodePtr struct {
 156         off       int64
 157         extentIdx int
 158         extentOff int
 159         repacked  int64
 160 }
 161
 162 // seek returns a ptr that is consistent with both startPtr.off and
 163 // the current state of fn. The caller must already hold fn.RLock() or
 164 // fn.Lock().
 165 //
 166 // If startPtr points beyond the end of the file, ptr will point to
 167 // exactly the end of the file.
 168 //
 169 // After seeking:
 170 //
 171 //     ptr.extentIdx == len(filenode.extents) // i.e., at EOF
 172 //     ||
 173 //     filenode.extents[ptr.extentIdx].Len() >= ptr.extentOff
 174 func (fn *filenode) seek(startPtr filenodePtr) (ptr filenodePtr) {
 175         ptr = startPtr
 176         if ptr.off < 0 {
 177                 // meaningless anyway
 178                 return
 179         } else if ptr.off >= fn.fileinfo.size {
 180                 ptr.off = fn.fileinfo.size
 181                 ptr.extentIdx = len(fn.extents)
 182                 ptr.extentOff = 0
 183                 ptr.repacked = fn.repacked
 184                 return
 185         } else if ptr.repacked == fn.repacked {
 186                 // extentIdx and extentOff accurately reflect ptr.off,
 187                 // but might have fallen off the end of an extent
 188                 if ptr.extentOff >= fn.extents[ptr.extentIdx].Len() {
 189                         ptr.extentIdx++
 190                         ptr.extentOff = 0
 191                 }
 192                 return
 193         }
 194         defer func() {
 195                 ptr.repacked = fn.repacked
 196         }()
 197         if ptr.off >= fn.fileinfo.size {
 198                 ptr.extentIdx, ptr.extentOff = len(fn.extents), 0
 199                 return
 200         }
 201         // Recompute extentIdx and extentOff.  We have already
 202         // established fn.fileinfo.size > ptr.off >= 0, so we don't
 203         // have to deal with edge cases here.
 204         var off int64
 205         for ptr.extentIdx, ptr.extentOff = 0, 0; off < ptr.off; ptr.extentIdx++ {
 206                 // This would panic (index out of range) if
 207                 // fn.fileinfo.size were larger than
 208                 // sum(fn.extents[i].Len()) -- but that can't happen
 209                 // because we have ensured fn.fileinfo.size is always
 210                 // accurate.
 211                 extLen := int64(fn.extents[ptr.extentIdx].Len())
 212                 if off+extLen > ptr.off {
 213                         ptr.extentOff = int(ptr.off - off)
 214                         break
 215                 }
 216                 off += extLen
 217         }
 218         return
 219 }
 220
 221 func (fn *filenode) appendExtent(e extent) {
 222         fn.Lock()
 223         defer fn.Unlock()
 224         fn.extents = append(fn.extents, e)
 225         fn.fileinfo.size += int64(e.Len())
 226 }
 227
 228 func (fn *filenode) Parent() inode {
 229         return fn.parent
 230 }
 231
 232 func (fn *filenode) Readdir() []os.FileInfo {
 233         return nil
 234 }
 235
 236 func (fn *filenode) Read(p []byte, startPtr filenodePtr) (n int, ptr filenodePtr, err error) {
 237         fn.RLock()
 238         defer fn.RUnlock()
 239         ptr = fn.seek(startPtr)
 240         if ptr.off < 0 {
 241                 err = ErrNegativeOffset
 242                 return
 243         }
 244         if ptr.extentIdx >= len(fn.extents) {
 245                 err = io.EOF
 246                 return
 247         }
 248         n, err = fn.extents[ptr.extentIdx].ReadAt(p, int64(ptr.extentOff))
 249         if n > 0 {
 250                 ptr.off += int64(n)
 251                 ptr.extentOff += n
 252                 if ptr.extentOff == fn.extents[ptr.extentIdx].Len() {
 253                         ptr.extentIdx++
 254                         ptr.extentOff = 0
 255                         if ptr.extentIdx < len(fn.extents) && err == io.EOF {
 256                                 err = nil
 257                         }
 258                 }
 259         }
 260         return
 261 }
 262
 263 func (fn *filenode) Truncate(size int64) error {
 264         fn.Lock()
 265         defer fn.Unlock()
 266         if size < fn.fileinfo.size {
 267                 ptr := fn.seek(filenodePtr{off: size, repacked: fn.repacked - 1})
 268                 if ptr.extentOff == 0 {
 269                         fn.extents = fn.extents[:ptr.extentIdx]
 270                 } else {
 271                         fn.extents = fn.extents[:ptr.extentIdx+1]
 272                         e := fn.extents[ptr.extentIdx]
 273                         if e, ok := e.(writableExtent); ok {
 274                                 e.Truncate(ptr.extentOff)
 275                         } else {
 276                                 fn.extents[ptr.extentIdx] = e.Slice(0, ptr.extentOff)
 277                         }
 278                 }
 279                 fn.fileinfo.size = size
 280                 fn.repacked++
 281                 return nil
 282         }
 283         for size > fn.fileinfo.size {
 284                 grow := size - fn.fileinfo.size
 285                 var e writableExtent
 286                 var ok bool
 287                 if len(fn.extents) == 0 {
 288                         e = &memExtent{}
 289                         fn.extents = append(fn.extents, e)
 290                 } else if e, ok = fn.extents[len(fn.extents)-1].(writableExtent); !ok || e.Len() >= maxBlockSize {
 291                         e = &memExtent{}
 292                         fn.extents = append(fn.extents, e)
 293                 } else {
 294                         fn.repacked++
 295                 }
 296                 if maxgrow := int64(maxBlockSize - e.Len()); maxgrow < grow {
 297                         grow = maxgrow
 298                 }
 299                 e.Truncate(e.Len() + int(grow))
 300                 fn.fileinfo.size += grow
 301         }
 302         return nil
 303 }
 304
 305 func (fn *filenode) Write(p []byte, startPtr filenodePtr) (n int, ptr filenodePtr, err error) {
 306         fn.Lock()
 307         defer fn.Unlock()
 308         ptr = fn.seek(startPtr)
 309         if ptr.off < 0 {
 310                 err = ErrNegativeOffset
 311                 return
 312         }
 313         for len(p) > 0 && err == nil {
 314                 cando := p
 315                 if len(cando) > maxBlockSize {
 316                         cando = cando[:maxBlockSize]
 317                 }
 318                 // Rearrange/grow fn.extents (and shrink cando if
 319                 // needed) such that cando can be copied to
 320                 // fn.extents[ptr.extentIdx] at offset ptr.extentOff.
 321                 cur := ptr.extentIdx
 322                 prev := ptr.extentIdx - 1
 323                 var curWritable bool
 324                 if cur < len(fn.extents) {
 325                         _, curWritable = fn.extents[cur].(writableExtent)
 326                 }
 327                 var prevAppendable bool
 328                 if prev >= 0 && fn.extents[prev].Len() < maxBlockSize {
 329                         _, prevAppendable = fn.extents[prev].(writableExtent)
 330                 }
 331                 if ptr.extentOff > 0 && !curWritable {
 332                         // Split a non-writable block.
 333                         if max := fn.extents[cur].Len() - ptr.extentOff; max <= len(cando) {
 334                                 // Truncate cur, and insert a new
 335                                 // extent after it.
 336                                 cando = cando[:max]
 337                                 fn.extents = append(fn.extents, nil)
 338                                 copy(fn.extents[cur+1:], fn.extents[cur:])
 339                         } else {
 340                                 // Split cur into two copies, truncate
 341                                 // the one on the left, shift the one
 342                                 // on the right, and insert a new
 343                                 // extent between them.
 344                                 fn.extents = append(fn.extents, nil, nil)
 345                                 copy(fn.extents[cur+2:], fn.extents[cur:])
 346                                 fn.extents[cur+2] = fn.extents[cur+2].Slice(ptr.extentOff+len(cando), -1)
 347                         }
 348                         cur++
 349                         prev++
 350                         e := &memExtent{}
 351                         e.Truncate(len(cando))
 352                         fn.extents[cur] = e
 353                         fn.extents[prev] = fn.extents[prev].Slice(0, ptr.extentOff)
 354                         ptr.extentIdx++
 355                         ptr.extentOff = 0
 356                         fn.repacked++
 357                         ptr.repacked++
 358                 } else if curWritable {
 359                         if fit := int(fn.extents[cur].Len()) - ptr.extentOff; fit < len(cando) {
 360                                 cando = cando[:fit]
 361                         }
 362                 } else {
 363                         if prevAppendable {
 364                                 // Shrink cando if needed to fit in prev extent.
 365                                 if cangrow := maxBlockSize - fn.extents[prev].Len(); cangrow < len(cando) {
 366                                         cando = cando[:cangrow]
 367                                 }
 368                         }
 369
 370                         if cur == len(fn.extents) {
 371                                 // ptr is at EOF, filesize is changing.
 372                                 fn.fileinfo.size += int64(len(cando))
 373                         } else if el := fn.extents[cur].Len(); el <= len(cando) {
 374                                 // cando is long enough that we won't
 375                                 // need cur any more. shrink cando to
 376                                 // be exactly as long as cur
 377                                 // (otherwise we'd accidentally shift
 378                                 // the effective position of all
 379                                 // extents after cur).
 380                                 cando = cando[:el]
 381                                 copy(fn.extents[cur:], fn.extents[cur+1:])
 382                                 fn.extents = fn.extents[:len(fn.extents)-1]
 383                         } else {
 384                                 // shrink cur by the same #bytes we're growing prev
 385                                 fn.extents[cur] = fn.extents[cur].Slice(len(cando), -1)
 386                         }
 387
 388                         if prevAppendable {
 389                                 // Grow prev.
 390                                 ptr.extentIdx--
 391                                 ptr.extentOff = fn.extents[prev].Len()
 392                                 fn.extents[prev].(writableExtent).Truncate(ptr.extentOff + len(cando))
 393                                 ptr.repacked++
 394                                 fn.repacked++
 395                         } else {
 396                                 // Insert an extent between prev and cur, and advance prev/cur.
 397                                 fn.extents = append(fn.extents, nil)
 398                                 if cur < len(fn.extents) {
 399                                         copy(fn.extents[cur+1:], fn.extents[cur:])
 400                                         ptr.repacked++
 401                                         fn.repacked++
 402                                 } else {
 403                                         // appending a new extent does
 404                                         // not invalidate any ptrs
 405                                 }
 406                                 e := &memExtent{}
 407                                 e.Truncate(len(cando))
 408                                 fn.extents[cur] = e
 409                                 cur++
 410                                 prev++
 411                         }
 412                 }
 413
 414                 // Finally we can copy bytes from cando to the current extent.
 415                 fn.extents[ptr.extentIdx].(writableExtent).WriteAt(cando, ptr.extentOff)
 416                 n += len(cando)
 417                 p = p[len(cando):]
 418
 419                 ptr.off += int64(len(cando))
 420                 ptr.extentOff += len(cando)
 421                 if fn.extents[ptr.extentIdx].Len() == ptr.extentOff {
 422                         ptr.extentOff = 0
 423                         ptr.extentIdx++
 424                 }
 425         }
 426         return
 427 }
 428
 429 // FileSystem returns a CollectionFileSystem for the collection.
 430 func (c *Collection) FileSystem(client *Client, kc keepClient) CollectionFileSystem {
 431         fs := &fileSystem{dirnode: dirnode{
 432                 cache:    &keepBlockCache{kc: kc},
 433                 client:   client,
 434                 kc:       kc,
 435                 fileinfo: fileinfo{name: ".", mode: os.ModeDir | 0755},
 436                 parent:   nil,
 437                 inodes:   make(map[string]inode),
 438         }}
 439         fs.dirnode.parent = &fs.dirnode
 440         fs.dirnode.loadManifest(c.ManifestText)
 441         return fs
 442 }
 443
 444 type file struct {
 445         inode
 446         ptr        filenodePtr
 447         append     bool
 448         writable   bool
 449         unreaddirs []os.FileInfo
 450 }
 451
 452 func (f *file) Read(p []byte) (n int, err error) {
 453         n, f.ptr, err = f.inode.Read(p, f.ptr)
 454         return
 455 }
 456
 457 func (f *file) Seek(off int64, whence int) (pos int64, err error) {
 458         size := f.inode.Size()
 459         ptr := f.ptr
 460         switch whence {
 461         case os.SEEK_SET:
 462                 ptr.off = off
 463         case os.SEEK_CUR:
 464                 ptr.off += off
 465         case os.SEEK_END:
 466                 ptr.off = size + off
 467         }
 468         if ptr.off < 0 {
 469                 return f.ptr.off, ErrNegativeOffset
 470         }
 471         if ptr.off > size {
 472                 ptr.off = size
 473         }
 474         if ptr.off != f.ptr.off {
 475                 f.ptr = ptr
 476                 // force filenode to recompute f.ptr fields on next
 477                 // use
 478                 f.ptr.repacked = -1
 479         }
 480         return f.ptr.off, nil
 481 }
 482
 483 func (f *file) Truncate(size int64) error {
 484         return f.inode.Truncate(size)
 485 }
 486
 487 func (f *file) Write(p []byte) (n int, err error) {
 488         if !f.writable {
 489                 return 0, ErrReadOnlyFile
 490         }
 491         n, f.ptr, err = f.inode.Write(p, f.ptr)
 492         return
 493 }
 494
 495 func (f *file) Readdir(count int) ([]os.FileInfo, error) {
 496         if !f.inode.IsDir() {
 497                 return nil, ErrInvalidOperation
 498         }
 499         if count <= 0 {
 500                 return f.inode.Readdir(), nil
 501         }
 502         if f.unreaddirs == nil {
 503                 f.unreaddirs = f.inode.Readdir()
 504         }
 505         if len(f.unreaddirs) == 0 {
 506                 return nil, io.EOF
 507         }
 508         if count > len(f.unreaddirs) {
 509                 count = len(f.unreaddirs)
 510         }
 511         ret := f.unreaddirs[:count]
 512         f.unreaddirs = f.unreaddirs[count:]
 513         return ret, nil
 514 }
 515
 516 func (f *file) Stat() (os.FileInfo, error) {
 517         return f.inode, nil
 518 }
 519
 520 func (f *file) Close() error {
 521         // FIXME: flush
 522         return nil
 523 }
 524
 525 type dirnode struct {
 526         fileinfo
 527         parent *dirnode
 528         client *Client
 529         kc     keepClient
 530         cache  blockCache
 531         inodes map[string]inode
 532         sync.RWMutex
 533 }
 534
 535 // caller must hold dn.Lock().
 536 func (dn *dirnode) sync() error {
 537         type shortBlock struct {
 538                 fn  *filenode
 539                 idx int
 540         }
 541         var pending []shortBlock
 542         var pendingLen int
 543
 544         flush := func(sbs []shortBlock) error {
 545                 if len(sbs) == 0 {
 546                         return nil
 547                 }
 548                 hash := md5.New()
 549                 size := 0
 550                 for _, sb := range sbs {
 551                         data := sb.fn.extents[sb.idx].(*memExtent).buf
 552                         if _, err := hash.Write(data); err != nil {
 553                                 return err
 554                         }
 555                         size += len(data)
 556                 }
 557                 // FIXME: write to keep
 558                 locator := fmt.Sprintf("%x+%d", hash.Sum(nil), size)
 559                 off := 0
 560                 for _, sb := range sbs {
 561                         data := sb.fn.extents[sb.idx].(*memExtent).buf
 562                         sb.fn.extents[sb.idx] = storedExtent{
 563                                 cache:   dn.cache,
 564                                 locator: locator,
 565                                 size:    size,
 566                                 offset:  off,
 567                                 length:  len(data),
 568                         }
 569                         off += len(data)
 570                 }
 571                 return nil
 572         }
 573
 574         names := make([]string, 0, len(dn.inodes))
 575         for name := range dn.inodes {
 576                 names = append(names, name)
 577         }
 578         sort.Strings(names)
 579
 580         for _, name := range names {
 581                 fn, ok := dn.inodes[name].(*filenode)
 582                 if !ok {
 583                         continue
 584                 }
 585                 fn.Lock()
 586                 defer fn.Unlock()
 587                 for idx, ext := range fn.extents {
 588                         ext, ok := ext.(*memExtent)
 589                         if !ok {
 590                                 continue
 591                         }
 592                         if ext.Len() > maxBlockSize/2 {
 593                                 if err := flush([]shortBlock{{fn, idx}}); err != nil {
 594                                         return err
 595                                 }
 596                                 continue
 597                         }
 598                         if pendingLen+ext.Len() > maxBlockSize {
 599                                 if err := flush(pending); err != nil {
 600                                         return err
 601                                 }
 602                                 pending = nil
 603                                 pendingLen = 0
 604                         }
 605                         pending = append(pending, shortBlock{fn, idx})
 606                         pendingLen += ext.Len()
 607                 }
 608         }
 609         return flush(pending)
 610 }
 611
 612 func (dn *dirnode) MarshalManifest(prefix string) (string, error) {
 613         dn.Lock()
 614         defer dn.Unlock()
 615         if err := dn.sync(); err != nil {
 616                 return "", err
 617         }
 618
 619         var streamLen int64
 620         type m1segment struct {
 621                 name   string
 622                 offset int64
 623                 length int64
 624         }
 625         var segments []m1segment
 626         var subdirs string
 627         var blocks []string
 628
 629         names := make([]string, 0, len(dn.inodes))
 630         for name := range dn.inodes {
 631                 names = append(names, name)
 632         }
 633         sort.Strings(names)
 634
 635         for _, name := range names {
 636                 node := dn.inodes[name]
 637                 switch node := node.(type) {
 638                 case *dirnode:
 639                         subdir, err := node.MarshalManifest(prefix + "/" + node.Name())
 640                         if err != nil {
 641                                 return "", err
 642                         }
 643                         subdirs = subdirs + subdir
 644                 case *filenode:
 645                         for _, e := range node.extents {
 646                                 switch e := e.(type) {
 647                                 case *memExtent:
 648                                         blocks = append(blocks, fmt.Sprintf("FIXME+%d", e.Len()))
 649                                         segments = append(segments, m1segment{
 650                                                 name:   node.Name(),
 651                                                 offset: streamLen,
 652                                                 length: int64(e.Len()),
 653                                         })
 654                                         streamLen += int64(e.Len())
 655                                 case storedExtent:
 656                                         if len(blocks) > 0 && blocks[len(blocks)-1] == e.locator {
 657                                                 streamLen -= int64(e.size)
 658                                         } else {
 659                                                 blocks = append(blocks, e.locator)
 660                                         }
 661                                         segments = append(segments, m1segment{
 662                                                 name:   node.Name(),
 663                                                 offset: streamLen + int64(e.offset),
 664                                                 length: int64(e.length),
 665                                         })
 666                                         streamLen += int64(e.size)
 667                                 default:
 668                                         panic(fmt.Sprintf("can't marshal extent type %T", e))
 669                                 }
 670                         }
 671                 default:
 672                         panic(fmt.Sprintf("can't marshal inode type %T", node))
 673                 }
 674         }
 675         var filetokens []string
 676         for _, s := range segments {
 677                 filetokens = append(filetokens, fmt.Sprintf("%d:%d:%s", s.offset, s.length, s.name))
 678         }
 679         if len(filetokens) == 0 {
 680                 return subdirs, nil
 681         } else if len(blocks) == 0 {
 682                 blocks = []string{"d41d8cd98f00b204e9800998ecf8427e+0"}
 683         }
 684         return prefix + " " + strings.Join(blocks, " ") + " " + strings.Join(filetokens, " ") + "\n" + subdirs, nil
 685 }
 686
 687 func (dn *dirnode) loadManifest(txt string) {
 688         // FIXME: faster
 689         var dirname string
 690         for _, stream := range strings.Split(txt, "\n") {
 691                 var extents []storedExtent
 692                 for i, token := range strings.Split(stream, " ") {
 693                         if i == 0 {
 694                                 dirname = manifestUnescape(token)
 695                                 continue
 696                         }
 697                         if !strings.Contains(token, ":") {
 698                                 toks := strings.SplitN(token, "+", 3)
 699                                 if len(toks) < 2 {
 700                                         // FIXME: broken
 701                                         continue
 702                                 }
 703                                 length, err := strconv.ParseInt(toks[1], 10, 32)
 704                                 if err != nil || length < 0 {
 705                                         // FIXME: broken
 706                                         continue
 707                                 }
 708                                 extents = append(extents, storedExtent{
 709                                         locator: token,
 710                                         size:    int(length),
 711                                         offset:  0,
 712                                         length:  int(length),
 713                                 })
 714                                 continue
 715                         }
 716                         toks := strings.Split(token, ":")
 717                         if len(toks) != 3 {
 718                                 // FIXME: broken manifest
 719                                 continue
 720                         }
 721                         offset, err := strconv.ParseInt(toks[0], 10, 64)
 722                         if err != nil || offset < 0 {
 723                                 // FIXME: broken manifest
 724                                 continue
 725                         }
 726                         length, err := strconv.ParseInt(toks[1], 10, 64)
 727                         if err != nil || length < 0 {
 728                                 // FIXME: broken manifest
 729                                 continue
 730                         }
 731                         name := path.Clean(dirname + "/" + manifestUnescape(toks[2]))
 732                         dn.makeParentDirs(name)
 733                         f, err := dn.OpenFile(name, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0700)
 734                         if err != nil {
 735                                 // FIXME: broken
 736                                 continue
 737                         }
 738                         if f.inode.Stat().IsDir() {
 739                                 f.Close()
 740                                 // FIXME: broken manifest
 741                                 continue
 742                         }
 743                         // Map the stream offset/range coordinates to
 744                         // block/offset/range coordinates and add
 745                         // corresponding storedExtents to the filenode
 746                         var pos int64
 747                         for _, e := range extents {
 748                                 next := pos + int64(e.Len())
 749                                 if next < offset {
 750                                         pos = next
 751                                         continue
 752                                 }
 753                                 if pos > offset+length {
 754                                         break
 755                                 }
 756                                 var blkOff int
 757                                 if pos < offset {
 758                                         blkOff = int(offset - pos)
 759                                 }
 760                                 blkLen := e.Len() - blkOff
 761                                 if pos+int64(blkOff+blkLen) > offset+length {
 762                                         blkLen = int(offset + length - pos - int64(blkOff))
 763                                 }
 764                                 f.inode.(*filenode).appendExtent(storedExtent{
 765                                         cache:   dn.cache,
 766                                         locator: e.locator,
 767                                         size:    e.size,
 768                                         offset:  blkOff,
 769                                         length:  blkLen,
 770                                 })
 771                                 pos = next
 772                         }
 773                         f.Close()
 774                 }
 775         }
 776 }
 777
 778 func (dn *dirnode) makeParentDirs(name string) (err error) {
 779         names := strings.Split(name, "/")
 780         for _, name := range names[:len(names)-1] {
 781                 f, err := dn.mkdir(name)
 782                 if err != nil {
 783                         return err
 784                 }
 785                 defer f.Close()
 786                 var ok bool
 787                 dn, ok = f.inode.(*dirnode)
 788                 if !ok {
 789                         return ErrFileExists
 790                 }
 791         }
 792         return nil
 793 }
 794
 795 func (dn *dirnode) mkdir(name string) (*file, error) {
 796         return dn.OpenFile(name, os.O_CREATE|os.O_EXCL, os.ModeDir|0755)
 797 }
 798
 799 func (dn *dirnode) Mkdir(name string, perm os.FileMode) error {
 800         f, err := dn.mkdir(name)
 801         if err != nil {
 802                 f.Close()
 803         }
 804         return err
 805 }
 806
 807 func (dn *dirnode) Remove(name string) error {
 808         dirname, name := path.Split(name)
 809         if name == "" || name == "." || name == ".." {
 810                 return ErrInvalidOperation
 811         }
 812         dn, ok := dn.lookupPath(dirname).(*dirnode)
 813         if !ok {
 814                 return os.ErrNotExist
 815         }
 816         dn.Lock()
 817         defer dn.Unlock()
 818         switch node := dn.inodes[name].(type) {
 819         case nil:
 820                 return os.ErrNotExist
 821         case *dirnode:
 822                 node.RLock()
 823                 defer node.RUnlock()
 824                 if len(node.inodes) > 0 {
 825                         return ErrDirectoryNotEmpty
 826                 }
 827         }
 828         delete(dn.inodes, name)
 829         return nil
 830 }
 831
 832 func (dn *dirnode) Parent() inode {
 833         dn.RLock()
 834         defer dn.RUnlock()
 835         return dn.parent
 836 }
 837
 838 func (dn *dirnode) Readdir() (fi []os.FileInfo) {
 839         dn.RLock()
 840         defer dn.RUnlock()
 841         fi = make([]os.FileInfo, 0, len(dn.inodes))
 842         for _, inode := range dn.inodes {
 843                 fi = append(fi, inode.Stat())
 844         }
 845         return
 846 }
 847
 848 func (dn *dirnode) Read(p []byte, ptr filenodePtr) (int, filenodePtr, error) {
 849         return 0, ptr, ErrInvalidOperation
 850 }
 851
 852 func (dn *dirnode) Write(p []byte, ptr filenodePtr) (int, filenodePtr, error) {
 853         return 0, ptr, ErrInvalidOperation
 854 }
 855
 856 func (dn *dirnode) Truncate(int64) error {
 857         return ErrInvalidOperation
 858 }
 859
 860 // lookupPath returns the inode for the file/directory with the given
 861 // name (which may contain "/" separators), along with its parent
 862 // node. If no such file/directory exists, the returned node is nil.
 863 func (dn *dirnode) lookupPath(path string) (node inode) {
 864         node = dn
 865         for _, name := range strings.Split(path, "/") {
 866                 dn, ok := node.(*dirnode)
 867                 if !ok {
 868                         return nil
 869                 }
 870                 if name == "." || name == "" {
 871                         continue
 872                 }
 873                 if name == ".." {
 874                         node = node.Parent()
 875                         continue
 876                 }
 877                 dn.RLock()
 878                 node = dn.inodes[name]
 879                 dn.RUnlock()
 880         }
 881         return
 882 }
 883
 884 func (dn *dirnode) OpenFile(name string, flag int, perm os.FileMode) (*file, error) {
 885         dirname, name := path.Split(name)
 886         dn, ok := dn.lookupPath(dirname).(*dirnode)
 887         if !ok {
 888                 return nil, os.ErrNotExist
 889         }
 890         writeMode := flag&(os.O_RDWR|os.O_WRONLY|os.O_CREATE) != 0
 891         if !writeMode {
 892                 // A directory can be opened via "foo/", "foo/.", or
 893                 // "foo/..".
 894                 switch name {
 895                 case ".", "":
 896                         return &file{inode: dn}, nil
 897                 case "..":
 898                         return &file{inode: dn.Parent()}, nil
 899                 }
 900         }
 901         createMode := flag&os.O_CREATE != 0
 902         if createMode {
 903                 dn.Lock()
 904                 defer dn.Unlock()
 905         } else {
 906                 dn.RLock()
 907                 defer dn.RUnlock()
 908         }
 909         n, ok := dn.inodes[name]
 910         if !ok {
 911                 if !createMode {
 912                         return nil, os.ErrNotExist
 913                 }
 914                 if perm.IsDir() {
 915                         n = &dirnode{
 916                                 parent: dn,
 917                                 client: dn.client,
 918                                 kc:     dn.kc,
 919                                 fileinfo: fileinfo{
 920                                         name: name,
 921                                         mode: os.ModeDir | 0755,
 922                                 },
 923                         }
 924                 } else {
 925                         n = &filenode{
 926                                 parent: dn,
 927                                 fileinfo: fileinfo{
 928                                         name: name,
 929                                         mode: 0755,
 930                                 },
 931                         }
 932                 }
 933                 if dn.inodes == nil {
 934                         dn.inodes = make(map[string]inode)
 935                 }
 936                 dn.inodes[name] = n
 937                 dn.fileinfo.size++
 938         } else if flag&os.O_EXCL != 0 {
 939                 return nil, ErrFileExists
 940         }
 941         return &file{
 942                 inode:    n,
 943                 append:   flag&os.O_APPEND != 0,
 944                 writable: flag&(os.O_WRONLY|os.O_RDWR) != 0,
 945         }, nil
 946 }
 947
 948 type extent interface {
 949         io.ReaderAt
 950         Len() int
 951         // Return a new extent with a subsection of the data from this
 952         // one. length<0 means length=Len()-off.
 953         Slice(off int, length int) extent
 954 }
 955
 956 type writableExtent interface {
 957         extent
 958         WriteAt(p []byte, off int)
 959         Truncate(n int)
 960 }
 961
 962 type memExtent struct {
 963         buf []byte
 964 }
 965
 966 func (me *memExtent) Len() int {
 967         return len(me.buf)
 968 }
 969
 970 func (me *memExtent) Slice(off, length int) extent {
 971         if length < 0 {
 972                 length = len(me.buf) - off
 973         }
 974         buf := make([]byte, length)
 975         copy(buf, me.buf[off:])
 976         return &memExtent{buf: buf}
 977 }
 978
 979 func (me *memExtent) Truncate(n int) {
 980         if n > cap(me.buf) {
 981                 newsize := 1024
 982                 for newsize < n {
 983                         newsize = newsize << 2
 984                 }
 985                 newbuf := make([]byte, n, newsize)
 986                 copy(newbuf, me.buf)
 987                 me.buf = newbuf
 988         } else {
 989                 // Zero unused part when shrinking, in case we grow
 990                 // and start using it again later.
 991                 for i := n; i < len(me.buf); i++ {
 992                         me.buf[i] = 0
 993                 }
 994         }
 995         me.buf = me.buf[:n]
 996 }
 997
 998 func (me *memExtent) WriteAt(p []byte, off int) {
 999         if off+len(p) > len(me.buf) {
1000                 panic("overflowed extent")
1001         }
1002         copy(me.buf[off:], p)
1003 }
1004
1005 func (me *memExtent) ReadAt(p []byte, off int64) (n int, err error) {
1006         if off > int64(me.Len()) {
1007                 err = io.EOF
1008                 return
1009         }
1010         n = copy(p, me.buf[int(off):])
1011         if n < len(p) {
1012                 err = io.EOF
1013         }
1014         return
1015 }
1016
1017 type storedExtent struct {
1018         cache   blockCache
1019         locator string
1020         size    int
1021         offset  int
1022         length  int
1023 }
1024
1025 func (se storedExtent) Len() int {
1026         return se.length
1027 }
1028
1029 func (se storedExtent) Slice(n, size int) extent {
1030         se.offset += n
1031         se.length -= n
1032         if size >= 0 && se.length > size {
1033                 se.length = size
1034         }
1035         return se
1036 }
1037
1038 func (se storedExtent) ReadAt(p []byte, off int64) (n int, err error) {
1039         if off > int64(se.length) {
1040                 return 0, io.EOF
1041         }
1042         maxlen := se.length - int(off)
1043         if len(p) > maxlen {
1044                 p = p[:maxlen]
1045                 n, err = se.cache.ReadAt(se.locator, p, int(off)+se.offset)
1046                 if err == nil {
1047                         err = io.EOF
1048                 }
1049                 return
1050         }
1051         return se.cache.ReadAt(se.locator, p, int(off)+se.offset)
1052 }
1053
1054 type blockCache interface {
1055         ReadAt(locator string, p []byte, off int) (n int, err error)
1056 }
1057
1058 type keepBlockCache struct {
1059         kc keepClient
1060 }
1061
1062 var scratch = make([]byte, 2<<26)
1063
1064 func (kbc *keepBlockCache) ReadAt(locator string, p []byte, off int) (int, error) {
1065         return kbc.kc.ReadAt(locator, p, off)
1066 }
1067
1068 func canonicalName(name string) string {
1069         name = path.Clean("/" + name)
1070         if name == "/" || name == "./" {
1071                 name = "."
1072         } else if strings.HasPrefix(name, "/") {
1073                 name = "." + name
1074         }
1075         return name
1076 }
1077
1078 var manifestEscapeSeq = regexp.MustCompile(`\\([0-9]{3}|\\)`)
1079
1080 func manifestUnescapeSeq(seq string) string {
1081         if seq == `\\` {
1082                 return `\`
1083         }
1084         i, err := strconv.ParseUint(seq[1:], 8, 8)
1085         if err != nil {
1086                 // Invalid escape sequence: can't unescape.
1087                 return seq
1088         }
1089         return string([]byte{byte(i)})
1090 }
1091
1092 func manifestUnescape(s string) string {
1093         return manifestEscapeSeq.ReplaceAllStringFunc(s, manifestUnescapeSeq)
1094 }