From 4f76342cbc8e6a2b48a3bfe6478170ec8f487f8b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 28 May 2019 00:51:01 -0700 Subject: [PATCH] WIP no errors, but not returning file content * the interval needs to use actual file zie * need to read the actual version instead of the current version --- weed/storage/erasure_coding/ec_volume.go | 4 +++- weed/storage/needle/needle_read_write.go | 8 +++---- weed/storage/store_ec.go | 27 +++++++++++++++--------- 3 files changed, 24 insertions(+), 15 deletions(-) diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index ecdc372bb..28f26d683 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -41,6 +41,8 @@ func NewEcVolume(dir string, collection string, vid needle.VolumeId) (ev *EcVolu } ev.ecxFileSize = ecxFi.Size() + ev.ShardLocations = make(map[ShardId][]string) + return } @@ -51,7 +53,7 @@ func (ev *EcVolume) AddEcVolumeShard(ecVolumeShard *EcVolumeShard) bool { } } ev.Shards = append(ev.Shards, ecVolumeShard) - sort.Slice(ev, func(i, j int) bool { + sort.Slice(ev.Shards, func(i, j int) bool { return ev.Shards[i].VolumeId < ev.Shards[j].VolumeId || ev.Shards[i].VolumeId == ev.Shards[j].VolumeId && ev.Shards[i].ShardId < ev.Shards[j].ShardId }) diff --git a/weed/storage/needle/needle_read_write.go b/weed/storage/needle/needle_read_write.go index f2d699d11..5444b076e 100644 --- a/weed/storage/needle/needle_read_write.go +++ b/weed/storage/needle/needle_read_write.go @@ -26,7 +26,7 @@ const ( ) func (n *Needle) DiskSize(version Version) int64 { - return getActualSize(n.Size, version) + return GetActualSize(n.Size, version) } func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32, actualSize int64, err error) { @@ -159,13 +159,13 @@ func (n *Needle) Append(w *os.File, version Version) (offset uint64, size uint32 _, err = w.Write(header[0 : NeedleChecksumSize+TimestampSize+padding]) } - return offset, n.DataSize, getActualSize(n.Size, version), err + return offset, n.DataSize, GetActualSize(n.Size, version), err } return 0, 0, 0, fmt.Errorf("Unsupported Version! (%d)", version) } func ReadNeedleBlob(r *os.File, offset int64, size uint32, version Version) (dataSlice []byte, err error) { - dataSlice = make([]byte, int(getActualSize(size, version))) + dataSlice = make([]byte, int(GetActualSize(size, version))) _, err = r.ReadAt(dataSlice, offset) return dataSlice, err } @@ -393,6 +393,6 @@ func (n *Needle) SetHasPairs() { n.Flags = n.Flags | FlagHasPairs } -func getActualSize(size uint32, version Version) int64 { +func GetActualSize(size uint32, version Version) int64 { return NeedleHeaderSize + NeedleBodyLength(size, version) } diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 41d885af1..fa1da31b4 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -101,13 +101,18 @@ func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *n return 0, err } - bytes, err := s.readEcShardIntervals(ctx, vid, localEcVolume, intervals) + glog.V(4).Infof("read ec volume %d offset %d size %d intervals:%+v", vid, offset.ToAcutalOffset(), size, intervals) + + // TODO need to read the version + version := needle.CurrentVersion + + // TODO the interval size should be the actual size + + bytes, err := s.readEcShardIntervals(ctx, vid, localEcVolume, version, intervals) if err != nil { return 0, fmt.Errorf("ReadEcShardIntervals: %v", err) } - version := needle.CurrentVersion - err = n.ReadBytes(bytes, offset.ToAcutalOffset(), size, version) if err != nil { return 0, fmt.Errorf("readbytes: %v", err) @@ -119,14 +124,14 @@ func (s *Store) ReadEcShardNeedle(ctx context.Context, vid needle.VolumeId, n *n return 0, fmt.Errorf("ec shard %d not found", vid) } -func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume, intervals []erasure_coding.Interval) (data []byte, err error) { +func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, ecVolume *erasure_coding.EcVolume, version needle.Version, intervals []erasure_coding.Interval) (data []byte, err error) { if err = s.cachedLookupEcShardLocations(ctx, ecVolume); err != nil { return nil, fmt.Errorf("failed to locate shard via master grpc %s: %v", s.MasterAddress, err) } for i, interval := range intervals { - if d, e := s.readOneEcShardInterval(ctx, ecVolume, interval); e != nil { + if d, e := s.readOneEcShardInterval(ctx, ecVolume, version, interval); e != nil { return nil, e } else { if i == 0 { @@ -139,11 +144,13 @@ func (s *Store) readEcShardIntervals(ctx context.Context, vid needle.VolumeId, e return } -func (s *Store) readOneEcShardInterval(ctx context.Context, ecVolume *erasure_coding.EcVolume, interval erasure_coding.Interval) (data []byte, err error) { +func (s *Store) readOneEcShardInterval(ctx context.Context, ecVolume *erasure_coding.EcVolume, version needle.Version, interval erasure_coding.Interval) (data []byte, err error) { shardId, actualOffset := interval.ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize) - data = make([]byte, interval.Size) + data = make([]byte, int(needle.GetActualSize(interval.Size, version))) if shard, found := ecVolume.FindEcVolumeShard(shardId); found { + glog.V(3).Infof("read local ec shard %d.%d", ecVolume.VolumeId, shardId) if _, err = shard.ReadAt(data, actualOffset); err != nil { + glog.V(0).Infof("read local ec shard %d.%d: %v", ecVolume.VolumeId, shardId, err) return } } else { @@ -153,6 +160,7 @@ func (s *Store) readOneEcShardInterval(ctx context.Context, ecVolume *erasure_co if !found || len(sourceDataNodes) == 0 { return nil, fmt.Errorf("failed to find ec shard %d.%d", ecVolume.VolumeId, shardId) } + glog.V(3).Infof("read remote ec shard %d.%d from %s", ecVolume.VolumeId, shardId, sourceDataNodes[0]) _, err = s.readOneRemoteEcShardInterval(ctx, sourceDataNodes[0], ecVolume.VolumeId, shardId, data, actualOffset) if err != nil { glog.V(1).Infof("failed to read from %s for ec shard %d.%d : %v", sourceDataNodes[0], ecVolume.VolumeId, shardId, err) @@ -168,8 +176,7 @@ func (s *Store) cachedLookupEcShardLocations(ctx context.Context, ecVolume *eras return nil } - ecVolume.ShardLocationsLock.Lock() - defer ecVolume.ShardLocationsLock.Unlock() + glog.V(3).Infof("lookup and cache ec volume %d locations", ecVolume.VolumeId) err = operation.WithMasterServerClient(s.MasterAddress, s.grpcDialOption, func(masterClient master_pb.SeaweedClient) error { req := &master_pb.LookupEcVolumeRequest{ @@ -183,7 +190,7 @@ func (s *Store) cachedLookupEcShardLocations(ctx context.Context, ecVolume *eras ecVolume.ShardLocationsLock.Lock() for _, shardIdLocations := range resp.ShardIdLocations { shardId := erasure_coding.ShardId(shardIdLocations.ShardId) - ecVolume.ShardLocations[shardId] = nil + delete(ecVolume.ShardLocations, shardId) for _, loc := range shardIdLocations.Locations { ecVolume.ShardLocations[shardId] = append(ecVolume.ShardLocations[shardId], loc.Url) }