โHackMD
mergesort-concurrent
github contributed by <Diana Ho>
tags: d0651 sys
ๆกไพๅๆ
ๅฐๅฎๅ Linked List
ๅ ฉๅคง้้ป
- ๆๅบ็ๅฐ่ฑกๆฏ singly-linked list
- ๅฉ็จ POSIX Thread ่็๏ผ้่ฆไธไฝตๆไฝ synchronization object
้ ๆ็ฎๆจ
- ไฝ็บ concurrency ็ๅฑ็คบๆกไพ
- ๅญธ็ฟ POSIX Thread Programming๏ผ็นๅฅๆฏ synchronization object
- ็บๆฅๅพๆ่ฝๅๆๅ scalability ็ ็ฉถๅปบๆงๅบ็คๅปบ่จญ
- ๅญธ็ฟ็จๅผๅ่ณชๅๆๅ็ธ้็้็ผๅทฅๅ ท
ไฝๆฅญ่ฆๆฑ
- ๅฐ merge sort ็ๅฏฆๅๆน็บๅฏๆฅๅ phonebook-concurrent ็ 35 ่ฌ็ญ่ณๆ่ผธๅ
ฅ็่ณๆๆช
- ๅญๅ
ธๆช่ณๆ้่ฆไบๅ
็จ
sort -R่็้ - ๆ่ๅฆไฝๅพๅฐๅๅปๅไฝ็ไบๆธๆๅ๏ผไธฆไธ่จญ่จ่ชๅๆธฌ่ฉฆ็ๆฉๅถ
- ๅญๅ
ธๆช่ณๆ้่ฆไบๅ
็จ
- ็ ็ฉถ thread pool ็ฎก็ worker thread ็ๅฏฆๅ๏ผๆๅบๅฏฆๅๅฑค้ข็ไธ่ถณ๏ผไธฆไธๅ็ ง concurrent-ll๏ผๆๅบ lock-free ็ๅฏฆๅ
- ๅญธ็ฟ concurrent-ll (concurrent linked-list ๅฏฆไฝ) ็ scalability ๅๆๆนๅผ๏ผ้้ gnuplot ่ฃฝๅๆฏ่ผ merge sort ๅจไธๅๅท่ก็ทๆธ้ๆไฝ็ๆ่ฝ
- ๆณจๆๅฐ linked list ๆฏๅ็ฏ้ป้ ็ฝฎ็่จๆถ้ซๅพๅพๆฏไธ้ฃ็บ๏ผๆ่้ๅฐๆ่ฝๅๆ็ๅฝฑ้ฟ
- ไธไฝตๅ่ฉฆ้ๆง (refactor) ็ตฆๅฎ็็จๅผ็ขผ๏ผไฝฟๅพ็จๅผๆดๅฎนๆ้ฑ่ฎๅ็ถญ่ญทใๅปถ็บ A05: introspect๏ผไธๅชๆฏๅจๅ ฑ็ญไธ็จๆๅญๆๅบ่ฏๆง่ฉณ็ก็ๆน่ฉ๏ผไน่ฉฒๅๆ ๅจ็จๅผ็ขผ็่ฎ้ฉ
- ๅ ฑ็ญ็ๅ งๅฎนๅ้็จ GraphViz ่ฃฝไฝ
POSIX Thread Programming
Model
- Boss/Worker Model
Boss ๆฅๆถๅฐไธๅๆฐ่ฆๆฑๆ๏ผๅๆ ๅปบ็ซไธๅ Worker ๅฎๆ่ฉฒ้ ่ฆๆฑ๏ผ้ฉ็จๆผไผบๆๅจไธใๅฏๆผ็จๅผ้ๅงๆๅ ๅปบ็ซๅฅฝไธๅฎๆธ้็ๅท่ก็ท(Thread Pool)๏ผๆธๅฐๅท่กๆๆ็่ฒ ๆใ - Peer Model (Workcrew model)
็ธ่ผๆผ Boss/Worker ๆจกๅผ๏ผๆญคๆจกๅผ็ๆฏๅท่ก็ทๆๆ่ชๅทฑ็่ผธๅ ฅ๏ผ้ฉ็จๆผ็ฉ้ฃ้็ฎใ่ณๆๅบซๆๅฐ็ญใ - Pipeline Model
ๅท่ก็ท็ฑไธไธ้ๆฎตๅๅพ่ผธๅ ฅ่ณๆ๏ผไธฆๅฐ็ตๆๅณ่ณไธไธ้ๆฎต็ๅท่ก็ทใๆด้ซ่ผธๅบๅๅถๆผๅท่กๆไน ็้ๆฎต๏ผ้ ๆณจๆๆฏไธ้ๆฎต็ๅทฅไฝ่ฒ ๆ่ฆ็กๅฏ่ฝๅนณๅใ
manager-worker
็บไบๅ้ ๅทฅไฝ๏ผๅจ worker thread ๅฏฆไฝ Task ็ๆฉๅถใๆฏๅไธป่ฆ็ๆไฝๆๅ ่ขซๆพ้ฒ task queue ่ฃก้ ญ๏ผ็ฉบ้็ thread ๅๅพ task queue ่ฃก้ ญๆๅ task ๅท่ก๏ผๅช่ฆๆๆๅ็ๆไฝๅฏซๆ task๏ผๅฐฑ่ฝ้ ๅฉๅท่กใ
Synchronization
-
pthread_join๏ผ ๆๅท่ก็ทๆซๅๅท่ก็ดๅฐๅฆไธ็ด่ก็ท็ตๆ
-
Mutex๏ผ ๅไธๆ้ๅ งๅชๆไธๅท่ก็ทๅฏไปฅไฟๆ่ฉฒ lock ๅไฟ่ญท่ณๆ็ๅญๅๆฌ
- Initialize
- Static๏ผ
PTHREAD_MUTEX_INITIALIZER - Dynamic๏ผ
pthread_mutex_init()
- Static๏ผ
- Lock
pthread_mutex_lock()
- Release
pthread_mutex_unlock()
- Initialize
-
Condition variable๏ผ ๅฐไบไปถๅฏฆ่ณชๅ๏ผไธฆๆไพๅฝๅผๅ้็ญๅพ ่ฉฒไบไปถ็ๅท่ก็ท
- Initialize
- Static๏ผ
PTHREAD_COND_INITIALIZER - Dynamic๏ผ
pthread_cond_init()
- Static๏ผ
- Waits for condition variables
- ็ญๅพ
็ดๅฐๅ้๏ผ
pthread_cond_wait() - ็ญๅพ
็นๅฎๆ้๏ผ
pthread_cond_timewait()
- ็ญๅพ
็ดๅฐๅ้๏ผ
- ๅ้็ญๅพ
ไธญๅท่ก็ท
- ๅ้ๅ
ถไธญไนไธ๏ผ
pthread_cond_signal() - ๅ้ๆๆ็ญๅพ
ไธญๅท่ก็ท๏ผ
pthread_cond_broadcast()
- ๅ้ๅ
ถไธญไนไธ๏ผ
- ็บ้ฟๅ Spurious Wake Ups ๆๅ ถไปๅ ่ขซๅ้็ๅท่ก็ทๅท่กๅฐ่ดๆขไปถ่ฎๆธๅๆฌกไธๆ็ซ๏ผไธ่ฌๆไปฅ่ฟดๅๅคๆทๆขไปถๆฏๅฆๆ็ซ๏ผไธๆ็ซๅๅๆฌก้ฒๅ ฅ็ญๅพ
- Initialize
-
pthread_once๏ผ ไฟ่ญๅๅงๅฝๅผ่ขซ่จฑๅค็ด่ก็ทๅผๅซๆๅ ๅท่กไธๆฌก
- ๅฎฃๅ pthread_once_t ไธฆ้ๆ
ๅๅง็บ
PTHREAD_ONCE_INIT - ไปฅ pthread_once() ่ once ๅๅๅผๅซ็ฎๆจๅฝๅผ
- ไธๅฎน่จฑๅณ้ๅๆธ่ณ once ๆไฟ่ญท็ๅฝๅผ
- ๅฎฃๅ pthread_once_t ไธฆ้ๆ
ๅๅง็บ
Pthreads Management
Key
ไธ็จฎๆๆจ๏ผๅฐ่ณๆๅๅท่ก็ท้ฒ่ก้่ฏ
Thread cancellation
- States๏ผ
PTHREAD_CANCEL_DISABLE- Type๏ผ ignored
PTHREAD_CANCEL_ENABLE- Types๏ผ
PTHREAD_CANCEL_ASYNCHRONOUS๏ผ็ซๅปๅๆถPTHREAD_CANCEL_DEFERRED๏ผ็ถๅท่กๅฐ Cancellation-point ๆ็ผ็
- Types๏ผ
- Cancellation-point
pthread_cond_wait(), pthread_cond_timewait(), pthread_join()- ไฝฟ็จ่
ๅฎ็พฉ
pthread_testcancel()
- Asynchronous Cancellation๏ผ ็ถๅท่ก็ทๅจๆญค็ๆ ๆ๏ผๅณไฝฟๅจๅท่กๅฝๅผๅบซๅผๅซๆๆฏ็ณป็ตฑๅผๅซๆไน่ฝๅค ่ขซๅๆถ๏ผ้ค้่ขซๅฎ็พฉ็บ cancellation-safe ๅฆๅๆ้ฒๆญข่ฉฒๆ ๅฝข็ผ็
mutex contention
mutrace ๅฏ็จไพๅตๆธฌ lock contention๏ผไฝฟ็จๆนไพฟ๏ผไธ้่ฆ้ๆฐ็ทจ่ญฏ็จๅผ็ขผใ
-
Lock contention: this occurs whenever one process or thread attempts to acquire a lock held by another process or thread. The more fine-grained the available locks, the less likely one process/thread will request a lock held by the other. (For example, locking a row rather than the entire table, or locking a cell rather than the entire row.)
-
mutrace v.s. valgrind/drd
In contrast to valgrind/drd it does not virtualize the CPU instruction set, making it a lot faster. In fact, the hooks mutrace relies on to profile mutex operations should only minimally influence application runtime. Mutrace is not useful for finding synchronizations bugs, it is solely useful for profiling locks.
-
About mutrace
- mutrace is implemented entirely in userspace.
- build your application with
-rdynamicto make the backtraces mutrace generates useful. -r| โtrack-rt: checks on each mutex operation wheter it is executed by a realtime thread or not.
=>use this to track down which mutexes are good candidates for priority inheritance.matracethat can be used to track down memory allocation operations in realtime threads.
้ฑ่ฎ็จๅผ็ขผ
list.[ch]
list.h
- node๏ผๅฒๅญๅฏฆ้็่ณๆ
typedef struct node {
val_t data;
struct node *next;
} node_t;
- linked list๏ผๅปบ็ซ linked list ้ ไพฟ็ด้้ทๅบฆ๏ผๆไปฅๅจๅๅฒ list ็ๆๅๅช้่ฆ่ฑ
typedef struct llist {
node_t *head;
uint32_t size;
} llist_t;
intptr_t ่ณๆๅๆ
When/Why to use intptr_t for type-casting in C?
What is the use of intptr_t?
- ๅ่ฝ
- ๅฏpointer่ฝๆๆintegerๅๆ
๏ผๅไฝๅ
้็ฎ่ๅ ๆธๆณใ
็ธ่ผๆผvoid *๏ผintptr_t็ไฝๅๅฏไปฅๅ bitwise operation ่void *ไธ่ฝใๅไฝๅ ้็ฎ๏ผ็ก่็uintptr_tๆๆฏ่ผๅฅฝใ
โโโโ/* Types for `void *' pointers. */ โโโโ#if __WORDSIZE == 64 โโโโ# ifndef __intptr_t_defined โโโโtypedef long int intptr_t; โโโโ# define __intptr_t_defined โโโโ# endif โโโโtypedef unsigned long int uintptr_t; โโโโ#else โโโโ# ifndef __intptr_t_defined โโโโtypedef int intptr_t; โโโโ# define __intptr_t_defined โโโโ# endif โโโโtypedef unsigned int uintptr_t; โโโโ#endif- pointer ่ฝๆ integer ๆ๏ผ32 bits ๆ 64 bits ็ machine ๆๆ pointer ้ทๅบฆ็ๅทฎๅฅใ่ฅไฝฟ็จๅฐ
intptr_t๏ผๅพๅพๆๅๅช้#include <stdint>ๅฐฑ่ฝๅ ้ค้บป็ ฉ็ๅๆ ่ฝๆใ้คๆญคไนๅค๏ผไนๆฏ็ขบไฟๅๆ ่ฝๆไธไธๆๅบๅ้ก็ไฟๅฎไฝๆณใ
โ#ifdef MACHINE 64 โ typedef long intptr; โ#else // MACHINE 32 โ typedef int intptr; โ#endif - ๅฏpointer่ฝๆๆintegerๅๆ
๏ผๅไฝๅ
้็ฎ่ๅ ๆธๆณใ
- ็ฎ็
- ่ฎ node ไธญ็่ณๆๅฏไปฅๆฏๆๆจๅๆ ๏ผไนๅฏไปฅๆฏไธ่ฌ่ณๆๅๆ ใ
- ็บไบ็ธๅฎนไธๅๅนณๅฐๆ่จญ่จใ
list.c
- node_new๏ผๅจ next ๅๆนๆพๅ ฅไธๅๆฐ็ node๏ผไธฆๅๅณๆๅ่ฉฒ node ไนๆๆจ
- list_new๏ผๅๅงๅไธไธฒๆฐ็ linked-list
- list_add๏ผๅจ linked-list ็ๅฐพ็ซฏๅ ๅ ฅไธๅๆฐ็ node
threadpool.[ch]
threadpool.h
- task๏ผ่ฎ thread ็ฅ้่ชๅทฑ่ฆๅไป้บผ
ๆก็จ็ doubly linked listใ็ด้ๅไธๅ task ็็จๆๅจๆผ๏ผ็ถๅพ queue ไธญๆฟๅบๅทฅไฝๅพ๏ผๅฏไปฅๅจ
typedef struct _task {
void (*func)(void *);
void *arg;
struct _task *next, *last;
} task_t;
- queue๏ผ่ฃก้ขๆพ็ฝฎ่็ญๅพ
่ฆ่ขซ่็็ task ๅ
ๅฉ็จ mutex ไพ็ขบไฟไธๆๆ race condition ็็ผ็ใ
่ condition variable ๅจ่ง็็ๆๅ็ผ็พๅชๆ่ขซๅๅงๅ่ๅทฒ๏ผๆฒๆๅ ถไปๅฐๆนไฝฟ็จๅฐใ
typedef struct {
task_t *head, *tail;
pthread_mutex_t mutex;
pthread_cond_t cond;
uint32_t size;
} tqueue_t;
- thread pool๏ผworker thread ็็ฎก็ๅก๏ผ่ฃก้ขๆพ็ฝฎไบๆๆ thread ๅ job queue
typedef struct {
pthread_t *threads;
uint32_t count;
tqueue_t *queue;
} tpool_t;
threadpool.c
- task_free๏ผ้ๆพๅฎไธ task ๅ ๅๆ้ ็ฝฎ็่จๆถ้ซ็ฉบ้
- tqueue_init๏ผๅๅงๅ queue
- tqueue_pop๏ผๅฐ queue ไธญ็ๅฎไธ task ๅๅบ๏ผๆดพ้ฃ็ตฆ thread
- tqueue_size๏ผๅๅณ queue ็ถๅ็ๅคงๅฐ
- tqueue_push๏ผๅฐๆฐ็ task ๅ ๅ ฅ queue ไธญ
- tqueue_free๏ผ้ๆพๅทฒ้ ็ฝฎ็ queue ็ฉบ้
- tpool_init๏ผๅๅงๅ threadpool
- tpool_free๏ผ้ๆพ threadpool ่จๆถ้ซ็ฉบ้
main.c
merge_list()ๅณๅ ฅๅ ฉๅlist๏ผ็ถๅพๅปๆฏ่ผ2ๅlistๆๅฐ็้ฃๅๅไธไธๆๅบ๏ผๆพๅฐๆฐ็list่ฃก้ข๏ผ็ดๅฐๅ ถไธญไธๅlist็ๆธ่ขซๆๅฎ๏ผ้ๆๆธๅญ็listๅฐฑๆฅๅฐๅทฒๆๅบๅฅฝ็listๅพ้ขใmerge_sort()ๅฆๆๅณๅ ฅ็list sizeๅฐๆผ2็ดๆฅๅๅณ่ฉฒlist๏ผๅฆๅๅฐฑๅฐไปๅพไธญ้ๅๆ2ๅlistไนๅพไธ้ฒmerge_list()ๅๆๅบไธฆๅๅณๆๅบๅฅฝ็listใmerge่ฃก้ขๅปๅคๆท็พๅจ็list็ธฝๅ ฑไธฒไบๅนพๅ๏ผๅฆๆๅฐๆผ่ผธๅ ฅ็่ณๆๆธ๏ผๅฐฑ็นผ็บๅๆๅบไธฒ่ตทไพ๏ผไธ็ถๅฐฑ่จญ็ฝฎ็ตๆญขtask็ถๅพๅฐๅบๆๅบๅฅฝ็listใ- ็ผ็พcut_func()่ทmerge_sort()้ฝๆฏๅๅๅๅฒlist๏ผ่cut_func่ฃก้ขๅคๆทcut_countๆฏๅฆๅฐๆผmax_cutๅ ถๅฏฆๆฒๆๅฟ ่ฆ๏ผๆไปฅๆcut_count็ญ็ธ้่ฎๆธๅชๆ๏ผไนๆmerge_sort()ๅชๆ๏ผ็ไธcut_func()ๅฐฑๅฅฝใ
task_runๅฐๆๅ่ฆๅ็ไบๅฏซๆtask๏ผๆพ้ฒjob queue่ฃก๏ผ็ฉบ้็threadๆๅพqueue่ฃกๆฟtaskๅบไพๅใ
Makefile
-M
็ๆๆไปถ้่ฏ็ไฟกๆฏใๅ
ๅซ็ฎๆจๆไปถๆไพ่ณด็ๆๆๆบไปฃ็ขผ
-MM
ๅ-Mไธๆจฃ๏ผไฝๆฏๅฎๅฐๅฟฝ็ฅ็ฑ#include<file>;้ ๆ็ไพ่ณด้ไฟใ
-MD
ๅ-M็ธๅ๏ผไฝๆฏ่ผธๅบๅฐๅฐๅ
ฅๅฐ.d็ๆไปถ่ฃก้ข
-MMD
ๅ-MM็ธๅ๏ผไฝๆฏ่ผธๅบๅฐๅฐๅ
ฅๅฐ.d็ๆไปถ่ฃก้ข
-MF
ๆๅฎ่ผธๅบๅฐๆๅๆชๆก,ๅฆๅ้ ่จญๆฏๅๅงๆชๆกๆชๅ.d
-rdynamic๏ผ
ๆๅฎ linker ๅฐๆๆ label ๅฏซๅ
ฅๅฐ dynamic symbol tableใ
็จๅผๅฏฆ้ฉ
mutrace
$ (for i in {1..8}; do echo $RANDOM; done) | mutrace ./sort 4 8
mutrace: 0.2 sucessfully initialized for process sort (pid 11679).
input unsorted data line-by-line
sorted results:
[300] [2022] [6225] [10989] [17303] [23622] [32251] [32551]
mutrace: Showing statistics for process sort (pid 11679).
mutrace: 3 mutexes used.
mutrace: Showing 3 most contended mutexes:
Mutex # Locked Changed Cont. tot.Time[ms] avg.Time[ms] max.Time[ms] Flags
0 145 40 30 0.023 0.000 0.001 Mx.--.
1 13 10 1 0.002 0.000 0.000 M-.--.
2 20 3 0 0.006 0.000 0.001 M-.--.
||||||
/|||||
Object: M = Mutex, W = RWLock /||||
State: x = dead, ! = inconsistent /|||
Use: R = used in realtime thread /||
Mutex Type: r = RECURSIVE, e = ERRRORCHECK, a = ADAPTIVE /|
Mutex Protocol: i = INHERIT, p = PROTECT /
RWLock Kind: r = PREFER_READER, w = PREFER_WRITER, W = PREFER_WRITER_NONREC
mutrace: Note that the flags column R is only valid in --track-rt mode!
mutrace: Total runtime is 0.883 ms.
mutrace: Results for SMP with 8 processors.
- ่กจๆ ผ:
Locked: how often the mutex was locked during the entire runtime.Changed: how often the owning thread of the mutex changed.
=>If the number is high this means the risk of contention is also high.Cont.: how often the lock was already taken when we tried to take it and we had to wait.
==> Mutex #1 is the most contended locktot.Time[ms]: for how long during the entire runtime the lock was lockedavg.Time[ms]: the average lock timemax.Time[ms]: the longest time the lock was held.Flags: what kind of mutex this is (recursive, normal or otherwise).
ๆ็
งไฝๆฅญ่ฆๆฑไธญๆๆ็ไฝฟ็จ addr2line ๆพๅฐๅฏฆ้ๅฐๆ็็จๅผ่กๆธ๏ผๅป้กฏ็คบ ??:0๏ผ ็ฎๅ้ๅจๅฐๆพๅๅ
$ addr2line -e sort 0x38
??:0
UNIX ๆไปค็ตๅ
ๅฐ phonebook-concurrent ่ฃก็ dictionary/words.txt ้้ UNIX ๆไปคๆไบ้ ๅบ๏ผไนๅพ้ๆฐๅฐๅๅฐๅฆไธๅๆชๆก input.txt๏ผ่ฎๆmerge sortๆฐ็่ณๆ่ผธๅ
ฅ
$ uniq words.txt | sort -R > input.txt
sort๏ผๅฐ่ผธๅ ฅ่ณๆๆๅบ-Rflag๏ผ้จๆฉๆๅบ
uniq๏ผๅฏไปฅ็จไพๅฐ้่ค็่กๅช้ค่ๅช้กฏ็คบไธๅ-cflag๏ผ้กฏ็คบ้่คๅญ่ฉ็ๅบ็พๆฌกๆธ
wc๏ผๅๅบๆชๆกๆๅคๅฐ่ก๏ผๅคๅฐๅฎๅญ๏ผๅคๅฐๅญๅ -l/w/m๏ผๅ ๅๅบ่กใๅฎๅญใๅญๅ
refactor
- list.[ch]
- list_add
- return ๆฐธ้ ๆฏ 0ใ
==> return list head ๅๅณๅปบๅฅฝๅพๆฐ็llistๆๆจ๏ผไฝฟ็จๆดๅฝๆงใ - ่จป่งฃๆๅฐ๏ผ่ฅๅณๅ ฅ็ val ๅทฒ็ถๅญๅจ๏ผๅไธๅ้ๆฐๅปบ็ซๆฐ็ node๏ผๅ็จๅผ็ขผไธญไธฆๆฒๆ็ธๅฐๆ็ๆชขๆฅๆฉๅถใ
โโโโโโโโllist_t list_add(llist_t *list, val_t val) โโโโโโโโ{ โโโโโโโโ //check if value already exist โโโโโโโโ node_t *cur = list->head; โโโโโโโโ while(cur) { โโโโโโโโ if(cur-> data == val) return list; โโโโโโโโ cur = cur->next; โโโโโโโโ } โโโโโโโโ //add to the head of the list โโโโโโโโ node_t *e = node_new(val, NULL); โโโโโโโโ e->next = list->head; โโโโโโโโ list->head = e; โโโโโโโโ list->size++; โโโโโโโโ return list; โโโโโโโโ} - return ๆฐธ้ ๆฏ 0ใ
- list_nth
- index out of range ๆขไปถๅคๆทๆ่ชค๏ผๆ nth ็ n ็ถไฝ array ็ index๏ผๆๅบ็บ 0 ~ list->size-1
==> ๅคๆท out of range ๆขไปถๆน็บif(idx > list->size-1)
==> ๅคๅ ไธๅๅคๆทif(idx == 0) return cur;
โโโโโโโโnode_t *list_nth(llist_t *list, uint32_t idx) โโโโโโโโ{ โโโโโโโโ if (idx > list->size-1) โโโโโโโโ return NULL; โโโโโโโโ node_t *head = list->head; โโโโโโโโ if(idx == 0) โโโโโโโโ return head; โโโโโโโโ while (idx--) โโโโโโโโ head = head->next; โโโโโโโโ return head; โโโโโโโโ} - index out of range ๆขไปถๅคๆทๆ่ชค๏ผๆ nth ็ n ็ถไฝ array ็ index๏ผๆๅบ็บ 0 ~ list->size-1
- list_add
- threadpool.[ch]
- tqueue_init๏ผ
- ๆช้ฒ่กๆๅณๅ
ฅ
tqueue_t็่จๆถ้ซ็ฉบ้้ ็ฝฎ
==> ็ดๆฅๅจๅฝๅผไธญ็จ malloc ้ ็ฝฎ็ฉบ้๏ผๆๅๅๅณ 0๏ผๅคฑๆๅๅณ -1
==> ๅพไพๆโโโโโโโโint tqueue_init(tqueue_t *the_queue) โโโโโโโโ{ โโโโโโโโ // if((the_queue = (tqueue_t *)malloc(sizeof(tqueue_t)) )== NULL) โโโโโโโโ // return -1; โโโโโโโโ the_queue->head = NULL; โโโโโโโโ the_queue->tail = NULL; โโโโโโโโ pthread_mutex_init(&(the_queue->mutex), NULL); โโโโโโโโ pthread_cond_init(&(the_queue->cond), NULL); โโโโโโโโ the_queue->size = 0; โโโโโโโโ return 0; โโโโโโโโ}if((the_queue = (tqueue_t *)malloc(sizeof(tqueue_t)) )== NULL) return -1;่จป่งฃๆ๏ผๆไธๆ็ผ็ segmentation fault(core dumped) - ๆช้ฒ่กๆๅณๅ
ฅ
- tqueue_pop๏ผ
- Queue ๆ่ฉฒๆฏ FIFO๏ผๅไพ็จๅผ็ขผ pop ๅบไพ็ไธๆฏ้ ญ่ๆฏๅฐพ
==> ๆtailๆนๆhead
โโโโโโโโtask_t *tqueue_pop(tqueue_t *the_queue) โโโโโโโโ{ โโโโโโโโ task_t *ret; โโโโโโโโ pthread_mutex_lock(&(the_queue->mutex)); โโโโโโโโ ret = the_queue->head; โโโโโโโโ if (ret) { โโโโโโโโ the_queue->head = ret->next; โโโโโโโโ if (the_queue->head) { โโโโโโโโ the_queue->head->last = NULL; โโโโโโโโ } else { โโโโโโโโ the_queue->tail = NULL; โโโโโโโโ } โโโโโโโโ the_queue->size--; โโโโโโโโ } โโโโโโโโ pthread_mutex_unlock(&(the_queue->mutex)); โโโโโโโโ return ret; โโโโโโโโ} - Queue ๆ่ฉฒๆฏ FIFO๏ผๅไพ็จๅผ็ขผ pop ๅบไพ็ไธๆฏ้ ญ่ๆฏๅฐพ
- tqueue_push๏ผ
- ๅๅณๅผๆฒๆ่็้ฏ่ชคๆ
ๆณ
==> ่ฅ task ไธๅญๅจๅๅๅณ -1๏ผๆๅ push ๅๅๅณ 0 - ๆฐๅ ๅ
ฅ็ task ๆ่ฉฒ็ฝฎๆผ queue ็ๅฐพ็ซฏ๏ผ้่ฃกๆพๅจ้ ญ
==> ๅฐheadๆน็บtail
โโโโโโโโint tqueue_push(tqueue_t *the_queue, task_t *task) โโโโโโโโ{ โโโโโโโโ if(task == NULL) return -1; โโโโโโโโ pthread_mutex_lock(&(the_queue->mutex)); โโโโโโโโ task->next = NULL; โโโโโโโโ task->last = the_queue->tail; โโโโโโโโ if (the_queue->tail) โโโโโโโโ the_queue->tail->next = task; โโโโโโโโ the_queue->tail = task; โโโโโโโโ if (the_queue->size++ == 0) โโโโโโโโ the_queue->head = task; โโโโโโโโ pthread_mutex_unlock(&(the_queue->mutex)); โโโโโโโโ return 0; โโโโโโโโ} - ๅๅณๅผๆฒๆ่็้ฏ่ชคๆ
ๆณ
- tqueue_free๏ผ
- ๅๅณๅผ็กๆ็พฉ
==> ๅคๆทthe_queueๆฏๅฆๅญๅจ็ๆ ๆณ๏ผๅฆๅๅๅณ -1๏ผไธฆไธไธๅจๅท่กไธ้ข้ๆพ่จๆถ้ซ้จๅ - condition variable ๆ่ขซๅๅง๏ผไฝๆฏๆฒๆ่ขซ destroy
==>thread_cond_detroy(&(the_queue->cond));
โโโโโโโโint tqueue_free(tqueue_t *the_queue) โโโโโโโโ{ โโโโโโโโ if(the_queue == NULL) return -1; โโโโโโโโ task_t *cur = the_queue->head; โโโโโโโโ while (cur) { โโโโโโโโ the_queue->head = the_queue->head->next; โโโโโโโโ free(cur); โโโโโโโโ cur = the_queue->head; โโโโโโโโ } โโโโโโโโ pthread_mutex_destroy(&(the_queue->mutex)); โโโโโโโโ pthread_cond_destroy(&(the_queue->cond)); โโโโโโโโ return 0; โโโโโโโโ} - ๅๅณๅผ็กๆ็พฉ
- tpool_free๏ผ
- ๆญคๅฝๅผๅๅณๅผ็กๆ็พฉ
==> ๅขๅthe_poolๆฏๅฆๅญๅจ็ๅคๆท๏ผไปฅๅpthread_join()ๆฏๅฆๆๅ็ๅคๆท
โโโโโโโโint tpool_free(tpool_t *the_pool) โโโโโโโโ{ โโโโโโโโ if(the_pool == NULL) return -1; โโโโโโโโ for (uint32_t i = 0; i < the_pool->count; ++i) โโโโโโโโ if(pthread_join(the_pool->threads[i], NULL) != 0) return -1; โโโโโโโโ free(the_pool->threads); โโโโโโโโ tqueue_free(the_pool->queue); โโโโโโโโ return 0; โโโโโโโโ} - ๆญคๅฝๅผๅๅณๅผ็กๆ็พฉ
- tqueue_init๏ผ
thread pool
concurrent-ll
lock-free
่จญ่จ่ชๅๆธฌ่ฉฆๅฏฆ้ฉ
ๅจ Makefile ไธญๅ ๅ ฅ่ฆๅ๏ผๅฉ็จ็ณป็ตฑๅ งๅปบ็ๅทฅๅ ทไพ้จๆฉ็ข็่ณๆไปฅๅๆฏ่ผ็ตๆใ
check: all
sort -R words.txt | ./sort $(THREAD_NUM) $(shell wc -l words.txt) > sorted.txt
diff -u words.txt sorted.txt && echo "Verified!" || echo "Failed!"
ๅ ถไปๆนๆณ
Scalability
- Horizontal scalar (scale out/in)๏ผๆฐๅข/ๆธๅฐ็ฏ้ปๅฐ็ณป็ตฑไธญใไพๅฆ๏ผๅๆฃๅผ้็ฎ๏ผ็ฑๅคๅฐ้ป่ ฆๅ ฑๅๅท่กไปปๅ
- Vertical scalar (scale up/down)๏ผๅขๅผท/ๆธๅผฑๅฎๅฐ้ป่ ฆ็ๆง่ฝใไพๅฆ๏ผไฝฟ็จๅคๆ ธๅฟ
