MMKV的原理-如何实现跨进程(2)

桃扇骨 2023-07-08 02:52 1阅读 0赞

接上一篇:[https://blog.csdn.net/lin20044140410/article/details/104450727][https_blog.csdn.net_lin20044140410_article_details_104450727]

在mmkv的使用中,肯定是有多线程,多进程的同步问题,有同步问题就肯定会用到锁,所以先从mmkv中锁的使用说起,mmkv处理线程的同步使用了mutex互斥锁, 比如在从集合中获取mmkv的c++层的对象时,就加了锁,因为可能会有多个线程同时操作的情况;

处理进程间的同步时使用了flock文件锁,比如在处理写指针的同步,内存重整时.

以下锁的使用都是在native层.

1,先看下互斥锁是怎么用的，这个demo是展示线程锁的使用，进一步引出mmkv对锁的使用时怎么封装的。

queue<int> m_queue;
    
    void *dequeue(void *args) {
        if (!m_queue.empty()) {
            printf("dequeue data :%d \n", m_queue.front());
            __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,dequeue data :%d \n", m_queue.front());
            m_queue.pop();
        } else {
            __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,no data");
        }
        return 0;
    }
    
    extern "C"
    JNIEXPORT void JNICALL
    Java_com_test_mmkvdemo_MainActivity_nativeThreadTest(JNIEnv *env, jobject instance) {
        for (int i = 0; i < 5; ++i) {
            m_queue.push(i);
        }
    
        pthread_t tid[10];
        for (int i = 0; i < 10; ++i) {
            pthread_create(&tid[i], 0, dequeue, &m_queue);
        }
    
       __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,m_queue," );
    }

在这个demo中，先往一个队列放了5个元素，然后起了10个线程来取队列中的元素，

2020-02-24 21:49:29.213 20930-20949/? D/mmap: nativeThreadTest,dequeue data :0 ,tid= 3529775472
    2020-02-24 21:49:29.214 20930-20948/? D/mmap: nativeThreadTest,dequeue data :1 ,tid= 3530815856
    2020-02-24 21:49:29.214 20930-20950/? D/mmap: nativeThreadTest,dequeue data :1 ,tid= 3528735088
    2020-02-24 21:49:29.214 20930-20951/? D/mmap: nativeThreadTest,dequeue data :3 ,tid= 3527694704
    2020-02-24 21:49:29.215 20930-20953/? D/mmap: nativeThreadTest,dequeue data :4 ,tid= 3525613936
    2020-02-24 21:49:29.215 20930-20954/? D/mmap: nativeThreadTest,dequeue data :4 ,tid= 3524573552
    2020-02-24 21:49:29.215 20930-20952/? D/mmap: nativeThreadTest,dequeue data :-374526740 ,tid= 3526654320
    2020-02-24 21:49:29.216 20930-20956/? D/mmap: nativeThreadTest,dequeue data :-374526724 ,tid= 3522492784
    2020-02-24 21:49:29.216 20930-20930/? D/mmap: nativeThreadTest,m_queue,
    2020-02-24 21:49:29.216 20930-20955/? D/mmap: nativeThreadTest,dequeue data :-374526708 ,tid= 3523533168
    2020-02-24 21:49:29.216 20930-20957/? D/mmap: nativeThreadTest,dequeue data :-374526692 ,tid= 3521452400

从输出看，线程取出的数据出现了重复，并且还可能取出未知的数据，这是因为多线程去操作共享数据没有加锁，导致的脏读。

给操作pop的方法加上互斥锁，可以解决脏读的问题，这里用的是pthead\_mutex\_t互斥量，这个锁默认是非递归锁，也就是不可重入，对于不可重入锁，如果一个线程内多次获取同一个这样的锁，就会产生死锁，所以在pthread\_mutex\_t锁初始化时，需要设置属性pthread\_mutexattr\_t，把它设置为递归锁。 pthread\_mutexattr\_settype(&m\_attr, PTHREAD\_MUTEX\_RECURSIVE);

queue<int> m_queue;
    pthread_mutex_t m_mutex;
    pthread_mutexattr_t m_attr;
    
    void *dequeue(void *args) {
        pthread_mutex_lock(&m_mutex);
        pthread_t tid = pthread_self();
        if (!m_queue.empty()) {
            printf("dequeue data :%d \n", m_queue.front());
            __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,dequeue data :%d ,tid= %u\n",
                    m_queue.front(), (unsigned int)tid);
            m_queue.pop();
        } else {
            __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,no data,tid= %u\n",(unsigned int)tid);
        }
        pthread_mutex_unlock(&m_mutex);
        return 0;
    }
    
    extern "C"
    JNIEXPORT void JNICALL
    Java_com_test_mmkvdemo_MainActivity_nativeThreadTest(JNIEnv *env, jobject instance) {
        pthread_mutexattr_init(&m_attr);
        pthread_mutexattr_settype(&m_attr, PTHREAD_MUTEX_RECURSIVE);
        pthread_mutex_init(&m_mutex, &m_attr);
        pthread_mutexattr_destroy(&m_attr);
        for (int i = 0; i < 5; ++i) {
            m_queue.push(i);
        }
    
        pthread_t tid[10];
        for (int i = 0; i < 10; ++i) {
            pthread_create(&tid[i], 0, dequeue, &m_queue);
        }
        pthread_mutex_destroy(&m_mutex);
    
       __android_log_print(ANDROID_LOG_DEBUG,"mmap","nativeThreadTest,m_queue," );
    }

锁使用过程中，调用pthread\_mutex\_lock(&m\_mutex); 上锁，一定要记得调用pthread\_mutex\_unlock(&m\_mutex);解锁，如果不小心忘记了解锁，那肯定就悲剧了，而且如果要加个锁，需要写那么多代码，肯定也会不爽，所以就有了对锁的封装，实现锁的自动管理。

先用c++类的构造方法和析构方法进行封装，在构造函数中进行锁及属性的初始化，在析构函数中销毁锁对象。

class ThreadLock {
    private:
        pthread_mutex_t m_lock;
    
    public:
        ThreadLock() {
            pthread_mutexattr_t attr;
            pthread_mutexattr_init(&attr);
            pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE);
            pthread_mutex_init(&m_lock, &attr);
            pthread_mutexattr_destroy(&attr);
        }
    
        ~ThreadLock() {
            pthread_mutex_destroy(&m_lock);
        }
    
        void lock() {
            auto result = pthread_mutex_lock(&m_lock);
            if (result != 0) {
                //failed.
            }
        }
    
        bool try_lock() {
            auto result = pthread_mutex_lock(&m_lock);
            return  (result == 0);
        }
    
        void unlock() {
            auto result = pthread_mutex_unlock(&m_lock);
            if (0 != result){
                //failed.
            }
        }
    };

这样封装了以后,尽管简单了很多,但是依然可能会因为忘记unlock,导致悲剧,为了确保使用过程中不至于lock之后,忘记unlock,就需要更进一步的封装,就是在创建一个class,让他持有一个ThreadLock的对象,在这个类的构造方法中上锁,在其析构方法中解锁.

template <typename T>
    class ScopedLock {
        T *m_lock;
    
        // just forbid it for possibly misuse
        ScopedLock(const ScopedLock<T> &other) = delete;
    
        ScopedLock &operator=(const ScopedLock<T> &other) = delete;
    
    public:
        ScopedLock(T *oLock) : m_lock(oLock) {
            assert(m_lock);
            lock();
        }
    
        ~ScopedLock() {
            unlock();
            m_lock = nullptr;
        }
    
        void lock() {
            if (m_lock) {
                m_lock->lock();
            }
        }
    
        bool try_lock() {
            if (m_lock) {
                return m_lock->try_lock();
            }
            return false;
        }
    
        void unlock() {
            if (m_lock) {
                m_lock->unlock();
            }
        }
    };

如此封装后,使用方法就是在代码块中仅需要创建锁对象即可实现上锁,在代码块结束,会随着锁对象生命周期结束,调用其析构函数释放锁.

{
       ThreadLocak * thread_lock = new ThreadLock();
       ScopedLock scoped_lock(thread_lock);
    }

ScopedLock类是一个模板类,实际在mmkv中,文件锁中读锁和写锁,及线程锁都是通过这个模板类使用的.

上面的封装对锁的使用已经非常方便了,然后mmkv又让锁的使用更加的便捷,那就是又定义了宏函数简化了锁的使用.

#define SCOPEDLOCK(lock) _SCOPEDLOCK(lock, __COUNTER__)
    #define _SCOPEDLOCK(lock, counter) __SCOPEDLOCK(lock, counter)
    #define __SCOPEDLOCK(lock, counter) ScopedLock<decltype(lock)> __scopedLock##counter(&lock)

第一个宏函数,把ThreadLock对象传进去,会调用第二个宏函数,

第二个宏函数,除了传入的ThreadLock参数外,还加了一个编译器中宏定义,\_\_COUNTER\_\_在编译时会被替换成具体的值0,1,2,...

它就相当于一个计数器,记录了这个宏函数在一个编译单元中被调用了多少次.

第三个宏函数,就是创建了一个具体的锁对象,

ScopedLock<decltype(lock)> \_\_scopedLock\#\#counter(&lock)

decltype(lock)获取lock的具体类型,加上计数值,就得到了一个ScopedLock类型的锁对象,

第一次调用 \_\_scopedLock0,

第二次调用 \_\_scopedLock1,

第三次调用 \_\_scopedLock2 ,....

2,接着看下文件锁.

尽管pthread\_mutex可以用作进程锁,但是因为andorid版本的pthread\_mutex不够健壮,如果加了pthread\_mutex锁的进程被kill了,系统不会进行清理工作,这个锁会一直存在下去,其他等待锁的进程会被饿死.

[https://github.com/Tencent/MMKV/wiki/android\_ipc][https_github.com_Tencent_MMKV_wiki_android_ipc]

多进程的实现中用到了文件锁,多进程同时操作一个文件时,可能存在A进程去写这个文件,而B进程去读这个文件,就会导致B进程读取到的是脏数据.为了保证B进程可以读取到最新的数据,并且保证数据的完整性,使用文件锁来完成多进程操作文件的同步.

文件锁的使用:

#include <sys/file.h>
    //通过open方法打开一个文件
    string m_path;
    int m_fd = open(m_path.c_str(), O_RDWR | O_CREAT, S_IRWXU);
    //通过文件句柄对文件上锁
    int flock(m_fd, operation);

其中的参数operation是上锁的类型,flock支持两中锁类型LOCK\_SH,LOCK\_EX

LOCK\_SH, 共享锁,多个进程可以同时使用,可以作为读锁.

LOCK\_EX,排他锁,同时只允许一个进程使用,可以作为写锁.

LOCK\_UN, 解锁

LOCK\_BN, 非阻塞请求, 与读写锁配合使用.

使用flock对一个文件上读锁,或者写锁,都是会阻塞的,比如A进程对持有一个文件的写锁,B进程想要对这个文件上写锁,就会阻塞住,如果不想被阻塞,可以配合LOCK\_BN属性使用,即LOCK\_BN | LOCK\_EX.

flock有几个特点:

1)flock支持对一个文件多次上锁,并且因为是状态锁,没有计数器,不管加了多少次锁,都只需要解锁一次.所以,mmkv中对flock封装时,加了计数器,就是保证上了几次锁,就要执行几次解锁.

2)锁升级,降级,当一个进程对一个文件加了读锁后,如果再次执行flock操作,传入的operation是LOCK\_EX,那么这个进程对文件的读锁就升级为了写锁,这就是锁升级,反之,就是锁降级,但是文件锁降级是无法进行的,因为他不支持递归,导致一降级就没锁了.

为了解决上面的问题,mmkv对文件锁进行了封装,增加了读写锁计数器,支持递归.

结合源码看mmkv是怎么多的.

InterProcessLock.h

class FileLock {
        int m_fd;
        size_t m_sharedLockCount;
        size_t m_exclusiveLockCount;
    
        bool doLock(LockType lockType, bool wait);
    }

增加了读写锁计数器,通过调用doLock对文件上锁.locktype决定了上锁的类型,第二个参数是否阻塞.所以重点就是doLock是怎么实现的.

上锁的处理逻辑:

1),//先判断上的是什么锁  
//上读锁,计数器先加1,如果加1后,读锁计数器大于1了,说明已经上过读锁了,为了实现递归,这种情况就不在执行上锁,只是增加了计数器,如果这个文件的写锁计数器大于0,说明已经上过写锁了,当前还在用写锁,所以不  
//执行上读锁的操作,会导致锁降级,所以直接返回  
//如果是第一次上读锁,后面会执行加锁操作

2),/上写锁,计数器先加1,判断如果已经上过写锁,直接return.  
//如果读锁计数器大于0,这时会先尝试上写锁,如果失败了,说明有别的进程在使用这个文件的读锁,  
//这种情况就要先把自己的读锁释放掉,再去以阻塞的形式上写锁,这么做的原因是为了避免死锁.  
//假设,A进程没有解自己的读锁,直接上写锁在这里阻塞着,如果别的进程B进程刚好释放了读锁,那么当前进程//A进程上写锁  
//就成功了,但是,如果别的进程B集成也要上写锁,因为A进程的读锁没有释放,会导致B进程上写锁的操作一直  
//阻塞着,这就是A,B进程都在等对方的读锁释放,而导致死锁.

bool FileLock::doLock(LockType lockType, bool wait) {
        if (!isFileLockValid()) {
            return false;
        }
        bool unLockFirstIfNeeded = false;
       //先判断上的是什么锁
    //上读锁,计数器先加1,如果加1后,读锁计数器大于1了,说明已经上过读锁了,为
    了实现递归,这种情况就不在执行上锁,只是增加了计数器,如果这个文件的写锁计数器大于0,
    说明已经上过写锁了,当前还在用写锁,所以不
    //执行上读锁的操作,会导致锁降级,所以直接返回
    //如果是第一次上读锁,后面会执行加锁操作
        if (lockType == SharedLockType) {
            m_sharedLockCount++;
            // don't want shared-lock to break any existing locks
            if (m_sharedLockCount > 1 || m_exclusiveLockCount > 0) {
                return true;
            }
        } else {
    //上写锁,计数器先加1,判断如果已经上过写锁,直接return.
    //如果读锁计数器大于0,这时会先尝试上写锁,如果失败了,说明有别的进程在使用这个文件的读锁,
    //这种情况就要先把自己的读锁释放掉,再去以阻塞的形式上写锁,这么做的原因是为了避免死锁.
    //假设,A进程没有解自己的读锁,直接上写锁在这里阻塞着,如果别的进程B进程刚好释放了读锁,那么当前进程//A进程上写锁
    //就成功了,但是,如果别的进程B集成也要上写锁,因为A进程的读锁没有释放,会导致B进程上写锁的操作一直
    //阻塞着,这就是A,B进程都在等对方的读锁释放,二导致死锁.
            m_exclusiveLockCount++;
            // don't want exclusive-lock to break existing exclusive-locks
            if (m_exclusiveLockCount > 1) {
                return true;
            }
            // prevent deadlock
            if (m_sharedLockCount > 0) {
                unLockFirstIfNeeded = true;
            }
        }
        return platformLock(lockType, wait, unLockFirstIfNeeded);
    }

再看解锁的处理逻辑:

1),//如果是解读锁,读锁计数器为0,直接返回,  
//如果读锁计数器 减1后,还大于了,为了实现递归锁,那么上几次读锁,就要解几次读锁,所以直接  
//返回,  
//如果这里有写锁,也直接返回,因为如果执行LOCK\_UN,就把写锁也解了.

2),//如果解写锁,计数器为0,直接返回,  
//如果计数器减1后大于0,直接返回  
//否则,如果写锁计数器为0了,但是有读锁,也不能直接LOCK\_UN,而是要做锁降级,

bool FileLock::unlock(LockType lockType) {
        if (!isFileLockValid()) {
            return false;
        }
        bool unlockToSharedLock = false;
    //如果是解读锁,读锁计数器为0,直接返回,
    //如果读锁计数器 减1后,还大于了,为了实现递归锁,那么上几次读锁,就要解几次读锁,所以直接
    //返回,
    //如果这里有写锁,也直接返回,因为如果执行LOCK_UN,就把写锁也解了.
        if (lockType == SharedLockType) {
            if (m_sharedLockCount == 0) {
                return false;
            }
            m_sharedLockCount--;
            // don't want shared-lock to break any existing locks
            if (m_sharedLockCount > 0 || m_exclusiveLockCount > 0) {
                return true;
            }
        } else {
    //如果解写锁,计数器为0,直接返回,
    //如果计数器减1后大于0,直接返回
    //否则,如果写锁计数器为0了,但是有读锁,也不能直接LOCK_UN,而是要做锁降级,
            if (m_exclusiveLockCount == 0) {
                return false;
            }
            m_exclusiveLockCount--;
            if (m_exclusiveLockCount > 0) {
                return true;
            }
            // restore shared-lock when all exclusive-locks are done
            if (m_sharedLockCount > 0) {
                unlockToSharedLock = true;
            }
        }
        return platformUnLock(unlockToSharedLock);
    }

3,这样文件锁就封装好了,利用这个文件锁就可以实现同一时间只有一个进程对file进行操作了,但是A进程修改了文件后,B进程怎么知道这个修改呢?

针对上面的问题,mmkv是做的呢? 实际跟我们想象的并不一样,mmkv并没有去对保存key-value数据的那个default文件枷锁,而是锁了个.crc校验文件.这个校验文件就是来解决上面的问题的.

struct MMKVMetaInfo {
        uint32_t m_crcDigest = 0;
        uint32_t m_version = 1;
        uint32_t m_sequence = 0; // full write-back count
        unsigned char m_vector[AES_KEY_LEN] = {0};
    }

这个mmkv.default.crc文件中有一个校验码,就是mmkv.default文件的MD5值,通过他可以判断文件是否合法.

还有一个序列号,当序列化文件进行了去重,扩容操作,这个序列号就会增加,

具体看源码是怎么做的?

1)当mmkv初始化时,会读取crc文件,记录其中的校验码,序列号,,

2)当要读取数据时,会通过checkLoadData来做校验.

void MMKV::checkLoadData() {
       //读取校验文件,对比序列号,如果序列号不一致,说明发生了内存重整,
    //这时,重新读取整个文件,
        MMKVMetaInfo metaInfo;
        metaInfo.read(m_metaFile.getMemory());
        if (m_metaInfo.m_sequence != metaInfo.m_sequence) {
            SCOPEDLOCK(m_sharedProcessLock);
    
            clearMemoryState();
            loadFromFile();
            notifyContentChanged();
        }else if (m_metaInfo.m_crcDigest != metaInfo.m_crcDigest) {
        //校验码不匹配,说明数据文件被修改了,可能时数据增加,也可能是去重操作,还可能时文件扩容了
    //如果文件大小不一致了,可能发生了扩容,这时重新加载整个文件,
    //文件大小一致,说明只是新增了k-v,也即是增量更新,这时只要加载新增加的数据.
            size_t fileSize = 0;
            if (m_isAshmem) {
                fileSize = m_size;
            } else {
                struct stat st = {0};
                if (fstat(m_fd, &st) != -1) {
                    fileSize = (size_t) st.st_size;
                }
            }
    
            if (m_size != fileSize) {
                MMKVInfo("file size has changed [%s] from %zu to %zu",             
               m_mmapID.c_str(), m_size,
                         fileSize);
                clearMemoryState();
                loadFromFile();
            } else {
                partialLoadFromFile();
            }
        }
    }

所以,通过校验文件,在读取数据时,来做校验,就实现了多个进程的数据同步.

[https_blog.csdn.net_lin20044140410_article_details_104450727]: https://blog.csdn.net/lin20044140410/article/details/104450727
[https_github.com_Tencent_MMKV_wiki_android_ipc]: https://github.com/Tencent/MMKV/wiki/android_ipc