MOD-6572 replace semaphore with wait in jobq #4446

meiravgri · 2024-02-12T13:03:52Z

This PR significantly modifies the threadpool's pulling mechanism, setting the stage for more flexible runtime configurations.

Changing pulling synchronization machanism

Previously, the threadpool was synchronized using a semaphore-based waiting mechanism for job retrieval. This PR replaces that mechanism with a condition variable.

The revised job pulling mechanism operates as follows:
2. Threads enter a waiting state on a condition variable when the job queue is both empty and running.
3. Upon adding a new job to the queue:

Single job addition triggers a signal to one thread.
Adding multiple jobs results in a broadcast signal, waking up all threads for processing.

Configure threadpool at runtime

Threadpool state

can be either initialized - has n_threads valid and ready to pull threads,
or uninitialized - some or all of the threads may have exited.
Threadpool state is set to uninitialized when terminate_when_empty or destroy are called.

Removed `threads_all_idle` condition variable

drain and wait are internally implemented with a busy wait.
calling wait() is equivalent to calling drain(threshold = 0, yieldCB = nullptr)

Thread state

A thread can be in one of three states: running, terminate_when_empty, or dead.
The thread state can be configured by pushing a change_state_job to the admin priority queue.

Pause and resume

When pausing the threadpool, the jobq state is changed to paused. pause function will return when there are no more jobs in progress (i.e num_working_threads == 0)

Intialization

The new design assumes that the threadpool is initialized by the main thread.
All threadpool initialization is lazy and occurs upon the first push to the queue.
Since the GC pushes to the jobq from the bg, it breaks this assumption. Hence, the GC thpool initialized upon module startup.

Auestethic changes

Add name to the threadpool

The threadpool name is used to set the thepool's threads names.
a thread name is <thpool_name>-<rand_id>.
As the maximum number of threads per thpool is 10,000, rand_id is a random number between 0 to 9,999.

Remove `thpool->threads` array

This array was unnecessary and has been removed.

codecov · 2024-02-12T13:31:42Z

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 86.25%. Comparing base (e907ece) to head (d147db0).

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #4446      +/-   ##
==========================================
- Coverage   86.25%   86.25%   -0.01%     
==========================================
  Files         190      190              
  Lines       34822    34822              
==========================================
- Hits        30036    30035       -1     
- Misses       4786     4787       +1

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

wait and drain are modified accordingly.

sembm om dim - 768

remove_sem_bm.py

deps/thpool/thpool.c

rename priority_queue->num_threads_working to num_threads_not_idle uncomment LOG remove script

deps/thpool/thpool.c

alonre24

NICE!
few small comments

deps/thpool/thpool.c

alonre24 · 2024-04-02T09:02:28Z

deps/thpool/thpool.c

@@ -86,15 +81,11 @@ typedef enum {
 typedef struct redisearch_thpool_t {
  thread** threads;                    /* pointer to threads        */
  size_t total_threads_count;
-  volatile size_t num_threads_alive;   /* threads currently alive   */
-  volatile size_t num_threads_working; /* threads currently working */
+  volatile atomic_size_t num_threads_alive;   /* threads currently alive   */


I think that atomic type implicitly means volatile as well.

Not as far as i know, those are different concepts.
Do you have a reference that claims something else?

They are different you are right, but do we need both? Isn't atomic enough?

Since this variable is accessed and modified across different threads, yes.
Having said that, should_run should be volatile as well.

alonre24 · 2024-04-02T09:08:48Z

deps/thpool/thpool.c

        LOG_IF_EXISTS("verbose", "Job queue is empty - terminating thread %d", thread_p->id);
        thpool_p->state = THPOOL_UNINITIALIZED;
+        thpool_p->jobqueue.should_run = false;
+        pthread_cond_broadcast(&thpool_p->jobqueue.has_jobs);


Just making sure, in THPOOL_TERMINATE_WHEN_EMPTY mode, all threads will come here eventually, right? (meaning pthread_cond_broadcast is called multiple times)

n_threads in the worst case.
Here we also change the thpool state so if we are lucky some threads will finish their jobs before the jobq is empty, try to continue to the next iteration (skipping the broadcast), and then a different thread will change the state and the first one will not enter the loop.

I see. Consider elaborating on the documentation in this sensitive place

deps/thpool/thpool.c

tests/cpptests/test_cpp_thpool.cpp

unify if (job_p) in thread_do remove unused var from sleep_and_set in test_cpp_thpool

* add admin priority queue *add threads manager to send admin jobs to threads *not lazy gc init to avoid bg intialization

remove threads array from thpool struct

change thpool state to THPOOL_UNINITIALIZED in redisearch_thpool_terminate_when_empty wait for num working threads to be 0 in redisearch_thpool_terminate_pause_threads thpool cpp tests: TestTerminateWhenEmpty: test recreating the threads after temination new test: TestPauseResume

GuyAv46

Nice!

tests/pytests/test_multithread.py

src/gc.c

GuyAv46 · 2024-05-26T15:03:07Z

src/concurrent_ctx.c

@@ -19,7 +19,7 @@ int ConcurrentSearch_CreatePool(int numThreads) {
  }
  int poolId = array_len(threadpools_g);
  threadpools_g = array_append(threadpools_g, redisearch_thpool_create(numThreads, DEFAULT_PRIVILEGED_THREADS_NUM,
-                                                                       LogCallback));
+                                                                       LogCallback, "coord"));


Consider passing the pool name as a parameter. I would at least add a comment about why we chose this name

2 rows above the array explictly initialized to size 1 with a comment "Only used by the coordinator, so 1 is enough"

deps/thpool/pthread_barrier.h

GuyAv46 · 2024-05-26T16:18:05Z

deps/thpool/thpool.c


+  /* Initialize threads if needed */
+  redisearch_thpool_verify_init(thpool_p);


Consider moving into priority_queue_push_chain to avoid duplications

I don't think it's the priority queue's responsibility to initialize the threads.

I added a wrapper function priority_queue_push_chain_init_threads to the threadpool section

priority_queue_push_chain_unsafe: replaced redisearch_thpool_t * argument that was used only to get the priority queue, with a priority_queue *

deps/thpool/thpool.c

GuyAv46 · 2024-05-27T06:54:59Z

deps/thpool/thpool.c

-  pthread_detach((*thread_p)->pthread);
+static int thread_init(redisearch_thpool_t *thpool_p) {
+  pthread_t thread_id;
+  pthread_create(&thread_id, NULL, (void *)thread_do, thpool_p);


is this cast needed?

deps/thpool/thpool.c

GuyAv46 · 2024-05-27T08:23:32Z

deps/thpool/thpool.c

+    rm_free(job_p);
+
+    // Both variables are atomic, so we can do this without a lock.
+    thpool_p->total_jobs_done += !job_ctx.is_admin;


Suggested change

thpool_p->total_jobs_done += !job_ctx.is_admin;

if (!job_ctx.is_admin) thpool_p->total_jobs_done++;

in the C Standard (ISO/IEC 9899:2018):

Section 6.5.3.3 Unary arithmetic operators: "The result of the logical negation operator ! is 0 if the value of its operand compares unequal to 0, 1 if the value of its operand compares equal to 0. The result has type int."

@GuyAv46

…lback, triggred by redis from the main thread remove thpool_init. not in use renaming of thpool structs' members remove priority_push_chain and introduce redisearch_thpool_push_chain_init_threads that encapsulate both safe push to the queue and threads verify init pthread_varrier: add volatile to cycle test_multithread: replace algo, datatype loop with an automatic generation of a function for each permutation (algo, datatype) as Test_burst_threads_sanity class attribute Thanks @GuyAv46

tests/pytests/test_multithread.py

Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com>

replace semaphore with wait in jobq

65f2e4a

meiravgri and others added 10 commits February 12, 2024 14:51

reset jobq run flag in verify init

4d1b6a6

Merge branch 'master' into meiravg_remove_thpool_sem_POC

499ae6d

add bm script

20c40e9

update script

937721d

Merge branch 'master' into meiravg_remove_thpool_sem_POC

5bfd42f

update num_wrking under the jobq lock

3310ed5

wait and drain are modified accordingly.

Merge branch 'master' into meiravg_remove_thpool_sem_POC

770768c

inistialize num_working in redisearch_thpool_verify_init

981c23f

remove logs

0c05512

signal one if we insert one job, otherwise broadcast

c0dd3a4

sembm om dim - 768

meiravgri changed the title ~~replace semaphore with wait in jobq~~ MOD-6572 replace semaphore with wait in jobq Feb 23, 2024

meiravgri and others added 7 commits March 6, 2024 13:45

print every 50K queries

20e7759

cleanups

fc4c534

Merge branch 'master' into meiravg_remove_thpool_sem_POC

093ded1

Merge branch 'master' into meiravg_remove_thpool_sem_POC

4b68709

move n_vectors and vecs_file to main

98b11d2

fix typo

c02b331

Merge branch 'master' into meiravg_remove_thpool_sem_POC

c57550b

meiravgri requested review from alonre24 and GuyAv46 March 24, 2024 06:39

script cleanups

472bd5f

GuyAv46 reviewed Mar 26, 2024

View reviewed changes

remove_sem_bm.py Outdated Show resolved Hide resolved

deps/thpool/thpool.c Outdated Show resolved Hide resolved

deps/thpool/thpool.c Show resolved Hide resolved

deps/thpool/thpool.c Outdated Show resolved Hide resolved

meiravgri added 2 commits March 26, 2024 10:29

Merge branch 'master' into meiravg_remove_thpool_sem_POC

cf71f08

rename priority_queue->cond to has_jobs

ba290bc

rename priority_queue->num_threads_working to num_threads_not_idle uncomment LOG remove script

GuyAv46 previously approved these changes Mar 26, 2024

View reviewed changes

deps/thpool/thpool.c Show resolved Hide resolved

fix identations

147ca3b

meiravgri dismissed GuyAv46’s stale review via 147ca3b March 26, 2024 11:31

GuyAv46 previously approved these changes Mar 26, 2024

View reviewed changes

deps/thpool/thpool.c Outdated Show resolved Hide resolved

meiravgri added 2 commits March 30, 2024 05:28

add seed to query in poc script

cf1247b

remove script

e1b37dd

GuyAv46 previously approved these changes Apr 1, 2024

View reviewed changes

alonre24 reviewed Apr 2, 2024

View reviewed changes

remove lock from redisearch_thpool_num_threads_working

0154074

unify if (job_p) in thread_do remove unused var from sleep_and_set in test_cpp_thpool

meiravgri dismissed GuyAv46’s stale review via 0154074 April 2, 2024 13:59

meiravgri added the backport 2.10 label Apr 2, 2024

meiravgri and others added 14 commits April 2, 2024 17:54

Merge branch 'master' into meiravg_remove_thpool_sem_POC

9047c2f

add volatile to should_run

9d0e017

Merge branch 'master' into meiravg_remove_thpool_sem_POC

f6408d2

Merge branch 'master' into meiravg_remove_thpool_sem_POC

22e0b66

Merge branch 'master' into meiravg_remove_thpool_sem_POC

3427a5a

refactor thpool

3c4ecab

* add admin priority queue *add threads manager to send admin jobs to threads *not lazy gc init to avoid bg intialization

little fix

2258835

add thpool name

5f9a9ba

remove threads array from thpool struct

format thpool file

8f70a31

Merge branch 'master' into meiravg_remove_thpool_sem_POC

5a57151

add pthread_barrier implmntation if not defined

d1edec6

update thpool.h

4c2e867

thpool.h format

368b495

meiravgri requested review from GuyAv46 and alonre24 May 26, 2024 14:53

GuyAv46 reviewed May 27, 2024

View reviewed changes

meiravgri added 2 commits May 27, 2024 09:04

Merge branch 'master' into meiravg_remove_thpool_sem_POC

0512e81

GuyAv46 reviewed May 27, 2024

View reviewed changes

tests/pytests/test_multithread.py Outdated Show resolved Hide resolved

tests/pytests/test_multithread.py Outdated Show resolved Hide resolved

tests/pytests/test_multithread.py Outdated Show resolved Hide resolved

meiravgri and others added 3 commits May 27, 2024 17:00

Update tests/pytests/test_multithread.py

a161065

Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com>

Update tests/pytests/test_multithread.py

d1f51d2

Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com>

Update tests/pytests/test_multithread.py

d147db0

Co-authored-by: GuyAv46 <47632673+GuyAv46@users.noreply.github.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MOD-6572 replace semaphore with wait in jobq #4446

MOD-6572 replace semaphore with wait in jobq #4446

meiravgri commented Feb 12, 2024 •

edited

codecov bot commented Feb 12, 2024 •

edited

alonre24 left a comment

alonre24 Apr 2, 2024

meiravgri Apr 2, 2024

alonre24 Apr 2, 2024

meiravgri Apr 2, 2024

alonre24 Apr 2, 2024

meiravgri Apr 2, 2024

alonre24 Apr 2, 2024

GuyAv46 left a comment

GuyAv46 May 26, 2024

meiravgri May 27, 2024

GuyAv46 May 26, 2024

meiravgri May 27, 2024

GuyAv46 May 27, 2024

GuyAv46 May 27, 2024

meiravgri May 27, 2024


		/* Initialize threads if needed */
		redisearch_thpool_verify_init(thpool_p);

	thpool_p->total_jobs_done += !job_ctx.is_admin;
	if (!job_ctx.is_admin) thpool_p->total_jobs_done++;

MOD-6572 replace semaphore with wait in jobq #4446

Are you sure you want to change the base?

MOD-6572 replace semaphore with wait in jobq #4446

Conversation

meiravgri commented Feb 12, 2024 • edited

Changing pulling synchronization machanism

Configure threadpool at runtime

Threadpool state

Removed threads_all_idle condition variable

Thread state

Pause and resume

Intialization

Auestethic changes

Add name to the threadpool

Remove thpool->threads array

codecov bot commented Feb 12, 2024 • edited

Codecov Report

alonre24 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GuyAv46 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

meiravgri commented Feb 12, 2024 •

edited

Removed `threads_all_idle` condition variable

Remove `thpool->threads` array

codecov bot commented Feb 12, 2024 •

edited