feat: Task execution pipelining #484

aneojgurhem · 2023-09-04T07:22:16Z

Some things I read to get to this point.

Common/src/Pollster/Pollster.cs

Common/src/Pollster/RunningTaskQueue.cs

lemaitre-aneo · 2023-09-08T08:54:28Z

Common/src/Pollster/RunningTaskQueue.cs

+  public bool RemoveException([MaybeNullWhen(false)] out Exception e)
+  {
+    var r = exceptions_.Count > 0;
+
+    e = r
+          ? exceptions_.Dequeue()
+          : null;
+
+    return r;
+  }


You could instead return an aggregate exception and use .Handle to filter out some exceptions out of it.

Not as clear and easy to use.

Common/src/Pollster/TaskProcessingChecker/TaskProcessingCheckerClient.cs

Common/tests/Pollster/TaskHandlerTest.cs

…me agent

… this agent

…or in LogStatsFromSessionAsync

…uisition timeout is not exceeded

… of Register

aneojgurhem · 2023-09-11T06:02:23Z

I did some tests on AWS with a c5.12xlarge instance.
Comparison vs main (e4ca29a)

Install dependencies

sudo yum install -y git docker yum-utils

# just
mkdir -p ~/bin
curl --proto '=https' --tlsv1.2 -sSf https://just.systems/install.sh | bash -s -- --to ~/bin

# terraform
sudo yum-config-manager --add-repo https://rpm.releases.hashicorp.com/AmazonLinux/hashicorp.repo
sudo yum -y install terraform

# docker
sudo systemctl enable docker.service
sudo systemctl start docker.service
sudo usermod -a -G docker ec2-user

Relog

Build clients

docker build -t dockerhubaneo/armonik_core_htcmock_test_client:0.0.0.0-test -f Tests/HtcMock/Client/src/Dockerfile .
docker build -t dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test -f Tests/Bench/Client/src/Dockerfile .

or use (prefered way)

just tag=0.0.0.0-test buildBenchClient
just tag=0.0.0.0-test buildHtcmockClient

How to read results ?

I measured throughput (tasks/seconds). Every table shows that. Higher results are better

Test with htcmock

just worker=mock replicas=30 partitions=1 build-deploy

Some warmup

docker run --net armonik_network --rm \
	-e HtcMock__TotalCalculationTime=00:00:10 \
	-e HtcMock__NTasks=300 \
	-e HtcMock__SubTasksLevels=4 \
	-e HtcMock__EnableFastCompute=true \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080\
	dockerhubaneo/armonik_core_htcmock_test_client:0.0.0.0-test

Test with complex dependencies

docker run --net armonik_network --rm \
	-e HtcMock__TotalCalculationTime=00:00:10 \
	-e HtcMock__NTasks=10000 \
	-e HtcMock__SubTasksLevels=100 \
	-e HtcMock__EnableFastCompute=true \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_htcmock_test_client:0.0.0.0-test

Result 480 tasks/s vs 365 on main

Tests with Bench Client/Worker

just destroy
just worker=bench replicas=30 partitions=1 build-deploy

Some warmup

docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10 \
	-e BenchOptions__PayloadSize=1 \
	-e BenchOptions__ResultSize=1 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test

Tests with 10 seconds tasks and input/output size increase. We should observe constant throughput until prefetching is longer than task execution.

docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=1 \
	-e BenchOptions__ResultSize=1 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=10 \
	-e BenchOptions__ResultSize=10 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=100 \
	-e BenchOptions__ResultSize=100 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=1000 \
	-e BenchOptions__ResultSize=1000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=10000 \
	-e BenchOptions__ResultSize=10000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=20000 \
	-e BenchOptions__ResultSize=20000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=10000 \
	-e BenchOptions__PayloadSize=30000 \
	-e BenchOptions__ResultSize=30000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test

Size	pipelining	main
1	2.971	2.971
10	2.976	2.971
100	2.975	2.97
1000	2.973	2.967
10000	2.873	2.800
20000	2.834	2.711

Well, it does not change much. Tasks may be too long. Let's use 1 second tasks.

docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=1 \
	-e BenchOptions__ResultSize=1 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=10 \
	-e BenchOptions__ResultSize=10 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=100 \
	-e BenchOptions__ResultSize=100 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=1000 \
	-e BenchOptions__ResultSize=1000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=10000 \
	-e BenchOptions__ResultSize=10000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=20000 \
	-e BenchOptions__ResultSize=20000 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test

Size	pipelining	main
1	27.9	27.8
10	27.5	27.3
100	27.7	27.5
1000	27.2	26.9
10000	22.0	18.0
20000	17.7	12.5

Without results

docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=1 \
	-e BenchOptions__ResultSize=1 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=10 \
	-e BenchOptions__ResultSize=1 \
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=100 \
	-e BenchOptions__ResultSize=1\
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=1000 \
	-e BenchOptions__ResultSize=1\
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=10000 \
	-e BenchOptions__ResultSize=1\
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test
docker run --net armonik_network --rm \
	-e BenchOptions__NTasks=300 \
	-e BenchOptions__TaskDurationMs=1000 \
	-e BenchOptions__PayloadSize=20000 \
	-e BenchOptions__ResultSize=1\
	-e BenchOptions__DegreeOfParallelism=100 \
	-e GrpcClient__Endpoint=http://armonik.control.submitter:1080 \
	dockerhubaneo/armonik_core_bench_test_client:0.0.0.0-test

Size	pipelining	main
1	27.6	27.6
10	28.2	27.2
100	28.1	27.1
1000	26.3	26.7
10000	24.0	20.8
20000	21.8	16.3

Flush redis

docker exec -it object redis-cli flushall sync

aneojgurhem requested review from ngruelaneo, lemaitre-aneo, melflitty-aneo and dbrasseur-aneo September 4, 2023 07:22

aneojgurhem self-assigned this Sep 4, 2023

aneojgurhem force-pushed the jg/pipelining branch from 68078c3 to e62c0b2 Compare September 4, 2023 13:05

lemaitre-aneo reviewed Sep 5, 2023

View reviewed changes

aneojgurhem force-pushed the jg/pipelining branch from e12beac to f280ce2 Compare September 6, 2023 06:33

lemaitre-aneo reviewed Sep 8, 2023

View reviewed changes

aneojgurhem added 21 commits September 11, 2023 06:44

feat: Task execution pipelining

75c71ca

fix: properly manage task handler lifecycle in RunningTaskProcessor

e510448

refactor: task deduplication and cancellation adapted to task pipelining

5b9e4db

tests: more robust CancelLongTaskShouldSucceed

2fd6620

fix: manage tasks already in a final state during acquisition

69412ab

fix: task acquisition when message is processed in parallel on the sa…

8f481e7

…me agent

refactor: add message handler id in scopes

09d316a

refactor: add an option for timeout before new task acquisition

7aab15b

refactor: use RethrowWithStacktrace

6f0115c

refactor: use concurrent dictionnary to store task handlers

dca869f

refactor: if task id is already in dict, task is already processed by…

5b5f628

… this agent

fix: properly dispose task handler when preprocessing throws

98ef1e5

refactor: log task data that cannot be processed when there is an err…

dc8f422

…or in LogStatsFromSessionAsync

fix: proper rethrows in TaskHandler

dca1b94

refactor: better logs for task starting

f91d125

refactor: retry check on other agent when connection is refused

286925c

fix: postpone task acquisition when other agent is unavailable if acq…

9dba90b

…uisition timeout is not exceeded

refactor: remove unneeded (Exception) in catch

dc8bd20

refactor: make StopCancelledTask a function

85b97cb

refactor: use CancellationTokenSource.CreateLinkedTokenSource instead…

ec055d7

… of Register

refactor: simpler use of async

274fd5d

aneojgurhem force-pushed the jg/pipelining branch from 09f086a to 274fd5d Compare September 11, 2023 04:44

fix: return task instead of ConfiguredTaskAwaitable

4287ae0

refactor: create TaskQueueBase to factorize code

d99eb9c

lemaitre-aneo previously approved these changes Sep 13, 2023

View reviewed changes

refactor: use const for number of retries

3f3a399

aneojgurhem dismissed lemaitre-aneo’s stale review via 3f3a399 September 13, 2023 08:58

lemaitre-aneo approved these changes Sep 13, 2023

View reviewed changes

aneojgurhem marked this pull request as ready for review September 13, 2023 09:17

aneojgurhem merged commit 8b5ea41 into main Sep 13, 2023

aneojgurhem deleted the jg/pipelining branch September 13, 2023 09:45

aneojgurhem mentioned this pull request Sep 18, 2023

feat: Agent and Worker send data with files #499

Merged

aneojgurhem mentioned this pull request Jan 8, 2024

feat: use .Net8.0 #570

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Task execution pipelining #484

feat: Task execution pipelining #484

aneojgurhem commented Sep 4, 2023 •

edited

Loading

lemaitre-aneo Sep 8, 2023

aneojgurhem Sep 10, 2023

aneojgurhem commented Sep 11, 2023 •

edited

Loading

feat: Task execution pipelining #484

feat: Task execution pipelining #484

Conversation

aneojgurhem commented Sep 4, 2023 • edited Loading

lemaitre-aneo Sep 8, 2023

Choose a reason for hiding this comment

aneojgurhem Sep 10, 2023

Choose a reason for hiding this comment

aneojgurhem commented Sep 11, 2023 • edited Loading

Install dependencies

How to read results ?

Test with htcmock

Tests with Bench Client/Worker

Flush redis

aneojgurhem commented Sep 4, 2023 •

edited

Loading

aneojgurhem commented Sep 11, 2023 •

edited

Loading