当前位置：首页 > article >正文

线程-4-线程库与线程封装

article 2025/1/5 18:08:01

线程ID&进程地址空间布局

不存在叫tcb的，pthread就是tcb

库中，所有tcb按照数组存放

struct pthread 就是tcb

pthread_create返回线程ID（tid）

线程ID是其pthread地址

每个新线程的栈是create时申请的

新线程的栈是动态申请出来的

线程的局部存储（TLS）__thread

thread local storage

__thread int share_val

每个线程各有一个此变量

其他进程拿到此地址也可以访问，修改（但一般不会这样做）

线程局部存储只能用来修饰内置类型，不能修饰自定义类型（struct，class）

pthread库属于glibc

malloc底层用mmap实现

使用该空间时才通过缺页中断建立叶表映射虚拟/物理内存

pthread_create 通过pthread库来初始化tcb等各参数，之后通过软中断系统调用建立LWP，所以叫pthread做用户级库

封装pthread

不能令routine为类内普通函数，可为类内static函数（this指针占参数了）

封装code

tips

类的成员变量就是全局变量（类内空间的全局变量）

code

mythread.cc

#include <iostream>
#include <string>
#include <cstdio>
#include <cstring>
#include <unistd.h>
#include <thread>

// 线程的局部存储
// __thread只能修饰内置类型
__thread int shared_value = 100;

std::string toHex(pthread_t tid)
{
    // 4. 进程内的函数，线程共享
    char buffer[64];
    snprintf(buffer, sizeof(buffer), "0x%lx", tid);
    return buffer;
}

void *start(void *args)
{
    std::string name = static_cast<const char *>(args);
    sleep(1);
    while (true)
    {
        printf("I am a new thread, name: %s, shared_value: %d,  &shared_value: %p\n", name.c_str(), shared_value, &shared_value);
        sleep(1);
    }

    return nullptr;
}

int main()
{
    pthread_attr_t attr;

    
    pthread_t tid;
    pthread_create(&tid, nullptr, start, (void *)"thread-1");
    std::cout << "I am a new thread, name: main, " << toHex(pthread_self())
              << ", NEW thread id: " << toHex(tid) << std::endl;

    while (true)
    {
        printf("main thread, shared_value: %d,  &shared_value: %p\n", shared_value, &shared_value);
        shared_value += 10;
        sleep(1);
    }
    pthread_join(tid, nullptr);
    return 0;
}

// int *addr = nullptr;

// void *start1(void *args)
// {
//     std::string name = static_cast<const char *>(args);
//     int a = 100;
//     while (true)
//     {
//         std::cout << name << " local val a: " << a << std::endl;
//         sleep(1);
//     }
// }

// void *start(void *args)
// {
//     // 可以的！
//     // pid_t id = fork();
//     // if(id == 0)
//     // {
//     //     // ....
//     // }
//     // pthread_detach(pthread_self());
//     // std::string name = static_cast<const char *>(args);
//     // while (true)
//     // {
//     //     // if(addr != nullptr)
//     //     //std::cout << name << " mod val a: " << (*addr)++ << std::endl;
//     //     std::cout << "I am a new thread" << std::endl;
//     //     sleep(1);
//     //     // break;
//     // }
//     // return 0; // 9. 新线程return表示该线程退出
//     // exit(1); // 任何地方调用exit，表示进程退出!
//     // pthread_exit((void*)10);
// }

// int main()
// {
//     // // pthread_t tid1, tid2;
//     // pthread_t tid;
//     // pthread_create(&tid, nullptr, start, (void *)"thread-1");
//     // // pthread_detach(tid);
//     // sleep(5);

//     // // int n = pthread_cancel(tid);
//     // // std::cout << "取消线程: " << tid << std::endl;

//     // // sleep(5);

//     // void *ret = nullptr;
//     // int n = pthread_join(tid, &ret); //PTHREAD_CANCELED;
//     // std::cout << "new thread exit code: " << (long long int)ret << " ,n: " << n << std::endl;

//     // // pthread_create(&tid1, nullptr, start2, (void *)"thread-1");
//     // // pthread_join(tid1, nullptr);
//     // return 0; // 9. 主线程return，表示进程结束！
// }

// class ThreadData
// {
// public:
//     ThreadData()
//     {}
//     void Init(const std::string &name, int a, int b)
//     {
//         _name = name;
//         _a = a;
//         _b = b;
//     }
//     void Excute()
//     {
//         _result = _a + _b;
//     }
//     int Result(){ return _result; }
//     std::string Name(){ return _name; }
//     void SetId(pthread_t tid) { _tid = tid;}
//     pthread_t Id() { return _tid; }
//     int A(){return _a;}
//     int B(){return _b;}
//     ~ThreadData()
//     {}
// private:
//     std::string _name;
//     int _a;
//     int _b;
//     int _result;
//     pthread_t _tid;
// };

// // class outData
// // {

// // };

// // 5. 全局变量在线程内部是共享的
// int gval = 100;
// // std::queue<int> q;
// // char buffer[4096];

// std::string toHex(pthread_t tid)
// {
//     // 4. 进程内的函数，线程共享
//     char buffer[64];
//     snprintf(buffer, sizeof(buffer), "0x%lx", tid);
//     return buffer;
// }

// void *routine2(void *args)
// {
//     std::string name = static_cast<const char *>(args);
//     while (true)
//     {
//         // 3. 不加保护的情况下，显示器文件就是共享资源！
//         std::cout << "我是新线程,我的名字是: " << name << ", my tid is : " << toHex(pthread_self()) << "， 全局变量(只是检测)：" << gval << std::endl;
//         sleep(1);
//         // 6. 线程一旦出现异常，可能会导致其他线程全部崩溃
//         // 6.1 信号
//         int *p = nullptr;
//         *p = 100;
//     }
//     return 0;
// }

// // 被重入了！！
// void *routine(void *args)
// {
//     //std::string name = static_cast<const char *>(args);
//     ThreadData *td = static_cast<ThreadData *>(args);
//     while (true)
//     {
//         // 3. 不加保护的情况下，显示器文件就是共享资源！
//         std::cout << "我是新线程,我的名字是: " << td->Name() << ", my tid is : " << toHex(pthread_self()) << "， 全局变量(会修改)：" << gval << std::endl;
//         gval++;
//         td->Excute();
//         sleep(1);
//         break;
//     }
//     // 8.返回值问题: 返回参数，可以是变量，数字，对象！
//     // 8.1: 理论上，堆空间也是共享的！谁拿着堆空间的入口地址，谁就能访问该堆区！
//     // int *p = new int(10);
//     // return (void*)p; // 线程退出方式1：线程入口函数return，表示线程退出

//     return td;
// }

// #define NUM 10

// int main()
// {

//     // ThreadData td[NUM];
//     // // 准备我们要加工处理的数据
//     // for(int i = 0; i < NUM; i++)
//     // {
//     //     char id[64];
//     //     snprintf(id, sizeof(id), "thread-%d", i);
//     //     td[i].Init(id, i*10, i*20);
//     // }

//     // // 创建多线程
//     // for(int i = 0; i < NUM; i++)
//     // {
//     //     pthread_t id;
//     //     pthread_create(&id, nullptr, routine, &td[i]);
//     //     td[i].SetId(id);
//     // }

//     // // 等待多个线程
//     // for(int i =0; i< NUM;i++)
//     // {
//     //     pthread_join(td[i].Id(), nullptr);
//     // }

//     // // 汇总处理结果
//     // for(int i =0;i < NUM; i++)
//     // {
//     //     printf("td[%d]: %d+%d=%d[%ld]\n", i, td[i].A(), td[i].B(), td[i].Result(), td[i].Id());
//     // }
//     // 1. 新线程和main线程谁先运行，不确定
//     // 2. 线程创建出来，要对进程的时间片进行瓜分
//     // 8. 传参问题: 传递参数，可以是变量，数字，对象！
//     // pthread_t tid1;
//     // ThreadData *td = new ThreadData("thread-1", 10, 20);
//     // pthread_create(&tid1, nullptr, routine1, td);

//     // // 7. 线程创建之后，也是要被等待和回收的！
//     // // 7.1 理由：a. 类似僵尸进程的问题 b. 为了知道新线程的执行结果
//     // ThreadData *rtd = nullptr;
//     // int n = pthread_join(tid1, (void**)&rtd); // 我们可以保证，执行完毕，任务一定处理完了，结果变量一定已经被写入了!
//     // if(n != 0)
//     // {
//     //     std::cerr << "join error: " << n << ", " << strerror(n) << std::endl;
//     //     return 1;
//     // }
//     // std::cout << "join success!, ret: " << rtd->Result() << std::endl;
//     // delete td;

//     // pthread_t tid2;
//     // pthread_create(&tid2, nullptr, routine2, (void *)"thread-2");

//     // pthread_t tid3;
//     // pthread_create(&tid3, nullptr, routine, (void *)"thread-3");

//     // pthread_t tid4;
//     // pthread_create(&tid4, nullptr, routine, (void *)"thread-4");

//     // std::cout << "new thread id: " << tid << std::endl;
//     // printf("new thread id: 0x%lx\n", tid1);
//     // printf("new thread id: 0x%lx\n", tid2);
//     // printf("new thread id: 0x%lx\n", tid3);
//     // printf("new thread id: 0x%lx\n", tid4);

//     // while (true)
//     // {
//     //     std::cout << "我是main线程..." << std::endl;
//     //     sleep(1);
//     // }
// }

// // void *routine(void *args)
// // {
// //     std::string name = static_cast<const char *>(args);
// //     while (true)
// //     {
// //         std::cout << "我是新线程,我的名字是: " << name << std::endl;
// //         sleep(1);
// //     }
// //     return 0;
// // }

// // int main()
// // {
// //     std::thread t([](){
// //         while (true)
// //         {
// //             std::cout << "我是新线程,我的名字是 : new thread "  << std::endl;
// //             sleep(1);
// //         }
// //     });

// //     while (true)
// //     {
// //         std::cout << "我是main线程..."  << std::endl;
// //         sleep(1);
// //     }

// //     // pthread_t tid;
// //     // int n = pthread_create(&tid, nullptr, routine, (void *)"thread-1");
// //     // if (n != 0)
// //     // {
// //     //     std::cout << "create thread error: " << strerror(n) << std::endl;
// //     //     return 1;
// //     // }
// //     // while (true)
// //     // {
// //     //     std::cout << "我是main线程..."  << std::endl;
// //     //     sleep(1);
// //     // }
// // }

main.cc

#include "Thread.hpp"
#include <unordered_map>
#include <memory>

#define NUM 10

// using thread_ptr_t = std::shared_ptr<ThreadModule::Thread>;

class threadData
{
public:
    int max;
    int start;
};

void Count(threadData td)
{
    for (int i = td.start; i < td.max; i++)
    {
        std::cout << "i == " << i << std::endl;
        sleep(1);
    }
}

int main()
{
    threadData td;
    td.max = 60;
    td.start = 50;
    ThreadModule::Thread<threadData> t(Count, td);
    // ThreadModule::Thread<int> t(Count, 10);

    t.Start();

    t.Join();

    // 先描述，在组织！
    // std::unordered_map<std::string, thread_ptr_t> threads;
    // // 如果我要创建多线程呢？？？
    // for (int i = 0; i < NUM; i++)
    // {
    //     thread_ptr_t t = std::make_shared<ThreadModule::Thread>([](){
    //         while(true)
    //         {
    //             std::cout << "hello world" << std::endl;
    //             sleep(1);
    //         }
    //     });
    //     threads[t->Name()] = t;
    // }

    // for(auto &thread:threads)
    // {
    //     thread.second->Start();
    // }

    // for(auto &thread:threads)
    // {
    //     thread.second->Join();
    // }

    // ThreadModule::Thread t([](){
    //     while(true)
    //     {
    //         std::cout << "hello world" << std::endl;
    //         sleep(1);
    //     }
    // });

    // t.Start();
    // std::cout << t.Name() << "is running" << std::endl;
    // sleep(5);

    // t.Stop();
    // std::cout << "Stop thread : " << t.Name()<< std::endl;
    // sleep(1);
    // t.Join();
    // std::cout << "Join thread : " << t.Name()<< std::endl;

    return 0;
}

thread.hpp

#ifndef _THREAD_HPP__
#define _THREAD_HPP__

#include <iostream>
#include <string>
#include <pthread.h>
#include <functional>
#include <sys/types.h>
#include <unistd.h>

// v2
namespace ThreadModule
{
    static int number = 1;
    enum class TSTATUS
    {
        NEW,
        RUNNING,
        STOP
    };

    template <typename T>
    class Thread
    {
        using func_t = std::function<void(T)>;
    private:
        // 成员方法！
        static void *Routine(void *args)
        {
            Thread<T> *t = static_cast<Thread<T> *>(args);
            t->_status = TSTATUS::RUNNING;
            t->_func(t->_data);
            return nullptr;
        }
        void EnableDetach() { _joinable = false; }

    public:
        Thread(func_t func, T data) : _func(func), _data(data), _status(TSTATUS::NEW), _joinable(true)
        {
            _name = "Thread-" + std::to_string(number++);
            _pid = getpid();
        }
        bool Start()
        {
            if (_status != TSTATUS::RUNNING)
            {
                int n = ::pthread_create(&_tid, nullptr, Routine, this); // TODO
                if (n != 0)
                    return false;
                return true;
            }
            return false;
        }
        bool Stop()
        {
            if (_status == TSTATUS::RUNNING)
            {
                int n = ::pthread_cancel(_tid);
                if (n != 0)
                    return false;
                _status = TSTATUS::STOP;
                return true;
            }
            return false;
        }
        bool Join()
        {
            if (_joinable)
            {
                int n = ::pthread_join(_tid, nullptr);
                if (n != 0)
                    return false;
                _status = TSTATUS::STOP;
                return true;
            }
            return false;
        }
        void Detach()
        {
            EnableDetach();
            pthread_detach(_tid);
        }
        bool IsJoinable() { return _joinable; }
        std::string Name() { return _name; }
        ~Thread()
        {
        }

    private:
        std::string _name;
        pthread_t _tid;
        pid_t _pid;
        bool _joinable; // 是否是分离的，默认不是
        func_t _func;
        TSTATUS _status;
        T _data;
    };
}

// v1
// namespace ThreadModule
// {
//     using func_t = std::function<void()>;
//     static int number = 1;
//     enum class TSTATUS
//     {
//         NEW,
//         RUNNING,
//         STOP
//     };

//     class Thread
//     {
//     private:
//         // 成员方法！
//         static void *Routine(void *args)
//         {
//             Thread *t = static_cast<Thread *>(args);
//             t->_status = TSTATUS::RUNNING;
//             t->_func();
//             return nullptr;
//         }
//         void EnableDetach() { _joinable = false; }

//     public:
//         Thread(func_t func) : _func(func), _status(TSTATUS::NEW), _joinable(true)
//         {
//             _name = "Thread-" + std::to_string(number++);
//             _pid = getpid();
//         }
//         bool Start()
//         {
//             if (_status != TSTATUS::RUNNING)
//             {
//                 int n = ::pthread_create(&_tid, nullptr, Routine, this); // TODO
//                 if (n != 0)
//                     return false;
//                 return true;
//             }
//             return false;
//         }
//         bool Stop()
//         {
//             if (_status == TSTATUS::RUNNING)
//             {
//                 int n = ::pthread_cancel(_tid);
//                 if (n != 0)
//                     return false;
//                 _status = TSTATUS::STOP;
//                 return true;
//             }
//             return false;
//         }
//         bool Join()
//         {
//             if (_joinable)
//             {
//                 int n = ::pthread_join(_tid, nullptr);
//                 if (n != 0)
//                     return false;
//                 _status = TSTATUS::STOP;
//                 return true;
//             }
//             return false;
//         }
//         void Detach()
//         {
//             EnableDetach();
//             pthread_detach(_tid);
//         }
//         bool IsJoinable() { return _joinable; }
//         std::string Name() {return _name;}
//         ~Thread()
//         {
//         }

//     private:
//         std::string _name;
//         pthread_t _tid;
//         pid_t _pid;
//         bool _joinable; // 是否是分离的，默认不是
//         func_t _func;
//         TSTATUS _status;
//     };
// }

#endif

ticket.cc

#include <iostream>
#include <vector>
#include "Thread.hpp"
#include <mutex>

using namespace ThreadModule;

#define NUM 4

// 1. 锁本身是全局的，那么锁也是共享资源！谁保证锁的安全？？
// pthread_mutex:加锁和解锁被设计成为原子的了 --- TODO??
// 2. 如何看待锁呢？二元信号量就是锁！
//    2.1 加锁本质就是对资源展开预订！
//    2.2 整体使用资源！！
// 3. 如果申请锁的时候，锁被别人已经拿走了，怎么办？其他线程要进行阻塞等待
// 4. 线程在访问临界区代码的时候，可以不可以切换？？可以切换！！
//    4.1 我被切走的时候，别人能进来吗？？不能！我是抱着锁，被切换的！！不就是串行吗！效率低的原因！原子性！
// 5. 不遵守这个约定？？bug！
// pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
int ticketnum = 10000; // 共享资源,临界资源
std::mutex gmtx;

class ThreadData
{
public:
    std::string name;
    pthread_mutex_t *lock_ptr;
};

void Ticket(ThreadData &td)
{
    while (true)
    {
        gmtx.lock();
        // pthread_mutex_lock(td.lock_ptr); // 加锁
        if (ticketnum > 0)
        {
            usleep(1000);

            // 1. 抢票
            printf("get a new ticket, who get it: %s, id: %d\n", td.name.c_str(), ticketnum--);
            gmtx.unlock();
            // pthread_mutex_unlock(td.lock_ptr); // ??
            // 2. 入库模拟
            usleep(50);
        }
        else
        {
            // pthread_mutex_unlock(td.lock_ptr); // ??
            gmtx.unlock();
            break;
        }
    }
}

int main()
{
    pthread_mutex_t lock;
    pthread_mutex_init(&lock, nullptr);

    // 1. 构建线程对象
    std::vector<Thread<ThreadData>> threads;

    for (int i = 0; i < NUM; i++)
    {
        ThreadData *td = new ThreadData();
        td->lock_ptr = &lock;
        threads.emplace_back(Ticket, *td);
        td->name = threads.back().Name();
    }

    // 2. 启动线程
    for (auto &thread : threads)
    {
        thread.Start();
    }

    // 3. 等待线程
    for (auto &thread : threads)
    {
        thread.Join();
    }

    pthread_mutex_destroy(&lock);

    return 0;
}

mutex.hpp

#pragma once
#include <iostream>
#include <pthread.h>

namespace LockModule
{
    class Mutex
    {
    public:
        Mutex(const Mutex&) = delete;
        const Mutex& operator = (const Mutex&) = delete;
        Mutex()
        {
            int n = ::pthread_mutex_init(&_lock, nullptr);
            (void)n;
        }
        ~Mutex()
        {
            int n = ::pthread_mutex_destroy(&_lock);
            (void)n;
        }
        void Lock()
        {
            int n = ::pthread_mutex_lock(&_lock);
            (void)n;
        }
        void Unlock()
        {
            int n = ::pthread_mutex_unlock(&_lock);
            (void)n;
        }

    private:
        pthread_mutex_t _lock;
    };

    class LockGuard
    {
    public:
        LockGuard(Mutex &mtx):_mtx(mtx)
        {
            _mtx.Lock();
        }
        ~LockGuard()
        {
            _mtx.Unlock();
        }
    private:
        Mutex &_mtx;
    };
}

main.cc

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sched.h>

int ticket = 0;
pthread_mutex_t mutex;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;

void *route(void *arg)
{
    char *id = (char *)arg;
    while (1)
    {
        pthread_mutex_lock(&mutex);
        if (ticket > 0)
        {
            usleep(1000);
            printf("%s sells ticket:%d\n", id, ticket);
            ticket--;
            pthread_mutex_unlock(&mutex);
        }
        else
        {
            printf("%s wait on cond!\n", id);
            pthread_cond_wait(&cond, &mutex); //醒来的时候，会重新申请锁！！
            printf("%s 被叫醒了\n", id);
        }
        pthread_mutex_unlock(&mutex);
    }
    return nullptr;
}

int main(void)
{
    pthread_t t1, t2, t3, t4;

    pthread_mutex_init(&mutex, NULL);

    pthread_create(&t1, NULL, route, (void *)"thread 1");
    pthread_create(&t2, NULL, route, (void *)"thread 2");
    pthread_create(&t3, NULL, route, (void *)"thread 3");
    pthread_create(&t4, NULL, route, (void *)"thread 4");

    int cnt = 10;

    while(true)
    {
        sleep(5);
        ticket += cnt;
        printf("主线程放票喽, ticket: %d\n", ticket);
        pthread_cond_signal(&cond);
    }


    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    pthread_mutex_destroy(&mutex);
}