mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/chrislusf/seaweedfs synced 2024-09-18 06:50:35 +02:00
Code Issues Projects Releases Wiki Activity

WIP

Browse source 
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
This commit is contained in:
Chris Lu 2019-05-28 00:51:01 -07:00
parent 2858a954b3
commit 4f76342cbc
3 changed files with 24 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
weed/storage/erasure_coding/ec_volume.go
View file

@ -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
})

8
weed/storage/needle/needle_read_write.go
View file

@ -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)
}

27
weed/storage/store_ec.go
View file

@ -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)
}