Skip to content

Commit 604b9ca

Browse files
jbern0rdjbern0rd
authored
Hotfix version 8.2.1 (#642)
Hotfix version 8.2.1
1 parent 417ccab commit 604b9ca

File tree

10 files changed

+90
-192
lines changed

10 files changed

+90
-192
lines changed

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,12 @@
22

33
All notable changes to this project will be documented in this file.
44

5+
## [[8.2.1]](https://github.com/iExecBlockchainComputing/iexec-core/releases/tag/v8.2.1) 2023-12-13
6+
7+
### Bug Fixes
8+
9+
- Do not write `lastAliveDate` or `lastReplicateDemandDate` to MongoDB to decrease load on the database. (#642)
10+
511
## [[8.2.0]](https://github.com/iExecBlockchainComputing/iexec-core/releases/tag/v8.2.0) 2023-09-29
612

713
### New Features

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
version=8.2.0
1+
version=8.2.1
22
iexecCommonVersion=8.3.0
33
iexecCommonsPocoVersion=3.1.0
44
iexecBlockchainAdapterVersion=8.2.0

‎src/main/java/com/iexec/core/worker/Worker.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ public class Worker {
5050
private List<String> participatingChainTaskIds;
5151
private List<String> computingChainTaskIds;
5252

53+
// TODO remove it cleanly in a release
5354
private Date lastAliveDate;
5455
private Date lastReplicateDemandDate;
5556

‎src/main/java/com/iexec/core/worker/WorkerController.java‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -31,11 +31,9 @@
3131
import org.web3j.crypto.Hash;
3232

3333
import java.util.ArrayList;
34-
import java.util.Date;
3534
import java.util.List;
3635

3736
import static org.springframework.http.ResponseEntity.ok;
38-
import static org.springframework.http.ResponseEntity.status;
3937

4038
@Slf4j
4139
@RestController
@@ -63,9 +61,8 @@ public ResponseEntity<String> ping(@RequestHeader("Authorization") String bearer
6361
return ResponseEntity.status(HttpStatus.UNAUTHORIZED).build();
6462
}
6563
final String publicConfigurationHash = publicConfigurationService.getPublicConfigurationHash();
66-
return workerService.updateLastAlive(workerWalletAddress)
67-
.map(worker -> ok(publicConfigurationHash))
68-
.orElseGet(() -> status(HttpStatus.NOT_FOUND).build());
64+
workerService.updateLastAlive(workerWalletAddress);
65+
return ok(publicConfigurationHash);
6966
}
7067

7168
@GetMapping(path = "/workers/challenge")
@@ -117,7 +114,6 @@ public ResponseEntity<Worker> registerWorker(@RequestHeader("Authorization") Str
117114
.memorySize(model.getMemorySize())
118115
.teeEnabled(model.isTeeEnabled())
119116
.gpuEnabled(model.isGpuEnabled())
120-
.lastAliveDate(new Date())
121117
.participatingChainTaskIds(new ArrayList<>())
122118
.computingChainTaskIds(new ArrayList<>())
123119
.build();

‎src/main/java/com/iexec/core/worker/WorkerRepository.java‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,19 +17,13 @@
1717
package com.iexec.core.worker;
1818

1919
import org.springframework.data.mongodb.repository.MongoRepository;
20-
import org.springframework.data.mongodb.repository.Query;
2120

22-
import java.util.Date;
2321
import java.util.List;
2422
import java.util.Optional;
2523

2624
interface WorkerRepository extends MongoRepository<Worker, String> {
2725

2826
Optional<Worker> findByWalletAddress(String walletAddress);
2927

30-
@Query("{'lastAliveDate': {$lt: ?0}}")
31-
List<Worker> findByLastAliveDateBefore(Date date);
32-
33-
@Query("{'lastAliveDate': {$gt: ?0}}")
34-
List<Worker> findByLastAliveDateAfter(Date date);
28+
List<Worker> findAllByWalletAddress(Iterable<String> walletAddress);
3529
}

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

Lines changed: 42 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,14 @@
1818

1919
import com.iexec.common.utils.ContextualLockRunner;
2020
import com.iexec.core.configuration.WorkerConfiguration;
21+
import lombok.Data;
22+
import lombok.Getter;
2123
import lombok.extern.slf4j.Slf4j;
2224
import org.springframework.stereotype.Service;
2325

24-
import java.util.Collections;
25-
import java.util.Date;
26-
import java.util.List;
27-
import java.util.Optional;
26+
import java.util.*;
27+
import java.util.concurrent.ConcurrentHashMap;
28+
import java.util.stream.Collectors;
2829

2930
import static com.iexec.common.utils.DateTimeUtils.addMinutesToDate;
3031

@@ -42,6 +43,19 @@ public class WorkerService {
4243
private final WorkerRepository workerRepository;
4344
private final WorkerConfiguration workerConfiguration;
4445
private final ContextualLockRunner<String> contextualLockRunner;
46+
@Getter
47+
private final ConcurrentHashMap<String, WorkerStats> workerStatsMap = new ConcurrentHashMap<>();
48+
49+
@Data
50+
public static class WorkerStats {
51+
private String walletAddress;
52+
private Date lastAliveDate;
53+
private Date lastReplicateDemandDate;
54+
55+
public WorkerStats(String walletAddress) {
56+
this.walletAddress = walletAddress;
57+
}
58+
}
4559

4660
public WorkerService(WorkerRepository workerRepository,
4761
WorkerConfiguration workerConfiguration) {
@@ -65,29 +79,19 @@ public boolean isAllowedToJoin(String workerAddress){
6579
}
6680

6781
public boolean isWorkerAllowedToAskReplicate(String walletAddress) {
68-
Optional<Date> oDate = getLastReplicateDemand(walletAddress);
69-
if (oDate.isEmpty()) {
82+
Date lastReplicateDemandDate = workerStatsMap.get(walletAddress).getLastReplicateDemandDate();
83+
if (lastReplicateDemandDate == null) {
7084
return true;
7185
}
7286

7387
// the difference between now and the last time the worker asked for work should be less than the period allowed
7488
// in the configuration (500ms since (now - lastAsk) can still be slightly too small even if the worker behave nicely)
7589
long now = new Date().getTime();
76-
long lastAsk = oDate.get().getTime();
90+
long lastAsk = lastReplicateDemandDate.getTime();
7791

7892
return (now - lastAsk) + 500 > workerConfiguration.getAskForReplicatePeriod();
7993
}
8094

81-
public Optional<Date> getLastReplicateDemand(String walletAddress) {
82-
Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
83-
if (optional.isEmpty()) {
84-
return Optional.empty();
85-
}
86-
Worker worker = optional.get();
87-
88-
return Optional.ofNullable(worker.getLastReplicateDemandDate());
89-
}
90-
9195
public List<String> getChainTaskIds(String walletAddress) {
9296
Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
9397
if (optional.isPresent()) {
@@ -110,13 +114,23 @@ public List<String> getComputingTaskIds(String walletAddress) {
110114
// worker is considered lost if it didn't ping for 1 minute
111115
public List<Worker> getLostWorkers() {
112116
Date oneMinuteAgo = addMinutesToDate(new Date(), -1);
113-
return workerRepository.findByLastAliveDateBefore(oneMinuteAgo);
117+
List<String> lostWorkers = workerStatsMap.entrySet()
118+
.stream()
119+
.filter(entry -> entry.getValue().getLastAliveDate().getTime() < oneMinuteAgo.getTime())
120+
.map(Map.Entry::getKey)
121+
.collect(Collectors.toList());
122+
return workerRepository.findAllByWalletAddress(lostWorkers);
114123
}
115124

116-
// worker is considered alive if it ping after 1 minute
125+
// worker is considered alive if it received a ping during the last minute
117126
public List<Worker> getAliveWorkers() {
118127
Date oneMinuteAgo = addMinutesToDate(new Date(), -1);
119-
return workerRepository.findByLastAliveDateAfter(oneMinuteAgo);
128+
List<String> aliveWorkers = workerStatsMap.entrySet()
129+
.stream()
130+
.filter(entry -> entry.getValue().getLastAliveDate().getTime() > oneMinuteAgo.getTime())
131+
.map(Map.Entry::getKey)
132+
.collect(Collectors.toList());
133+
return workerRepository.findAllByWalletAddress(aliveWorkers);
120134
}
121135

122136
public boolean canAcceptMoreWorks(String walletAddress) {
@@ -130,7 +144,7 @@ public boolean canAcceptMoreWorks(String walletAddress) {
130144
int runningReplicateNb = worker.getComputingChainTaskIds().size();
131145

132146
if (runningReplicateNb >= workerMaxNbTasks) {
133-
log.debug("Worker asking for too many replicates [walletAddress: {}, runningReplicateNb:{}, workerMaxNbTasks:{}]",
147+
log.debug("Worker asking for too many replicates [walletAddress:{}, runningReplicateNb:{}, workerMaxNbTasks:{}]",
134148
walletAddress, runningReplicateNb, workerMaxNbTasks);
135149
return false;
136150
}
@@ -191,6 +205,7 @@ public int getAliveAvailableGpu () {
191205

192206
// region Read-and-write methods
193207
public Worker addWorker(Worker worker) {
208+
updateLastAlive(worker.getWalletAddress());
194209
return contextualLockRunner.applyWithLock(
195210
worker.getWalletAddress(),
196211
address -> addWorkerWithoutThreadSafety(worker)
@@ -213,42 +228,14 @@ private Worker addWorkerWithoutThreadSafety(Worker worker) {
213228
return workerRepository.save(worker);
214229
}
215230

216-
public Optional<Worker> updateLastAlive(String walletAddress) {
217-
return contextualLockRunner.applyWithLock(
218-
walletAddress,
219-
this::updateLastAliveWithoutThreadSafety
220-
);
231+
public void updateLastAlive(String walletAddress) {
232+
workerStatsMap.computeIfAbsent(walletAddress, WorkerStats::new)
233+
.setLastAliveDate(new Date());
221234
}
222235

223-
private Optional<Worker> updateLastAliveWithoutThreadSafety(String walletAddress) {
224-
Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
225-
if (optional.isPresent()) {
226-
Worker worker = optional.get();
227-
worker.setLastAliveDate(new Date());
228-
workerRepository.save(worker);
229-
return Optional.of(worker);
230-
}
231-
232-
return Optional.empty();
233-
}
234-
235-
public Optional<Worker> updateLastReplicateDemandDate(String walletAddress) {
236-
return contextualLockRunner.applyWithLock(
237-
walletAddress,
238-
this::updateLastReplicateDemandDateWithoutThreadSafety
239-
);
240-
}
241-
242-
private Optional<Worker> updateLastReplicateDemandDateWithoutThreadSafety(String walletAddress) {
243-
Optional<Worker> optional = workerRepository.findByWalletAddress(walletAddress);
244-
if (optional.isPresent()) {
245-
Worker worker = optional.get();
246-
worker.setLastReplicateDemandDate(new Date());
247-
workerRepository.save(worker);
248-
return Optional.of(worker);
249-
}
250-
251-
return Optional.empty();
236+
public void updateLastReplicateDemandDate(String walletAddress) {
237+
workerStatsMap.computeIfAbsent(walletAddress, WorkerStats::new)
238+
.setLastReplicateDemandDate(new Date());
252239
}
253240

254241
public Optional<Worker> addChainTaskIdToWorker(String chainTaskId, String walletAddress) {

‎src/test/java/com/iexec/core/detector/WorkerLostDetectorTests.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,8 @@
3232
import org.mockito.MockitoAnnotations;
3333

3434
import java.util.Collections;
35-
import java.util.Date;
3635
import java.util.Optional;
3736

38-
import static com.iexec.common.utils.DateTimeUtils.addMinutesToDate;
3937
import static org.mockito.Mockito.*;
4038

4139
class WorkerLostDetectorTests {
@@ -71,15 +69,11 @@ void shouldNotDetectAnyWorkerLost(){
7169

7270
@Test
7371
void shouldUpdateOneReplicateToWorkerLost(){
74-
Date twoMinutesAgo = addMinutesToDate(new Date(), -2);
75-
7672
Worker worker = Worker.builder()
7773
.walletAddress(WALLET_WORKER)
78-
.lastAliveDate(twoMinutesAgo)
7974
.participatingChainTaskIds(Collections.singletonList(CHAIN_TASK_ID))
8075
.build();
8176

82-
8377
Replicate replicate = new Replicate(WALLET_WORKER, CHAIN_TASK_ID);
8478
replicate.updateStatus(ReplicateStatus.STARTING, ReplicateStatusModifier.WORKER);
8579

@@ -156,11 +150,8 @@ void shouldNotUpdateToWorkerLostSinceCompleted(){
156150

157151
@Test
158152
void shouldNotUpdateOneReplicateToWorkerLostTaskIsExpired(){
159-
Date twoMinutesAgo = addMinutesToDate(new Date(), -2);
160-
161153
Worker worker = Worker.builder()
162154
.walletAddress(WALLET_WORKER)
163-
.lastAliveDate(twoMinutesAgo)
164155
.participatingChainTaskIds(Collections.singletonList(CHAIN_TASK_ID))
165156
.build();
166157

‎src/test/java/com/iexec/core/replicate/ReplicateSupplyServiceTests.java‎

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,6 @@ void shouldNotGetReplicateSinceNoReplicatesList() {
131131
.walletAddress(WALLET_WORKER_2)
132132
.cpuNb(4)
133133
.teeEnabled(false)
134-
.lastAliveDate(new Date())
135134
.build();
136135

137136
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -162,7 +161,6 @@ void shouldNotGetReplicateSinceConsensusReachedOnChain() {
162161
.walletAddress(WALLET_WORKER_2)
163162
.cpuNb(4)
164163
.teeEnabled(false)
165-
.lastAliveDate(new Date())
166164
.build();
167165
final Replicate replicate = new Replicate(WALLET_WORKER_2, CHAIN_TASK_ID);
168166
replicate.updateStatus(CONTRIBUTED, ReplicateStatusModifier.WORKER);
@@ -222,7 +220,6 @@ void shouldNotGetAnyReplicateSinceWorkerAlreadyParticipated() {
222220
.id("1")
223221
.walletAddress(WALLET_WORKER_1)
224222
.cpuNb(2)
225-
.lastAliveDate(new Date())
226223
.build();
227224

228225
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -259,7 +256,6 @@ void shouldNotGetReplicateSinceDoesNotNeedMoreContributionsForConsensus() {
259256
.id("1")
260257
.walletAddress(WALLET_WORKER_2)
261258
.cpuNb(2)
262-
.lastAliveDate(new Date())
263259
.build();
264260

265261
int trust = 5;
@@ -302,7 +298,6 @@ void shouldNotGetReplicateSinceEnclaveChallengeNeededButNotGenerated() {
302298
.walletAddress(WALLET_WORKER_1)
303299
.cpuNb(2)
304300
.teeEnabled(true)
305-
.lastAliveDate(new Date())
306301
.build();
307302

308303
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -341,7 +336,6 @@ void shouldGetOnlyOneReplicateSinceOtherOneReachedConsensusDeadline() {
341336
.walletAddress(WALLET_WORKER_1)
342337
.cpuNb(4)
343338
.teeEnabled(false)
344-
.lastAliveDate(new Date())
345339
.build();
346340

347341
int trust = 5;
@@ -389,7 +383,6 @@ void shouldNotGetReplicateWhenTaskAlreadyAccessed() {
389383
.walletAddress(WALLET_WORKER_1)
390384
.cpuNb(2)
391385
.teeEnabled(false)
392-
.lastAliveDate(new Date())
393386
.build();
394387

395388
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -421,7 +414,6 @@ void shouldGetReplicateWithNoTee() {
421414
.walletAddress(WALLET_WORKER_1)
422415
.cpuNb(2)
423416
.teeEnabled(false)
424-
.lastAliveDate(new Date())
425417
.build();
426418

427419
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -461,7 +453,6 @@ void shouldGetReplicateWithTee() {
461453
.walletAddress(WALLET_WORKER_1)
462454
.cpuNb(2)
463455
.teeEnabled(true)
464-
.lastAliveDate(new Date())
465456
.build();
466457

467458
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -501,7 +492,6 @@ void shouldTeeNeededTaskNotBeGivenToTeeDisabledWorker() {
501492
.walletAddress(WALLET_WORKER_1)
502493
.cpuNb(2)
503494
.teeEnabled(false)
504-
.lastAliveDate(new Date())
505495
.build();
506496

507497
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);
@@ -530,7 +520,6 @@ void shouldTeeNeededTaskBeGivenToTeeEnabledWorker() {
530520
.walletAddress(WALLET_WORKER_1)
531521
.cpuNb(2)
532522
.teeEnabled(true)
533-
.lastAliveDate(new Date())
534523
.build();
535524

536525
Task runningTask = new Task(DAPP_NAME, COMMAND_LINE, 5, CHAIN_TASK_ID);

0 commit comments

Comments
 (0)