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.

282 lines
8.0 KiB

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