摘要:我们将这样的一个内存数据称为,数据字段第二,由于内存本身是要被维护的,所以我们的策略是随需分配,已分配的不在回收,我们将作为一种可重用的资源,用于追踪那些共享内存。
由于我对于Java并发库JUC的深入,一直以来有个想法,能不能把Java并发库移植到纯C语言环境下,并且在实现、使用方式上都与Java平台保持相当程度的相似性呢?
纯C环境下内存模型与Java平台不一致?加上内存屏障(fence)或者lock指令就行。
C环境下缺少对象模型?无非是给每个数据块提供个*init方法(比如pthread_mutex_t的pthread_mutex_init,pthread_barrier_t的pthread_barrier_init),之后再将逻辑的部分直接写到函数里(phtread_mutex_lock、pthread_barrier_wait),无非是这些逻辑并不是像Java平台上一样被绑定到某个"对象"。这些都不是本质上困难的地方。
真正的困难在于内存回收。
Java平台,我们能够对内存做的唯一事情就是申请、创建(new关键字),如此一来便得到一个新的对象。之后,我们无法直接针对这块内存释放。我们最多只能把自己目所能及范围内的关于这块内存的"引用"置为null,从而期待GC去回收(前提是这块内存不存在其他引用)。同时,我们根本无法知道究竟哪个时刻这块内存会被回收,我们只能认为,Runtime environment能够选择最适当的时刻。
同时,Java虚拟机给出了一个更强的保证:只要你的对象(引用)obj != null,那么这个引用所指示的对象便是肯定存在的,我们可以绝对安全地调用obj.method()而不用害怕任何意外。
把这个场景放在C环境下比喻似乎就在说:如果你能看到某个地址(数值),那么就放心对他做任何合理的事吧!Runtime environment确保了你的安全!
总之,Java运行时给出了强大的保证:
1 看得到的对象都存在,你可以放心对它操作。
2 那些正在被回收的、你不能安全操作的对象(内存),你绝对无法看到。
如此一来,无论是内存独占或共享、线程并发或非并发,我们都无需担心内存本身的问题(Java GC回收所有内存,全平台的垃圾回收器)。
那么现在我们回头来看C环境。它的运行时环境根本不帮你做任何事,你甚至可以虚构一个内存地址,然后将它强制转化为虚构的struct类型,接着对它操作。只不过这样很可能破坏了内存,导致不可预期的结果,同时这个错误会一直潜伏,你根本不知道何时出现。甚至于,都不需要定义结构体信息,假如你了解运行机器的具体信息,你都可以直接在一个地址上做地址偏移来操作内存......当然,前提是这块内存是安全的。
那么问题就很清楚了,
Java环境下内存安全回收由虚拟机完全负责。
C环境下,内存安全回收由逻辑单元(线程)本身来负责。
也就是说C环境下,我们不仅要处理算法本身的逻辑,同时也要额外去处理内存回收的问题。
那么C环境下内存回收困难在哪呢?
首先,我们不考虑不需要回收的内存。我们知道,独占的内存容易回收,对于共享的内存,假如有个逻辑点,我们确保所有线程当下以及将来都不会使用这块内存,那么也是可以安全回收的。
所以真正难于回收的是满足以下3个条件的内存:
共享的。
需要回收的(有些程序设计成不回收)。
没有明确的可以安全回收内存的逻辑点(存在被并发读写,但是有明确逻辑点回收的设计,比如主逻辑join)
OK,我们接着引出Maged M. Michael,在2004发表的《Hazard Pointers:Safe Memory Reclamation for Lock-Free Objects》。
Hazard Pointer可以说是最知名的一种共享内存的回收方案。借用Erez Petrank的ppt,它可以概括如下:
以及后续变化的Lock-Free Data Structures with Hazard Pointers、Hazard Eras、Maged M. Michael等。
对于每一块被共享的内存,这些技术都需要为它配置另外N(MAX_THREAD)个内存来标记它是否被对应的线程访问,同时为了释放一块内存,都需要遍历这N个内存位置从而判定是否可以安全回收(尽管可以采用先排序,二分搜索等方式降低检索的数据量,但是检索过程的代价依然是随着MAX_THREAD而增长的,就算你总操作数一样。尽管它的算法是lock-free(甚至wait-free)。这里的缺陷在于每一块共享数据都与线程本身紧紧耦合,几乎没有扩展性。
而另一种被广泛讨论的回收方案是epoch-based ReclamationPractical lock-freedom以及相似的技术Performance of memory reclamation for lockless synchronization,它的性能很好,但是如内存管理规则所说,它并不是无阻塞的算法,它在本质上就会阻塞,所以progress无法得到保证。
其中2005年,哥德堡大学的研究团队发表的方案Practical and Efficient Lock-Free Garbage Collection Based on Reference Counting,尽管也采用了引用计数的方式(与我将要介绍的方式类似),但是却并没有解除线程本身与共享数据之间的依赖,并且scan时依然要遍历整个线程组。
在不计较lock指令或者fence的情况下,是否存在一种方法能够同时做到:
解除共享数据与线程之间的强制依赖,不需要为每一块共享数据造另N个标记(正因如此,通常线程数N较小),从而极大得增强算法的扩展性,并且节省内存。
减少甚至取消判断内存是否可被回收的遍历过程,消除MAX_THREAD的限制(算法视角),从而极大地减少了算法的搜索代价,同时增强了算法的灵活性和可用性(无MAX_THREAD限制)。如此一来,很可能不需要有个阈值来判断何时该回收。
同时,它保持non-blocking的progress。
另外,能不能提供一种方式,在我们开发某些并发数据结构时,它能够正确回收对象层次内部的数据呢?
能不能给让C语言像其他支持并发的语言一样,做一些比较高级的事。比如:某个线程构造了一个队列A,往里面put几个队列,其他线程从A中获取队列,再通过获取的队列与主线程通信呢?
为了达成以上几个目标,我向大家介绍自创的SHP(Scalable Hazard Pointers)。
Scalable Hazard Pointers分为如下三个部分来讲述:
3RE&S协议(每一块满足以上三个条件的内存,都通过3RE协议规定的方式回收)
并发无锁带引用计数的有序单链表(我们组织被记录地址的方式)
可扩充的Reocrd资源库(我们维护Record的方式)
不断的优化(局部保留Record+每线程每内存变量、链表遍历保留前驱、批量获取Record、统一回收的大块数据、Record分布计数)
带参数的retire,通过传入定制的freeMemory方法,从而在回收当前内存前,先回收内部指针开辟的数据
3RE&S Protocol3RE&S指的是如下四个,针对被分配内存地址(pointer)的抽象操作:
RECORD(增加一个记录计数)
REMOVE(移除一个记录计数)
RETIRE(标记该记录的retire,表明之后若记录计数为0则可回收)
SCAN:遍历已经retire的所有record,有策略地回收内存
同时,对于共享数据的读、写为下面两种方式:
read: read共享变量+RECORD+再检验共享变量
write: write共享变量+RETIRE共享变量(write操作一般的应用场景会在read之后)
这个协议总结了,包含了hazardpointer以及许多类似技术的处理方式,RECORD操作时将共享内存地址本身也作为传输传入。
几乎任何一种算法,并发共享的数据都可通过以上方式回收。
这里的关键点是:
一块内存是否被回收这件事情最好由SCAN操作中的原子操作来完成,原因是我们肯定是在retire之后才考虑回收内存,同时RECORD可以回退。
RECORD之后一般有个再判断的操作。
这里给出个针对Michael原始论文的对比例子:
void enqueue(int value){ NodeType* node; posix_memalign(&node, 64, sizeof(NodeType)); memset(node, 0, sizeof(NodeType)); node->value = value; node->next = NULL; NodeType* t; for(;;){ t = Tail; psly_record(&Tail, t); if(Tail != t) { psly_remove(t); continue; } NodeType* next = t->next; if(Tail != t) { psly_remove(t); continue; } if(next != NULL){ psly_remove(t); __sync_bool_compare_and_swap(&Tail, t, next); continue; } if(__sync_bool_compare_and_swap(&t->next, NULL, node)) { psly_remove(t); break; } psly_remove(t); } __sync_bool_compare_and_swap(&Tail, t, node); }
int dequeue(){ int data; NodeType* h; for(;;){ h = Head; //myhprec->HP[0] = h; psly_record(&Tail, h); if(Head != h) { psly_remove(h); continue; } NodeType* t = Tail; NodeType* next = h->next; psly_record(&(h->next), next); //myhprec->HP[1] = next; if(Head != h) { psly_remove(next); psly_remove(h); continue; } if(next == NULL) { psly_remove(next); psly_remove(h); return -1000000; } if(h == t){ psly_remove(next); psly_remove(h); __sync_bool_compare_and_swap(&Tail, t, next); continue; } data = next->value; //myhprec->HP[1] = NULL; //myhprec->HP[0] = NULL; if(__sync_bool_compare_and_swap(&Head, h, next)) { psly_remove(next); psly_remove(h); retireNode(h); break; } psly_remove(next); psly_remove(h); } //myhprec->HP[0] = NULL; //myhprec->HP[1] = NULL; return data; }concurrent lock-free ordered singly linked-list with reference counting
首先,为什么要为每块共享内存维护N个变量,这样做不仅浪费内存而且增加搜索代价。理想情况下我们应该只需要一个内存数据来处理一个共享内存。那么我们怎么来处理多个线程引用它的情况呢?这里的一个自然的想法是引入引用计数(reference count),(注意,这里的引用计数是线程引用共享数据,与另一个对象层次间的引用无关),我们用refcount代表当下访问它的线程数,refCount>0的内存绝对不会被回收,refCount == 0代表没有线程引用它,处于可以被回收的状态。我们将这样的一个内存数据称为Record,
struct Record{ //数据字段 void* pointer; int refcount; }
第二,由于Record内存本身是要被维护的,所以我们的策略是随需分配,已分配的不在回收,我们将Record作为一种可重用的资源,用于追踪那些共享内存。那么意味着Record能够被高效地从资源库并发获取和返回。
以上斜体字是我之前的观点,我有了点突破,现在我有了技术方案,能以较小代价实现Record的回收了!:)
这里的技巧是用一个固定的self字段,标记Record本身,(比如低10位作为indexNum,接下来4位作为arrayNum),用于唯一的标记Record自身。这么做的目的是为了支持后面的map。
想象一下,我们已经有了很多共享的pointer(地址),每个分配一个Record,接着我们要如何组织它们呢?从而高效的支持add,remove,search操作。
这里的方式是用一个大的map,它具有一个大的buckets数目比如1024。
typedef struct RecordList { Record* volatile head ; Record* volatile tail ; } RecordList ; typedef struct RecordMap { RecordList* lists[1024] ; } RecordMap ;
RecordList初始化之后久拥有了固定的head/tail。
很明显,我们这里自然的策略是将具有相同后缀的地址base到同一个list上,同时由于开辟的内存地址一般是单调递增,并且保持16字节对齐,我们可以根据地址本身来避开了hits。
我们可以抽象出如下的接口如下:
struct Record{ //Record库维护字段 int volatile next ; int self; //数据字段 void* pointer; int refcount; } int record(void* pointer) { long key = ((long) pointer) >> 4 ; RecordList* list = map.lists[key & (1024-1)]; return handle_records(list, pointer, RECORD); }
那么接下来的问题是如何在一个list上组织那些带有同样后缀地址的Record?
这里推荐著名的Harris" list,它做到用一个并发无锁单链表来组织有序数据,支持增加、删除、搜索。
我们需要给Record再加一个retire"bit和一个long字段nextRecord,它里面包含四个字段:
下一个Record的self(位移)(最右23位 next 8百万+节点)
下一个Record本身的版本号(20位 nextV 百万+个版本号)
该Record本身的版本号(20位 nodeV 百万+个版本号)
该Record是否已经被逻辑删除的标记(最高位 isDeleted)
retireBit代表该Record是否被retire
struct Record{ //Record库维护字段 int volatile next ; int self; //数据字段 void* pointer; int refcount; //retire bool retireBit; long nextRecord; }
然后,对它进行了相当程度的改造,改造的地方如下:
不将next字段(harris"s list的next字段,不是Record资源库的)作为地址,将它作为数值,把Record的self字段原子交换过去。(CAS nextRecord)
对于被切除的Record,我们要自己回收这些Record并将它重新放入资源库(Harris"list基于GC可以不用管)。(nextRecord)
因为Record是可复用,所以我们为它增加版本号字段(nodeV),同时它的next字段也需要版本号来回避ABA问题(nextV)
原来的insert操作有对象的情况下什么都不做,我们给引用计数+1 (refcount),如果发现已经retire那么回退
需要一个bit来标记这块内存被retire了(retireBit),同时需要另一个bit来标记是否被逻辑删除(逻辑删除后,该块内存就可以被回收了,nextRecord最高位)
通过对前驱节点的nextRecord中版本号和DELETE的判断来准确判断当前节点值是否是在链上的值
注意,由于以上的操作存在相互依赖,比如新增的Record不能链接到已经逻辑删除的Record上。
所以,当一个Record被从资源库get到之后,生命周期为:
它的nodeVersion+1,refCount+1,clear DELETE,同时将它的nextValue设置为后续节点。
接着它会经历refCount +1/-1的序列操作。
nextVersion+1 & nextValue改变 的序列操作。(与2没有先后关系)
retire的bit被设置的操作。(会导致RECORD的失败或则回退)
最高位DELETE被cmpandswap。(之后该pointer指示的地址可以被回收)
包含该Record在内连在一起被DELETE的节点,都从list上切除,依次回收,并将Record返回资源库。
该Record被return回资源库。
这里最核心的技术就是基于两点判定节点没有失效:
共享变量的值依旧是传入地址值。
节点的 nodeVersion 以及 DELETE bit维持原状。
可扩充的Reocrd资源库我们需要一些Record去持有地址,我们采用随需批量malloc的方式。这种方式的特点是:
可重复利用:大量malloc和free的方式对操作系统的内存维护也不友好,我们替换为支持并发的getRecord/returnRecord/idx_Record。
多用途(也就是nextRecord的工作方式):因为我们开辟的是一块大内存,使用的是其中一小块数据。那么每小块数据都可以唯一的标记为:大内存首地址+大内存中的offset。这样,在64位机器上,我们不需要再使用64位地址,而只需要用较小的位数便可以,从而带来了极大的好处(当原子操作时,64位中剩下的位数可以用于其他作用)。同时,20位便可以标记(1<<20,百万多个小块内存)。
也就是说,我们拿record本身的self字段做定位: records[(record->self & ARRBIT) >> IDXNUM] + (record->self & IDXBIT) == record
支持多种方式实现,只需要提供getRecord/returnRecord/idx_Record接口(下面会给出具体的一种实现),
而我们付出的代价仅仅只是一定固定长度的数组。
以上方式,可以如图所示:
下面给出一个大概例子,采用的经典的MSQueue,当然,存在其他更高效的方式lcrq:
int PSLY_Record_IDXNUM = 16; int PSLY_Record_IDXBIT = ((1 << 16) - 1); int PSLY_Record_ARRAYNUM_MAX = (1 << 4); int PSLY_Record_ARRAYNUM = (1 << 2); int PSLY_Record_ARRAYBITS = ((1 << 4) -1); int PSLY_Record_ARRBIT = (((1 << 4) - 1) << 16); int PSLY_Record_ARRBITR = ((1 << 4) - 1); int PSLY_Record_ARRIDXBIT = ((((1 << 4) - 1) << 16) | ((1 << 16) - 1)); int PSLY_Record_NEXTIDXNUM = 16; int PSLY_Record_NEXTIDXBIT = ((1 << 16) - 1); int PSLY_Record_NEXTTAILNUM = 1; int PSLY_Record_NEXTTAILBIT = (((1 << 1) - 1) << 16); int PSLY_Record_NEXTVERSIONNUM = (32 - 1 - 16); int PSLY_Record_NEXTVERSIONBIT = ((~0)^((((1 << 1) - 1) << 16) | ((1 << 16) - 1))); int PSLY_Record_NEXTVERSIONONE = (1 + ((((1 << 1) - 1) << 16) | ((1 << 16) - 1))); int PSLY_Record_TAILIDXNUM = 16; int PSLY_Record_TAILIDXBIT = ((1 << 16) - 1); int PSLY_Record_TAILVERSIONNUM = (32 - 16); int PSLY_Record_TAILVERSIONBIT = ((~0) ^ ((1 << 16) - 1)); int PSLY_Record_TAILVERSIONONE = (1 + ((1 << 16) - 1)); int PSLY_Record_HEADIDXNUM = 16; int PSLY_Record_HEADIDXBIT = ((1 << 16) - 1); int PSLY_Record_HEADVERSIONNUM = (32 - 16); int PSLY_Record_HEADVERSIONBIT = ((~0) ^ ((1 << 16) - 1)); int PSLY_Record_HEADVERSIONONE = (1 + ((1 << 16) - 1)); typedef struct Record { int volatile next __attribute__((aligned(128))); int self ; long volatile nextRecord __attribute__((aligned(128))); void* volatile pointer ; } Record __attribute__((aligned(128))); typedef struct RecordQueue { int volatile head ; int volatile tail ; } RecordQueue ; static Record* volatile psly_Records[1 << 4]; static RecordQueue volatile psly_Record_queues[1 << 4]; static int volatile recordTake = 0; Record* idx_Record(int index) { return psly_Records[(index & PSLY_Record_ARRBIT) >> PSLY_Record_IDXNUM] + (index & PSLY_Record_IDXBIT); } Record* get_Record() { for(;;) { int localArrayNum = PSLY_Record_ARRAYNUM; //取最高队列 int array = localArrayNum - 1; RecordQueue* queue = psly_Record_queues + array; Record* arr = psly_Records[array]; for(;;){ int headIndex = (queue->head); int indexHead = headIndex & PSLY_Record_HEADIDXBIT; Record* head = arr + indexHead; int tailIndex = (queue->tail); int indexTail = tailIndex & PSLY_Record_TAILIDXBIT; int nextIndex = (head->next); if(headIndex == (queue->head)) { if(indexHead == indexTail){ if((nextIndex & PSLY_Record_NEXTTAILBIT) == PSLY_Record_NEXTTAILBIT) break; __sync_bool_compare_and_swap(&queue->tail, tailIndex, (((tailIndex & PSLY_Record_TAILVERSIONBIT) + PSLY_Record_TAILVERSIONONE ) & PSLY_Record_TAILVERSIONBIT)|(nextIndex & PSLY_Record_TAILIDXBIT)); } else { if(__sync_bool_compare_and_swap(&queue->head, headIndex, (((headIndex & PSLY_Record_HEADVERSIONBIT) + PSLY_Record_HEADVERSIONONE) & PSLY_Record_HEADVERSIONBIT)|(nextIndex & PSLY_Record_HEADIDXBIT))) { return head; } else { break; } } } } // 轮询某些队列 for(int i = 0; i < localArrayNum; ++i) { int array = __sync_fetch_and_add(&recordTake, 1) % localArrayNum; RecordQueue* queue = psly_Record_queues + array; Record* arr = psly_Records[array]; for(;;){ int headIndex = (queue->head); int indexHead = headIndex & PSLY_Record_HEADIDXBIT; Record* head = arr + indexHead; int tailIndex = (queue->tail); int indexTail = tailIndex & PSLY_Record_TAILIDXBIT; int nextIndex = (head->next); if(headIndex == (queue->head)) { if(indexHead == indexTail){ if((nextIndex & PSLY_Record_NEXTTAILBIT) == PSLY_Record_NEXTTAILBIT) break; __sync_bool_compare_and_swap(&queue->tail, tailIndex, (((tailIndex & PSLY_Record_TAILVERSIONBIT) + PSLY_Record_TAILVERSIONONE ) & PSLY_Record_TAILVERSIONBIT)|(nextIndex & PSLY_Record_TAILIDXBIT)); } else { if(__sync_bool_compare_and_swap(&queue->head, headIndex, (((headIndex & PSLY_Record_HEADVERSIONBIT) + PSLY_Record_HEADVERSIONONE) & PSLY_Record_HEADVERSIONBIT)|(nextIndex & PSLY_Record_HEADIDXBIT))) { return head; } else { break; } } } } } // 遍历所有队列 for(int i = 0; i < localArrayNum; ++i) { int array = i; RecordQueue* queue = psly_Record_queues + array; Record* arr = psly_Records[array]; for(;;){ int headIndex = (queue->head); int indexHead = headIndex & PSLY_Record_HEADIDXBIT; Record* head = arr + indexHead; int tailIndex = (queue->tail); int indexTail = tailIndex & PSLY_Record_TAILIDXBIT; int nextIndex = (head->next); if(headIndex == (queue->head)) { if(indexHead == indexTail){ if((nextIndex & PSLY_Record_NEXTTAILBIT) == PSLY_Record_NEXTTAILBIT) break; __sync_bool_compare_and_swap(&queue->tail, tailIndex, (((tailIndex & PSLY_Record_TAILVERSIONBIT) + PSLY_Record_TAILVERSIONONE ) & PSLY_Record_TAILVERSIONBIT)|(nextIndex & PSLY_Record_TAILIDXBIT)); } else { if(__sync_bool_compare_and_swap(&queue->head, headIndex, (((headIndex & PSLY_Record_HEADVERSIONBIT) + PSLY_Record_HEADVERSIONONE) & PSLY_Record_HEADVERSIONBIT)|(nextIndex & PSLY_Record_HEADIDXBIT))) { return head; } } } } } //不够增加 if(localArrayNum == PSLY_Record_ARRAYNUM_MAX) return NULL; if(localArrayNum == PSLY_Record_ARRAYNUM) { if(psly_Records[localArrayNum] == NULL) { int array_ = localArrayNum; Record* record; void * ptr; int ret = posix_memalign(&ptr, 4096, (1 << PSLY_Record_IDXNUM) * sizeof(Record)); record = ptr; memset(record, 0, (1 << PSLY_Record_IDXNUM) * sizeof(Record)); for(int j = 0; j < (1 << PSLY_Record_IDXNUM) - 1; ++j){ record->self = (array_ << PSLY_Record_IDXNUM) | j; record->next = j+1; record->pointer = NULL; record->nextRecord = 0; record += 1; } record->self = (array_ << PSLY_Record_IDXNUM) | ((1 << PSLY_Record_IDXNUM) - 1); record->next = PSLY_Record_NEXTTAILBIT; record->pointer = NULL; record->nextRecord = 0; //printf("I"m here %d %ld ", localArrayNum, pthread_self()); if(!__sync_bool_compare_and_swap(&psly_Records[array_], NULL, ptr)) {free(ptr);} else /*printf("extend to %d ", localArrayNum + 1)*/; } if(localArrayNum == PSLY_Record_ARRAYNUM) __sync_bool_compare_and_swap(&PSLY_Record_ARRAYNUM, localArrayNum, localArrayNum + 1); } } } void return_Record(Record* record) { long local = (record->next); local |= PSLY_Record_NEXTTAILBIT; record->next = local; int self = record->self; int array = (self >> PSLY_Record_IDXNUM) & PSLY_Record_ARRBITR; Record* arr = psly_Records[array]; RecordQueue* queue = psly_Record_queues + array; for(;;) { int tailIndex = (queue->tail); int indexTail = tailIndex & PSLY_Record_TAILIDXBIT; Record* tail = arr + indexTail; int nextIndex = (tail->next); if(tailIndex == (queue->tail)){ if((nextIndex & PSLY_Record_NEXTTAILBIT) == PSLY_Record_NEXTTAILBIT) { if(__sync_bool_compare_and_swap(&tail->next, nextIndex, (((nextIndex & PSLY_Record_NEXTVERSIONBIT) + PSLY_Record_NEXTVERSIONONE) & PSLY_Record_NEXTVERSIONBIT)|(self & PSLY_Record_NEXTIDXBIT))){ __sync_bool_compare_and_swap(&queue->tail, tailIndex, (((tailIndex & PSLY_Record_TAILVERSIONBIT) + PSLY_Record_TAILVERSIONONE) & PSLY_Record_TAILVERSIONBIT)|(self & PSLY_Record_TAILIDXBIT)); return; } } else { __sync_bool_compare_and_swap(&queue->tail, tailIndex, (((tailIndex & PSLY_Record_TAILVERSIONBIT) + PSLY_Record_TAILVERSIONONE) & PSLY_Record_TAILVERSIONBIT)|(nextIndex & PSLY_Record_TAILIDXBIT)); } } } } typedef struct RecordList { Record* volatile head ; Record* volatile tail ; } RecordList ; typedef struct RecordMap { volatile RecordList* lists[131070] ; } RecordMap ; static volatile RecordMap map; #define INIT_RESOURCE(listNum) for(int i = 0; i < (PSLY_Record_ARRAYNUM); ++i){ Record* record; void * ptr; int ret = posix_memalign(&ptr, 4096, (1 << PSLY_Record_IDXNUM) * sizeof(Record)); psly_Records[i] = record = ptr; memset(record, 0, (1 << PSLY_Record_IDXNUM) * sizeof(Record)); for(int j = 0; j < (1 << PSLY_Record_IDXNUM) - 1; ++j){ record->self = (i << PSLY_Record_IDXNUM) | j; record->next = j+1; record->pointer = NULL; record->nextRecord = 0; record += 1; } record->self = (i << PSLY_Record_IDXNUM) | ((1 << PSLY_Record_IDXNUM) - 1); record->next = PSLY_Record_NEXTTAILBIT; record->pointer = NULL; record->nextRecord = 0; } for(int i = 0; i < PSLY_Record_ARRAYNUM_MAX; ++i){ psly_Record_queues[i].head = 0; psly_Record_queues[i].tail = (1 << PSLY_Record_IDXNUM) - 1; } for(int i = 0; i < listNum; ++i) { void* ptr; int ret = posix_memalign(&ptr, 4096, sizeof(RecordList)); Record* head = get_Record(); Record* tail = get_Record(); head->nextRecord = newNext(head->nextRecord, tail); map.lists[i] = ptr; map.lists[i]->head = head; map.lists[i]->tail = tail; } #define UNINIT_RESOURCE(listNum) for(int i = 0; i < (PSLY_Record_ARRAYNUM); ++i){ free(psly_Records[i]); } for(int i = 0; i < listNum; ++i) { free(map.lists[i]); }
这里的psly_Records[1 << 4],psly_Record_queues[1 << 4] 代表总共可以提供16组,每组65536个数据(PSLY_Record_IDXNUM = 16),初始分配4组数据(PSLY_Record_ARRAYNUM = (1 << 2)),之后如果不够就扩充一组。
不断的优化局部变量:
对于一块共享内存S,我们为它的Record保持一个局部变量reordS,只需要在RECORD时候返回,后续的REMOVE跟RETIRE就不需要去查询了,相对于之前的enqueue给出的例子如下:
+++为变动代码
void enqueue(int value){ NodeType* node; posix_memalign(&node, 64, sizeof(NodeType)); memset(node, 0, sizeof(NodeType)); node->value = value; node->next = NULL; NodeType* t; for(;;){ t = Tail; +++ Record* recordT = psly_record(&Tail, t); +++ if(recordT == NULL) +++ continue; if(Tail != t) { +++ psly_remove(recordT); continue; } NodeType* next = t->next; if(Tail != t) { +++ psly_remove(recordT); continue; } if(next != NULL){ +++ psly_remove(recordT); __sync_bool_compare_and_swap(&Tail, t, next); continue; } if(__sync_bool_compare_and_swap(&t->next, NULL, node)) { +++ psly_remove(recordT); break; } +++ psly_remove(recordT); } __sync_bool_compare_and_swap(&Tail, t, node); }
这样以来,每次操作,最需要在RECORD时候遍历一次。
我们还可以做的更好
线程私有数据:
有些场景的共享内存,会在一段长期时间内不会改变,这种情况的话每次都去查询maplist显得很浪费,我们可以做个缓存来节省查询。
对于一块确定的共享内存S,我们尝试为每线程配置一个私有变量(static __tread),用一段结构化的代码跟踪它的Record,这样以来就不需要每次都查询maplist了,
虽然不是非常合适,但我们还是拿前面Hazard pointer的例子做个示例,代码如下:
void enqueue(int value){ NodeType* node; posix_memalign(&node, 64, sizeof(NodeType)); memset(node, 0, sizeof(NodeType)); node->value = value; node->next = NULL; NodeType* t; for(;;){ t = Tail; +++ static __thread LocalRecord localRecordT; Record* recordT; +++ if(localRecordT.pointer == NULL) { +++ recordT = psly_record(&Tail, t, NULL, NULL); if(recordT == NULL) continue; +++ else { +++ localRecordT.pointer = t; +++ localRecordT.record = recordT; +++ localRecordT.nextRecord = recordT->nextRecord; +++ } +++ } else { +++ if(t != localRecordT.pointer || isChange(localRecordT.record, localRecordT.nextRecord) || t != localRecordT.record->pointer) { +++ localRecordT.pointer = NULL; +++ continue; +++ } +++ recordT = psly_record(&Tail, t, localRecordT.record, localRecordT.nextRecord); +++ if(recordT == NULL) { +++ localRecordT.pointer = NULL; +++ continue; +++ } +++ } if(Tail != t) { psly_remove(recordT); continue; } NodeType* next = t->next; if(Tail != t) { psly_remove(recordT); continue; } if(next != NULL){ psly_remove(recordT); __sync_bool_compare_and_swap(&Tail, t, next); continue; } if(__sync_bool_compare_and_swap(&t->next, NULL, node)) { psly_remove(recordT); break; } psly_remove(recordT); } __sync_bool_compare_and_swap(&Tail, t, node); }
这种场景下,假如我们系统共有N个线程,那么对于一个内存,在它的整个生命周期里,需要查询maplist的次数上限为N+1次!
这种方式极大地减少了查询的次数,从而为设计某些高效的共享数据结构提供了可能。
数据结构本身带Record
void enqueue(int value){ NodeType* node; posix_memalign(&node, 64, sizeof(NodeType)); memset(node, 0, sizeof(NodeType)); node->value = value; node->next = NULL; NodeType* t; for(;;){ +++ Record* recordT = TailRecord; +++ int versionT= recordT->version; t = Tail; +++ if(recordT->pointer != t) { +++ if(recordT == TailRecord && recordT->version == versionT) { +++ Record* localRecordT = NULL; //udpate记录Record并给引用计数设置1,Atomicity; +++ if(CAS(&TailRecord, recordT, localRecordT = get_Record(t, ONE))) { +++ return_Record(recordT); +++ } else { +++ return_Record(localRecordT); +++ continue; +++ } +++ } else { +++ continue; +++ } +++ } else if(recordT != TailRecord || recordT->version != versionT) { +++ continue; +++ } else { +++ psly_addOneRecord(recordT); // 增加引用计数 +++ } //now the t and RecordT is match!; if(Tail != t) { psly_remove(recordT); continue; } NodeType* next = t->next; if(Tail != t) { psly_remove(recordT); continue; } if(next != NULL){ psly_remove(recordT); __sync_bool_compare_and_swap(&Tail, t, next); continue; } if(__sync_bool_compare_and_swap(&t->next, NULL, node)) { psly_remove(recordT); break; } psly_remove(recordT); } __sync_bool_compare_and_swap(&Tail, t, node); }
这里的TailRecord伴随Tail更新。
链表遍历保留前驱:
当我们查询maplist时候,有可能会因为前驱节点的失效,而要重新在该list的head开始遍历,假如链表过长会代价较大,所以我们在遍历过程中维护些前驱节点可能会好点。
示例代码如下:
保留:
if(currKey != key) { int bucket; if((steps & STEPS_) == 0 && (bucket = (steps >> STEPBIT)) < MAXPREV) { Prevs* step = &prevs_[bucket]; step->r = prev; step->rNext = prevNext; } ++steps; prev = curr; prevNext = currNext; }
失效之后启用保留的前驱:
--steps; for(;;) { int bucket = steps >> STEPBIT; bucket = bucket < MAXPREV ? bucket: (MAXPREV - 1); Prevs* prevs = &prevs_[bucket]; prev = prevs->r; prevNext = prev->nextRecord; long prevNextKeep = prevs->rNext; if((prevNextKeep & NODEBITS) != (prevNext & NODEBITS) || (prevNext & REFCBITS) == DELETED) { steps -= STEPS; } else { prevs->rNext = prevNext; curr = idx_Record(prevNext); break; } } steps = steps & (~STEPS_);
这里的steps记录我们目前所在的位置,STEPS表达我们隔几个节点记录一次(极端情况下可以拷贝所有遍历过的节点)。
批量获取Record批量获取Record方式指的是,由于获取Record的竞争过于激烈,我们不再每次获取一个,而是每次获取一批,剩余的作为线程私有之后使用,维护好数据的 未使用/使用 状态,以及作为整体返回给资源库。从而极大地减少了线程间的竞争。
当然,最好的是我们不再让线程竞争Record了,我们重新组织Record,每个队列尺寸小一点。对于一个全新的线程,直接给它一个Record队列,这个队列如同之前一样是全局作用域。不够用继续获取整个队列,维护好单个Record元素的使用状况。
这样以来,对于每个线程而言,它拥有的就是Record[0],Record[2]....等整个队列了。如果一来,每个资源队列都是被某个线程私有,即单读多写的队列。极大地减少了竞争。
对于某些场景,许多小块的共享数据同时产生,又可以同时回收。我们不再为每个内存地址分配一个Record,我们尝试将一块大内存的分割为许多小内存来使用,如此一来小内存统一映射到大内存首地址的Record,直接省去了插入链表的操作。我需要对Record进行改造,retireBit不再作为一个元素使用,这里可以换成short
struct Record{ //Record库维护字段 int volatile next ; int self; //数据字段 void* pointer; int refcount; //retire +++ short retireNum; long nextRecord; }
同时,我们的psy_record接口增加一个参数:
recordT = psly_record(void** ppointer, void* pointer, Record* record, long nextRecord, short retireNum);
我们对于某个内存pointer,我们映射首地址的方式要依赖于如何分配内存。
retire操作现在要给retireNum减一。
最后讲一下,唯一需要注意的是,如果内存已经处于可回收状态:
所有分片内存都已经retire。
观察到一次refcount为0.
那么我们便可以立即回收内部,因为对于企图使用该内存的线程而言,要么正在递增引用计数,要么已经完成访问。完成访问的没关系,根据我们的设计,递增引用计数的线程稍后会回退减一,从而不再访问这块内存。
Record分布计数假如竞争激烈,那么我们不再将引用计数聚集到一个refcount字段上,我们可以采用多个refcount[M]来分开计数,从而改善竞争情况,有效提高吞吐量。
具体做法可以参考Striped64。
假如我们在开发一个并发数据结构,它本身将会被共享/动态开辟/回收,数据本身带有指针,指针指向的数据随着程序的执行变得不满足需求,从而我们要重新配置这一数据,并且回收原数据。
这种情况下我们能不能正确回收所有数据呢?
答案是可以的。
对于所有内存,如果满足
共享的 / 需要回收的 / 没有明确的可以安全回收内存的逻辑点
我们都采用3RE&S Protocol提供的语义来回收内存。
因为
只要采用这种方式,内部内存同样能够被record/remove/retire/scan维护好,而新开辟的内存将正确地作为内部内存。
任何时候,数据结构本身都是完整的。同时,当整体可被回收时,不会存在线程读写内部数据。安全性得到保证
最后,我们必须自己提供freeMemory函数用于先回收内部内存,再回收外部对象的内存。
文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/33305.html
摘要:两个函数之间的关系生存分析与新用户行为然而,生存分析的定义本身让它和很多用户行为事件本身发生联系。在此我们只做性别和用户年龄的回归分析。结语这里,我们用生存分析解决并使用在用户行为分析中。 生存分析的来历 生存分析(Survival Analysis)来源于基础医学领域,最早用来研究各种治疗方案对病人寿命的影响。而寿命则用一个end event(死亡)的方式衡量。基本定义如下: T为标...
摘要:原题链接广度优先搜索复杂度时间空间思路相当于是遍历二叉树。代码记录本层节点的个数最后一个节点的是,不做处理递归法复杂度时间空间递归栈空间思路由于父节点多了指针,我们不用记录父节点的父节点就能直到它相邻的节点。 Populating Next Right Pointers in Each Node I Given a binary tree struct TreeLinkNode { ...
摘要:题目要求给一个完全二叉树,将当前节点的值指向其右边的节点。而在中则是提供了一个不完全的二叉树,其它需求没有发生改变。我们需要使用一个变量来存储父节点,再使用一个变量来存储当前操作行的,将前一个指针指向当前父节点的子节点。 题目要求 Given a binary tree struct TreeLinkNode { TreeLinkNode *left; ...
摘要:复杂度思路设置一个指向下一层要遍历的节点的开头的位置。 LeetCode[117] Population Next Right Pointers in Each Node II 1 / 2 3 / 4 5 7 After calling your function, the tree should look like: ...
题目:Follow up for problem Populating Next Right Pointers in Each Node. What if the given tree could be any binary tree? Would your previous solution still work? Note: You may only use constant extra sp...
阅读 740·2023-04-25 19:35
阅读 2194·2021-11-22 09:34
阅读 3106·2021-10-09 09:44
阅读 1348·2021-09-22 15:25
阅读 2757·2019-08-29 14:00
阅读 3243·2019-08-29 11:01
阅读 2410·2019-08-26 13:26
阅读 1625·2019-08-23 18:08