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.

391 lines
12 KiB

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