From ac2727853f94c8fc4e301a995997f02b70047853 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 19 Apr 2019 00:39:34 -0700 Subject: [PATCH] fix needle map entry size --- unmaintained/fix_dat/fix_dat.go | 4 +- weed/command/backup.go | 4 +- weed/pb/volume_server.proto | 5 +- weed/pb/volume_server_pb/volume_server.pb.go | 196 +++++++++--------- weed/server/volume_grpc_copy.go | 19 +- weed/server/volume_grpc_copy_incremental.go | 4 +- weed/storage/disk_location.go | 3 +- weed/storage/needle/needle_read_write.go | 26 +-- .../needle_map/compact_map_perf_test.go | 4 +- weed/storage/needle_map_memory.go | 6 +- weed/storage/needle_map_metric.go | 12 +- weed/storage/types/needle_types.go | 13 +- weed/storage/volume.go | 33 ++- weed/storage/volume_backup.go | 26 +-- weed/storage/volume_checking.go | 6 +- weed/storage/volume_loading.go | 2 +- weed/storage/volume_read_write.go | 14 +- weed/storage/volume_vacuum.go | 2 +- 18 files changed, 195 insertions(+), 184 deletions(-) diff --git a/unmaintained/fix_dat/fix_dat.go b/unmaintained/fix_dat/fix_dat.go index f8a535aaf..a72a78eed 100644 --- a/unmaintained/fix_dat/fix_dat.go +++ b/unmaintained/fix_dat/fix_dat.go @@ -115,7 +115,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *needle.Needle 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) } }() @@ -125,7 +125,7 @@ func iterateEntries(datFile, idxFile *os.File, visitNeedle func(n *needle.Needle } visitNeedle(n, offset) - offset += types.NeedleEntrySize + rest + offset += types.NeedleHeaderSize + rest //fmt.Printf("==> new entry offset %d\n", offset) if n, _, rest, err = needle.ReadNeedleHeader(datFile, version, offset); err != nil { if err == io.EOF { diff --git a/weed/command/backup.go b/weed/command/backup.go index e2b0da7dd..0e14e03ec 100644 --- a/weed/command/backup.go +++ b/weed/command/backup.go @@ -103,7 +103,9 @@ func runBackup(cmd *Command, args []string) bool { v.DataFile().WriteAt(v.SuperBlock.Bytes(), 0) } - if uint64(v.Size()) > stats.TailOffset { + datSize, _, _ := v.FileStat() + + if datSize > stats.TailOffset { // remove the old data v.Destroy() // recreate an empty volume diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 70f391959..a00ccc053 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -155,6 +155,7 @@ message VolumeCopyRequest { string source_data_node = 5; } message VolumeCopyResponse { + uint64 last_append_at_ns = 1; } message CopyFileRequest { @@ -182,9 +183,9 @@ message ReadVolumeFileStatusRequest { } message ReadVolumeFileStatusResponse { uint32 volume_id = 1; - uint64 idx_file_timestamp = 2; + uint64 idx_file_timestamp_seconds = 2; uint64 idx_file_size = 3; - uint64 dat_file_timestamp = 4; + uint64 dat_file_timestamp_seconds = 4; uint64 dat_file_size = 5; uint64 file_count = 6; } diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 4df6db272..193d6d089 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -581,6 +581,7 @@ func (m *VolumeCopyRequest) GetSourceDataNode() string { } 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{} } @@ -588,6 +589,13 @@ func (m *VolumeCopyResponse) String() string { return proto.CompactTe func (*VolumeCopyResponse) ProtoMessage() {} 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 { 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"` @@ -717,12 +725,12 @@ func (m *ReadVolumeFileStatusRequest) GetVolumeId() uint32 { } type ReadVolumeFileStatusResponse struct { - 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"` - 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"` - 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"` + VolumeId uint32 `protobuf:"varint,1,opt,name=volume_id,json=volumeId" json:"volume_id,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"` + 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"` + FileCount uint64 `protobuf:"varint,6,opt,name=file_count,json=fileCount" json:"file_count,omitempty"` } func (m *ReadVolumeFileStatusResponse) Reset() { *m = ReadVolumeFileStatusResponse{} } @@ -737,9 +745,9 @@ func (m *ReadVolumeFileStatusResponse) GetVolumeId() uint32 { return 0 } -func (m *ReadVolumeFileStatusResponse) GetIdxFileTimestamp() uint64 { +func (m *ReadVolumeFileStatusResponse) GetIdxFileTimestampSeconds() uint64 { if m != nil { - return m.IdxFileTimestamp + return m.IdxFileTimestampSeconds } return 0 } @@ -751,9 +759,9 @@ func (m *ReadVolumeFileStatusResponse) GetIdxFileSize() uint64 { return 0 } -func (m *ReadVolumeFileStatusResponse) GetDatFileTimestamp() uint64 { +func (m *ReadVolumeFileStatusResponse) GetDatFileTimestampSeconds() uint64 { if m != nil { - return m.DatFileTimestamp + return m.DatFileTimestampSeconds } return 0 } @@ -1571,87 +1579,89 @@ var _VolumeServer_serviceDesc = grpc.ServiceDesc{ func init() { proto.RegisterFile("volume_server.proto", fileDescriptor0) } 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, } diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index b7789f88d..0cb5bb455 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -3,15 +3,15 @@ package weed_server import ( "context" "fmt" + "io" + "os" + "time" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/operation" "github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb" "github.com/chrislusf/seaweedfs/weed/storage" "github.com/chrislusf/seaweedfs/weed/storage/needle" - - "io" - "os" ) // 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 &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.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() return resp, nil } diff --git a/weed/server/volume_grpc_copy_incremental.go b/weed/server/volume_grpc_copy_incremental.go index 5977c44f6..f56fbeef4 100644 --- a/weed/server/volume_grpc_copy_incremental.go +++ b/weed/server/volume_grpc_copy_incremental.go @@ -17,7 +17,7 @@ func (vs *VolumeServer) VolumeIncrementalCopy(req *volume_server_pb.VolumeIncrem return fmt.Errorf("not found volume id %d", req.VolumeId) } - stopOffset := v.Size() + stopOffset, _, _ := v.FileStat() foundOffset, isLastOne, err := v.BinarySearchByAppendAtNs(req.SinceNs) if err != nil { 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() buf := make([]byte, 1024*1024*2) - return sendFileContent(v.DataFile(), buf, startOffset, stopOffset, stream) + return sendFileContent(v.DataFile(), buf, startOffset, int64(stopOffset), stream) } diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index 680cc6097..6bfb1ff76 100644 --- a/weed/storage/disk_location.go +++ b/weed/storage/disk_location.go @@ -54,8 +54,9 @@ func (l *DiskLocation) loadExistingVolume(dir os.FileInfo, needleMapKind NeedleM mutex.Lock() l.volumes[vid] = v mutex.Unlock() + size, _, _ := v.FileStat() 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 { glog.V(0).Infof("new volume %s error %s", name, e) } diff --git a/weed/storage/needle/needle_read_write.go b/weed/storage/needle/needle_read_write.go index 9876b2f1c..d1df7d4e2 100644 --- a/weed/storage/needle/needle_read_write.go +++ b/weed/storage/needle/needle_read_write.go @@ -45,7 +45,7 @@ func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32 } switch version { case Version1: - header := make([]byte, NeedleEntrySize) + header := make([]byte, NeedleHeaderSize) CookieToBytes(header[0:CookieSize], n.Cookie) NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id) 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 { return } - actualSize = NeedleEntrySize + int64(n.Size) + actualSize = NeedleHeaderSize + int64(n.Size) padding := PaddingLength(n.Size, version) util.Uint32toBytes(header[0:NeedleChecksumSize], n.Checksum.Value()) _, err = w.Write(header[0 : NeedleChecksumSize+padding]) return 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) NeedleIdToBytes(header[CookieSize:CookieSize+NeedleIdSize], n.Id) 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 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 } if n.DataSize > 0 { @@ -181,21 +181,21 @@ func (n *Needle) ReadData(r *os.File, offset int64, size uint32, version Version } switch version { case Version1: - n.Data = bytes[NeedleEntrySize : NeedleEntrySize+size] + n.Data = bytes[NeedleHeaderSize : NeedleHeaderSize+size] 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 { return err } - checksum := util.BytesToUint32(bytes[NeedleEntrySize+size : NeedleEntrySize+size+NeedleChecksumSize]) + checksum := util.BytesToUint32(bytes[NeedleHeaderSize+size : NeedleHeaderSize+size+NeedleChecksumSize]) newChecksum := NewCRC(n.Data) if checksum != newChecksum.Value() { return errors.New("CRC error! Data On Disk Corrupted") } n.Checksum = newChecksum if version == Version3 { - tsOffset := NeedleEntrySize + size + NeedleChecksumSize + tsOffset := NeedleHeaderSize + size + NeedleChecksumSize n.AppendAtNs = util.BytesToUint64(bytes[tsOffset : tsOffset+TimestampSize]) } 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) { n.Cookie = BytesToCookie(bytes[0:CookieSize]) 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) { @@ -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) { n = new(Needle) if version == Version1 || version == Version2 || version == Version3 { - bytes = make([]byte, NeedleEntrySize) + bytes = make([]byte, NeedleHeaderSize) var count int count, err = r.ReadAt(bytes, offset) 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 { if version == Version3 { // 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 { @@ -386,6 +386,6 @@ func (n *Needle) SetHasPairs() { } func getActualSize(size uint32, version Version) int64 { - return NeedleEntrySize + NeedleBodyLength(size, version) + return NeedleHeaderSize + NeedleBodyLength(size, version) } diff --git a/weed/storage/needle_map/compact_map_perf_test.go b/weed/storage/needle_map/compact_map_perf_test.go index a447375a4..3a3648641 100644 --- a/weed/storage/needle_map/compact_map_perf_test.go +++ b/weed/storage/needle_map/compact_map_perf_test.go @@ -52,11 +52,11 @@ func TestMemoryUsage(t *testing.T) { func loadNewNeedleMap(file *os.File) (*CompactMap, uint64) { m := NewCompactMap() - bytes := make([]byte, NeedleEntrySize) + bytes := make([]byte, NeedleMapEntrySize) rowCount := uint64(0) count, e := file.Read(bytes) for count > 0 && e == nil { - for i := 0; i < count; i += NeedleEntrySize { + for i := 0; i < count; i += NeedleMapEntrySize { rowCount++ key := BytesToNeedleId(bytes[i : i+NeedleIdSize]) offset := BytesToOffset(bytes[i+NeedleIdSize : i+NeedleIdSize+OffsetSize]) diff --git a/weed/storage/needle_map_memory.go b/weed/storage/needle_map_memory.go index 060920cea..727049a13 100644 --- a/weed/storage/needle_map_memory.go +++ b/weed/storage/needle_map_memory.go @@ -73,7 +73,7 @@ func doLoading(file *os.File, nm *NeedleMap) (*NeedleMap, error) { // stops with the error returned by the fn function func WalkIndexFile(r *os.File, fn func(key NeedleId, offset Offset, size uint32) error) error { var readerOffset int64 - bytes := make([]byte, NeedleEntrySize*RowsToRead) + bytes := make([]byte, NeedleMapEntrySize*RowsToRead) count, e := r.ReadAt(bytes, readerOffset) glog.V(3).Infoln("file", r.Name(), "readerOffset", readerOffset, "count", count, "e", e) 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 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 { return e } diff --git a/weed/storage/needle_map_metric.go b/weed/storage/needle_map_metric.go index 8cd16aedf..5c251793c 100644 --- a/weed/storage/needle_map_metric.go +++ b/weed/storage/needle_map_metric.go @@ -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) } fileSize := fi.Size() - if fileSize%NeedleEntrySize != 0 { + if fileSize%NeedleMapEntrySize != 0 { return fmt.Errorf("unexpected file %s size: %d", r.Name(), fileSize) } - entryCount := fileSize / NeedleEntrySize + entryCount := fileSize / NeedleMapEntrySize initFn(entryCount) batchSize := int64(1024 * 4) - bytes := make([]byte, NeedleEntrySize*batchSize) + bytes := make([]byte, NeedleMapEntrySize*batchSize) nextBatchSize := entryCount % batchSize if nextBatchSize == 0 { nextBatchSize = batchSize @@ -113,13 +113,13 @@ func reverseWalkIndexFile(r *os.File, initFn func(entryCount int64), fn func(key remainingCount := entryCount - nextBatchSize 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 { return e } 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 { return e } diff --git a/weed/storage/types/needle_types.go b/weed/storage/types/needle_types.go index 2815acaa2..2ebb392db 100644 --- a/weed/storage/types/needle_types.go +++ b/weed/storage/types/needle_types.go @@ -22,12 +22,13 @@ type OffsetLower struct { type Cookie uint32 const ( - SizeSize = 4 // uint32 size - NeedleEntrySize = CookieSize + NeedleIdSize + SizeSize - TimestampSize = 8 // int64 size - NeedlePaddingSize = 8 - TombstoneFileSize = math.MaxUint32 - CookieSize = 4 + SizeSize = 4 // uint32 size + NeedleHeaderSize = CookieSize + NeedleIdSize + SizeSize + NeedleMapEntrySize = NeedleIdSize + OffsetSize + SizeSize + TimestampSize = 8 // int64 size + NeedlePaddingSize = 8 + TombstoneFileSize = math.MaxUint32 + CookieSize = 4 ) func CookieToBytes(bytes []byte, cookie Cookie) { diff --git a/weed/storage/volume.go b/weed/storage/volume.go index e33d90d28..616a2aee8 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -27,8 +27,9 @@ type Volume struct { SuperBlock - dataFileAccessLock sync.Mutex - lastModifiedTime uint64 //unix time in seconds + dataFileAccessLock sync.Mutex + lastModifiedTsSeconds uint64 //unix time in seconds + lastAppendAtNs uint64 //unix time in nanoseconds lastCompactIndexOffset uint64 lastCompactRevision uint16 @@ -66,37 +67,26 @@ func (v *Volume) Version() needle.Version { return v.SuperBlock.Version() } -func (v *Volume) Size() int64 { +func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) { v.dataFileAccessLock.Lock() defer v.dataFileAccessLock.Unlock() if v.dataFile == nil { - return 0 + return } stat, e := v.dataFile.Stat() 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) - 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 { 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 { return uint64(v.nm.FileCount()) } @@ -138,8 +128,8 @@ func (v *Volume) expired(volumeSizeLimit uint64) bool { if v.Ttl == nil || v.Ttl.Minutes() == 0 { 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) if int64(v.Ttl.Minutes()) < livedMinutes { return true @@ -157,16 +147,17 @@ func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool { 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 false } func (v *Volume) ToVolumeInformationMessage() *master_pb.VolumeInformationMessage { + size, _, _ := v.FileStat() return &master_pb.VolumeInformationMessage{ Id: uint32(v.Id), - Size: uint64(v.Size()), + Size: size, Collection: v.Collection, FileCount: uint64(v.nm.FileCount()), DeleteCount: uint64(v.nm.DeletedCount()), diff --git a/weed/storage/volume_backup.go b/weed/storage/volume_backup.go index d25f945f8..8b7238a8c 100644 --- a/weed/storage/volume_backup.go +++ b/weed/storage/volume_backup.go @@ -60,7 +60,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial ctx := context.Background() - startFromOffset := v.Size() + startFromOffset, _, _ := v.FileStat() appendAtNs, err := v.findLastAppendAtNs() if err != nil { return err @@ -76,7 +76,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial return err } - v.dataFile.Seek(startFromOffset, io.SeekStart) + v.dataFile.Seek(int64(startFromOffset), io.SeekStart) for { resp, recvErr := stream.Recv() @@ -103,7 +103,7 @@ func (v *Volume) IncrementalBackup(volumeServer string, grpcDialOption grpc.Dial } // 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) } fileSize := fi.Size() - if fileSize%NeedleEntrySize != 0 { + if fileSize%NeedleMapEntrySize != 0 { return Offset{}, fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) } if fileSize == 0 { 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) } _, offset, _ := IdxFileEntry(bytes) @@ -153,7 +153,7 @@ func (v *Volume) readAppendAtNs(offset Offset) (uint64, error) { if err != nil { 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 { 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 } fileSize := fi.Size() - if fileSize%NeedleEntrySize != 0 { + if fileSize%NeedleMapEntrySize != 0 { err = fmt.Errorf("unexpected file %s size: %d", indexFile.Name(), fileSize) return } - bytes := make([]byte, NeedleEntrySize) - entryCount := fileSize / NeedleEntrySize + bytes := make([]byte, NeedleMapEntrySize) + entryCount := fileSize / NeedleMapEntrySize l := int64(0) 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) { - 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 } _, offset, _ := IdxFileEntry(bytes) diff --git a/weed/storage/volume_checking.go b/weed/storage/volume_checking.go index 0f419f2c0..9599885ec 100644 --- a/weed/storage/volume_checking.go +++ b/weed/storage/volume_checking.go @@ -19,7 +19,7 @@ func CheckVolumeDataIntegrity(v *Volume, indexFile *os.File) error { return nil } 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) } 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) { 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) } } @@ -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) return } - bytes = make([]byte, NeedleEntrySize) + bytes = make([]byte, NeedleMapEntrySize) _, err = indexFile.ReadAt(bytes, offset) return } diff --git a/weed/storage/volume_loading.go b/weed/storage/volume_loading.go index 82eb98b45..eb82ffe98 100644 --- a/weed/storage/volume_loading.go +++ b/weed/storage/volume_loading.go @@ -30,7 +30,7 @@ func (v *Volume) load(alsoLoadIndex bool, createDatIfMissing bool, needleMapKind } if canWrite { 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 { glog.V(0).Infoln("opening " + fileName + ".dat in READONLY mode") v.dataFile, e = os.Open(fileName + ".dat") diff --git a/weed/storage/volume_read_write.go b/weed/storage/volume_read_write.go index 45888d53b..222f11d66 100644 --- a/weed/storage/volume_read_write.go +++ b/weed/storage/volume_read_write.go @@ -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 { return } + v.lastAppendAtNs = n.AppendAtNs nv, ok := v.nm.Get(n.Id) 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) } } - if v.lastModifiedTime < n.LastModified { - v.lastModifiedTime = n.LastModified + if v.lastModifiedTsSeconds < n.LastModified { + v.lastModifiedTsSeconds = n.LastModified } return } @@ -125,6 +126,7 @@ func (v *Volume) deleteNeedle(n *needle.Needle) (uint32, error) { if err != nil { return size, err } + v.lastAppendAtNs = n.AppendAtNs if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil { return size, err } @@ -205,7 +207,7 @@ func ScanVolumeFileFrom(version needle.Version, dataFile *os.File, offset int64, } for n != nil { 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) //err = fmt.Errorf("cannot read needle body: %v", err) //return @@ -218,7 +220,7 @@ func ScanVolumeFileFrom(version needle.Version, dataFile *os.File, offset int64, if err != nil { glog.V(0).Infof("visit needle error: %v", err) } - offset += NeedleEntrySize + rest + offset += NeedleHeaderSize + rest glog.V(4).Infof("==> new entry offset %d", offset) if n, _, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil { if err == io.EOF { @@ -241,7 +243,7 @@ func ScanVolumeFileNeedleFrom(version needle.Version, dataFile *os.File, offset } for n != nil { 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) //err = fmt.Errorf("cannot read needle body: %v", err) //return @@ -251,7 +253,7 @@ func ScanVolumeFileNeedleFrom(version needle.Version, dataFile *os.File, offset glog.V(0).Infof("visit needle error: %v", err) return } - offset += NeedleEntrySize + rest + offset += NeedleHeaderSize + rest glog.V(4).Infof("==> new entry offset %d", offset) if n, nh, rest, err = needle.ReadNeedleHeader(dataFile, version, offset); err != nil { if err == io.EOF { diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index 109955bb5..3287ae391 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -138,7 +138,7 @@ func (v *Volume) makeupDiff(newDatFileName, newIdxFileName, oldDatFileName, oldI } 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 if IdxEntry, err = readIndexEntryAtOffset(oldIdxFile, idxOffset); err != nil { return fmt.Errorf("readIndexEntry %s at offset %d failed: %v", oldIdxFileName, idxOffset, err)