diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 5a62ad075..50de94dfc 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -479,6 +479,7 @@ message UpdateEntryRequest { // the write, under the filer's per-path lock. The caller must route the // key's writes to this entry's owner filer for the check to be authoritative. WriteCondition condition = 6; + bool is_moved = 7; // set on a forwarded request so the owner applies it locally } message UpdateEntryResponse { SubscribeMetadataResponse metadata_event = 1; diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 4fda304f2..f9bd74c4c 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -2378,6 +2378,7 @@ type UpdateEntryRequest struct { // the write, under the filer's per-path lock. The caller must route the // key's writes to this entry's owner filer for the check to be authoritative. Condition *WriteCondition `protobuf:"bytes,6,opt,name=condition,proto3" json:"condition,omitempty"` + IsMoved bool `protobuf:"varint,7,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"` // set on a forwarded request so the owner applies it locally unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -2454,6 +2455,13 @@ func (x *UpdateEntryRequest) GetCondition() *WriteCondition { return nil } +func (x *UpdateEntryRequest) GetIsMoved() bool { + if x != nil { + return x.IsMoved + } + return false +} + type UpdateEntryResponse struct { state protoimpl.MessageState `protogen:"open.v1"` MetadataEvent *SubscribeMetadataResponse `protobuf:"bytes,1,opt,name=metadata_event,json=metadataEvent,proto3" json:"metadata_event,omitempty"` @@ -7429,7 +7437,7 @@ const file_filer_proto_rawDesc = "" + "\n" + "error_code\x18\x03 \x01(\x0e2\x14.filer_pb.FilerErrorR\terrorCode\x12\x1a\n" + "\tlog_ts_ns\x18\x04 \x01(\x03R\alogTsNs\x12#\n" + - "\rlog_signature\x18\x05 \x01(\x05R\flogSignature\"\x8a\x03\n" + + "\rlog_signature\x18\x05 \x01(\x05R\flogSignature\"\xa5\x03\n" + "\x12UpdateEntryRequest\x12\x1c\n" + "\tdirectory\x18\x01 \x01(\tR\tdirectory\x12%\n" + "\x05entry\x18\x02 \x01(\v2\x0f.filer_pb.EntryR\x05entry\x121\n" + @@ -7438,7 +7446,8 @@ const file_filer_proto_rawDesc = "" + "signatures\x18\x04 \x03(\x05R\n" + "signatures\x12_\n" + "\x11expected_extended\x18\x05 \x03(\v22.filer_pb.UpdateEntryRequest.ExpectedExtendedEntryR\x10expectedExtended\x126\n" + - "\tcondition\x18\x06 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x1aC\n" + + "\tcondition\x18\x06 \x01(\v2\x18.filer_pb.WriteConditionR\tcondition\x12\x19\n" + + "\bis_moved\x18\a \x01(\bR\aisMoved\x1aC\n" + "\x15ExpectedExtendedEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\fR\x05value:\x028\x01\"\xa2\x01\n" + diff --git a/weed/pb/filer_pb/filer_vtproto.pb.go b/weed/pb/filer_pb/filer_vtproto.pb.go index 9e3f98762..249f1d14c 100644 --- a/weed/pb/filer_pb/filer_vtproto.pb.go +++ b/weed/pb/filer_pb/filer_vtproto.pb.go @@ -2097,6 +2097,16 @@ func (m *UpdateEntryRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { i -= len(m.unknownFields) copy(dAtA[i:], m.unknownFields) } + if m.IsMoved { + i-- + if m.IsMoved { + dAtA[i] = 1 + } else { + dAtA[i] = 0 + } + i-- + dAtA[i] = 0x38 + } if m.Condition != nil { size, err := m.Condition.MarshalToSizedBufferVT(dAtA[:i]) if err != nil { @@ -7346,6 +7356,9 @@ func (m *UpdateEntryRequest) SizeVT() (n int) { l = m.Condition.SizeVT() n += 1 + l + protohelpers.SizeOfVarint(uint64(l)) } + if m.IsMoved { + n += 2 + } n += len(m.unknownFields) return n } @@ -15117,6 +15130,26 @@ func (m *UpdateEntryRequest) UnmarshalVT(dAtA []byte) error { return err } iNdEx = postIndex + case 7: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field IsMoved", wireType) + } + var v int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + m.IsMoved = bool(v != 0) default: iNdEx = preIndex skippy, err := protohelpers.Skip(dAtA[iNdEx:]) diff --git a/weed/server/filer_grpc_server.go b/weed/server/filer_grpc_server.go index b0cf88345..e99c9abe2 100644 --- a/weed/server/filer_grpc_server.go +++ b/weed/server/filer_grpc_server.go @@ -662,6 +662,32 @@ func (fs *FilerServer) UpdateEntry(ctx context.Context, req *filer_pb.UpdateEntr fullpath := util.Join(req.Directory, req.Entry.Name) + // A conditional or preconditioned update is a read-then-write that the + // per-path lock below only makes atomic on this filer. Route it to the + // entry's ring owner so one filer's lock arbitrates every writer + // cluster-wide; is_moved bounds this to one hop. + if !req.IsMoved && (conditionIsSet(req.Condition) || len(req.ExpectedExtended) > 0) { + var ownerResp *filer_pb.UpdateEntryResponse + handled, forwardErr := fs.forwardToWriteOwner(ctx, entryRouteKey(util.FullPath(fullpath)), func(owner pb.ServerAddress) error { + glog.V(2).InfofCtx(ctx, "UpdateEntry %s: forwarding to owner %s", fullpath, owner) + req.IsMoved = true + return pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + forwarded, e := client.UpdateEntry(ctx, req) + if e != nil { + return e + } + ownerResp = forwarded + return nil + }) + }) + if handled { + if forwardErr != nil { + return &filer_pb.UpdateEntryResponse{}, forwardErr + } + return ownerResp, nil + } + } + // Serialize concurrent mutations to the same path on this filer so the // read (preconditions, garbage diff) and the write are atomic. Callers // route a key's writes to this owner filer, making this local lock