kraken

1
// Copyright (c) 2016-2019 Uber Technologies, Inc.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
package originstorage
15

16
import (
17
	"errors"
18
	"fmt"
19

20
	"github.com/uber/kraken/core"
21
	"github.com/uber/kraken/lib/store"
22
	"github.com/uber/kraken/lib/torrent/storage"
23
	"github.com/uber/kraken/lib/torrent/storage/piecereader"
24

25
	"github.com/willf/bitset"
26
	"go.uber.org/atomic"
27
)
28

29
// Torrent errors.
30
var (
31
	ErrReadOnly = errors.New("Read-only torrent is being written to")
32
)
33

34
// Torrent is a read-only storage.Torrent. It allows concurrent reads on all
35
// pieces.
36
type Torrent struct {
37
	metaInfo    *core.MetaInfo
38
	cas         *store.CAStore
39
	numComplete *atomic.Int32
40
}
41

42
// NewTorrent creates a new Torrent.
43
func NewTorrent(cas *store.CAStore, mi *core.MetaInfo) (*Torrent, error) {
44
	return &Torrent{
45
		cas:         cas,
46
		metaInfo:    mi,
47
		numComplete: atomic.NewInt32(int32(mi.NumPieces())),
48
	}, nil
49
}
50

51
// Digest returns the digest of the target blob.
52
func (t *Torrent) Digest() core.Digest {
53
	return t.metaInfo.Digest()
54
}
55

56
// Stat returns the TorrentInfo for t.
57
func (t *Torrent) Stat() *storage.TorrentInfo {
58
	return storage.NewTorrentInfo(t.metaInfo, t.Bitfield())
59
}
60

61
// InfoHash returns the torrent metainfo hash.
62
func (t *Torrent) InfoHash() core.InfoHash {
63
	return t.metaInfo.InfoHash()
64
}
65

66
// NumPieces returns the number of pieces in the torrent.
67
func (t *Torrent) NumPieces() int {
68
	return t.metaInfo.NumPieces()
69
}
70

71
// Length returns the length of the target file.
72
func (t *Torrent) Length() int64 {
73
	return t.metaInfo.Length()
74
}
75

76
// PieceLength returns the length of piece pi.
77
func (t *Torrent) PieceLength(pi int) int64 {
78
	return t.metaInfo.GetPieceLength(pi)
79
}
80

81
// MaxPieceLength returns the longest piece length of the torrent.
82
func (t *Torrent) MaxPieceLength() int64 {
83
	return t.PieceLength(0)
84
}
85

86
// Complete is always true.
87
func (t *Torrent) Complete() bool {
88
	return true
89
}
90

91
// BytesDownloaded always returns the total number of bytes.
92
func (t *Torrent) BytesDownloaded() int64 {
93
	return t.metaInfo.Length()
94
}
95

96
// WritePiece returns error, since Torrent is read-only.
97
func (t *Torrent) WritePiece(src storage.PieceReader, pi int) error {
98
	return ErrReadOnly
99
}
100

101
// Bitfield always returns a completed bitfield.
102
func (t *Torrent) Bitfield() *bitset.BitSet {
103
	return bitset.New(uint(t.NumPieces())).Complement()
104
}
105

106
func (t *Torrent) String() string {
107
	downloaded := int(float64(t.BytesDownloaded()) / float64(t.metaInfo.Length()) * 100)
108
	return fmt.Sprintf("torrent(hash=%s, downloaded=%d%%)", t.InfoHash().Hex(), downloaded)
109
}
110

111
type opener struct {
112
	torrent *Torrent
113
}
114

115
func (o *opener) Open() (store.FileReader, error) {
116
	return o.torrent.cas.GetCacheFileReader(o.torrent.Digest().Hex())
117
}
118

119
// GetPieceReader returns a reader for piece pi.
120
func (t *Torrent) GetPieceReader(pi int) (storage.PieceReader, error) {
121
	if pi >= t.NumPieces() {
122
		return nil, fmt.Errorf("invalid piece index %d: num pieces = %d", pi, t.NumPieces())
123
	}
124
	return piecereader.NewFileReader(t.getFileOffset(pi), t.PieceLength(pi), &opener{t}), nil
125
}
126

127
// HasPiece returns if piece pi is complete.
128
// For Torrent it's always true.
129
func (t *Torrent) HasPiece(pi int) bool {
130
	return true
131
}
132

133
// MissingPieces always returns empty list.
134
func (t *Torrent) MissingPieces() []int {
135
	return []int{}
136
}
137

138
// getFileOffset calculates the offset in the torrent file given piece index.
139
// Assumes pi is a valid piece index.
140
func (t *Torrent) getFileOffset(pi int) int64 {
141
	return t.metaInfo.PieceLength() * int64(pi)
142
}
143