seaweedfs/weed/server/filer_grpc_server_posix_lock.go

package weed_server

import (
	"context"
	"fmt"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/filer/posixlock"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
)

const (
	// posixLockSessionTTL is how long a mount's lease survives without a
	// keepalive before its locks are reaped; posixLockSweepInterval is how often
	// each filer checks. The mount renews well within the TTL.
	posixLockSessionTTL    = 15 * time.Second
	posixLockSweepInterval = 5 * time.Second
	// posixCoolingProbeTimeout bounds the dual-read probe to the prior owner so a
	// slow peer can't stall a non-blocking lock call during the cooling window.
	posixCoolingProbeTimeout = 2 * time.Second
)

// startPosixLockSweeper periodically reaps the locks of leased sessions (mounts)
// that stopped sending keepalives. Sessions that never renew are never reaped, so
// this is inert until mounts run with -posixLock.
func (fs *FilerServer) startPosixLockSweeper() {
	fs.posixLockSweeperStop = make(chan struct{})
	go func() {
		ticker := time.NewTicker(posixLockSweepInterval)
		defer ticker.Stop()
		for {
			select {
			case <-fs.posixLockSweeperStop:
				return
			case <-ticker.C:
				if reaped := fs.posixLocks.ReapExpired(posixLockSessionTTL); len(reaped) > 0 {
					glog.V(2).Infof("posix lock: reaped %d expired session(s): %v", len(reaped), reaped)
				}
			}
		}
	}()
}

// PosixLock applies one advisory lock operation against the in-memory lock table
// of the inode's owner filer. The owner is resolved from req.Key on the same
// ring the gateway and DLM use; a non-owner filer forwards the request one hop
// (is_moved bounds it), so the owner's table stays the single authority even when
// the caller's ring view is stale. Blocking (SetLkw) is the caller's job — this
// is strictly non-blocking try/release/query.
func (fs *FilerServer) PosixLock(ctx context.Context, req *filer_pb.PosixLockRequest) (*filer_pb.PosixLockResponse, error) {
	if req.Key == "" {
		return &filer_pb.PosixLockResponse{}, fmt.Errorf("key is required")
	}
	if req.Lock == nil {
		return &filer_pb.PosixLockResponse{}, fmt.Errorf("lock is required")
	}

	if !req.IsMoved && fs.filer.Dlm != nil {
		if owner := fs.filer.Dlm.LockRing.GetPrimary(req.Key); owner != "" && owner != fs.option.Host {
			forwarded := &filer_pb.PosixLockRequest{
				Key:     req.Key,
				IsMoved: true,
				Op:      req.Op,
				Lock:    req.Lock,
				Locks:   req.Locks,
			}
			glog.V(4).InfofCtx(ctx, "PosixLock %s op=%v: forwarding to owner %s", req.Key, req.Op, owner)
			var resp *filer_pb.PosixLockResponse
			err := pb.WithFilerClient(false, 0, owner, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
				var e error
				resp, e = client.PosixLock(ctx, forwarded)
				return e
			})
			if err != nil {
				return &filer_pb.PosixLockResponse{}, err
			}
			return resp, nil
		}
	}

	lk := posixlock.Range{
		Start:   req.Lock.GetStart(),
		End:     req.Lock.GetEnd(),
		Type:    req.Lock.GetType(),
		Sid:     req.Lock.GetSid(),
		Owner:   req.Lock.GetOwner(),
		Pid:     req.Lock.GetPid(),
		IsFlock: req.Lock.GetIsFlock(),
	}

	resp := &filer_pb.PosixLockResponse{}
	switch req.Op {
	case filer_pb.PosixLockOp_TRY_LOCK:
		// During a ring change, a previous owner may still hold a conflicting lock
		// this fresh owner has not rebuilt yet; consult it before granting.
		if !req.CoolingProbe {
			if c, found := fs.coolingConflict(ctx, req.Key, lk); found {
				resp.HasConflict = true
				resp.Conflict = c
				break
			}
		}
		if c, granted := fs.posixLocks.TryLock(req.Key, lk); granted {
			resp.Granted = true
		} else {
			resp.HasConflict = true
			resp.Conflict = posixRangeToPb(c)
		}
	case filer_pb.PosixLockOp_UNLOCK:
		fs.posixLocks.Unlock(req.Key, lk)
	case filer_pb.PosixLockOp_GET_LK:
		if !req.CoolingProbe {
			if c, found := fs.coolingConflict(ctx, req.Key, lk); found {
				resp.HasConflict = true
				resp.Conflict = c
				break
			}
		}
		if c, found := fs.posixLocks.GetLk(req.Key, lk); found {
			resp.HasConflict = true
			resp.Conflict = posixRangeToPb(c)
		}
	case filer_pb.PosixLockOp_RELEASE_POSIX_OWNER:
		fs.posixLocks.ReleasePosixOwner(req.Key, lk.Sid, lk.Owner)
	case filer_pb.PosixLockOp_RELEASE_FLOCK_OWNER:
		fs.posixLocks.ReleaseFlockOwner(req.Key, lk.Sid, lk.Owner)
	case filer_pb.PosixLockOp_KEEP_ALIVE:
		// A re-assertion carries the mount's held locks on this key so the owner
		// can rebuild its in-memory state after an ownership change or restart; a
		// bare keepalive just renews the lease.
		if len(req.Locks) > 0 {
			held := make([]posixlock.Range, 0, len(req.Locks))
			for _, l := range req.Locks {
				held = append(held, posixRangeFromPb(l))
			}
			if conflicts := fs.posixLocks.Reassert(req.Key, lk.Sid, held); len(conflicts) > 0 {
				glog.Warningf("posix reassert %s sid %d: %d lock(s) lost to another session: %+v", req.Key, lk.Sid, len(conflicts), conflicts)
			}
		} else {
			fs.posixLocks.Renew(lk.Sid)
		}
	default:
		return &filer_pb.PosixLockResponse{}, fmt.Errorf("unknown posix lock op %v", req.Op)
	}
	return resp, nil
}

// coolingConflict asks the key's previous owner, during a ring-change cooling
// window, whether it still holds a lock that blocks lk — so a fresh owner does
// not double-grant before re-assertion rebuilds its local state. Best-effort: if
// the previous owner is unreachable (e.g. it left, which caused the change, so
// its locks are gone) the caller proceeds. The probe is marked cooling_probe so
// the previous owner answers from local state without itself cooling off.
func (fs *FilerServer) coolingConflict(ctx context.Context, key string, lk posixlock.Range) (*filer_pb.PosixLockRange, bool) {
	if fs.filer.Dlm == nil {
		return nil, false
	}
	prior := fs.filer.Dlm.LockRing.PriorOwner(key)
	if prior == "" || prior == fs.option.Host {
		return nil, false
	}
	// Bound the probe with its own short deadline: it runs on the non-blocking
	// lock path, so a slow prior owner must not stall the caller. On timeout the
	// probe is treated as unreachable (best-effort).
	probeCtx, cancel := context.WithTimeout(ctx, posixCoolingProbeTimeout)
	defer cancel()
	var resp *filer_pb.PosixLockResponse
	err := pb.WithFilerClient(false, 0, prior, fs.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
		var e error
		resp, e = client.PosixLock(probeCtx, &filer_pb.PosixLockRequest{
			Key:          key,
			IsMoved:      true,
			CoolingProbe: true,
			Op:           filer_pb.PosixLockOp_GET_LK,
			Lock:         posixRangeToPb(lk),
		})
		return e
	})
	if err != nil {
		glog.V(2).InfofCtx(ctx, "posix cooling probe %s -> %s: %v", key, prior, err)
		return nil, false
	}
	if resp.GetHasConflict() {
		return resp.GetConflict(), true
	}
	return nil, false
}

func posixRangeFromPb(l *filer_pb.PosixLockRange) posixlock.Range {
	return posixlock.Range{
		Start:   l.GetStart(),
		End:     l.GetEnd(),
		Type:    l.GetType(),
		Sid:     l.GetSid(),
		Owner:   l.GetOwner(),
		Pid:     l.GetPid(),
		IsFlock: l.GetIsFlock(),
	}
}

func posixRangeToPb(r posixlock.Range) *filer_pb.PosixLockRange {
	return &filer_pb.PosixLockRange{
		Start:   r.Start,
		End:     r.End,
		Type:    r.Type,
		Sid:     r.Sid,
		Owner:   r.Owner,
		Pid:     r.Pid,
		IsFlock: r.IsFlock,
	}
}