Method UploadPersistentCacheTaskStarted

scheduler/service/service_v2.go:4043–4096

UploadPersistentCacheTaskStarted uploads the metadata of the persistent cache task started.

(ctx context.Context, req *schedulerv2.UploadPersistentCacheTaskStartedRequest)

// UploadPersistentCacheTaskStarted uploads the metadata of the persistent cache task started.
func (v *V2) UploadPersistentCacheTaskStarted(ctx context.Context, req *schedulerv2.UploadPersistentCacheTaskStartedRequest) error {
	if v.persistentCacheResource == nil {
		return status.Error(codes.FailedPrecondition, "redis is not enabled")
	}

	log := logger.WithPeer(req.GetHostId(), req.GetTaskId(), req.GetPeerId())
	log.Info("upload persistent cache task started")

	host, loaded := v.persistentCacheResource.HostManager().Load(ctx, req.GetHostId())
	if !loaded {
		log.Error("host not found")
		return status.Errorf(codes.NotFound, "host %s not found", req.GetHostId())
	}

	// Handle task with task started request, new task and store it.
	task, loaded := v.persistentCacheResource.TaskManager().Load(ctx, req.GetTaskId())
	if loaded && !task.FSM.Can(persistentcache.TaskEventUpload) {
		log.Errorf("persistent cache task %s is %s cannot upload", task.ID, task.FSM.Current())
		return status.Errorf(codes.AlreadyExists, "persistent cache task %s is %s cannot upload", task.ID, task.FSM.Current())
	}

	task = persistentcache.NewTask(req.GetTaskId(), req.GetTag(), req.GetApplication(), persistentcache.TaskStatePending, req.GetPersistentReplicaCount(),
		req.GetPieceLength(), req.GetContentLength(), req.GetPieceCount(), req.GetTtl().AsDuration(), time.Now(), time.Now(), log)

	if err := task.FSM.Event(ctx, persistentcache.TaskEventUpload); err != nil {
		log.Errorf("task fsm event failed: %s", err.Error())
		return status.Error(codes.Internal, err.Error())
	}

	if err := v.persistentCacheResource.TaskManager().Store(ctx, task); err != nil {
		log.Errorf("store persistent cache task %s error %s", task.ID, err)
		return status.Error(codes.Internal, err.Error())
	}

	// Handle peer with task started request, new peer and store it.
	if peer, loaded := v.persistentCacheResource.PeerManager().Load(ctx, req.GetPeerId()); loaded {
		log.Error("persistent cache peer already exists")
		return status.Errorf(codes.AlreadyExists, "persistent cache peer %s already exists", peer.ID)
	}

	peer := persistentcache.NewPeer(req.GetPeerId(), persistentcache.PeerStatePending, true, bitset.New(uint(req.GetPieceCount())), nil, task, host, 0, time.Now(), time.Now(), log)

	if err := peer.FSM.Event(ctx, persistentcache.PeerEventUpload); err != nil {
		log.Errorf("peer fsm event failed: %s", err.Error())
		return status.Error(codes.Internal, err.Error())
	}

	if err := v.persistentCacheResource.PeerManager().Store(ctx, peer); err != nil {
		log.Errorf("store persistent cache peer %s error %s", peer.ID, err)
		return status.Error(codes.Internal, err.Error())
	}

	return nil
}

// UploadPersistentCacheTaskFinished uploads the metadata of the persistent cache task finished.
func (v *V2) UploadPersistentCacheTaskFinished(ctx context.Context, req *schedulerv2.UploadPersistentCacheTaskFinishedRequest) (*commonv2.PersistentCacheTask, error) {