线程同步1-锁 @synchroinzed

作者: 答案不止一个 | 来源:发表于2020-11-10 23:21 被阅读0次

os_unfair_lock #import <os/lock.h>

os_unfair_lock 是用来替代 OSSPinLock 的一个互斥锁。OSSpinLock 是采用自选忙等待的方式，而 os_unfair_lock 则会刮起等待，可以通过 Debug->Debug Worlflow->Always Show Disassembly 去断点一个死锁的 OSSpinLock 是否是循环的执行。

@synchronized

@synchronized 的作用是创建一个递归锁，保证此时没有其它线程对self对象进行修改，保证代码的安全性。也就是包装这段代码是原子性的，安全的。这个是objective-c的一个锁定令牌，防止self对象在同一时间内被其他线程访问，起到保护线程安全的作用。

将.m 编译成.cpp
使用view debug overflow
可以看到，@synchronized 的编译之后的处理会在在代码块前后添加 objc_sync_enter 和 objc_sync_exit方法

添加 symbol breakpoint 找到符号 libobjc.A.dylib`objc_sync_enter
继续探究一下其源码

objc_sync_enter

主要时获取obj对一个的SuncData，然后再对SyncData 里的 recursive_mutex_t 递归锁进行加锁

typedef struct alignas(CacheLineSize) SyncData {
    struct SyncData* nextData;
    DisguisedPtr<objc_object> object;
    int32_t threadCount;  // number of THREADS using this block
    // 一个自旋锁
    recursive_mutex_t mutex;
} SyncData;

int objc_sync_enter(id obj)
{
    int result = OBJC_SYNC_SUCCESS;

    if (obj) {
        // 获取objc 对应的SyncData 。其中会有一个自旋锁用来对 代码块进行加锁
        SyncData* data = id2data(obj, ACQUIRE);
        ASSERT(data);
        // 使用递归锁进行加锁 
        data->mutex.lock();
    } else {
        // @synchronized(nil) does nothing
        if (DebugNilSync) {
            _objc_inform("NIL SYNC DEBUG: @synchronized(nil); set a breakpoint on objc_sync_nil to debug");
        }
        objc_sync_nil();
    }

    return result;
}

id2data 的函数时根据obj 对象，以及 ACQUIRE / RELEASE 的标志，获取对应的SyncData，以及对SyncData 的lockCout 和threadCount 进行处理。

lockCount 时lock 的次数。最少为0 ， threadCount 则是 0 / 1 。

objc_sync_exit

也是获取 obj对应的 SyncData（并对lockcount 和threadCount内进行操作，如果threadCount 为0 就不会返回了）。然后对 SyncData 中的递归锁进行解锁。

// End synchronizing on 'obj'. 
// Returns OBJC_SYNC_SUCCESS or OBJC_SYNC_NOT_OWNING_THREAD_ERROR
int objc_sync_exit(id obj)
{
    int result = OBJC_SYNC_SUCCESS;
    
    if (obj) {
        SyncData* data = id2data(obj, RELEASE); 
        if (!data) {
            result = OBJC_SYNC_NOT_OWNING_THREAD_ERROR;
        } else {
            bool okay = data->mutex.tryUnlock();
            if (!okay) {
                result = OBJC_SYNC_NOT_OWNING_THREAD_ERROR;
            }
        }
    } else {
        // @synchronized(nil) does nothing
    }
    

    return result;
}

id2data

id2data 是用来获取对象对应的SyncData。

先从tls 中获取缓存，查找obj对应的数据
从cache 中获取缓存，遍历查找obj对应的数据，查找，
如果找不到，就使用 os_unfair_lock 互斥锁，加锁，保证线程安全的去创建新的 SyncData，然后设置

// sDataLists 是一个静态的全局变量，StripedMap有一个下标算法，根据obj的地址 去获取下标，然后找到对应的数据
#define LOCK_FOR_OBJ(obj) sDataLists[obj].lock
#define LIST_FOR_OBJ(obj) sDataLists[obj].data
static StripedMap<SyncList> sDataLists;

static SyncData* id2data(id object, enum usage why)
{
    // 去 sDataList 查询 objc 对应的锁
    spinlock_t *lockp = &LOCK_FOR_OBJ(object); 
    //// 去 sDataList 查询对应的data 数据，可以看到是一个 SyncData 的二维数组的指针
    SyncData **listp = &LIST_FOR_OBJ(object);
    SyncData* result = NULL;

    // Check per-thread single-entry fast cache for matching object tls 当前的线程
    //。
    bool fastCacheOccupied = NO;
    // 获取当前线程的局部静态变量
    SyncData *data = (SyncData *)tls_get_direct(SYNC_DATA_DIRECT_KEY);
    
    if (data) { // 加入数据缓存存在
        fastCacheOccupied = YES;
        // 如果对应的 对象是当前对象
        if (data->object == object) {
            // Found a match in fast cache.
            uintptr_t lockCount;

            result = data;
            lockCount = (uintptr_t)tls_get_direct(SYNC_COUNT_DIRECT_KEY);
            if (result->threadCount <= 0  ||  lockCount <= 0) {
                _objc_fatal("id2data fastcache is buggy");
            }

            switch(why) {
            case ACQUIRE: {
                lockCount++;
                tls_set_direct(SYNC_COUNT_DIRECT_KEY, (void*)lockCount);
                break;
            }
            case RELEASE:
                lockCount--;
                tls_set_direct(SYNC_COUNT_DIRECT_KEY, (void*)lockCount);
                if (lockCount == 0) {
                    // remove from fast cache
                    tls_set_direct(SYNC_DATA_DIRECT_KEY, NULL);
                    // atomic because may collide with concurrent ACQUIRE
                    OSAtomicDecrement32Barrier(&result->threadCount);
                }
                break;
            case CHECK:
                // do nothing
                break;
            }
            // 找到就直接返回，
            return result;
        }
    }

    // Check per-thread cache of already-owned locks for matching object
    SyncCache *cache = fetch_cache(NO);
    if (cache) {
        unsigned int i;
        for (i = 0; i < cache->used; i++) {
            SyncCacheItem *item = &cache->list[i];
            if (item->data->object != object) continue;

            // Found a match.
            result = item->data;
            if (result->threadCount <= 0  ||  item->lockCount <= 0) {
                _objc_fatal("id2data cache is buggy");
            }
                
            switch(why) {
            case ACQUIRE:
                item->lockCount++;
                break;
            case RELEASE:
                item->lockCount--;
                if (item->lockCount == 0) {
                    // remove from per-thread cache
                    cache->list[i] = cache->list[--cache->used];
                    // atomic because may collide with concurrent ACQUIRE
                     // 对threadCount 进行原子操作，-1 只有当lockcount 位0 就全部都解锁是，晴空threadCout  threadCount 最大为1
                    OSAtomicDecrement32Barrier(&result->threadCount);
                }
                break;
            case CHECK:
                // do nothing
                break;
            }

            return result;
        }
    }

    // Thread cache didn't find anything.
    // Walk in-use list looking for matching object
    // Spinlock prevents multiple threads from creating multiple 
    // locks for the same new object.
    // We could keep the nodes in some hash table if we find that there are
    // more than 20 or so distinct locks active, but we don't do that now.
    
    // 加锁 使用的 os_unfair_lock 一种自旋锁 （等待的自旋锁） 
    lockp->lock();

    { 
        // 在listp中查找 listp 是多个obj 对应SyncData 的列表数据
        SyncData* p;
        SyncData* firstUnused = NULL;
        for (p = *listp; p != NULL; p = p->nextData) { 
            if ( p->object == object ) { 
                //找到listp中存储的数据，就返回跳转done
                result = p;
                // atomic because may collide with concurrent RELEASE
                // 对threadCount 进行原子操作，增加1（其实最大就会是1）
                OSAtomicIncrement32Barrier(&result->threadCount);
                goto done;
            }
            // 用来找到最后一个 数据被删除的空间，用来存储
            if ( (firstUnused == NULL) && (p->threadCount == 0) )
                firstUnused = p; 
        }
        
        // no SyncData currently associated with object
        // 找不到还去删除 就会result 位nil， 然后done，会在外面assert 是报错
        if ( (why == RELEASE) || (why == CHECK) ) 
            goto done;
    
        // an unused one was found, use it
        if ( firstUnused != NULL ) {
            result = firstUnused;
            result->object = (objc_object *)object;
            result->threadCount = 1;
            goto done;
        }
    }

    // Allocate a new SyncData and add to list.
    // XXX allocating memory with a global lock held is bad practice,
    // might be worth releasing the lock, allocating, and searching again.
    // But since we never free these guys we won't be stuck in allocation very often.
    // 创建内存，然后设置数据，并加到listp 中 即 sDataLists 中
    posix_memalign((void **)&result, alignof(SyncData), sizeof(SyncData));
    result->object = (objc_object *)object;
    result->threadCount = 1;
    new (&result->mutex) recursive_mutex_t(fork_unsafe_lock);
    result->nextData = *listp;
    *listp = result;
    
 done:
    lockp->unlock();
    if (result) {
        // Only new ACQUIRE should get here.
        // All RELEASE and CHECK and recursive ACQUIRE are 
        // handled by the per-thread caches above.
        if (why == RELEASE) {
            // Probably some thread is incorrectly exiting 
            // while the object is held by another thread.
            return nil;
        }
        if (why != ACQUIRE) _objc_fatal("id2data is buggy");
        if (result->object != object) _objc_fatal("id2data is buggy");

#if SUPPORT_DIRECT_THREAD_KEYS
        if (!fastCacheOccupied) {
            // Save in fast thread cache
            tls_set_direct(SYNC_DATA_DIRECT_KEY, result);
            tls_set_direct(SYNC_COUNT_DIRECT_KEY, (void*)1);
        } else 
#endif
        {
            // Save in thread cache
            if (!cache) cache = fetch_cache(YES);
            cache->list[cache->used].data = result;
            cache->list[cache->used].lockCount = 1;
            cache->used++;
        }
    }

    return result;
}

有限去查当前线程的tls中obj对用的SyncData，找到直接返回（对lockcount/threadCount的操作）
tls 找不到，再去cache（所有线程的缓存）中查找，找到返回也对对lockcount/threadCount 进行操作。当unlock 且 lockcout == 0 时，会从 tls 中移除数据。 listp 不会移除数据，但是会设置threadCount 为0
lock是对 lockcount 进行加1 ，threadcount 不操作， unlock时，lockcount -1 ，如果 lockcount == 0 就将threadcount -1 （原子操作） threadCount 最大是1，最小时0，
如果都能查不到，就去加锁，
然后在 listp 中遍历查询（全局变量 sDataLists 的StripedMap 中，Stringp 返回的是多个obj对应的数据（hash冲突的原因，或者说StrpeMap的地址计算逻辑）。同时找是否有空余的内存
如果找到数据，就跳转 8 done （解锁，然后保存数据到 tls ，和全局的cache中）
如果没找到，且有空余的内存，就设置数据放到空余的内存地址。然后跳转 8 done
还是没有就创建内存，然后设置数据，跳转 8 done
done 解锁，然后保存数据到tls。 cache，返回查找到的result。（没枷锁却去解锁，会返回nil，不会报错OBJC_SYNC_NOT_OWNING_THREAD_ERROR）

fetch_cache

根据 _objc_pthread_key key 去线程局部静态变量查询数据，根据create 是否在没有事，自动创建一个

static SyncCache *fetch_cache(bool create)
{
    _objc_pthread_data *data;
    // 根据 _objc_pthread_key 这个key 获取，如果没有且 create = true 就创建一个数据存入
    data = _objc_fetch_pthread_data(create);
    if (!data) return NULL;

    if (!data->syncCache) {
        if (!create) {
            return NULL;
        } else {
            int count = 4;
            data->syncCache = (SyncCache *)
                calloc(1, sizeof(SyncCache) + count*sizeof(SyncCacheItem));
            data->syncCache->allocated = count;
        }
    }

    // Make sure there's at least one open slot in the list.
    if (data->syncCache->allocated == data->syncCache->used) {
        data->syncCache->allocated *= 2;
        data->syncCache = (SyncCache *)
            realloc(data->syncCache, sizeof(SyncCache) 
                    + data->syncCache->allocated * sizeof(SyncCacheItem));
    }

    return data->syncCache;
}

TLS 线程局部存储

操作系统位线程单独提供的私有空间，用来将数据与一个正在执行的指定线程关联起来。通常只有有限的容量。

进程中的全局变量与函数内定义的静态(static)变量，是各个线程都可以访问的共享变量。

如果需要在一个线程内部的各个函数调用都能访问、但其它线程不能访问的变量（被称为static memory local to a thread 线程局部静态变量），就需要新的机制来实现。这就是TLS。

Linux 中TLS的实现和操作方法

linux:
　　int pthread_key_create(pthread_key_t *key, void (*destructor)(void*));
　　int pthread_key_delete(pthread_key_t key);
　　void *pthread_getspecific(pthread_key_t key);
　　int pthread_setspecific(pthread_key_t key, const void *value);

获取线程呢哦的局部静态变量

static inline void *tls_get_direct(tls_key_t k)
{ 
    ASSERT(is_valid_direct_key(k));

    if (_pthread_has_direct_tsd()) {
        return _pthread_getspecific_direct(k);
    } else {
        return pthread_getspecific(k);
    }
}

线程同步1-锁 @synchroinzed

os_unfair_lock #import <os/lock.h>

@synchronized

objc_sync_enter

objc_sync_exit

id2data

fetch_cache

TLS 线程局部存储

Linux 中TLS的实现和操作方法

参考

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读