Check if Worker can still accept more work right before giving it new replicate

Author: mcornaton

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [[8.2.3]](https://github.com/iExecBlockchainComputing/iexec-core/releases/tag/v8.2.3) 2023-12-13
+
+### Bug Fixes
+
+- Check if Worker can still accept more work right before giving it new replicate. (#644)
+
 ## [[8.2.2]](https://github.com/iExecBlockchainComputing/iexec-core/releases/tag/v8.2.2) 2023-12-13
 
 ### Bug Fixes

gradle.properties

@@ -1,4 +1,4 @@
-version=8.2.2
+version=8.2.3
 iexecCommonVersion=8.3.0
 iexecCommonsPocoVersion=3.1.0
 iexecBlockchainAdapterVersion=8.2.0

src/main/java/com/iexec/core/replicate/ReplicateSupplyService.java

@@ -86,14 +86,9 @@ public ReplicateSupplyService(ReplicatesService replicatesService,
      */
     @Retryable(value = {OptimisticLockingFailureException.class}, maxAttempts = 5)
     Optional<ReplicateTaskSummary> getAvailableReplicateTaskSummary(long workerLastBlock, String walletAddress) {
-        // return empty if max computing task is reached or if the worker is not found
-        if (!workerService.canAcceptMoreWorks(walletAddress)) {
-            return Optional.empty();
-        }
-
         // return empty if the worker is not sync
         //TODO Check if worker node is sync
-        boolean isWorkerLastBlockAvailable = workerLastBlock > 0;
+        final boolean isWorkerLastBlockAvailable = workerLastBlock > 0;
         if (!isWorkerLastBlockAvailable) {
             return Optional.empty();
         }
@@ -104,11 +99,16 @@ Optional<ReplicateTaskSummary> getAvailableReplicateTaskSummary(long workerLastB
 
         // TODO : Remove this, the optional can never be empty
         // This is covered in workerService.canAcceptMoreWorks
-        Optional<Worker> optional = workerService.getWorker(walletAddress);
+        final Optional<Worker> optional = workerService.getWorker(walletAddress);
         if (optional.isEmpty()) {
             return Optional.empty();
         }
-        Worker worker = optional.get();
+        final Worker worker = optional.get();
+
+        // return empty if max computing task is reached or if the worker is not found
+        if (!workerService.canAcceptMoreWorks(worker)) {
+            return Optional.empty();
+        }
 
         return getReplicateTaskSummaryForAnyAvailableTask(
                 walletAddress,
@@ -161,8 +161,8 @@ private Optional<ReplicateTaskSummary> getReplicateTaskSummary(Task task, String
                 chainTaskId,
                 task.getEnclaveChallenge());
         ReplicateTaskSummaryBuilder replicateTaskSummary = ReplicateTaskSummary.builder()
-            .workerpoolAuthorization(authorization);
-        if(task.isTeeTask()){
+                .workerpoolAuthorization(authorization);
+        if (task.isTeeTask()) {
             replicateTaskSummary.smsUrl(task.getSmsUrl());
         }
         return Optional.of(replicateTaskSummary.build());
@@ -173,7 +173,7 @@ private Optional<ReplicateTaskSummary> getReplicateTaskSummary(Task task, String
      * tries to accept the task - i.e. create a new {@link Replicate}
      * for that task on that worker.
      *
-     * @param task  {@link Task} needing at least one new {@link Replicate}.
+     * @param task          {@link Task} needing at least one new {@link Replicate}.
      * @param walletAddress Wallet address of a worker looking for new {@link Task}.
      * @return {@literal true} if the task has been accepted,
      * {@literal false} otherwise.
@@ -220,8 +220,8 @@ private boolean acceptOrRejectTask(Task task, String walletAddress) {
                 return false;
             }
 
-            replicatesService.addNewReplicate(chainTaskId, walletAddress);
-            workerService.addChainTaskIdToWorker(chainTaskId, walletAddress);
+            workerService.addChainTaskIdToWorker(chainTaskId, walletAddress)
+                    .ifPresent(worker -> replicatesService.addNewReplicate(chainTaskId, walletAddress));
         } finally {
             // We should always unlock the task
             // so that it could be taken by another replicate
@@ -234,8 +234,8 @@ private boolean acceptOrRejectTask(Task task, String walletAddress) {
 
     /**
      * Get notifications missed by the worker during the time it was absent.
-     * 
-     * @param blockNumber last seen blocknumber by the worker
+     *
+     * @param blockNumber   last seen blocknumber by the worker
      * @param walletAddress of the worker
      * @return list of missed notifications. Can be empty if no notification is found
      */
@@ -264,7 +264,7 @@ public List<TaskNotification> getMissedTaskNotifications(long blockNumber, Strin
                 continue;
             }
             TaskNotificationExtra taskNotificationExtra =
-                    getTaskNotificationExtra(task, taskNotificationType.get(),  walletAddress, enclaveChallenge);
+                    getTaskNotificationExtra(task, taskNotificationType.get(), walletAddress, enclaveChallenge);
 
             TaskNotification taskNotification = TaskNotification.builder()
                     .chainTaskId(chainTaskId)
@@ -286,7 +286,7 @@ public List<TaskNotification> getMissedTaskNotifications(long blockNumber, Strin
     private TaskNotificationExtra getTaskNotificationExtra(Task task, TaskNotificationType taskNotificationType, String walletAddress, String enclaveChallenge) {
         TaskNotificationExtra taskNotificationExtra = TaskNotificationExtra.builder().build();
 
-        switch (taskNotificationType){
+        switch (taskNotificationType) {
             case PLEASE_CONTRIBUTE:
                 WorkerpoolAuthorization authorization = signatureService.createAuthorization(
                         walletAddress, task.getChainTaskId(), enclaveChallenge);
@@ -312,7 +312,7 @@ public Optional<TaskNotificationType> getTaskNotificationType(Task task, Replica
         // CONTRIBUTION_TIMEOUT or CONSENSUS_REACHED without contribution
         if (task.getCurrentStatus().equals(TaskStatus.CONTRIBUTION_TIMEOUT)
                 || (task.getCurrentStatus().equals(TaskStatus.CONSENSUS_REACHED)
-                        && !replicate.containsContributedStatus())) {
+                && !replicate.containsContributedStatus())) {
             return Optional.of(TaskNotificationType.PLEASE_ABORT);
         }
 

src/main/java/com/iexec/core/replicate/ReplicatesService.java

@@ -72,17 +72,20 @@ public ReplicatesService(ReplicatesRepository replicatesRepository,
     }
 
     public void addNewReplicate(String chainTaskId, String walletAddress) {
-        if (getReplicate(chainTaskId, walletAddress).isEmpty()) {
-            Optional<ReplicatesList> optional = getReplicatesList(chainTaskId);
-            if (optional.isPresent()) {
-                ReplicatesList replicatesList = optional.get();
-                Replicate replicate = new Replicate(walletAddress, chainTaskId);
-                replicate.setWorkerWeight(iexecHubService.getWorkerWeight(walletAddress));// workerWeight value for pendingWeight estimate
-                replicatesList.getReplicates().add(replicate);
-
-                replicatesRepository.save(replicatesList);
-                log.info("New replicate saved [chainTaskId:{}, walletAddress:{}]", chainTaskId, walletAddress);
-            }
+        final Optional<ReplicatesList> oReplicatesList = getReplicatesList(chainTaskId);
+        if (oReplicatesList.isEmpty()) {
+            log.warn("Can't add replicate to unknown ReplicatesList [chainTaskId:{}, workerName:{}]", chainTaskId, walletAddress);
+            return;
+        }
+
+        final ReplicatesList replicatesList = oReplicatesList.get();
+        if (replicatesList.getReplicateOfWorker(walletAddress).isEmpty()) {
+            Replicate replicate = new Replicate(walletAddress, chainTaskId);
+            replicate.setWorkerWeight(iexecHubService.getWorkerWeight(walletAddress));// workerWeight value for pendingWeight estimate
+            replicatesList.getReplicates().add(replicate);
+
+            replicatesRepository.save(replicatesList);
+            log.info("New replicate saved [chainTaskId:{}, walletAddress:{}]", chainTaskId, walletAddress);
         } else {
             log.error("Replicate already saved [chainTaskId:{}, walletAddress:{}]", chainTaskId, walletAddress);
         }
@@ -635,4 +638,4 @@ public void setRevealTimeoutStatusIfNeeded(String chainTaskId, Replicate replica
             updateReplicateStatus(chainTaskId, replicate.getWalletAddress(), statusUpdate);
         }
     }
-}
+}

src/main/java/com/iexec/core/worker/WorkerService.java

@@ -69,10 +69,10 @@ public Optional<Worker> getWorker(String walletAddress) {
         return workerRepository.findByWalletAddress(walletAddress);
     }
 
-    public boolean isAllowedToJoin(String workerAddress){
+    public boolean isAllowedToJoin(String workerAddress) {
         List<String> whitelist = workerConfiguration.getWhitelist();
         // if the whitelist is empty, there is no restriction on the workers
-        if (whitelist.isEmpty()){
+        if (whitelist.isEmpty()) {
             return true;
         }
         return whitelist.contains(workerAddress);
@@ -135,17 +135,17 @@ public List<Worker> getAliveWorkers() {
 
     public boolean canAcceptMoreWorks(String walletAddress) {
         Optional<Worker> optionalWorker = getWorker(walletAddress);
-        if (optionalWorker.isEmpty()){
-            return false;
-        }
+        return optionalWorker.filter(this::canAcceptMoreWorks).isPresent();
+
+    }
 
-        Worker worker = optionalWorker.get();
+    public boolean canAcceptMoreWorks(Worker worker) {
         int workerMaxNbTasks = worker.getMaxNbTasks();
         int runningReplicateNb = worker.getComputingChainTaskIds().size();
 
         if (runningReplicateNb >= workerMaxNbTasks) {
             log.debug("Worker asking for too many replicates [walletAddress:{}, runningReplicateNb:{}, workerMaxNbTasks:{}]",
-                    walletAddress, runningReplicateNb, workerMaxNbTasks);
+                    worker.getWalletAddress(), runningReplicateNb, workerMaxNbTasks);
             return false;
         }
 
@@ -154,44 +154,44 @@ public boolean canAcceptMoreWorks(String walletAddress) {
 
     public int getAliveAvailableCpu() {
         int availableCpus = 0;
-        for (Worker worker: getAliveWorkers()) {
+        for (Worker worker : getAliveWorkers()) {
             if (worker.isGpuEnabled()) {
                 continue;
             }
 
             int workerCpuNb = worker.getCpuNb();
             int computingReplicateNb = worker.getComputingChainTaskIds().size();
             int availableCpu = workerCpuNb - computingReplicateNb;
-            availableCpus+= availableCpu;
+            availableCpus += availableCpu;
         }
         return availableCpus;
     }
 
     public int getAliveTotalCpu() {
         int totalCpus = 0;
-        for (Worker worker: getAliveWorkers()){
-            if(worker.isGpuEnabled()) {
+        for (Worker worker : getAliveWorkers()) {
+            if (worker.isGpuEnabled()) {
                 continue;
             }
-            totalCpus+= worker.getCpuNb();
+            totalCpus += worker.getCpuNb();
         }
         return totalCpus;
     }
 
     // We suppose for now that 1 Gpu enabled worker has only one GPU
     public int getAliveTotalGpu() {
         int totalGpus = 0;
-        for(Worker worker: getAliveWorkers()) {
+        for (Worker worker : getAliveWorkers()) {
             if (worker.isGpuEnabled()) {
                 totalGpus++;
             }
         }
         return totalGpus;
     }
 
-    public int getAliveAvailableGpu () {
+    public int getAliveAvailableGpu() {
         int availableGpus = getAliveTotalGpu();
-        for (Worker worker: getAliveWorkers()) {
+        for (Worker worker : getAliveWorkers()) {
             if (worker.isGpuEnabled()) {
                 boolean isWorking = !worker.getComputingChainTaskIds().isEmpty();
                 if (isWorking) {
@@ -246,13 +246,20 @@ public Optional<Worker> addChainTaskIdToWorker(String chainTaskId, String wallet
     }
 
     private Optional<Worker> addChainTaskIdToWorkerWithoutThreadSafety(String chainTaskId, String walletAddress) {
-        Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
+        final Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
         if (optional.isPresent()) {
-            Worker worker = optional.get();
+            final Worker worker = optional.get();
+            if (!canAcceptMoreWorks(worker)) {
+                log.warn("Can't add chainTaskId to worker when already full [chainTaskId:{}, workerName:{}]",
+                        chainTaskId, walletAddress);
+                return Optional.empty();
+            }
             worker.addChainTaskId(chainTaskId);
             log.info("Added chainTaskId to worker [chainTaskId:{}, workerName:{}]", chainTaskId, walletAddress);
             return Optional.of(workerRepository.save(worker));
         }
+        log.warn("Can't add chainTaskId to worker when unknown worker [chainTaskId:{}, workerName:{}]",
+                chainTaskId, walletAddress);
         return Optional.empty();
     }