Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

fix needle map entry size

pull/934/head
Chris Lu 6 years ago
parent
e5506152c0
commit
ac2727853f
18 changed files with 195 additions and 184 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      unmaintained/fix_dat/fix_dat.go
  2. 4
      weed/command/backup.go
  3. 5
      weed/pb/volume_server.proto
  4. 188
      weed/pb/volume_server_pb/volume_server.pb.go
  5. 19
      weed/server/volume_grpc_copy.go
  6. 4
      weed/server/volume_grpc_copy_incremental.go
  7. 3
      weed/storage/disk_location.go
  8. 26
      weed/storage/needle/needle_read_write.go
  9. 4
      weed/storage/needle_map/compact_map_perf_test.go
  10. 6
      weed/storage/needle_map_memory.go
  11. 12
      weed/storage/needle_map_metric.go
  12. 3
      weed/storage/types/needle_types.go
  13. 31
      weed/storage/volume.go
  14. 26
      weed/storage/volume_backup.go
  15. 6
      weed/storage/volume_checking.go
  16. 2
      weed/storage/volume_loading.go
  17. 14
      weed/storage/volume_read_write.go
  18. 2
      weed/storage/volume_vacuum.go

4
unmaintained/fix_dat/fix_dat.go
View File

@ -115,7 +115,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *needle.Needle
fmt.Println("Recovered in f", r) fmt.Println("Recovered in f", r)
} }
}() }()
if _, err = n.ReadNeedleBody(datFile, version, offset+int64(types.NeedleEntrySize), rest); err != nil {
if _, err = n.ReadNeedleBody(datFile, version, offset+int64(types.NeedleHeaderSize), rest); err != nil {
fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err) fmt.Printf("cannot read needle body: offset %d body %d %v\n", offset, rest, err)
} }
}() }()
@ -125,7 +125,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *needle.Needle
} }
visitNeedle(n, offset) visitNeedle(n, offset)
offset += types.NeedleEntrySize + rest
offset += types.NeedleHeaderSize + rest
//fmt.Printf("==> new entry offset %d\n", offset) //fmt.Printf("==> new entry offset %d\n", offset)
if n, _, rest, err = needle.ReadNeedleHeader(datFile, version, offset); err != nil { if n, _, rest, err = needle.ReadNeedleHeader(datFile, version, offset); err != nil {
if err == io.EOF { if err == io.EOF {

4
weed/command/backup.go
View File

@ -103,7 +103,9 @@ func runBackup(cmd *Command, args []string) bool {
v.DataFile().WriteAt(v.SuperBlock.Bytes(), 0) v.DataFile().WriteAt(v.SuperBlock.Bytes(), 0)
} }
if uint64(v.Size()) > stats.TailOffset {
datSize, _, _ := v.FileStat()
if datSize > stats.TailOffset {
// remove the old data // remove the old data
v.Destroy() v.Destroy()
// recreate an empty volume // recreate an empty volume

5
weed/pb/volume_server.proto
View File

@ -155,6 +155,7 @@ message VolumeCopyRequest {
string source_data_node = 5; string source_data_node = 5;
} }
message VolumeCopyResponse { message VolumeCopyResponse {
uint64 last_append_at_ns = 1;
} }
message CopyFileRequest { message CopyFileRequest {
@ -182,9 +183,9 @@ message ReadVolumeFileStatusRequest {
} }
message ReadVolumeFileStatusResponse { message ReadVolumeFileStatusResponse {
uint32 volume_id = 1; uint32 volume_id = 1;
uint64 idx_file_timestamp = 2;
uint64 idx_file_timestamp_seconds = 2;
uint64 idx_file_size = 3; uint64 idx_file_size = 3;
uint64 dat_file_timestamp = 4;
uint64 dat_file_timestamp_seconds = 4;
uint64 dat_file_size = 5; uint64 dat_file_size = 5;
uint64 file_count = 6; uint64 file_count = 6;
} }

188
weed/pb/volume_server_pb/volume_server.pb.go
View File

@ -581,6 +581,7 @@ func (m *VolumeCopyRequest) GetSourceDataNode() string {
} }
type VolumeCopyResponse struct { type VolumeCopyResponse struct {
LastAppendAtNs uint64 `protobuf:"varint,1,opt,name=last_append_at_ns,json=lastAppendAtNs" json:"last_append_at_ns,omitempty"`
} }
func (m *VolumeCopyResponse) Reset() { *m = VolumeCopyResponse{} } func (m *VolumeCopyResponse) Reset() { *m = VolumeCopyResponse{} }
@ -588,6 +589,13 @@ func (m *VolumeCopyResponse) String() string { return proto.CompactTe
func (*VolumeCopyResponse) ProtoMessage() {} func (*VolumeCopyResponse) ProtoMessage() {}
func (*VolumeCopyResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{27} } func (*VolumeCopyResponse) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{27} }
func (m *VolumeCopyResponse) GetLastAppendAtNs() uint64 {
if m != nil {
return m.LastAppendAtNs
}
return 0
}
type CopyFileRequest struct { type CopyFileRequest struct {
VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"`
IsIdxFile bool `protobuf:"varint,2,opt,name=is_idx_file,json=isIdxFile" json:"is_idx_file,omitempty"` IsIdxFile bool `protobuf:"varint,2,opt,name=is_idx_file,json=isIdxFile" json:"is_idx_file,omitempty"`
@ -718,9 +726,9 @@ func (m *ReadVolumeFileStatusRequest) GetVolumeId() uint32 {
type ReadVolumeFileStatusResponse struct { type ReadVolumeFileStatusResponse struct {
VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"` VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,omitempty"`
IdxFileTimestamp uint64 `protobuf:"varint,2,opt,name=idx_file_timestamp,json=idxFileTimestamp" json:"idx_file_timestamp,omitempty"`
IdxFileTimestampSeconds uint64 `protobuf:"varint,2,opt,name=idx_file_timestamp_seconds,json=idxFileTimestampSeconds" json:"idx_file_timestamp_seconds,omitempty"`
IdxFileSize uint64 `protobuf:"varint,3,opt,name=idx_file_size,json=idxFileSize" json:"idx_file_size,omitempty"` IdxFileSize uint64 `protobuf:"varint,3,opt,name=idx_file_size,json=idxFileSize" json:"idx_file_size,omitempty"`
DatFileTimestamp uint64 `protobuf:"varint,4,opt,name=dat_file_timestamp,json=datFileTimestamp" json:"dat_file_timestamp,omitempty"`
DatFileTimestampSeconds uint64 `protobuf:"varint,4,opt,name=dat_file_timestamp_seconds,json=datFileTimestampSeconds" json:"dat_file_timestamp_seconds,omitempty"`
DatFileSize uint64 `protobuf:"varint,5,opt,name=dat_file_size,json=datFileSize" json:"dat_file_size,omitempty"` DatFileSize uint64 `protobuf:"varint,5,opt,name=dat_file_size,json=datFileSize" json:"dat_file_size,omitempty"`
FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount" json:"file_count,omitempty"` FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount" json:"file_count,omitempty"`
} }
@ -737,9 +745,9 @@ func (m *ReadVolumeFileStatusResponse) GetVolumeId() uint32 {
return 0 return 0
} }
func (m *ReadVolumeFileStatusResponse) GetIdxFileTimestamp() uint64 {
func (m *ReadVolumeFileStatusResponse) GetIdxFileTimestampSeconds() uint64 {
if m != nil { if m != nil {
return m.IdxFileTimestamp
return m.IdxFileTimestampSeconds
} }
return 0 return 0
} }
@ -751,9 +759,9 @@ func (m *ReadVolumeFileStatusResponse) GetIdxFileSize() uint64 {
return 0 return 0
} }
func (m *ReadVolumeFileStatusResponse) GetDatFileTimestamp() uint64 {
func (m *ReadVolumeFileStatusResponse) GetDatFileTimestampSeconds() uint64 {
if m != nil { if m != nil {
return m.DatFileTimestamp
return m.DatFileTimestampSeconds
} }
return 0 return 0
} }
@ -1571,87 +1579,89 @@ var _VolumeServer_serviceDesc = grpc.ServiceDesc{
func init() { proto.RegisterFile("volume_server.proto", fileDescriptor0) } func init() { proto.RegisterFile("volume_server.proto", fileDescriptor0) }
var fileDescriptor0 = []byte{ var fileDescriptor0 = []byte{
// 1305 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xa4, 0x58, 0x5b, 0x6f, 0xd4, 0xc6,
0x17, 0x8f, 0xd9, 0xcd, 0xed, 0xec, 0x2e, 0x2c, 0x93, 0x40, 0x16, 0x73, 0xf9, 0x2f, 0x03, 0x7f,
0x58, 0x2e, 0x0d, 0x2d, 0xa8, 0x2d, 0xed, 0x53, 0x4b, 0xd2, 0xaa, 0x91, 0x0a, 0x48, 0x0e, 0xa0,
0x56, 0x54, 0xb2, 0x26, 0xf6, 0x24, 0x19, 0xc5, 0x6b, 0x2f, 0x9e, 0x31, 0x25, 0x55, 0xd5, 0x2f,
0x53, 0xf5, 0xa5, 0x0f, 0xfd, 0x82, 0x95, 0xaa, 0x6a, 0x2e, 0x76, 0x7c, 0xcd, 0x9a, 0xf2, 0x36,
0x3e, 0x73, 0xce, 0xef, 0x5c, 0xe6, 0xcc, 0x99, 0xdf, 0x2e, 0xac, 0xbd, 0x8d, 0x82, 0x64, 0x4a,
0x5d, 0x4e, 0xe3, 0xb7, 0x34, 0xde, 0x9c, 0xc5, 0x91, 0x88, 0xd0, 0xb0, 0x20, 0x74, 0x67, 0x7b,
0xf8, 0x01, 0xa0, 0x27, 0x44, 0x78, 0x87, 0xdb, 0x34, 0xa0, 0x82, 0x3a, 0xf4, 0x4d, 0x42, 0xb9,
0x40, 0x97, 0x60, 0x65, 0x9f, 0x05, 0xd4, 0x65, 0x3e, 0x1f, 0x59, 0xe3, 0xce, 0x64, 0xd5, 0x59,
0x96, 0xdf, 0x3b, 0x3e, 0xc7, 0xcf, 0x61, 0xad, 0x60, 0xc0, 0x67, 0x51, 0xc8, 0x29, 0x7a, 0x0c,
0xcb, 0x31, 0xe5, 0x49, 0x20, 0xb4, 0x41, 0xef, 0xe1, 0xb5, 0xcd, 0xb2, 0xaf, 0xcd, 0xcc, 0x24,
0x09, 0x84, 0x93, 0xaa, 0x63, 0x06, 0xfd, 0xfc, 0x06, 0xda, 0x80, 0x65, 0xe3, 0x7b, 0x64, 0x8d,
0xad, 0xc9, 0xaa, 0xb3, 0xa4, 0x5d, 0xa3, 0x8b, 0xb0, 0xc4, 0x05, 0x11, 0x09, 0x1f, 0x9d, 0x19,
0x5b, 0x93, 0x45, 0xc7, 0x7c, 0xa1, 0x75, 0x58, 0xa4, 0x71, 0x1c, 0xc5, 0xa3, 0x8e, 0x52, 0xd7,
0x1f, 0x08, 0x41, 0x97, 0xb3, 0x5f, 0xe8, 0xa8, 0x3b, 0xb6, 0x26, 0x03, 0x47, 0xad, 0xf1, 0x32,
0x2c, 0x7e, 0x33, 0x9d, 0x89, 0x63, 0xfc, 0x39, 0x8c, 0x5e, 0x11, 0x2f, 0x49, 0xa6, 0xaf, 0x54,
0x8c, 0x5b, 0x87, 0xd4, 0x3b, 0x4a, 0x73, 0xbf, 0x0c, 0xab, 0x26, 0x72, 0x13, 0xc1, 0xc0, 0x59,
0xd1, 0x82, 0x1d, 0x1f, 0x7f, 0x05, 0x97, 0x6a, 0x0c, 0x4d, 0x0d, 0x6e, 0xc0, 0xe0, 0x80, 0xc4,
0x7b, 0xe4, 0x80, 0xba, 0x31, 0x11, 0x2c, 0x52, 0xd6, 0x96, 0xd3, 0x37, 0x42, 0x47, 0xca, 0xf0,
0x6b, 0xb0, 0x0b, 0x08, 0xd1, 0x74, 0x46, 0x3c, 0xd1, 0xc6, 0x39, 0x1a, 0x43, 0x6f, 0x16, 0x53,
0x12, 0x04, 0x91, 0x47, 0x04, 0x55, 0x55, 0xe8, 0x38, 0x79, 0x11, 0xbe, 0x0a, 0x97, 0x6b, 0xc1,
0x75, 0x80, 0xf8, 0x71, 0x29, 0xfa, 0x68, 0x3a, 0x65, 0xad, 0x5c, 0xe3, 0x2b, 0x95, 0xa8, 0x95,
0xa5, 0xc1, 0xfd, 0xa2, 0xb4, 0x1b, 0x50, 0x12, 0x26, 0xb3, 0x56, 0xc0, 0xe5, 0x88, 0x53, 0xd3,
0x0c, 0x79, 0x43, 0x37, 0xc7, 0x56, 0x14, 0x04, 0xd4, 0x13, 0x2c, 0x0a, 0x53, 0xd8, 0x6b, 0x00,
0x5e, 0x26, 0x34, 0xad, 0x92, 0x93, 0x60, 0x1b, 0x46, 0x55, 0x53, 0x03, 0xfb, 0xa7, 0x05, 0x17,
0xbe, 0x36, 0x45, 0xd3, 0x8e, 0x5b, 0x1d, 0x40, 0xd1, 0xe5, 0x99, 0xb2, 0xcb, 0xf2, 0x01, 0x75,
0x2a, 0x07, 0x24, 0x35, 0x62, 0x3a, 0x0b, 0x98, 0x47, 0x14, 0x44, 0x57, 0x41, 0xe4, 0x45, 0x68,
0x08, 0x1d, 0x21, 0x82, 0xd1, 0xa2, 0xda, 0x91, 0x4b, 0x3c, 0x82, 0x8b, 0xe5, 0x58, 0x4d, 0x1a,
0x9f, 0xc1, 0x86, 0x96, 0xec, 0x1e, 0x87, 0xde, 0xae, 0xba, 0x0d, 0xad, 0x8a, 0xfe, 0xb7, 0x05,
0xa3, 0xaa, 0xa1, 0xe9, 0xe2, 0x0f, 0xad, 0xc0, 0xfb, 0xe6, 0x87, 0xfe, 0x07, 0x3d, 0x41, 0x58,
0xe0, 0x46, 0xfb, 0xfb, 0x9c, 0x8a, 0xd1, 0xd2, 0xd8, 0x9a, 0x74, 0x1d, 0x90, 0xa2, 0xe7, 0x4a,
0x82, 0xee, 0xc0, 0xd0, 0xd3, 0x9d, 0xec, 0xc6, 0xf4, 0x2d, 0xe3, 0x12, 0x79, 0x59, 0x05, 0x76,
0xce, 0x4b, 0x3b, 0x5c, 0x8b, 0x11, 0x86, 0x01, 0xf3, 0xdf, 0xb9, 0x6a, 0x80, 0xa8, 0xeb, 0xbf,
0xa2, 0xd0, 0x7a, 0xcc, 0x7f, 0xf7, 0x2d, 0x0b, 0xe8, 0xae, 0x9c, 0x02, 0xaf, 0xe0, 0x8a, 0x4e,
0x7e, 0x27, 0xf4, 0x62, 0x3a, 0xa5, 0xa1, 0x20, 0xc1, 0x56, 0x34, 0x3b, 0x6e, 0xd5, 0x02, 0x97,
0x60, 0x85, 0xb3, 0xd0, 0xa3, 0x6e, 0xa8, 0xc7, 0x50, 0xd7, 0x59, 0x56, 0xdf, 0xcf, 0x38, 0x7e,
0x02, 0x57, 0x1b, 0x70, 0x4d, 0x65, 0xaf, 0x43, 0x5f, 0x05, 0xe6, 0x45, 0xa1, 0xa0, 0xa1, 0x50,
0xd8, 0x7d, 0xa7, 0x27, 0x65, 0x5b, 0x5a, 0x84, 0x3f, 0x01, 0xa4, 0x31, 0x9e, 0x46, 0x49, 0xd8,
0xee, 0x6a, 0x5e, 0x80, 0xb5, 0x82, 0x89, 0xe9, 0x8d, 0x47, 0xb0, 0xae, 0xc5, 0x2f, 0xc3, 0x69,
0x6b, 0xac, 0x0d, 0xb8, 0x50, 0x32, 0x32, 0x68, 0x0f, 0x53, 0x27, 0xc5, 0x77, 0xe2, 0x54, 0xb0,
0x8b, 0x69, 0x04, 0xc5, 0xa7, 0x02, 0xff, 0x65, 0xc1, 0xf9, 0x74, 0x8c, 0xb4, 0xac, 0xfa, 0x7b,
0xb6, 0x5d, 0xa7, 0xb1, 0xed, 0xba, 0x27, 0x6d, 0x37, 0x81, 0x21, 0x8f, 0x92, 0xd8, 0xa3, 0xae,
0x4f, 0x04, 0x71, 0xc3, 0xc8, 0xa7, 0xa6, 0x2b, 0xcf, 0x6a, 0xf9, 0x36, 0x11, 0xe4, 0x59, 0xe4,
0x53, 0xbc, 0x9e, 0x1e, 0x4a, 0xfe, 0x34, 0x71, 0x08, 0xe7, 0xe4, 0xb7, 0x6c, 0xab, 0x96, 0x39,
0xf4, 0x18, 0x77, 0xd3, 0xee, 0x54, 0x49, 0xac, 0x38, 0xab, 0x8c, 0xef, 0xe8, 0xd6, 0x34, 0xfb,
0x3e, 0x11, 0x7a, 0xbf, 0x93, 0xee, 0x6f, 0x13, 0x21, 0xf7, 0xf1, 0xa7, 0x30, 0x3c, 0xf1, 0xd7,
0xbe, 0xa3, 0x92, 0xb4, 0xd8, 0x2f, 0x08, 0x0b, 0x3e, 0xb0, 0xc5, 0xd1, 0x04, 0xce, 0xf9, 0x31,
0x61, 0x21, 0x0b, 0x0f, 0x76, 0xa9, 0x17, 0x85, 0x3e, 0x57, 0x71, 0x0e, 0x9c, 0xb2, 0x18, 0xff,
0x96, 0xd6, 0x4c, 0xbb, 0x3d, 0x79, 0x21, 0x43, 0x4a, 0xfd, 0x80, 0xba, 0x87, 0x94, 0xf8, 0x34,
0x36, 0x01, 0xf7, 0xb5, 0xf0, 0x3b, 0x25, 0x93, 0xf3, 0xc0, 0x28, 0xed, 0x45, 0xfe, 0xb1, 0x0a,
0xa1, 0xef, 0x80, 0x16, 0x3d, 0x89, 0xfc, 0x63, 0x75, 0xc9, 0xb9, 0x1b, 0x10, 0x2e, 0x5c, 0xef,
0x30, 0x09, 0x8f, 0x4c, 0xad, 0x7a, 0x8c, 0x7f, 0x4f, 0xb8, 0xd8, 0x92, 0x22, 0xfc, 0x25, 0x5c,
0x76, 0x28, 0xf1, 0x75, 0x0c, 0xea, 0xea, 0xb7, 0x1f, 0x8f, 0xff, 0x58, 0x70, 0xa5, 0xde, 0xb8,
0xcd, 0x88, 0xbc, 0x0f, 0x28, 0x1b, 0x41, 0x82, 0x4d, 0x29, 0x17, 0x64, 0x3a, 0x33, 0x85, 0x1c,
0x9a, 0x39, 0xf4, 0x22, 0x95, 0x57, 0x07, 0x56, 0xa7, 0x32, 0xb0, 0x24, 0x62, 0xda, 0x16, 0x39,
0xc4, 0xae, 0x46, 0xf4, 0x75, 0x7b, 0x14, 0x10, 0x33, 0x6d, 0x85, 0xb8, 0xa8, 0x11, 0x8d, 0xa2,
0x42, 0xbc, 0x0a, 0x60, 0xfa, 0x26, 0x09, 0xd3, 0x89, 0xbb, 0xaa, 0xbb, 0x26, 0x09, 0x05, 0xfe,
0x01, 0x60, 0x9b, 0xf1, 0x23, 0x9d, 0xb5, 0xbc, 0x3a, 0x3e, 0x8b, 0xcd, 0x0b, 0x2b, 0x97, 0x52,
0x42, 0x82, 0xc0, 0xe4, 0x24, 0x97, 0x92, 0x6d, 0x25, 0x9c, 0xfa, 0x26, 0x7a, 0xb5, 0x96, 0xb2,
0xfd, 0x98, 0x52, 0x13, 0xa8, 0x5a, 0xe3, 0xdf, 0x2d, 0x58, 0x7d, 0x4a, 0xa7, 0x06, 0xf9, 0x1a,
0xc0, 0x41, 0x14, 0x47, 0x89, 0x60, 0x21, 0xe5, 0xca, 0xc1, 0xa2, 0x93, 0x93, 0xfc, 0x77, 0x3f,
0x8a, 0xfd, 0xd1, 0x60, 0xdf, 0xe4, 0xae, 0xd6, 0x52, 0x76, 0x48, 0xc9, 0xcc, 0xa4, 0xab, 0xd6,
0x92, 0x3b, 0x72, 0x41, 0xbc, 0x23, 0xf5, 0x9e, 0x74, 0x1d, 0xfd, 0xf1, 0xf0, 0x8f, 0x01, 0xf4,
0xcd, 0xfb, 0xa8, 0xc8, 0x2b, 0xfa, 0x09, 0x7a, 0x39, 0xd2, 0x8b, 0x6e, 0x56, 0xb9, 0x6d, 0x95,
0x44, 0xdb, 0xff, 0x9f, 0xa3, 0x65, 0xe6, 0xc8, 0x02, 0x0a, 0xe1, 0x7c, 0x85, 0x54, 0xa2, 0xbb,
0x55, 0xeb, 0x26, 0xca, 0x6a, 0xdf, 0x6b, 0xa5, 0x9b, 0xf9, 0x13, 0xb0, 0x56, 0xc3, 0x12, 0xd1,
0xfd, 0x39, 0x28, 0x05, 0xa6, 0x6a, 0x7f, 0xd4, 0x52, 0x3b, 0xf3, 0xfa, 0x06, 0x50, 0x95, 0x42,
0xa2, 0x7b, 0x73, 0x61, 0x4e, 0x28, 0xaa, 0x7d, 0xbf, 0x9d, 0x72, 0x63, 0xa2, 0x9a, 0x5c, 0xce,
0x4d, 0xb4, 0x40, 0x5f, 0xe7, 0x26, 0x5a, 0x62, 0xac, 0x0b, 0xe8, 0x08, 0x86, 0x65, 0xe2, 0x89,
0xee, 0x34, 0xfd, 0x1a, 0xaa, 0xf0, 0x5a, 0xfb, 0x6e, 0x1b, 0xd5, 0xcc, 0x19, 0x85, 0xb3, 0x45,
0x72, 0x88, 0x6e, 0x57, 0xed, 0x6b, 0xa9, 0xae, 0x3d, 0x99, 0xaf, 0x98, 0xcf, 0xa9, 0x4c, 0x18,
0xeb, 0x72, 0x6a, 0x60, 0xa3, 0x75, 0x39, 0x35, 0xf1, 0x4f, 0xbc, 0x80, 0x7e, 0x4d, 0x59, 0x48,
0x89, 0x48, 0xa1, 0xcd, 0x26, 0x98, 0x7a, 0x26, 0x67, 0x3f, 0x68, 0xad, 0x9f, 0xfa, 0xfe, 0xd8,
0x92, 0x77, 0x3d, 0xc7, 0xa7, 0xea, 0xee, 0x7a, 0x95, 0xa1, 0xd5, 0xdd, 0xf5, 0x3a, 0x52, 0xb6,
0x80, 0xf6, 0x60, 0x50, 0x60, 0x58, 0xe8, 0x56, 0x93, 0x65, 0x91, 0xb7, 0xd9, 0xb7, 0xe7, 0xea,
0x65, 0x3e, 0xdc, 0x74, 0x7a, 0x99, 0x71, 0xd5, 0x18, 0x5c, 0x71, 0x5e, 0xdd, 0x9a, 0xa7, 0x96,
0x39, 0xf8, 0x11, 0xe0, 0x84, 0x10, 0xa1, 0x1b, 0x4d, 0x76, 0xf9, 0xa3, 0xb8, 0x79, 0xba, 0x52,
0x06, 0xfd, 0x33, 0xac, 0xd7, 0x3d, 0xbd, 0xa8, 0xe6, 0x16, 0x9e, 0xf2, 0xbe, 0xdb, 0x9b, 0x6d,
0xd5, 0x33, 0xc7, 0x2f, 0x61, 0x25, 0xa5, 0x57, 0xe8, 0x7a, 0xd5, 0xba, 0x44, 0xf5, 0x6c, 0x7c,
0x9a, 0x4a, 0xae, 0x9b, 0x5e, 0xa7, 0xa5, 0x92, 0x3c, 0xa8, 0xb9, 0x54, 0x39, 0x72, 0xd6, 0x5c,
0xaa, 0x3c, 0x95, 0x92, 0xe0, 0x7b, 0x4b, 0xea, 0x5f, 0x9d, 0x47, 0xff, 0x06, 0x00, 0x00, 0xff,
0xff, 0x71, 0xd9, 0xf3, 0x7e, 0xec, 0x11, 0x00, 0x00,
// 1339 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x09, 0x6e, 0x88, 0x02, 0xff, 0xa4, 0x58, 0xdd, 0x73, 0x14, 0x45,
0x10, 0xcf, 0x72, 0x97, 0xaf, 0xbe, 0x3b, 0x08, 0x13, 0x20, 0xc7, 0xf2, 0xe1, 0x31, 0x20, 0x1c,
0x1f, 0x06, 0x85, 0x52, 0x51, 0x1f, 0x14, 0x12, 0x2d, 0x53, 0x25, 0x50, 0xb5, 0x01, 0x4a, 0x0b,
0xab, 0xb6, 0x26, 0xbb, 0x93, 0x64, 0x2b, 0x7b, 0xb3, 0xcb, 0xce, 0x6c, 0x24, 0x96, 0xe5, 0x5f,
0xe2, 0x9b, 0xe5, 0x8b, 0x0f, 0xfe, 0x83, 0xbe, 0x58, 0xf3, 0xb1, 0x97, 0xfd, 0xcc, 0x2d, 0xf2,
0x36, 0xdb, 0xd3, 0xdd, 0xbf, 0xee, 0x9e, 0x9e, 0x9e, 0xdf, 0x1d, 0xac, 0x1e, 0x46, 0x61, 0x3a,
0xa1, 0x2e, 0xa7, 0xc9, 0x21, 0x4d, 0xd6, 0xe3, 0x24, 0x12, 0x11, 0x5a, 0x29, 0x08, 0xdd, 0x78,
0x07, 0xdf, 0x07, 0xf4, 0x84, 0x08, 0x6f, 0x7f, 0x93, 0x86, 0x54, 0x50, 0x87, 0xbe, 0x49, 0x29,
0x17, 0xe8, 0x22, 0x2c, 0xed, 0x06, 0x21, 0x75, 0x03, 0x9f, 0x0f, 0xad, 0x51, 0x67, 0xbc, 0xec,
0x2c, 0xca, 0xef, 0x2d, 0x9f, 0xe3, 0xe7, 0xb0, 0x5a, 0x30, 0xe0, 0x71, 0xc4, 0x38, 0x45, 0x8f,
0x60, 0x31, 0xa1, 0x3c, 0x0d, 0x85, 0x36, 0xe8, 0x3d, 0xb8, 0xba, 0x5e, 0xc6, 0x5a, 0x9f, 0x9a,
0xa4, 0xa1, 0x70, 0x32, 0x75, 0x1c, 0x40, 0x3f, 0xbf, 0x81, 0xd6, 0x60, 0xd1, 0x60, 0x0f, 0xad,
0x91, 0x35, 0x5e, 0x76, 0x16, 0x34, 0x34, 0xba, 0x00, 0x0b, 0x5c, 0x10, 0x91, 0xf2, 0xe1, 0xa9,
0x91, 0x35, 0x9e, 0x77, 0xcc, 0x17, 0x3a, 0x07, 0xf3, 0x34, 0x49, 0xa2, 0x64, 0xd8, 0x51, 0xea,
0xfa, 0x03, 0x21, 0xe8, 0xf2, 0xe0, 0x57, 0x3a, 0xec, 0x8e, 0xac, 0xf1, 0xc0, 0x51, 0x6b, 0xbc,
0x08, 0xf3, 0xdf, 0x4e, 0x62, 0x71, 0x84, 0x3f, 0x87, 0xe1, 0x2b, 0xe2, 0xa5, 0xe9, 0xe4, 0x95,
0x8a, 0x71, 0x63, 0x9f, 0x7a, 0x07, 0x59, 0xee, 0x97, 0x60, 0xd9, 0x44, 0x6e, 0x22, 0x18, 0x38,
0x4b, 0x5a, 0xb0, 0xe5, 0xe3, 0x6f, 0xe0, 0x62, 0x8d, 0xa1, 0xa9, 0xc1, 0x75, 0x18, 0xec, 0x91,
0x64, 0x87, 0xec, 0x51, 0x37, 0x21, 0x22, 0x88, 0x94, 0xb5, 0xe5, 0xf4, 0x8d, 0xd0, 0x91, 0x32,
0xfc, 0x1a, 0xec, 0x82, 0x87, 0x68, 0x12, 0x13, 0x4f, 0xb4, 0x01, 0x47, 0x23, 0xe8, 0xc5, 0x09,
0x25, 0x61, 0x18, 0x79, 0x44, 0x50, 0x55, 0x85, 0x8e, 0x93, 0x17, 0xe1, 0x2b, 0x70, 0xa9, 0xd6,
0xb9, 0x0e, 0x10, 0x3f, 0x2a, 0x45, 0x1f, 0x4d, 0x26, 0x41, 0x2b, 0x68, 0x7c, 0xb9, 0x12, 0xb5,
0xb2, 0x34, 0x7e, 0xbf, 0x28, 0xed, 0x86, 0x94, 0xb0, 0x34, 0x6e, 0xe5, 0xb8, 0x1c, 0x71, 0x66,
0x3a, 0xf5, 0xbc, 0xa6, 0x9b, 0x63, 0x23, 0x0a, 0x43, 0xea, 0x89, 0x20, 0x62, 0x99, 0xdb, 0xab,
0x00, 0xde, 0x54, 0x68, 0x5a, 0x25, 0x27, 0xc1, 0x36, 0x0c, 0xab, 0xa6, 0xc6, 0xed, 0xdf, 0x16,
0x9c, 0x7f, 0x6c, 0x8a, 0xa6, 0x81, 0x5b, 0x1d, 0x40, 0x11, 0xf2, 0x54, 0x19, 0xb2, 0x7c, 0x40,
0x9d, 0xca, 0x01, 0x49, 0x8d, 0x84, 0xc6, 0x61, 0xe0, 0x11, 0xe5, 0xa2, 0xab, 0x5c, 0xe4, 0x45,
0x68, 0x05, 0x3a, 0x42, 0x84, 0xc3, 0x79, 0xb5, 0x23, 0x97, 0x78, 0x08, 0x17, 0xca, 0xb1, 0x9a,
0x34, 0x3e, 0x83, 0x35, 0x2d, 0xd9, 0x3e, 0x62, 0xde, 0xb6, 0xba, 0x0d, 0xad, 0x8a, 0xfe, 0xaf,
0x05, 0xc3, 0xaa, 0xa1, 0xe9, 0xe2, 0xf7, 0xad, 0xc0, 0xbb, 0xe6, 0x87, 0x3e, 0x80, 0x9e, 0x20,
0x41, 0xe8, 0x46, 0xbb, 0xbb, 0x9c, 0x8a, 0xe1, 0xc2, 0xc8, 0x1a, 0x77, 0x1d, 0x90, 0xa2, 0xe7,
0x4a, 0x82, 0x6e, 0xc3, 0x8a, 0xa7, 0x3b, 0xd9, 0x4d, 0xe8, 0x61, 0xc0, 0xa5, 0xe7, 0x45, 0x15,
0xd8, 0x19, 0x2f, 0xeb, 0x70, 0x2d, 0x46, 0x18, 0x06, 0x81, 0xff, 0xd6, 0x55, 0x03, 0x44, 0x5d,
0xff, 0x25, 0xe5, 0xad, 0x17, 0xf8, 0x6f, 0xbf, 0x0b, 0x42, 0xba, 0x2d, 0xa7, 0xc0, 0x2b, 0xb8,
0xac, 0x93, 0xdf, 0x62, 0x5e, 0x42, 0x27, 0x94, 0x09, 0x12, 0x6e, 0x44, 0xf1, 0x51, 0xab, 0x16,
0xb8, 0x08, 0x4b, 0x3c, 0x60, 0x1e, 0x75, 0x99, 0x1e, 0x43, 0x5d, 0x67, 0x51, 0x7d, 0x3f, 0xe3,
0xf8, 0x09, 0x5c, 0x69, 0xf0, 0x6b, 0x2a, 0x7b, 0x0d, 0xfa, 0x2a, 0x30, 0x2f, 0x62, 0x82, 0x32,
0xa1, 0x7c, 0xf7, 0x9d, 0x9e, 0x94, 0x6d, 0x68, 0x11, 0xfe, 0x04, 0x90, 0xf6, 0xf1, 0x34, 0x4a,
0x59, 0xbb, 0xab, 0x79, 0x1e, 0x56, 0x0b, 0x26, 0xa6, 0x37, 0x1e, 0xc2, 0x39, 0x2d, 0x7e, 0xc9,
0x26, 0xad, 0x7d, 0xad, 0xc1, 0xf9, 0x92, 0x91, 0xf1, 0xf6, 0x20, 0x03, 0x29, 0xbe, 0x13, 0x27,
0x3a, 0xbb, 0x90, 0x45, 0x50, 0x7c, 0x2a, 0xf0, 0x3f, 0x16, 0x9c, 0xcd, 0xc6, 0x48, 0xcb, 0xaa,
0xbf, 0x63, 0xdb, 0x75, 0x1a, 0xdb, 0xae, 0x7b, 0xdc, 0x76, 0x63, 0x58, 0xe1, 0x51, 0x9a, 0x78,
0xd4, 0xf5, 0x89, 0x20, 0x2e, 0x8b, 0x7c, 0x6a, 0xba, 0xf2, 0xb4, 0x96, 0x6f, 0x12, 0x41, 0x9e,
0x45, 0x3e, 0xc5, 0x5f, 0x67, 0x87, 0x52, 0x38, 0xcd, 0xdb, 0x70, 0x36, 0x24, 0x5c, 0xb8, 0x24,
0x8e, 0x29, 0xf3, 0x5d, 0x22, 0x64, 0x4b, 0x58, 0xaa, 0x25, 0x4e, 0xcb, 0x8d, 0xc7, 0x4a, 0xfe,
0x58, 0x3c, 0xe3, 0x98, 0xc1, 0x19, 0x69, 0x2a, 0x3b, 0xb0, 0x65, 0xba, 0xbd, 0x80, 0xbb, 0x59,
0x23, 0xab, 0x7c, 0x97, 0x9c, 0xe5, 0x80, 0x6f, 0xe9, 0x2e, 0x36, 0xfb, 0x3e, 0x11, 0x7a, 0xbf,
0x93, 0xed, 0x6f, 0x12, 0x21, 0xf7, 0xf1, 0xa7, 0xb0, 0x72, 0x8c, 0xd7, 0xbe, 0xf9, 0xd2, 0xec,
0x5c, 0x5e, 0x90, 0x20, 0x7c, 0xcf, 0xdb, 0x80, 0xc6, 0x70, 0xc6, 0x4f, 0x48, 0xc0, 0x02, 0xb6,
0xb7, 0x4d, 0xbd, 0x88, 0xf9, 0x5c, 0xc5, 0x39, 0x70, 0xca, 0x62, 0xfc, 0x7b, 0x56, 0x5e, 0x0d,
0x7b, 0xfc, 0x98, 0x32, 0x4a, 0xfd, 0x90, 0xba, 0xfb, 0x94, 0xf8, 0x34, 0x31, 0x01, 0xf7, 0xb5,
0xf0, 0x7b, 0x25, 0x93, 0xa3, 0xc3, 0x28, 0xed, 0x44, 0xfe, 0x91, 0x0a, 0xa1, 0xef, 0x80, 0x16,
0x3d, 0x89, 0xfc, 0x23, 0x35, 0x0f, 0xb8, 0xab, 0xce, 0xc9, 0xdb, 0x4f, 0xd9, 0x81, 0xa9, 0x55,
0x2f, 0xe0, 0x3f, 0x10, 0x2e, 0x36, 0xa4, 0x08, 0x7f, 0x09, 0x97, 0x1c, 0x4a, 0x7c, 0x1d, 0x83,
0x9a, 0x12, 0xed, 0x27, 0xe9, 0x1f, 0xa7, 0xe0, 0x72, 0xbd, 0x71, 0x9b, 0x69, 0xfa, 0x15, 0xd8,
0xd3, 0x69, 0x25, 0x82, 0x09, 0xe5, 0x82, 0x4c, 0x62, 0x97, 0x9b, 0x72, 0xe9, 0x82, 0xae, 0x99,
0xd1, 0xf5, 0x22, 0xdb, 0x37, 0x65, 0xab, 0x8e, 0xba, 0x4e, 0x65, 0xd4, 0x49, 0x80, 0xac, 0x4b,
0x6a, 0x00, 0xba, 0x1a, 0xc0, 0xd7, 0x5d, 0x53, 0x07, 0x30, 0x35, 0x56, 0x00, 0xf3, 0x1a, 0xc0,
0xe8, 0x2b, 0x80, 0x2b, 0x00, 0xa6, 0xab, 0x52, 0x96, 0x8d, 0xee, 0x65, 0xdd, 0x53, 0x29, 0x13,
0xf8, 0x47, 0x80, 0xcd, 0x80, 0x1f, 0xe8, 0x9a, 0xc8, 0x3b, 0xe8, 0x07, 0x89, 0x79, 0xaa, 0xe5,
0x52, 0x4a, 0x48, 0x18, 0x9a, 0x4c, 0xe5, 0x52, 0xd2, 0xb6, 0x94, 0x53, 0xdf, 0x24, 0xa3, 0xd6,
0x52, 0xb6, 0x9b, 0x50, 0x6a, 0xe2, 0x55, 0x6b, 0xfc, 0xa7, 0x05, 0xcb, 0x4f, 0xe9, 0xc4, 0x78,
0xbe, 0x0a, 0xb0, 0x17, 0x25, 0x51, 0x2a, 0x02, 0x46, 0xf5, 0x25, 0x9c, 0x77, 0x72, 0x92, 0xff,
0x8f, 0xa3, 0x68, 0x24, 0x0d, 0x77, 0x4d, 0xee, 0x6a, 0x2d, 0x65, 0xfb, 0x94, 0xc4, 0x26, 0x5d,
0xb5, 0x96, 0x24, 0x94, 0x0b, 0xe2, 0x1d, 0xa8, 0x87, 0xa9, 0xeb, 0xe8, 0x8f, 0x07, 0x7f, 0x0d,
0xa0, 0x6f, 0x1e, 0x5a, 0xc5, 0x82, 0xd1, 0xcf, 0xd0, 0xcb, 0xb1, 0x67, 0x74, 0xa3, 0x4a, 0x92,
0xab, 0x6c, 0xdc, 0xfe, 0x70, 0x86, 0x96, 0x99, 0xab, 0x73, 0x88, 0xc1, 0xd9, 0x0a, 0x3b, 0x45,
0x77, 0xaa, 0xd6, 0x4d, 0xdc, 0xd7, 0xbe, 0xdb, 0x4a, 0x77, 0x8a, 0x27, 0x60, 0xb5, 0x86, 0x6e,
0xa2, 0x7b, 0x33, 0xbc, 0x14, 0x28, 0xaf, 0xfd, 0x51, 0x4b, 0xed, 0x29, 0xea, 0x1b, 0x40, 0x55,
0x2e, 0x8a, 0xee, 0xce, 0x74, 0x73, 0xcc, 0x75, 0xed, 0x7b, 0xed, 0x94, 0x1b, 0x13, 0xd5, 0x2c,
0x75, 0x66, 0xa2, 0x05, 0x1e, 0x3c, 0x33, 0xd1, 0x12, 0xf5, 0x9d, 0x43, 0x07, 0xb0, 0x52, 0x66,
0xb0, 0xe8, 0x76, 0xd3, 0xcf, 0xaa, 0x0a, 0x41, 0xb6, 0xef, 0xb4, 0x51, 0x9d, 0x82, 0x51, 0x38,
0x5d, 0x64, 0x99, 0xe8, 0x56, 0xd5, 0xbe, 0x96, 0x33, 0xdb, 0xe3, 0xd9, 0x8a, 0xf9, 0x9c, 0xca,
0xcc, 0xb3, 0x2e, 0xa7, 0x06, 0x5a, 0x5b, 0x97, 0x53, 0x13, 0x91, 0xc5, 0x73, 0xe8, 0xb7, 0x8c,
0xce, 0x94, 0x18, 0x19, 0x5a, 0x6f, 0x72, 0x53, 0x4f, 0x09, 0xed, 0xfb, 0xad, 0xf5, 0x33, 0xec,
0x8f, 0x2d, 0x79, 0xd7, 0x73, 0xc4, 0xac, 0xee, 0xae, 0x57, 0xa9, 0x5e, 0xdd, 0x5d, 0xaf, 0x63,
0x77, 0x73, 0x68, 0x07, 0x06, 0x05, 0xaa, 0x86, 0x6e, 0x36, 0x59, 0x16, 0x09, 0xa0, 0x7d, 0x6b,
0xa6, 0xde, 0x14, 0xc3, 0xcd, 0xa6, 0x97, 0x19, 0x57, 0x8d, 0xc1, 0x15, 0xe7, 0xd5, 0xcd, 0x59,
0x6a, 0x53, 0x80, 0x9f, 0x00, 0x8e, 0x99, 0x15, 0xba, 0xde, 0x64, 0x97, 0x3f, 0x8a, 0x1b, 0x27,
0x2b, 0x4d, 0x5d, 0xff, 0x02, 0xe7, 0xea, 0x1e, 0x66, 0x54, 0x73, 0x0b, 0x4f, 0x78, 0xfd, 0xed,
0xf5, 0xb6, 0xea, 0x53, 0xe0, 0x97, 0xb0, 0x94, 0x91, 0x2f, 0x74, 0xad, 0x6a, 0x5d, 0x22, 0x82,
0x36, 0x3e, 0x49, 0x25, 0xd7, 0x4d, 0xaf, 0xb3, 0x52, 0x49, 0x96, 0xd4, 0x5c, 0xaa, 0x1c, 0x75,
0x6b, 0x2e, 0x55, 0x9e, 0x68, 0x49, 0xe7, 0x3b, 0x0b, 0xea, 0xef, 0xa1, 0x87, 0xff, 0x05, 0x00,
0x00, 0xff, 0xff, 0xfc, 0x26, 0x3f, 0x74, 0x35, 0x12, 0x00, 0x00,
} }

19
weed/server/volume_grpc_copy.go
View File

@ -3,15 +3,15 @@ package weed_server
import ( import (
"context" "context"
"fmt" "fmt"
"io"
"os"
"time"
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/operation"
"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb" "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
"github.com/chrislusf/seaweedfs/weed/storage" "github.com/chrislusf/seaweedfs/weed/storage"
"github.com/chrislusf/seaweedfs/weed/storage/needle" "github.com/chrislusf/seaweedfs/weed/storage/needle"
"io"
"os"
) )
// VolumeCopy copy the .idx .dat files, and mount the volume // VolumeCopy copy the .idx .dat files, and mount the volume
@ -96,7 +96,9 @@ func (vs *VolumeServer) VolumeCopy(ctx context.Context, req *volume_server_pb.Vo
return nil, fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err) return nil, fmt.Errorf("failed to mount volume %d: %v", req.VolumeId, err)
} }
return &volume_server_pb.VolumeCopyResponse{}, err
return &volume_server_pb.VolumeCopyResponse{
LastAppendAtNs:volFileInfoResp.DatFileTimestampSeconds*uint64(time.Second),
}, err
} }
/** /**
@ -153,10 +155,11 @@ func (vs *VolumeServer) ReadVolumeFileStatus(ctx context.Context, req *volume_se
} }
resp.VolumeId = req.VolumeId resp.VolumeId = req.VolumeId
resp.DatFileSize = v.DataFileSize()
resp.IdxFileSize = v.IndexFileSize()
resp.DatFileTimestamp = v.LastModifiedTime()
resp.IdxFileTimestamp = v.LastModifiedTime()
datSize, idxSize, modTime := v.FileStat()
resp.DatFileSize = datSize
resp.IdxFileSize = idxSize
resp.DatFileTimestampSeconds = uint64(modTime.Unix())
resp.IdxFileTimestampSeconds = uint64(modTime.Unix())
resp.FileCount = v.FileCount() resp.FileCount = v.FileCount()
return resp, nil return resp, nil
} }

4
weed/server/volume_grpc_copy_incremental.go
View File

@ -17,7 +17,7 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem
return fmt.Errorf("not found volume id %d", req.VolumeId) return fmt.Errorf("not found volume id %d", req.VolumeId)
} }
stopOffset := v.Size()
stopOffset, _, _ := v.FileStat()
foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs) foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs)
if err != nil { if err != nil {
return fmt.Errorf("fail to locate by appendAtNs %d: %s", req.SinceNs, err) return fmt.Errorf("fail to locate by appendAtNs %d: %s", req.SinceNs, err)
@ -30,7 +30,7 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem
startOffset := foundOffset.ToAcutalOffset() startOffset := foundOffset.ToAcutalOffset()
buf := make([]byte, 1024*1024*2) buf := make([]byte, 1024*1024*2)
return sendFileContent(v.DataFile(), buf, startOffset, stopOffset, stream)
return sendFileContent(v.DataFile(), buf, startOffset, int64(stopOffset), stream)
} }

3
weed/storage/disk_location.go
View File

@ -54,8 +54,9 @@ func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleM
mutex.Lock() mutex.Lock()
l.volumes[vid] = v l.volumes[vid] = v
mutex.Unlock() mutex.Unlock()
size, _, _ := v.FileStat()
glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s", glog.V(0).Infof("data file %s, replicaPlacement=%s v=%d size=%d ttl=%s",
l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), v.Size(), v.Ttl.String())
l.Directory+"/"+name, v.ReplicaPlacement, v.Version(), size, v.Ttl.String())
} else { } else {
glog.V(0).Infof("new volume %s error %s", name, e) glog.V(0).Infof("new volume %s error %s", name, e)
} }

26
weed/storage/needle/needle_read_write.go
View File

@ -45,7 +45,7 @@ func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32
} }
switch version { switch version {
case Version1: case Version1:
header := make([]byte, NeedleEntrySize)
header := make([]byte, NeedleHeaderSize)
CookieToBytes(header[0:CookieSize], n.Cookie) CookieToBytes(header[0:CookieSize], n.Cookie)
NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id) NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
n.Size = uint32(len(n.Data)) n.Size = uint32(len(n.Data))
@ -57,13 +57,13 @@ func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32
if _, err = w.Write(n.Data); err != nil { if _, err = w.Write(n.Data); err != nil {
return return
} }
actualSize = NeedleEntrySize + int64(n.Size)
actualSize = NeedleHeaderSize + int64(n.Size)
padding := PaddingLength(n.Size, version) padding := PaddingLength(n.Size, version)
util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value()) util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value())
_, err = w.Write(header[0 : NeedleChecksumSize+padding]) _, err = w.Write(header[0 : NeedleChecksumSize+padding])
return return
case Version2, Version3: case Version2, Version3:
header := make([]byte, NeedleEntrySize+TimestampSize) // adding timestamp to reuse it and avoid extra allocation
header := make([]byte, NeedleHeaderSize+TimestampSize) // adding timestamp to reuse it and avoid extra allocation
CookieToBytes(header[0:CookieSize], n.Cookie) CookieToBytes(header[0:CookieSize], n.Cookie)
NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id) NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id)
if len(n.Name) >= math.MaxUint8 { if len(n.Name) >= math.MaxUint8 {
@ -94,7 +94,7 @@ func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32
} }
size = n.DataSize size = n.DataSize
util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size) util.Uint32toBytes(header[CookieSize+NeedleIdSize:CookieSize+NeedleIdSize+SizeSize], n.Size)
if _, err = w.Write(header[0:NeedleEntrySize]); err != nil {
if _, err = w.Write(header[0:NeedleHeaderSize]); err != nil {
return return
} }
if n.DataSize > 0 { if n.DataSize > 0 {
@ -181,21 +181,21 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version
} }
switch version { switch version {
case Version1: case Version1:
n.Data = bytes[NeedleEntrySize : NeedleEntrySize+size]
n.Data = bytes[NeedleHeaderSize : NeedleHeaderSize+size]
case Version2, Version3: case Version2, Version3:
err = n.readNeedleDataVersion2(bytes[NeedleEntrySize : NeedleEntrySize+int(n.Size)])
err = n.readNeedleDataVersion2(bytes[NeedleHeaderSize : NeedleHeaderSize+int(n.Size)])
} }
if size == 0 || err != nil { if size == 0 || err != nil {
return err return err
} }
checksum := util.BytesToUint32(bytes[NeedleEntrySize+size : NeedleEntrySize+size+NeedleChecksumSize])
checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize])
newChecksum := NewCRC(n.Data) newChecksum := NewCRC(n.Data)
if checksum != newChecksum.Value() { if checksum != newChecksum.Value() {
return errors.New("CRC error! Data On Disk Corrupted") return errors.New("CRC error! Data On Disk Corrupted")
} }
n.Checksum = newChecksum n.Checksum = newChecksum
if version == Version3 { if version == Version3 {
tsOffset := NeedleEntrySize + size + NeedleChecksumSize
tsOffset := NeedleHeaderSize + size + NeedleChecksumSize
n.AppendAtNs = util.BytesToUint64(bytes[tsOffset : tsOffset+TimestampSize]) n.AppendAtNs = util.BytesToUint64(bytes[tsOffset : tsOffset+TimestampSize])
} }
return nil return nil
@ -204,7 +204,7 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version
func (n *Needle) ParseNeedleHeader(bytes []byte) { func (n *Needle) ParseNeedleHeader(bytes []byte) {
n.Cookie = BytesToCookie(bytes[0:CookieSize]) n.Cookie = BytesToCookie(bytes[0:CookieSize])
n.Id = BytesToNeedleId(bytes[CookieSize : CookieSize+NeedleIdSize]) n.Id = BytesToNeedleId(bytes[CookieSize : CookieSize+NeedleIdSize])
n.Size = util.BytesToUint32(bytes[CookieSize+NeedleIdSize : NeedleEntrySize])
n.Size = util.BytesToUint32(bytes[CookieSize+NeedleIdSize : NeedleHeaderSize])
} }
func (n *Needle) readNeedleDataVersion2(bytes []byte) (err error) { func (n *Needle) readNeedleDataVersion2(bytes []byte) (err error) {
@ -271,7 +271,7 @@ func (n *Needle) readNeedleDataVersion2(bytes []byte) (err error) {
func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bytes []byte, bodyLength int64, err error) { func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, bytes []byte, bodyLength int64, err error) {
n = new(Needle) n = new(Needle)
if version == Version1 || version == Version2 || version == Version3 { if version == Version1 || version == Version2 || version == Version3 {
bytes = make([]byte, NeedleEntrySize)
bytes = make([]byte, NeedleHeaderSize)
var count int var count int
count, err = r.ReadAt(bytes, offset) count, err = r.ReadAt(bytes, offset)
if count <= 0 || err != nil { if count <= 0 || err != nil {
@ -286,9 +286,9 @@ func ReadNeedleHeader(r *os.File, version Version, offset int64) (n *Needle, byt
func PaddingLength(needleSize uint32, version Version) uint32 { func PaddingLength(needleSize uint32, version Version) uint32 {
if version == Version3 { if version == Version3 {
// this is same value as version2, but just listed here for clarity // this is same value as version2, but just listed here for clarity
return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize + TimestampSize) % NeedlePaddingSize)
return NeedlePaddingSize - ((NeedleHeaderSize + needleSize + NeedleChecksumSize + TimestampSize) % NeedlePaddingSize)
} }
return NeedlePaddingSize - ((NeedleEntrySize + needleSize + NeedleChecksumSize) % NeedlePaddingSize)
return NeedlePaddingSize - ((NeedleHeaderSize + needleSize + NeedleChecksumSize) % NeedlePaddingSize)
} }
func NeedleBodyLength(needleSize uint32, version Version) int64 { func NeedleBodyLength(needleSize uint32, version Version) int64 {
@ -386,6 +386,6 @@ func (n *Needle) SetHasPairs() {
} }
func getActualSize(size uint32, version Version) int64 { func getActualSize(size uint32, version Version) int64 {
return NeedleEntrySize + NeedleBodyLength(size, version)
return NeedleHeaderSize + NeedleBodyLength(size, version)
} }

4
weed/storage/needle_map/compact_map_perf_test.go
View File

@ -52,11 +52,11 @@ func TestMemoryUsage(t *testing.T) {
func loadNewNeedleMap(file *os.File) (*CompactMap, uint64) { func loadNewNeedleMap(file *os.File) (*CompactMap, uint64) {
m := NewCompactMap() m := NewCompactMap()
bytes := make([]byte, NeedleEntrySize)
bytes := make([]byte, NeedleMapEntrySize)
rowCount := uint64(0) rowCount := uint64(0)
count, e := file.Read(bytes) count, e := file.Read(bytes)
for count > 0 && e == nil { for count > 0 && e == nil {
for i := 0; i < count; i += NeedleEntrySize {
for i := 0; i < count; i += NeedleMapEntrySize {
rowCount++ rowCount++
key := BytesToNeedleId(bytes[i : i+NeedleIdSize]) key := BytesToNeedleId(bytes[i : i+NeedleIdSize])
offset := BytesToOffset(bytes[i+NeedleIdSize : i+NeedleIdSize+OffsetSize]) offset := BytesToOffset(bytes[i+NeedleIdSize : i+NeedleIdSize+OffsetSize])

6
weed/storage/needle_map_memory.go
View File

@ -73,7 +73,7 @@ func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) {
// stops with the error returned by the fn function // stops with the error returned by the fn function
func WalkIndexFile(r *os.File, fn func(key NeedleId, offset Offset, size uint32) error) error { func WalkIndexFile(r *os.File, fn func(key NeedleId, offset Offset, size uint32) error) error {
var readerOffset int64 var readerOffset int64
bytes := make([]byte, NeedleEntrySize*RowsToRead)
bytes := make([]byte, NeedleMapEntrySize*RowsToRead)
count, e := r.ReadAt(bytes, readerOffset) count, e := r.ReadAt(bytes, readerOffset)
glog.V(3).Infoln("file", r.Name(), "readerOffset", readerOffset, "count", count, "e", e) glog.V(3).Infoln("file", r.Name(), "readerOffset", readerOffset, "count", count, "e", e)
readerOffset += int64(count) readerOffset += int64(count)
@ -85,8 +85,8 @@ func WalkIndexFile(r *os.File, fn func(key NeedleId, offset Offset, size uint32)
) )
for count > 0 && e == nil || e == io.EOF { for count > 0 && e == nil || e == io.EOF {
for i = 0; i+NeedleEntrySize <= count; i += NeedleEntrySize {
key, offset, size = IdxFileEntry(bytes[i : i+NeedleEntrySize])
for i = 0; i+NeedleMapEntrySize <= count; i += NeedleMapEntrySize {
key, offset, size = IdxFileEntry(bytes[i : i+NeedleMapEntrySize])
if e = fn(key, offset, size); e != nil { if e = fn(key, offset, size); e != nil {
return e return e
} }

12
weed/storage/needle_map_metric.go
View File

@ -96,16 +96,16 @@ func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key
return fmt.Errorf("file %s stat error: %v", r.Name(), err) return fmt.Errorf("file %s stat error: %v", r.Name(), err)
} }
fileSize := fi.Size() fileSize := fi.Size()
if fileSize%NeedleEntrySize != 0 {
if fileSize%NeedleMapEntrySize != 0 {
return fmt.Errorf("unexpected file %s size: %d", r.Name(), fileSize) return fmt.Errorf("unexpected file %s size: %d", r.Name(), fileSize)
} }
entryCount := fileSize / NeedleEntrySize
entryCount := fileSize / NeedleMapEntrySize
initFn(entryCount) initFn(entryCount)
batchSize := int64(1024 * 4) batchSize := int64(1024 * 4)
bytes := make([]byte, NeedleEntrySize*batchSize)
bytes := make([]byte, NeedleMapEntrySize*batchSize)
nextBatchSize := entryCount % batchSize nextBatchSize := entryCount % batchSize
if nextBatchSize == 0 { if nextBatchSize == 0 {
nextBatchSize = batchSize nextBatchSize = batchSize
@ -113,13 +113,13 @@ func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key
remainingCount := entryCount - nextBatchSize remainingCount := entryCount - nextBatchSize
for remainingCount >= 0 { for remainingCount >= 0 {
_, e := r.ReadAt(bytes[:NeedleEntrySize*nextBatchSize], NeedleEntrySize*remainingCount)
// glog.V(0).Infoln("file", r.Name(), "readerOffset", NeedleEntrySize*remainingCount, "count", count, "e", e)
_, e := r.ReadAt(bytes[:NeedleMapEntrySize*nextBatchSize], NeedleMapEntrySize*remainingCount)
// glog.V(0).Infoln("file", r.Name(), "readerOffset", NeedleMapEntrySize*remainingCount, "count", count, "e", e)
if e != nil { if e != nil {
return e return e
} }
for i := int(nextBatchSize) - 1; i >= 0; i-- { for i := int(nextBatchSize) - 1; i >= 0; i-- {
key, offset, size := IdxFileEntry(bytes[i*NeedleEntrySize : i*NeedleEntrySize+NeedleEntrySize])
key, offset, size := IdxFileEntry(bytes[i*NeedleMapEntrySize : i*NeedleMapEntrySize+NeedleMapEntrySize])
if e = fn(key, offset, size); e != nil { if e = fn(key, offset, size); e != nil {
return e return e
} }

3
weed/storage/types/needle_types.go
View File

@ -23,7 +23,8 @@ type Cookie uint32
const ( const (
SizeSize = 4 // uint32 size SizeSize = 4 // uint32 size
NeedleEntrySize = CookieSize + NeedleIdSize + SizeSize
NeedleHeaderSize = CookieSize + NeedleIdSize + SizeSize
NeedleMapEntrySize = NeedleIdSize + OffsetSize + SizeSize
TimestampSize = 8 // int64 size TimestampSize = 8 // int64 size
NeedlePaddingSize = 8 NeedlePaddingSize = 8
TombstoneFileSize = math.MaxUint32 TombstoneFileSize = math.MaxUint32

31
weed/storage/volume.go
View File

@ -28,7 +28,8 @@ type Volume struct {
SuperBlock SuperBlock
dataFileAccessLock sync.Mutex dataFileAccessLock sync.Mutex
lastModifiedTime uint64 //unix time in seconds
lastModifiedTsSeconds uint64 //unix time in seconds
lastAppendAtNs uint64 //unix time in nanoseconds
lastCompactIndexOffset uint64 lastCompactIndexOffset uint64
lastCompactRevision uint16 lastCompactRevision uint16
@ -66,37 +67,26 @@ func (v *Volume) Version() needle.Version {
return v.SuperBlock.Version() return v.SuperBlock.Version()
} }
func (v *Volume) Size() int64 {
func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
v.dataFileAccessLock.Lock() v.dataFileAccessLock.Lock()
defer v.dataFileAccessLock.Unlock() defer v.dataFileAccessLock.Unlock()
if v.dataFile == nil { if v.dataFile == nil {
return 0
return
} }
stat, e := v.dataFile.Stat() stat, e := v.dataFile.Stat()
if e == nil { if e == nil {
return stat.Size()
return uint64(stat.Size()), v.nm.IndexFileSize(), stat.ModTime()
} }
glog.V(0).Infof("Failed to read file size %s %v", v.dataFile.Name(), e) glog.V(0).Infof("Failed to read file size %s %v", v.dataFile.Name(), e)
return 0 // -1 causes integer overflow and the volume to become unwritable.
return // -1 causes integer overflow and the volume to become unwritable.
} }
func (v *Volume) IndexFileSize() uint64 { func (v *Volume) IndexFileSize() uint64 {
return v.nm.IndexFileSize() return v.nm.IndexFileSize()
} }
func (v *Volume) DataFileSize() uint64 {
return uint64(v.Size())
}
/**
unix time in seconds
*/
func (v *Volume) LastModifiedTime() uint64 {
return v.lastModifiedTime
}
func (v *Volume) FileCount() uint64 { func (v *Volume) FileCount() uint64 {
return uint64(v.nm.FileCount()) return uint64(v.nm.FileCount())
} }
@ -138,8 +128,8 @@ func (v *Volume) expired(volumeSizeLimit uint64) bool {
if v.Ttl == nil || v.Ttl.Minutes() == 0 { if v.Ttl == nil || v.Ttl.Minutes() == 0 {
return false return false
} }
glog.V(1).Infof("now:%v lastModified:%v", time.Now().Unix(), v.lastModifiedTime)
livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTime)) / 60
glog.V(1).Infof("now:%v lastModified:%v", time.Now().Unix(), v.lastModifiedTsSeconds)
livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
glog.V(1).Infof("ttl:%v lived:%v", v.Ttl, livedMinutes) glog.V(1).Infof("ttl:%v lived:%v", v.Ttl, livedMinutes)
if int64(v.Ttl.Minutes()) < livedMinutes { if int64(v.Ttl.Minutes()) < livedMinutes {
return true return true
@ -157,16 +147,17 @@ func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
removalDelay = maxDelayMinutes removalDelay = maxDelayMinutes
} }
if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTime < uint64(time.Now().Unix()) {
if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
return true return true
} }
return false return false
} }
func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage { func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage {
size, _, _ := v.FileStat()
return &master_pb.VolumeInformationMessage{ return &master_pb.VolumeInformationMessage{
Id: uint32(v.Id), Id: uint32(v.Id),
Size: uint64(v.Size()),
Size: size,
Collection: v.Collection, Collection: v.Collection,
FileCount: uint64(v.nm.FileCount()), FileCount: uint64(v.nm.FileCount()),
DeleteCount: uint64(v.nm.DeletedCount()), DeleteCount: uint64(v.nm.DeletedCount()),

26
weed/storage/volume_backup.go
View File

@ -60,7 +60,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial
ctx := context.Background() ctx := context.Background()
startFromOffset := v.Size()
startFromOffset, _, _ := v.FileStat()
appendAtNs, err := v.findLastAppendAtNs() appendAtNs, err := v.findLastAppendAtNs()
if err != nil { if err != nil {
return err return err
@ -76,7 +76,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial
return err return err
} }
v.dataFile.Seek(startFromOffset, io.SeekStart)
v.dataFile.Seek(int64(startFromOffset), io.SeekStart)
for { for {
resp, recvErr := stream.Recv() resp, recvErr := stream.Recv()
@ -103,7 +103,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial
} }
// add to needle map // add to needle map
return ScanVolumeFileFrom(v.version, v.dataFile, startFromOffset, &VolumeFileScanner4GenIdx{v: v})
return ScanVolumeFileFrom(v.version, v.dataFile, int64(startFromOffset), &VolumeFileScanner4GenIdx{v: v})
} }
@ -130,16 +130,16 @@ func (v *Volume) locateLastAppendEntry() (Offset, error) {
return Offset{}, fmt.Errorf("file %s stat error: %v", indexFile.Name(), err) return Offset{}, fmt.Errorf("file %s stat error: %v", indexFile.Name(), err)
} }
fileSize := fi.Size() fileSize := fi.Size()
if fileSize%NeedleEntrySize != 0 {
if fileSize%NeedleMapEntrySize != 0 {
return Offset{}, fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) return Offset{}, fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize)
} }
if fileSize == 0 { if fileSize == 0 {
return Offset{}, nil return Offset{}, nil
} }
bytes := make([]byte, NeedleEntrySize)
n, e := indexFile.ReadAt(bytes, fileSize-NeedleEntrySize)
if n != NeedleEntrySize {
bytes := make([]byte, NeedleMapEntrySize)
n, e := indexFile.ReadAt(bytes, fileSize-NeedleMapEntrySize)
if n != NeedleMapEntrySize {
return Offset{}, fmt.Errorf("file %s read error: %v", indexFile.Name(), e) return Offset{}, fmt.Errorf("file %s read error: %v", indexFile.Name(), e)
} }
_, offset, _ := IdxFileEntry(bytes) _, offset, _ := IdxFileEntry(bytes)
@ -153,7 +153,7 @@ func (v *Volume) readAppendAtNs(offset Offset) (uint64, error) {
if err != nil { if err != nil {
return 0, fmt.Errorf("ReadNeedleHeader: %v", err) return 0, fmt.Errorf("ReadNeedleHeader: %v", err)
} }
_, err = n.ReadNeedleBody(v.dataFile, v.SuperBlock.version, offset.ToAcutalOffset()+int64(NeedleEntrySize), bodyLength)
_, err = n.ReadNeedleBody(v.dataFile, v.SuperBlock.version, offset.ToAcutalOffset()+int64(NeedleHeaderSize), bodyLength)
if err != nil { if err != nil {
return 0, fmt.Errorf("ReadNeedleBody offset %d, bodyLength %d: %v", offset.ToAcutalOffset(), bodyLength, err) return 0, fmt.Errorf("ReadNeedleBody offset %d, bodyLength %d: %v", offset.ToAcutalOffset(), bodyLength, err)
} }
@ -176,13 +176,13 @@ func (v *Volume) BinarySearchByAppendAtNs(sinceNs uint64) (offset Offset, isLast
return return
} }
fileSize := fi.Size() fileSize := fi.Size()
if fileSize%NeedleEntrySize != 0 {
if fileSize%NeedleMapEntrySize != 0 {
err = fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) err = fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize)
return return
} }
bytes := make([]byte, NeedleEntrySize)
entryCount := fileSize / NeedleEntrySize
bytes := make([]byte, NeedleMapEntrySize)
entryCount := fileSize / NeedleMapEntrySize
l := int64(0) l := int64(0)
h := entryCount h := entryCount
@ -225,9 +225,9 @@ func (v *Volume) BinarySearchByAppendAtNs(sinceNs uint64) (offset Offset, isLast
} }
// bytes is of size NeedleEntrySize
// bytes is of size NeedleMapEntrySize
func (v *Volume) readAppendAtNsForIndexEntry(indexFile *os.File, bytes []byte, m int64) (Offset, error) { func (v *Volume) readAppendAtNsForIndexEntry(indexFile *os.File, bytes []byte, m int64) (Offset, error) {
if _, readErr := indexFile.ReadAt(bytes, m*NeedleEntrySize); readErr != nil && readErr != io.EOF {
if _, readErr := indexFile.ReadAt(bytes, m*NeedleMapEntrySize); readErr != nil && readErr != io.EOF {
return Offset{}, readErr return Offset{}, readErr
} }
_, offset, _ := IdxFileEntry(bytes) _, offset, _ := IdxFileEntry(bytes)

6
weed/storage/volume_checking.go
View File

@ -19,7 +19,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) error {
return nil return nil
} }
var lastIdxEntry []byte var lastIdxEntry []byte
if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleEntrySize); e != nil {
if lastIdxEntry, e = readIndexEntryAtOffset(indexFile, indexSize-NeedleMapEntrySize); e != nil {
return fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e) return fmt.Errorf("readLastIndexEntry %s failed: %v", indexFile.Name(), e)
} }
key, offset, size := IdxFileEntry(lastIdxEntry) key, offset, size := IdxFileEntry(lastIdxEntry)
@ -35,7 +35,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) error {
func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) { func verifyIndexFileIntegrity(indexFile *os.File) (indexSize int64, err error) {
if indexSize, err = util.GetFileSize(indexFile); err == nil { if indexSize, err = util.GetFileSize(indexFile); err == nil {
if indexSize%NeedleEntrySize != 0 {
if indexSize%NeedleMapEntrySize != 0 {
err = fmt.Errorf("index file's size is %d bytes, maybe corrupted", indexSize) err = fmt.Errorf("index file's size is %d bytes, maybe corrupted", indexSize)
} }
} }
@ -47,7 +47,7 @@ func readIndexEntryAtOffset(indexFile *os.File, offset int64) (bytes []byte, err
err = fmt.Errorf("offset %d for index file is invalid", offset) err = fmt.Errorf("offset %d for index file is invalid", offset)
return return
} }
bytes = make([]byte, NeedleEntrySize)
bytes = make([]byte, NeedleMapEntrySize)
_, err = indexFile.ReadAt(bytes, offset) _, err = indexFile.ReadAt(bytes, offset)
return return
} }

2
weed/storage/volume_loading.go
View File

@ -30,7 +30,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind
} }
if canWrite { if canWrite {
v.dataFile, e = os.OpenFile(fileName+".dat", os.O_RDWR|os.O_CREATE, 0644) v.dataFile, e = os.OpenFile(fileName+".dat", os.O_RDWR|os.O_CREATE, 0644)
v.lastModifiedTime = uint64(modifiedTime.Unix())
v.lastModifiedTsSeconds = uint64(modifiedTime.Unix())
} else { } else {
glog.V(0).Infoln("opening " + fileName + ".dat in READONLY mode") glog.V(0).Infoln("opening " + fileName + ".dat in READONLY mode")
v.dataFile, e = os.Open(fileName + ".dat") v.dataFile, e = os.Open(fileName + ".dat")

14
weed/storage/volume_read_write.go
View File

@ -95,6 +95,7 @@ func (v *Volume) writeNeedle(n *needle.Needle) (offset uint64, size uint32, err
if offset, size, _, err = n.Append(v.dataFile, v.Version()); err != nil { if offset, size, _, err = n.Append(v.dataFile, v.Version()); err != nil {
return return
} }
v.lastAppendAtNs = n.AppendAtNs
nv, ok := v.nm.Get(n.Id) nv, ok := v.nm.Get(n.Id)
if !ok || uint64(nv.Offset.ToAcutalOffset()) < offset { if !ok || uint64(nv.Offset.ToAcutalOffset()) < offset {
@ -102,8 +103,8 @@ func (v *Volume) writeNeedle(n *needle.Needle) (offset uint64, size uint32, err
glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err) glog.V(4).Infof("failed to save in needle map %d: %v", n.Id, err)
} }
} }
if v.lastModifiedTime < n.LastModified {
v.lastModifiedTime = n.LastModified
if v.lastModifiedTsSeconds < n.LastModified {
v.lastModifiedTsSeconds = n.LastModified
} }
return return
} }
@ -125,6 +126,7 @@ func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) {
if err != nil { if err != nil {
return size, err return size, err
} }
v.lastAppendAtNs = n.AppendAtNs
if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil { if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil {
return size, err return size, err
} }
@ -205,7 +207,7 @@ func ScanVolumeFileFrom(version needle.Version, dataFile *os.File, offset int64,
} }
for n != nil { for n != nil {
if volumeFileScanner.ReadNeedleBody() { if volumeFileScanner.ReadNeedleBody() {
if _, err = n.ReadNeedleBody(dataFile, version, offset+NeedleEntrySize, rest); err != nil {
if _, err = n.ReadNeedleBody(dataFile, version, offset+NeedleHeaderSize, rest); err != nil {
glog.V(0).Infof("cannot read needle body: %v", err) glog.V(0).Infof("cannot read needle body: %v", err)
//err = fmt.Errorf("cannot read needle body: %v", err) //err = fmt.Errorf("cannot read needle body: %v", err)
//return //return
@ -218,7 +220,7 @@ func ScanVolumeFileFrom(version needle.Version, dataFile *os.File, offset int64,
if err != nil { if err != nil {
glog.V(0).Infof("visit needle error: %v", err) glog.V(0).Infof("visit needle error: %v", err)
} }
offset += NeedleEntrySize + rest
offset += NeedleHeaderSize + rest
glog.V(4).Infof("==> new entry offset %d", offset) glog.V(4).Infof("==> new entry offset %d", offset)
if n, _, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil { if n, _, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil {
if err == io.EOF { if err == io.EOF {
@ -241,7 +243,7 @@ func ScanVolumeFileNeedleFrom(version needle.Version, dataFile *os.File, offset
} }
for n != nil { for n != nil {
var needleBody []byte var needleBody []byte
if needleBody, err = n.ReadNeedleBody(dataFile, version, offset+NeedleEntrySize, rest); err != nil {
if needleBody, err = n.ReadNeedleBody(dataFile, version, offset+NeedleHeaderSize, rest); err != nil {
glog.V(0).Infof("cannot read needle body: %v", err) glog.V(0).Infof("cannot read needle body: %v", err)
//err = fmt.Errorf("cannot read needle body: %v", err) //err = fmt.Errorf("cannot read needle body: %v", err)
//return //return
@ -251,7 +253,7 @@ func ScanVolumeFileNeedleFrom(version needle.Version, dataFile *os.File, offset
glog.V(0).Infof("visit needle error: %v", err) glog.V(0).Infof("visit needle error: %v", err)
return return
} }
offset += NeedleEntrySize + rest
offset += NeedleHeaderSize + rest
glog.V(4).Infof("==> new entry offset %d", offset) glog.V(4).Infof("==> new entry offset %d", offset)
if n, nh, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil { if n, nh, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil {
if err == io.EOF { if err == io.EOF {

2
weed/storage/volume_vacuum.go
View File

@ -138,7 +138,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI
} }
incrementedHasUpdatedIndexEntry := make(map[NeedleId]keyField) incrementedHasUpdatedIndexEntry := make(map[NeedleId]keyField)
for idxOffset := indexSize - NeedleEntrySize; uint64(idxOffset) >= v.lastCompactIndexOffset; idxOffset -= NeedleEntrySize {
for idxOffset := indexSize - NeedleMapEntrySize; uint64(idxOffset) >= v.lastCompactIndexOffset; idxOffset -= NeedleMapEntrySize {
var IdxEntry []byte var IdxEntry []byte
if IdxEntry, err = readIndexEntryAtOffset(oldIdxFile, idxOffset); err != nil { if IdxEntry, err = readIndexEntryAtOffset(oldIdxFile, idxOffset); err != nil {
return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idxOffset, err) return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idxOffset, err)

Write Preview
Loading…
Cancel
Save