C/C++ WebServer 3 Linux 多线程开发
Meqt
  2022-12-17 18:40:08 2022-12-17 18:40      2.7k Words  13 Mins  

线程概述

  • 与进程 (process)类似。线程(thread）是允许应用程序并发执行多个任务的一种机制。一个进程可以包含多个线程。同一个程序中的所有线程均会独立执行相同程序，且共享同一份全局内存区城，其中包括初始化数据段、未初始化数据段，以及堆内存段。（传统意义上的 UNIX 进程只是多线程程序的一个特例，该进程只包含一个线程）
  • 进程是 CPU 分配资源的最小单位
  • 线程是操作系统调度执行的最小单位。
  • 线程是轻量级的进程 (IWP：Light weight Process)，在Iinux 环境下线程的本质仍是进程。
  • 查看指定进程的 IWP 号：ps -Lf $(pid)
  • 进程间的信息难以共享。由于除去只读代码段外，父子进程并末共享内存，因此必须采用一些进程间通信方式，在进程问进行信息交换。
  • 调用fork(）来创建进程的代价相对较高，即便利用马时复制技术，仍热需要复制诸如内存页表和文件描述符表之类的多种进程属性，这意味着 foIk(）调用在时间上的开销依然不菲。
  • 线程之间能够方便、快速地共享信息。只需将数据复制到共享（全局或堆）变量中即可。
  • 创建线程比创建进程通常要快 10倍甚至更多。线程间是共享虚拟地址空间的，无需采用写时复制来复制内存，也无需复制页表。
  • 共享资源
    • 进程ID，父进程ID
    • 进程组ID，会话ID
    • 用户ID，用户组ID
    • 文件描述符表
    • 信号处置
    • 文件系统的相关信息umask，path
    • 虚拟空间地址（除栈，.text）
  • 非共享资源
    • 线程ID
    • 信号掩码
    • 线程特有数据
    • error变量
    • 实时调度策略和优先级
    • 栈、本地变量和调用链接信息
      NPTL：Native POSIX Thread Library
      getconf GNU_LIBPTHREAD_VERSION

创建线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#include <pthread.h>
int pthread_create(pthread_t *thread, const pthread_attr_t *attr,
                          void *(*start_routine) (void *), void *arg);

// pthread_create.c
#include <stdio.h>
#include <pthread.h>
#include <string.h>

void * callback( void * arg) {
    printf("child thread ...");
}

int main() {
    pthread_t tid;
    int ret = pthread_create(&tid, NULL, callback, NULL);
    if (ret==-1) {
        char* errstr = strerror(ret);
        printf("error: %s", errstr);
    }
    
    for (int i=0; i<5; i++) {
        printf("%d\n", i);
    }
    return 0;
}
1
2
gcc pthread_create.c -o create -pthread
# or gcc pthread_create.c -o create -l pthread

终止线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
// pthread_exit.c
#include <stdio.h>
#include <pthread.h>
#include <string.h>

void * callback(void* args) {
    printf("child thread: tid:%ld\n", pthread_self());
}

int main() {
    pthread_t tid;
    int ret = pthread_create(&tid, NULL, callback, NULL);
    if (ret==-1) {
        char *errstr = strerror(ret);
        printf("%s\n", errstr);
    }

    printf("father thread, tid: %ld, child thread id:%ld\n", pthread_self(), tid);
    for (int i=0; i<5; i++) {
        printf("%d\n", i);
    }

    // 让主线程退出，当主线程退出时，不会影响其他正常运行的线程，
    pthread_exit(NULL);
    return 0;
}

连接已终止的线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#include <stdio.h>
#include <pthread.h>
#include <string.h>
#include <unistd.h>
int value = 10;

void * callback(void* args) {
    printf("child thread: tid:%ld\n", pthread_self());
    //sleep(3);
//    int value = 10;
// don't pass local variable
    pthread_exit(&value);
}

int main() {
    pthread_t tid;
    int ret = pthread_create(&tid, NULL, callback, NULL);
    if (ret==-1) {
        char *errstr = strerror(ret);
        printf("%s\n", errstr);
    }

    printf("father thread, tid: %ld, child thread id:%ld\n", pthread_self(), tid);
    for (int i=0; i<5; i++) {
        printf("%d\n", i);
    }

    int *thread_ret;
    ret = pthread_join(tid, (void**)&thread_ret);
    if (ret!=0) {
        char *errstr = strerror(ret);
        printf("%s\n", errstr);
    }

    printf("child thread exit value:%d\n", *thread_ret);
    return 0;
}

线程的分离

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// pthread_detach
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

void* callback(void* args) {
    printf("child thread id:%ld\n", pthread_self());
}
int main() {
    pthread_t tid;
    int ret = pthread_create(&tid, NULL, callback, NULL);
    if (ret!=0) {
        printf("%s\n", strerror(ret));
    }

    printf("tid: %ld, main thread id:%ld\n", tid, pthread_self());
    // 设置子线程分离，子线程分离后，子线程结束时对应的资源就不需要主线程释放,不能再链接
    pthread_detach(tid);
    pthread_exit(NULL);
    return 0;
}

线程取消

只有当子线程执行到取消点时，线程才会终止（系统调用，用户区到内核区）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

void* callback(void* args) {
    printf("child thread id:%ld\n", pthread_self());
    for (int i=0; i<5; i++) {
        printf("child thread id:%ld\n", pthread_self());
    }
}

int main() {
    pthread_t tid;
    int ret = pthread_create(&tid, NULL, callback, NULL);
    if (ret!=0) {
        printf("%s\n", strerror(ret));
    }

    pthread_cancel(tid);
    for (int i=0; i<5; i++) {
        printf("%d\n", i);
    }
    printf("tid: %ld, main thread id:%ld\n", tid, pthread_self());
    // 设置子线程分离，子线程分离后，子线程结束时对应的资源就不需要主线程释放,不能再链接
    pthread_detach(tid);
    pthread_exit(NULL);
    return 0;
}

线程属性

#include <pthread.h>
int
pthread_attr_init(pthread_attr_t *attr);
int
pthread_attr_destroy(pthread_attr_t *attr);
int
pthread_attr_setstack(pthread_attr_t *attr, void *stackaddr,
size_t stacksize);
int
pthread_attr_getstack(const pthread_attr_t * restrict attr,
void ** restrict stackaddr, size_t * restrict stacksize);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
// pthread_attr.c
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <string.h>

void* callback(void* args) {
    printf("child thread id:%ld\n", pthread_self());
}
int main() {
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    size_t size;
    pthread_t tid;
    int ret = pthread_create(&tid, &attr, callback, NULL);
    if (ret!=0) {
        printf("%s\n", strerror(ret));
    }

    pthread_attr_getstacksize(&attr, &size);
    printf("stacksize %ld\n", size);

    printf("tid: %ld, main thread id:%ld\n", tid, pthread_self());
    // 设置子线程分离，子线程分离后，子线程结束时对应的资源就不需要主线程释放,不能再链接
    pthread_detach(tid);
    pthread_attr_destroy(&attr); 
    pthread_exit(NULL);
    return 0;
}

线程同步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
// sellTickets.c
#include <stdio.h>
#include <pthread.h>
#include <unistd.h>

int tickets = 100;
void *sell(void * args) {
    while(tickets>0) {
        usleep(5000);
        printf("tid: %ld, Selling ticket :%d\n", pthread_self(), tickets);
        tickets--;
    }
    return NULL;
}
int main() {
    pthread_t tid1, tid2, tid3;
    pthread_create(&tid1, NULL, sell, NULL);
    pthread_create(&tid2, NULL, sell, NULL);
    pthread_create(&tid3, NULL, sell, NULL);
    
    pthread_join(tid1, NULL);
    pthread_join(tid2, NULL);
    pthread_join(tid3, NULL);

    pthread_detach(tid1);
    pthread_detach(tid2);
    pthread_detach(tid3);


    pthread_exit(NULL);
}

临界区
Atomic

互斥锁

Mutual exclusion
Locked, unlocked

  • 一旦线程锁定互斥量，随即成为该互斥量的所有者，只有所有者才能给互斥量解锁。一般情况下，对每一共享资源（可能由多个相关变量组成）会使用不同的互斥量，每一线程在访问同一资源时将采用如下协议：
    • 针对共享资源锁定互斥量
    • 访问共享资源
    • 对互斥量解锁
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      #include <stdio.h>
      #include <pthread.h>
      #include <unistd.h>

      int tickets = 1000;
      pthread_mutex_t mutex;
      void *sell(void * args) {
          // lock
          while(tickets>0) {
              
              pthread_mutex_lock(&mutex);
              if (tickets>0) {
                  usleep(5000);
                  printf("tid: %ld, Selling ticket :%d\n", pthread_self(), tickets);
                  tickets--;
              }else{
                  pthread_mutex_unlock(&mutex);
                  break;
              }
              pthread_mutex_unlock(&mutex);
          }
          return NULL;
      }
      int main() {
          pthread_mutex_init(&mutex, NULL);
          pthread_t tid1, tid2, tid3;
          pthread_create(&tid1, NULL, sell, NULL);
          pthread_create(&tid2, NULL, sell, NULL);
          pthread_create(&tid3, NULL, sell, NULL);
          
          pthread_join(tid1, NULL);
          pthread_join(tid2, NULL);
          pthread_join(tid3, NULL);

          pthread_detach(tid1);
          pthread_detach(tid2);
          pthread_detach(tid3);

          pthread_mutex_destroy(&mutex);
          pthread_exit(NULL);
      }

死锁

  • 有时，一个线程需要同时访问两个或更多不同的共享资源，而每个资源又都由不同的互斥量管理。当超过一个线程加锁同一组互斥量时，就有可能发生死锁。
  • 两个或两个以上的进程在执行过程中，因年李共享资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁。
  • 死锁的几种场景：
    • 忘记释放锁
    • 重复加锁
    • 多线程多锁，抢占锁资源
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      // deadlock.c
      #include <stdio.h>
      #include <pthread.h>
      #include <unistd.h>

      int tickets = 1000;
      pthread_mutex_t mutex;
      void *sell(void * args) {
          // lock
          while(tickets>0) {
              
              pthread_mutex_lock(&mutex);
              if (tickets>0) {
                  usleep(5000);
                  printf("tid: %ld, Selling ticket :%d\n", pthread_self(), tickets);
                  tickets--;
              }else{
                  pthread_mutex_unlock(&mutex);
                  break;
              }
              // pthread_mutex_unlock(&mutex);
          }
          return NULL;
      }
      int main() {
          pthread_mutex_init(&mutex, NULL);
          pthread_t tid1, tid2, tid3;
          pthread_create(&tid1, NULL, sell, NULL);
          pthread_create(&tid2, NULL, sell, NULL);
          pthread_create(&tid3, NULL, sell, NULL);
          
          pthread_join(tid1, NULL);
          pthread_join(tid2, NULL);
          pthread_join(tid3, NULL);

          pthread_detach(tid1);
          pthread_detach(tid2);
          pthread_detach(tid3);

          pthread_mutex_destroy(&mutex);
          pthread_exit(NULL);
      }
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      52
      53
      54
      55
      56
      // deadlock1.c
      #include <stdio.h>
      #include <pthread.h>
      #include <unistd.h>

      // 创建2个互斥量
      pthread_mutex_t mutex1, mutex2;

      void * workA(void * arg) {

          pthread_mutex_lock(&mutex1);
          sleep(1);
          pthread_mutex_lock(&mutex2);

          printf("workA....\n");

          pthread_mutex_unlock(&mutex2);
          pthread_mutex_unlock(&mutex1);
          return NULL;
      }


      void * workB(void * arg) {
          pthread_mutex_lock(&mutex2);
          sleep(1);
          pthread_mutex_lock(&mutex1);

          printf("workB....\n");

          pthread_mutex_unlock(&mutex1);
          pthread_mutex_unlock(&mutex2);

          return NULL;
      }

      int main() {

          // 初始化互斥量
          pthread_mutex_init(&mutex1, NULL);
          pthread_mutex_init(&mutex2, NULL);

          // 创建2个子线程
          pthread_t tid1, tid2;
          pthread_create(&tid1, NULL, workA, NULL);
          pthread_create(&tid2, NULL, workB, NULL);

          // 回收子线程资源
          pthread_join(tid1, NULL);
          pthread_join(tid2, NULL);

          // 释放互斥量资源
          pthread_mutex_destroy(&mutex1);
          pthread_mutex_destroy(&mutex2);

          return 0;
      }

读写锁

  • 在对数据的读写操作中，更多的是读操作：写操作较少，例如对数据库数据的读写应用。为了满足当前能够允许多个读出，但只允许一个写入的需求，线程提供了读写锁来实现。
  • 读写锁的特点：
    • 如果有其它线程读数据，则允许其它线程执行读操作，但不允许写操作。
    • 如果有其它线程写数据，则其它线程都不允许读、写操作。
    • 写是独占的，写的优先级高。
      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      20
      21
      22
      23
      24
      25
      26
      27
      28
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      // rwlock.c
      #include <stdio.h>
      #include <pthread.h>
      #include <unistd.h>

      int num=0;
      pthread_rwlock_t rwlock;
      void * writeNum(void* args) {
          while(1) {
              pthread_rwlock_wrlock(&rwlock);
              num++;
              printf("write, tid:%ld, num:%d\n", pthread_self(), num);
              pthread_rwlock_unlock(&rwlock);
               usleep(100);
         }

          return NULL;
      }

      void * readNum(void* args) {
          while(1) {
              pthread_rwlock_rdlock(&rwlock);
              printf("write, tid:%ld, num:%d\n", pthread_self(), num);
              pthread_rwlock_unlock(&rwlock);
              usleep(100);
          }

          return NULL;
      }
      int main() {
          pthread_rwlock_init(&rwlock, NULL); 
          pthread_t wtids[3], rtids[5];
          for (int i=0; i<3; i++) {
              pthread_create(&wtids[i], NULL, writeNum, NULL);
          }
          
          for (int i=0; i<5; i++) {
              pthread_create(&rtids[i], NULL, readNum, NULL);
          }

          for (int i=0; i<3; i++) {
              pthread_detach(wtids[i]);
          }
          for (int i=0; i<5; i++) {
              pthread_detach(rtids[i]);
          }
          
          pthread_exit(NULL);
          pthread_rwlock_destroy(&rwlock);
          return 0;
      }

生产者和消费者模型

条件变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
// pthread_cond.c
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
typedef struct Node{
    int num;
    struct Node* next;
}Node;

Node* head = NULL;
pthread_mutex_t mutex;
pthread_cond_t cond;
void *producor(void* args) {
    while(1) {
        pthread_mutex_lock(&mutex);
         Node* newNode = (Node* )malloc(sizeof(Node));
         newNode->next = head;
         head = newNode;
         newNode->num = rand() % 1000;
         printf("producor tid:%ld, add node:%d\n", pthread_self(), newNode->num); 
         pthread_cond_signal(&cond);
         pthread_mutex_unlock(&mutex);
         usleep(1000);
    }
    return NULL;
}

void *customer(void* args) {
    while(1) {
        pthread_mutex_lock(&mutex);
        if(!head) {
            // when calling wait blocked lock, after unblock, unlock.
            pthread_cond_wait(&cond, &mutex); 
            pthread_mutex_unlock(&mutex);
            continue;
        }
        Node* node = head;
        head = head->next;
        printf("del node, num:%d, tid:%ld\n", node->num, pthread_self());
        free(node);
        pthread_mutex_unlock(&mutex);
    }
    return NULL;
}

int main() {

    pthread_mutex_init(&mutex, NULL);
    pthread_cond_init(&cond, NULL);
    pthread_t ptids[5], ctids[5]; 
    for (int i=0; i<5; i++) {
        pthread_create(&ptids[i], NULL, producor, NULL);
        pthread_detach(ptids[i]);
        pthread_create(&ctids[i], NULL, customer, NULL);
        pthread_detach(ctids[i]);
    }

    while(1) {
        sleep(10);
    }

    pthread_exit(NULL);
    pthread_cond_destroy(&cond);
    pthread_mutex_destroy(&mutex);
    return 0;
}

信号量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
// semaphore.c
#include <stdio.h>
#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
#include <semaphore.h>
typedef struct Node{
    int num;
    struct Node* next;
}Node;

Node* head = NULL;
sem_t psem, csem ;
pthread_mutex_t mutex;

void *producor(void* args) {
    while(1) {
        sem_wait(&psem);
        pthread_mutex_lock(&mutex);
         Node* newNode = (Node* )malloc(sizeof(Node));
         newNode->next = head;
         head = newNode;
         newNode->num = rand() % 1000;
         printf("producor tid:%ld, add node:%d\n", pthread_self(), newNode->num); 
         pthread_mutex_unlock(&mutex);
         sem_post(&csem);
         usleep(1000);
    }
    return NULL;
}

void *customer(void* args) {
    while(1) {
        sem_wait(&csem);
        pthread_mutex_lock(&mutex);
        Node* node = head;
        head = head->next;
        printf("del node, num:%d, tid:%ld\n", node->num, pthread_self());
        free(node);
        pthread_mutex_unlock(&mutex);
        sem_post(&psem);
    }
    return NULL;
}

int main() {

    pthread_mutex_init(&mutex, NULL);
    sem_init(&psem, 0, 8);
    sem_init(&csem, 0, 0);
    pthread_t ptids[5], ctids[5]; 
    for (int i=0; i<5; i++) {
        pthread_create(&ptids[i], NULL, producor, NULL);
        pthread_detach(ptids[i]);
        pthread_create(&ctids[i], NULL, customer, NULL);
        pthread_detach(ctids[i]);
    }

    while(1) {
        sleep(10);
    }

    pthread_exit(NULL);
    pthread_mutex_destroy(&mutex);
    return 0;
}
  • Post title：C/C++ WebServer 3 Linux 多线程开发
  • Post author：Meqt
  • Create time：2022-12-17 18:40:08
  • Post link：https://meqtmac.github.io/2022/12/17/WebServerNote/webserver-note3/
  • Copyright Notice：All articles in this blog are licensed under BY-NC-SA unless stating additionally.
Contents
  1. 1. 线程概述
  2. 2. 创建线程
  3. 3. 终止线程
  4. 4. 连接已终止的线程
  5. 5. 线程的分离
  6. 6. 线程取消
  7. 7. 线程属性
  8. 8. 线程同步
  9. 9. 互斥锁
  10. 10. 死锁
  11. 11. 读写锁
  12. 12. 生产者和消费者模型
  13. 13. 条件变量
  14. 14. 信号量