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.

291 lines
8.2 KiB

12 years ago
12 years ago
12 years ago
12 years ago
9 years ago
10 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 years ago
12 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. "io"
  6. "os"
  7. "github.com/chrislusf/seaweedfs/go/glog"
  8. "github.com/chrislusf/seaweedfs/go/util"
  9. )
  10. const (
  11. FlagGzip = 0x01
  12. FlagHasName = 0x02
  13. FlagHasMime = 0x04
  14. FlagHasLastModifiedDate = 0x08
  15. FlagHasTtl = 0x10
  16. FlagIsChunkManifest = 0x80
  17. LastModifiedBytesLength = 5
  18. TtlBytesLength = 2
  19. )
  20. func (n *Needle) DiskSize() int64 {
  21. padding := NeedlePaddingSize - ((NeedleHeaderSize + int64(n.Size) + NeedleChecksumSize) % NeedlePaddingSize)
  22. return NeedleHeaderSize + int64(n.Size) + padding + NeedleChecksumSize
  23. }
  24. func (n *Needle) Append(w io.Writer, version Version) (size uint32, err error) {
  25. if s, ok := w.(io.Seeker); ok {
  26. if end, e := s.Seek(0, 1); e == nil {
  27. defer func(s io.Seeker, off int64) {
  28. if err != nil {
  29. if _, e = s.Seek(off, 0); e != nil {
  30. glog.V(0).Infof("Failed to seek %s back to %d with error: %v", w, off, e)
  31. }
  32. }
  33. }(s, end)
  34. } else {
  35. err = fmt.Errorf("Cannot Read Current Volume Position: %v", e)
  36. return
  37. }
  38. }
  39. switch version {
  40. case Version1:
  41. header := make([]byte, NeedleHeaderSize)
  42. util.Uint32toBytes(header[0:4], n.Cookie)
  43. util.Uint64toBytes(header[4:12], n.Id)
  44. n.Size = uint32(len(n.Data))
  45. size = n.Size
  46. util.Uint32toBytes(header[12:16], 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. padding := NeedlePaddingSize - ((NeedleHeaderSize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
  54. util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
  55. _, err = w.Write(header[0 : NeedleChecksumSize+padding])
  56. return
  57. case Version2:
  58. header := make([]byte, NeedleHeaderSize)
  59. util.Uint32toBytes(header[0:4], n.Cookie)
  60. util.Uint64toBytes(header[4:12], n.Id)
  61. n.DataSize, n.NameSize, n.MimeSize = uint32(len(n.Data)), uint8(len(n.Name)), uint8(len(n.Mime))
  62. if n.DataSize > 0 {
  63. n.Size = 4 + n.DataSize + 1
  64. if n.HasName() {
  65. n.Size = n.Size + 1 + uint32(n.NameSize)
  66. }
  67. if n.HasMime() {
  68. n.Size = n.Size + 1 + uint32(n.MimeSize)
  69. }
  70. if n.HasLastModifiedDate() {
  71. n.Size = n.Size + LastModifiedBytesLength
  72. }
  73. if n.HasTtl() {
  74. n.Size = n.Size + TtlBytesLength
  75. }
  76. } else {
  77. n.Size = 0
  78. }
  79. size = n.DataSize
  80. util.Uint32toBytes(header[12:16], n.Size)
  81. if _, err = w.Write(header); err != nil {
  82. return
  83. }
  84. if n.DataSize > 0 {
  85. util.Uint32toBytes(header[0:4], n.DataSize)
  86. if _, err = w.Write(header[0:4]); err != nil {
  87. return
  88. }
  89. if _, err = w.Write(n.Data); err != nil {
  90. return
  91. }
  92. util.Uint8toBytes(header[0:1], n.Flags)
  93. if _, err = w.Write(header[0:1]); err != nil {
  94. return
  95. }
  96. if n.HasName() {
  97. util.Uint8toBytes(header[0:1], n.NameSize)
  98. if _, err = w.Write(header[0:1]); err != nil {
  99. return
  100. }
  101. if _, err = w.Write(n.Name); err != nil {
  102. return
  103. }
  104. }
  105. if n.HasMime() {
  106. util.Uint8toBytes(header[0:1], n.MimeSize)
  107. if _, err = w.Write(header[0:1]); err != nil {
  108. return
  109. }
  110. if _, err = w.Write(n.Mime); err != nil {
  111. return
  112. }
  113. }
  114. if n.HasLastModifiedDate() {
  115. util.Uint64toBytes(header[0:8], n.LastModified)
  116. if _, err = w.Write(header[8-LastModifiedBytesLength : 8]); err != nil {
  117. return
  118. }
  119. }
  120. if n.HasTtl() && n.Ttl != nil {
  121. n.Ttl.ToBytes(header[0:TtlBytesLength])
  122. if _, err = w.Write(header[0:TtlBytesLength]); err != nil {
  123. return
  124. }
  125. }
  126. }
  127. padding := NeedlePaddingSize - ((NeedleHeaderSize + n.Size + NeedleChecksumSize) % NeedlePaddingSize)
  128. util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
  129. _, err = w.Write(header[0 : NeedleChecksumSize+padding])
  130. return n.DataSize, err
  131. }
  132. return 0, fmt.Errorf("Unsupported Version! (%d)", version)
  133. }
  134. func ReadNeedleBlob(r *os.File, offset int64, size uint32) (bytes []byte, err error) {
  135. padding := NeedlePaddingSize - ((NeedleHeaderSize + size + NeedleChecksumSize) % NeedlePaddingSize)
  136. bytes = make([]byte, NeedleHeaderSize+size+NeedleChecksumSize+padding)
  137. _, err = r.ReadAt(bytes, offset)
  138. return
  139. }
  140. func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version) (err error) {
  141. bytes, err := ReadNeedleBlob(r, offset, size)
  142. if err != nil {
  143. return err
  144. }
  145. n.ParseNeedleHeader(bytes)
  146. if n.Size != size {
  147. return fmt.Errorf("File Entry Not Found. Needle %d Memory %d", n.Size, size)
  148. }
  149. switch version {
  150. case Version1:
  151. n.Data = bytes[NeedleHeaderSize : NeedleHeaderSize+size]
  152. case Version2:
  153. n.readNeedleDataVersion2(bytes[NeedleHeaderSize : NeedleHeaderSize+int(n.Size)])
  154. }
  155. checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize])
  156. newChecksum := NewCRC(n.Data)
  157. if checksum != newChecksum.Value() {
  158. return errors.New("CRC error! Data On Disk Corrupted")
  159. }
  160. n.Checksum = newChecksum
  161. return nil
  162. }
  163. func (n *Needle) ParseNeedleHeader(bytes []byte) {
  164. n.Cookie = util.BytesToUint32(bytes[0:4])
  165. n.Id = util.BytesToUint64(bytes[4:12])
  166. n.Size = util.BytesToUint32(bytes[12:NeedleHeaderSize])
  167. }
  168. func (n *Needle) readNeedleDataVersion2(bytes []byte) {
  169. index, lenBytes := 0, len(bytes)
  170. if index < lenBytes {
  171. n.DataSize = util.BytesToUint32(bytes[index : index+4])
  172. index = index + 4
  173. if int(n.DataSize)+index > lenBytes {
  174. // this if clause is due to bug #87 and #93, fixed in v0.69
  175. // remove this clause later
  176. return
  177. }
  178. n.Data = bytes[index : index+int(n.DataSize)]
  179. index = index + int(n.DataSize)
  180. n.Flags = bytes[index]
  181. index = index + 1
  182. }
  183. if index < lenBytes && n.HasName() {
  184. n.NameSize = uint8(bytes[index])
  185. index = index + 1
  186. n.Name = bytes[index : index+int(n.NameSize)]
  187. index = index + int(n.NameSize)
  188. }
  189. if index < lenBytes && n.HasMime() {
  190. n.MimeSize = uint8(bytes[index])
  191. index = index + 1
  192. n.Mime = bytes[index : index+int(n.MimeSize)]
  193. index = index + int(n.MimeSize)
  194. }
  195. if index < lenBytes && n.HasLastModifiedDate() {
  196. n.LastModified = util.BytesToUint64(bytes[index : index+LastModifiedBytesLength])
  197. index = index + LastModifiedBytesLength
  198. }
  199. if index < lenBytes && n.HasTtl() {
  200. n.Ttl = LoadTTLFromBytes(bytes[index : index+TtlBytesLength])
  201. index = index + TtlBytesLength
  202. }
  203. }
  204. func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bodyLength uint32, err error) {
  205. n = new(Needle)
  206. if version == Version1 || version == Version2 {
  207. bytes := make([]byte, NeedleHeaderSize)
  208. var count int
  209. count, err = r.ReadAt(bytes, offset)
  210. if count <= 0 || err != nil {
  211. return nil, 0, err
  212. }
  213. n.ParseNeedleHeader(bytes)
  214. padding := NeedlePaddingSize - ((n.Size + NeedleHeaderSize + NeedleChecksumSize) % NeedlePaddingSize)
  215. bodyLength = n.Size + NeedleChecksumSize + padding
  216. }
  217. return
  218. }
  219. //n should be a needle already read the header
  220. //the input stream will read until next file entry
  221. func (n *Needle) ReadNeedleBody(r *os.File, version Version, offset int64, bodyLength uint32) (err error) {
  222. if bodyLength <= 0 {
  223. return nil
  224. }
  225. switch version {
  226. case Version1:
  227. bytes := make([]byte, bodyLength)
  228. if _, err = r.ReadAt(bytes, offset); err != nil {
  229. return
  230. }
  231. n.Data = bytes[:n.Size]
  232. n.Checksum = NewCRC(n.Data)
  233. case Version2:
  234. bytes := make([]byte, bodyLength)
  235. if _, err = r.ReadAt(bytes, offset); err != nil {
  236. return
  237. }
  238. n.readNeedleDataVersion2(bytes[0:n.Size])
  239. n.Checksum = NewCRC(n.Data)
  240. default:
  241. err = fmt.Errorf("Unsupported Version! (%d)", version)
  242. }
  243. return
  244. }
  245. func (n *Needle) IsGzipped() bool {
  246. return n.Flags&FlagGzip > 0
  247. }
  248. func (n *Needle) SetGzipped() {
  249. n.Flags = n.Flags | FlagGzip
  250. }
  251. func (n *Needle) HasName() bool {
  252. return n.Flags&FlagHasName > 0
  253. }
  254. func (n *Needle) SetHasName() {
  255. n.Flags = n.Flags | FlagHasName
  256. }
  257. func (n *Needle) HasMime() bool {
  258. return n.Flags&FlagHasMime > 0
  259. }
  260. func (n *Needle) SetHasMime() {
  261. n.Flags = n.Flags | FlagHasMime
  262. }
  263. func (n *Needle) HasLastModifiedDate() bool {
  264. return n.Flags&FlagHasLastModifiedDate > 0
  265. }
  266. func (n *Needle) SetHasLastModifiedDate() {
  267. n.Flags = n.Flags | FlagHasLastModifiedDate
  268. }
  269. func (n *Needle) HasTtl() bool {
  270. return n.Flags&FlagHasTtl > 0
  271. }
  272. func (n *Needle) SetHasTtl() {
  273. n.Flags = n.Flags | FlagHasTtl
  274. }
  275. func (n *Needle) IsChunkedManifest() bool {
  276. return n.Flags&FlagIsChunkManifest > 0
  277. }
  278. func (n *Needle) SetIsChunkManifest() {
  279. n.Flags = n.Flags | FlagIsChunkManifest
  280. }