You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

354 lines
10 KiB

12 years ago
12 years ago
12 years ago
7 years ago
12 years ago
9 years ago
12 years ago
12 years ago
7 years ago
6 years ago
7 years ago
7 years ago
7 years ago
12 years ago
12 years ago
12 years ago
7 years ago
7 years ago
12 years ago
7 years ago
7 years ago
12 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
7 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
9 years ago
9 years ago
  1. package storage
  2. import (
  3. "errors"
  4. "fmt"
  5. "os"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  8. "github.com/chrislusf/seaweedfs/weed/util"
  9. "math"
  10. )
  11. const (
  12. FlagGzip = 0x01
  13. FlagHasName = 0x02
  14. FlagHasMime = 0x04
  15. FlagHasLastModifiedDate = 0x08
  16. FlagHasTtl = 0x10
  17. FlagHasPairs = 0x20
  18. FlagIsChunkManifest = 0x80
  19. LastModifiedBytesLength = 5
  20. TtlBytesLength = 2
  21. )
  22. func (n *Needle) DiskSize(version Version) int64 {
  23. return getActualSize(n.Size, version)
  24. }
  25. func (n *Needle) Append(w *os.File, version Version) (offset Offset, size uint32, actualSize int64, err error) {
  26. if end, e := w.Seek(0, 2); e == nil {
  27. defer func(w *os.File, off int64) {
  28. if err != nil {
  29. if te := w.Truncate(end); te != nil {
  30. glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", w.Name(), end, te)
  31. }
  32. }
  33. }(w, end)
  34. offset = Offset(end)
  35. } else {
  36. err = fmt.Errorf("Cannot Read Current Volume Position: %v", e)
  37. return
  38. }
  39. switch version {
  40. case Version1:
  41. header := make([]byte, NeedleEntrySize)
  42. CookieToBytes(header[0:CookieSize], n.Cookie)
  43. NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
  44. n.Size = uint32(len(n.Data))
  45. size = n.Size
  46. util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
  47. if _, err = w.Write(header); err != nil {
  48. return
  49. }
  50. if _, err = w.Write(n.Data); err != nil {
  51. return
  52. }
  53. actualSize = NeedleEntrySize + int64(n.Size)
  54. padding := PaddingLength(n.Size, version)
  55. util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
  56. _, err = w.Write(header[0 : NeedleChecksumSize+padding])
  57. return
  58. case Version2, Version3:
  59. header := make([]byte, NeedleEntrySize+TimestampSize) // adding timestamp to reuse it and avoid extra allocation
  60. CookieToBytes(header[0:CookieSize], n.Cookie)
  61. NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
  62. if len(n.Name) >= math.MaxUint8 {
  63. n.NameSize = math.MaxUint8
  64. } else {
  65. n.NameSize = uint8(len(n.Name))
  66. }
  67. n.DataSize, n.MimeSize = uint32(len(n.Data)), uint8(len(n.Mime))
  68. if n.DataSize > 0 {
  69. n.Size = 4 + n.DataSize + 1
  70. if n.HasName() {
  71. n.Size = n.Size + 1 + uint32(n.NameSize)
  72. }
  73. if n.HasMime() {
  74. n.Size = n.Size + 1 + uint32(n.MimeSize)
  75. }
  76. if n.HasLastModifiedDate() {
  77. n.Size = n.Size + LastModifiedBytesLength
  78. }
  79. if n.HasTtl() {
  80. n.Size = n.Size + TtlBytesLength
  81. }
  82. if n.HasPairs() {
  83. n.Size += 2 + uint32(n.PairsSize)
  84. }
  85. } else {
  86. n.Size = 0
  87. }
  88. size = n.DataSize
  89. util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
  90. if _, err = w.Write(header[0:NeedleEntrySize]); err != nil {
  91. return
  92. }
  93. if n.DataSize > 0 {
  94. util.Uint32toBytes(header[0:4], n.DataSize)
  95. if _, err = w.Write(header[0:4]); err != nil {
  96. return
  97. }
  98. if _, err = w.Write(n.Data); err != nil {
  99. return
  100. }
  101. util.Uint8toBytes(header[0:1], n.Flags)
  102. if _, err = w.Write(header[0:1]); err != nil {
  103. return
  104. }
  105. if n.HasName() {
  106. util.Uint8toBytes(header[0:1], n.NameSize)
  107. if _, err = w.Write(header[0:1]); err != nil {
  108. return
  109. }
  110. if _, err = w.Write(n.Name[:n.NameSize]); err != nil {
  111. return
  112. }
  113. }
  114. if n.HasMime() {
  115. util.Uint8toBytes(header[0:1], n.MimeSize)
  116. if _, err = w.Write(header[0:1]); err != nil {
  117. return
  118. }
  119. if _, err = w.Write(n.Mime); err != nil {
  120. return
  121. }
  122. }
  123. if n.HasLastModifiedDate() {
  124. util.Uint64toBytes(header[0:8], n.LastModified)
  125. if _, err = w.Write(header[8-LastModifiedBytesLength : 8]); err != nil {
  126. return
  127. }
  128. }
  129. if n.HasTtl() && n.Ttl != nil {
  130. n.Ttl.ToBytes(header[0:TtlBytesLength])
  131. if _, err = w.Write(header[0:TtlBytesLength]); err != nil {
  132. return
  133. }
  134. }
  135. if n.HasPairs() {
  136. util.Uint16toBytes(header[0:2], n.PairsSize)
  137. if _, err = w.Write(header[0:2]); err != nil {
  138. return
  139. }
  140. if _, err = w.Write(n.Pairs); err != nil {
  141. return
  142. }
  143. }
  144. }
  145. padding := PaddingLength(n.Size, version)
  146. util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
  147. if version == Version2 {
  148. _, err = w.Write(header[0 : NeedleChecksumSize+padding])
  149. } else {
  150. // version3
  151. util.Uint64toBytes(header[NeedleChecksumSize:NeedleChecksumSize+TimestampSize], n.AppendAtNs)
  152. _, err = w.Write(header[0 : NeedleChecksumSize+TimestampSize+padding])
  153. }
  154. return offset, n.DataSize, getActualSize(n.Size, version), err
  155. }
  156. return 0, 0, 0, fmt.Errorf("Unsupported Version! (%d)", version)
  157. }
  158. func ReadNeedleBlob(r *os.File, offset int64, size uint32, version Version) (dataSlice []byte, err error) {
  159. dataSlice = make([]byte, int(getActualSize(size, version)))
  160. _, err = r.ReadAt(dataSlice, offset)
  161. return dataSlice, err
  162. }
  163. func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version) (err error) {
  164. bytes, err := ReadNeedleBlob(r, offset, size, version)
  165. if err != nil {
  166. return err
  167. }
  168. n.ParseNeedleHeader(bytes)
  169. if n.Size != size {
  170. return fmt.Errorf("File Entry Not Found. Needle id %d expected size %d Memory %d", n.Id, n.Size, size)
  171. }
  172. switch version {
  173. case Version1:
  174. n.Data = bytes[NeedleEntrySize : NeedleEntrySize+size]
  175. case Version2, Version3:
  176. n.readNeedleDataVersion2(bytes[NeedleEntrySize : NeedleEntrySize+int(n.Size)])
  177. }
  178. if size == 0 {
  179. return nil
  180. }
  181. checksum := util.BytesToUint32(bytes[NeedleEntrySize+size : NeedleEntrySize+size+NeedleChecksumSize])
  182. newChecksum := NewCRC(n.Data)
  183. if checksum != newChecksum.Value() {
  184. return errors.New("CRC error! Data On Disk Corrupted")
  185. }
  186. n.Checksum = newChecksum
  187. if version == Version3 {
  188. tsOffset := NeedleEntrySize + size + NeedleChecksumSize
  189. n.AppendAtNs = util.BytesToUint64(bytes[tsOffset : tsOffset+TimestampSize])
  190. }
  191. return nil
  192. }
  193. func (n *Needle) ParseNeedleHeader(bytes []byte) {
  194. n.Cookie = BytesToCookie(bytes[0:CookieSize])
  195. n.Id = BytesToNeedleId(bytes[CookieSize : CookieSize+NeedleIdSize])
  196. n.Size = util.BytesToUint32(bytes[CookieSize+NeedleIdSize : NeedleEntrySize])
  197. }
  198. func (n *Needle) readNeedleDataVersion2(bytes []byte) {
  199. index, lenBytes := 0, len(bytes)
  200. if index < lenBytes {
  201. n.DataSize = util.BytesToUint32(bytes[index : index+4])
  202. index = index + 4
  203. if int(n.DataSize)+index > lenBytes {
  204. // this if clause is due to bug #87 and #93, fixed in v0.69
  205. // remove this clause later
  206. return
  207. }
  208. n.Data = bytes[index : index+int(n.DataSize)]
  209. index = index + int(n.DataSize)
  210. n.Flags = bytes[index]
  211. index = index + 1
  212. }
  213. if index < lenBytes && n.HasName() {
  214. n.NameSize = uint8(bytes[index])
  215. index = index + 1
  216. n.Name = bytes[index : index+int(n.NameSize)]
  217. index = index + int(n.NameSize)
  218. }
  219. if index < lenBytes && n.HasMime() {
  220. n.MimeSize = uint8(bytes[index])
  221. index = index + 1
  222. n.Mime = bytes[index : index+int(n.MimeSize)]
  223. index = index + int(n.MimeSize)
  224. }
  225. if index < lenBytes && n.HasLastModifiedDate() {
  226. n.LastModified = util.BytesToUint64(bytes[index : index+LastModifiedBytesLength])
  227. index = index + LastModifiedBytesLength
  228. }
  229. if index < lenBytes && n.HasTtl() {
  230. n.Ttl = LoadTTLFromBytes(bytes[index : index+TtlBytesLength])
  231. index = index + TtlBytesLength
  232. }
  233. if index < lenBytes && n.HasPairs() {
  234. n.PairsSize = util.BytesToUint16(bytes[index : index+2])
  235. index += 2
  236. end := index + int(n.PairsSize)
  237. n.Pairs = bytes[index:end]
  238. index = end
  239. }
  240. }
  241. func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bodyLength int64, err error) {
  242. n = new(Needle)
  243. if version == Version1 || version == Version2 || version == Version3 {
  244. bytes := make([]byte, NeedleEntrySize)
  245. var count int
  246. count, err = r.ReadAt(bytes, offset)
  247. if count <= 0 || err != nil {
  248. return nil, 0, err
  249. }
  250. n.ParseNeedleHeader(bytes)
  251. bodyLength = NeedleBodyLength(n.Size, version)
  252. }
  253. return
  254. }
  255. func PaddingLength(needleSize uint32, version Version) uint32 {
  256. if version == Version3 {
  257. // this is same value as version2, but just listed here for clarity
  258. return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize + TimestampSize) % NeedlePaddingSize)
  259. }
  260. return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize) % NeedlePaddingSize)
  261. }
  262. func NeedleBodyLength(needleSize uint32, version Version) int64 {
  263. if version == Version3 {
  264. return int64(needleSize) + NeedleChecksumSize + TimestampSize + int64(PaddingLength(needleSize, version))
  265. }
  266. return int64(needleSize) + NeedleChecksumSize + int64(PaddingLength(needleSize, version))
  267. }
  268. //n should be a needle already read the header
  269. //the input stream will read until next file entry
  270. func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength int64) (err error) {
  271. if bodyLength <= 0 {
  272. return nil
  273. }
  274. switch version {
  275. case Version1:
  276. bytes := make([]byte, bodyLength)
  277. if _, err = r.ReadAt(bytes, offset); err != nil {
  278. return
  279. }
  280. n.Data = bytes[:n.Size]
  281. n.Checksum = NewCRC(n.Data)
  282. case Version2, Version3:
  283. bytes := make([]byte, bodyLength)
  284. if _, err = r.ReadAt(bytes, offset); err != nil {
  285. return
  286. }
  287. n.readNeedleDataVersion2(bytes[0:n.Size])
  288. n.Checksum = NewCRC(n.Data)
  289. default:
  290. err = fmt.Errorf("Unsupported Version! (%d)", version)
  291. }
  292. return
  293. }
  294. func (n *Needle) IsGzipped() bool {
  295. return n.Flags&FlagGzip > 0
  296. }
  297. func (n *Needle) SetGzipped() {
  298. n.Flags = n.Flags | FlagGzip
  299. }
  300. func (n *Needle) HasName() bool {
  301. return n.Flags&FlagHasName > 0
  302. }
  303. func (n *Needle) SetHasName() {
  304. n.Flags = n.Flags | FlagHasName
  305. }
  306. func (n *Needle) HasMime() bool {
  307. return n.Flags&FlagHasMime > 0
  308. }
  309. func (n *Needle) SetHasMime() {
  310. n.Flags = n.Flags | FlagHasMime
  311. }
  312. func (n *Needle) HasLastModifiedDate() bool {
  313. return n.Flags&FlagHasLastModifiedDate > 0
  314. }
  315. func (n *Needle) SetHasLastModifiedDate() {
  316. n.Flags = n.Flags | FlagHasLastModifiedDate
  317. }
  318. func (n *Needle) HasTtl() bool {
  319. return n.Flags&FlagHasTtl > 0
  320. }
  321. func (n *Needle) SetHasTtl() {
  322. n.Flags = n.Flags | FlagHasTtl
  323. }
  324. func (n *Needle) IsChunkedManifest() bool {
  325. return n.Flags&FlagIsChunkManifest > 0
  326. }
  327. func (n *Needle) SetIsChunkManifest() {
  328. n.Flags = n.Flags | FlagIsChunkManifest
  329. }
  330. func (n *Needle) HasPairs() bool {
  331. return n.Flags&FlagHasPairs != 0
  332. }
  333. func (n *Needle) SetHasPairs() {
  334. n.Flags = n.Flags | FlagHasPairs
  335. }