Changeset 14c9aa6 in mainline for kernel/generic/src/adt/cht.c


Ignore:
Timestamp:
2012-07-27T13:40:19Z (12 years ago)
Author:
Adam Hraska <adam.hraska+hos@…>
Branches:
lfn, master, serial, ticket/834-toolchain-update, topic/msim-upgrade, topic/simplify-dev-export
Children:
0949b7a
Parents:
4ec9ea41
Message:

cht: Added initial working concurrent hash table. Builds and runs.

File:
1 edited

Legend:

Unmodified
Added
Removed

  • kernel/generic/src/adt/cht.c

    r4ec9ea41 r14c9aa6  
    2727 */
    2828
     29
    2930/** @addtogroup genericadt
    3031 * @{
    … …  
    3334/**
    3435 * @file
    35  * @brief Concurrent resizable lock-free hash table.
     36 * @brief Scalable resizable concurrent lock-free hash table.
    3637 *
    3738 */
    3839
    3940#include <adt/cht.h>
     41#include <adt/hash.h>
    4042#include <debug.h>
    4143#include <memstr.h>
    4244#include <mm/slab.h>
    43 #include <barrier.h>
     45#include <arch/barrier.h>
    4446#include <compiler/barrier.h>
    4547#include <atomic.h>
    4648#include <synch/rcu.h>
    4749
    48 /* Logarithm of the min bucket count. */
     50
     51/* Logarithm of the min bucket count. 2^6 == 64 buckets. */
    4952#define CHT_MIN_ORDER 6
    5053/* Logarithm of the max bucket count. */
    … …  
    6164        N_NORMAL = 0,
    6265        N_DELETED = 1,
    63         N_INVALID = 1,
    64         N_CONST = 3,
     66        N_CONST = 1,
     67        N_INVALID = 3,
    6568        N_JOIN = 2,
    6669        N_JOIN_FOLLOWS = 2,
    … …  
    8083} wnd_t;
    8184
    82 
    83 static size_t size_to_order(size_t bucket_cnt);
    84 static cht_buckets_t *alloc_buckets(size_t order);
    85 
    86 static marked_ptr_t make_link(cht_link_t *next, mark_t mark);
     85/* Fwd decl. */
     86static size_t size_to_order(size_t bucket_cnt, size_t min_order);
     87static cht_buckets_t *alloc_buckets(size_t order, bool set_invalid);
     88static cht_link_t *search_bucket(cht_t *h, marked_ptr_t head, void *key,
     89        size_t search_hash);
     90static cht_link_t *find_resizing(cht_t *h, void *key, size_t hash,
     91        marked_ptr_t old_head, size_t old_idx);
     92static bool insert_impl(cht_t *h, cht_link_t *item, bool unique);
     93static bool insert_at(cht_link_t *item, const wnd_t *wnd, walk_mode_t walk_mode,
     94        bool *resizing);
     95static bool has_duplicates(cht_t *h, const cht_link_t *item, size_t hash,
     96        const wnd_t *cwnd);
     97static cht_link_t *find_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
     98        cht_link_t *start);
     99static bool remove_pred(cht_t *h, size_t hash, equal_pred_t pred, void *pred_arg);
     100static bool delete_at(cht_t *h, wnd_t *wnd, walk_mode_t walk_mode,
     101        bool *deleted_but_gc, bool *resizing);
     102static bool mark_deleted(cht_link_t *cur, walk_mode_t walk_mode, bool *resizing);
     103static bool unlink_from_pred(wnd_t *wnd, walk_mode_t walk_mode, bool *resizing);
     104static bool find_wnd_and_gc_pred(cht_t *h, size_t hash, walk_mode_t walk_mode,
     105        equal_pred_t pred, void *pred_arg, wnd_t *wnd, bool *resizing);
     106static bool find_wnd_and_gc(cht_t *h, size_t hash, walk_mode_t walk_mode,
     107        wnd_t *wnd, bool *resizing);
     108static bool gc_deleted_node(cht_t *h, walk_mode_t walk_mode, wnd_t *wnd,
     109        bool *resizing);
     110static bool join_completed(cht_t *h, const wnd_t *wnd);
     111static void upd_resizing_head(cht_t *h, size_t hash, marked_ptr_t **phead,
     112        bool *join_finishing,  walk_mode_t *walk_mode);
     113static void item_removed(cht_t *h);
     114static void item_inserted(cht_t *h);
     115static void free_later(cht_t *h, cht_link_t *item);
     116static void help_head_move(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head);
     117static void start_head_move(marked_ptr_t *psrc_head);
     118static void mark_const(marked_ptr_t *psrc_head);
     119static void complete_head_move(marked_ptr_t *psrc_head, marked_ptr_t *pdest_head);
     120static void split_bucket(cht_t *h, marked_ptr_t *psrc_head,
     121        marked_ptr_t *pdest_head, size_t split_hash);
     122static void mark_join_follows(cht_t *h, marked_ptr_t *psrc_head,
     123        size_t split_hash, wnd_t *wnd);
     124static void mark_join_node(cht_link_t *join_node);
     125static void join_buckets(cht_t *h, marked_ptr_t *psrc_head,
     126        marked_ptr_t *pdest_head, size_t split_hash);
     127static void link_to_join_node(cht_t *h, marked_ptr_t *pdest_head,
     128        cht_link_t *join_node, size_t split_hash);
     129static void resize_table(work_t *arg);
     130static void grow_table(cht_t *h);
     131static void shrink_table(cht_t *h);
     132static void cleanup_join_node(cht_t *h, marked_ptr_t *new_head);
     133static void clear_join_and_gc(cht_t *h, cht_link_t *join_node,
     134        marked_ptr_t *new_head);
     135static void cleanup_join_follows(cht_t *h, marked_ptr_t *new_head);
     136static marked_ptr_t make_link(const cht_link_t *next, mark_t mark);
    87137static cht_link_t * get_next(marked_ptr_t link);
    88138static mark_t get_mark(marked_ptr_t link);
    89 
     139static void next_wnd(wnd_t *wnd);
     140static bool same_node_pred(void *node, const cht_link_t *item2);
    90141static size_t key_hash(cht_t *h, void *key);
    91142static size_t node_hash(cht_t *h, const cht_link_t *item);
    92 
    93143static size_t calc_split_hash(size_t split_idx, size_t order);
    94144static size_t calc_bucket_idx(size_t hash, size_t order);
     145static size_t grow_to_split_idx(size_t old_idx);
    95146static size_t grow_idx(size_t idx);
    96147static size_t shrink_idx(size_t idx);
    97 
    98 
    99 
    100 bool cht_create(cht_t *h, size_t init_size, cht_ops_t *op)
     148static marked_ptr_t cas_link(marked_ptr_t *link, const cht_link_t *cur_next,
     149        mark_t cur_mark, const cht_link_t *new_next, mark_t new_mark);
     150static marked_ptr_t _cas_link(marked_ptr_t *link, marked_ptr_t cur,
     151        marked_ptr_t new);
     152static void cas_order_barrier(void);
     153
     154
     155bool cht_create(cht_t *h, size_t init_size, size_t min_size, cht_ops_t *op)
    101156{
    102157        ASSERT(h);
    … …  
    107162                return false;
    108163       
    109         size_t order = size_to_order(init_size);
    110        
    111         h->b = alloc_buckets(order);
     164        size_t min_order = size_to_order(min_size, CHT_MIN_ORDER);
     165        size_t order = size_to_order(init_size, min_order);
     166       
     167        h->b = alloc_buckets(order, false);
    112168       
    113169        if (!h->b)
    114170                return false;
    115171       
     172        h->min_order = min_order;
    116173        h->new_b = 0;
    117174        h->op = op;
    … …  
    127184{
    128185        size_t bucket_cnt = (1 << order);
    129         cht_buckets_t *b = malloc(
    130                 sizeof(cht_buckets_t) + (bucket_cnt - 1) * sizeof(marked_ptr_t));
     186        size_t bytes =
     187                sizeof(cht_buckets_t) + (bucket_cnt - 1) * sizeof(marked_ptr_t);
     188        cht_buckets_t *b = malloc(bytes, FRAME_ATOMIC);
    131189       
    132190        if (!b)
    … …  
    145203}
    146204
    147 static size_t size_to_order(size_t bucket_cnt)
    148 {
    149         size_t order = CHT_MIN_ORDER;
     205static size_t size_to_order(size_t bucket_cnt, size_t min_order)
     206{
     207        size_t order = min_order;
    150208
    151209        /* Find a power of two such that bucket_cnt <= 2^order */
    152210        do {
    153                 if (bucket_cnt <= (1 << order))
     211                if (bucket_cnt <= ((size_t)1 << order))
    154212                        return order;
    155213               
    … …  
    163221void cht_destroy(cht_t *h)
    164222{
    165         /* todo: impl */
     223        /* Wait for resize to complete. */
     224        while (0 < atomic_get(&h->resize_reqs)) {
     225                rcu_barrier();
     226        }
     227       
     228        /* Wait for all remove_callback()s to complete. */
     229        rcu_barrier();
     230       
     231        free(h->b);
     232        h->b = 0;
    166233}
    167234
    168235cht_link_t *cht_find(cht_t *h, void *key)
    169236{
    170         /* Make the most recent changes of the table visible. */
     237        /* Make the most recent changes to the table visible. */
    171238        read_barrier();
    172239        return cht_find_lazy(h, key);
    … …  
    195262}
    196263
     264cht_link_t *cht_find_next(cht_t *h, const cht_link_t *item)
     265{
     266        /* Make the most recent changes to the table visible. */
     267        read_barrier();
     268        return cht_find_next_lazy(h, item);
     269}
     270
     271cht_link_t *cht_find_next_lazy(cht_t *h, const cht_link_t *item)
     272{
     273        ASSERT(h);
     274        ASSERT(rcu_read_locked());
     275        ASSERT(item);
     276       
     277        return find_duplicate(h, item, node_hash(h, item), get_next(item->link));
     278}
    197279
    198280static cht_link_t *search_bucket(cht_t *h, marked_ptr_t head, void *key,
    … …  
    207289                 * may find by following the next pointers is allocated.
    208290                 */
    209                 size_t cur_hash = node_hash(cur);
     291                size_t cur_hash = node_hash(h, cur);
    210292
    211293                if (cur_hash >= search_hash) {
    … …  
    363445void cht_insert(cht_t *h, cht_link_t *item)
    364446{
     447        insert_impl(h, item, false);
     448}
     449
     450bool cht_insert_unique(cht_t *h, cht_link_t *item)
     451{
    365452        return insert_impl(h, item, true);
    366453}
    367454
    368 bool cht_insert_unique(cht_t *h, cht_link_t *item)
    369 {
    370         insert_impl(h, item, false);
    371 }
    372 
    373 bool insert_impl(cht_t *h, cht_link_t *item, bool unique)
     455static bool insert_impl(cht_t *h, cht_link_t *item, bool unique)
    374456{
    375457        rcu_read_lock();
    … …  
    391473                /* The table is resizing. Get the correct bucket head. */
    392474                if (resizing) {
    393                         upd_resizing_head(hash, &phead, &join_finishing, &walk_mode);
     475                        upd_resizing_head(h, hash, &phead, &join_finishing, &walk_mode);
    394476                }
    395477               
    … …  
    405487                }
    406488               
    407                 if (unique && has_duplicates(h, item, hash, wnd)) {
     489                if (unique && has_duplicates(h, item, hash, &wnd)) {
    408490                        rcu_read_unlock();
    409491                        return false;
    410492                }
    411493               
    412                 inserted = insert_at(item, wnd, walk_mode, &resizing);         
     494                inserted = insert_at(item, &wnd, walk_mode, &resizing);         
    413495        } while (!inserted);
    414496       
     497        rcu_read_unlock();
     498
    415499        item_inserted(h);
    416        
    417         rcu_read_unlock();
    418500        return true;
    419501}
    … …  
    435517                        return true;
    436518                } else {
     519                        /* This includes an invalid head but not a const head. */
    437520                        *resizing = ((N_JOIN_FOLLOWS | N_JOIN) & get_mark(ret));
    438521                        return false;
    … …  
    465548}
    466549
    467 static bool has_duplicates(cht_t *h, cht_link_t *item, size_t hash,
    468         const wnd_t *cwnd)
     550static bool has_duplicates(cht_t *h, const cht_link_t *item, size_t hash,
     551        const wnd_t *wnd)
    469552{
    470553        ASSERT(0 == wnd->cur || hash <= node_hash(h, wnd->cur));
    … …  
    479562         */
    480563        read_barrier();
    481 
    482         cht_link_t *cur = wnd->cur;
    483        
    484         do {
     564        return 0 != find_duplicate(h, item, hash, wnd->cur);
     565}
     566
     567static cht_link_t *find_duplicate(cht_t *h, const cht_link_t *item, size_t hash,
     568        cht_link_t *start)
     569{
     570        ASSERT(0 == start || hash <= node_hash(h, start));
     571
     572        cht_link_t *cur = start;
     573       
     574        while (cur && node_hash(h, cur) == hash) {
    485575                bool deleted = (N_DELETED & get_mark(cur->link));
    486576               
    487577                /* Skip logically deleted nodes. */
    488578                if (!deleted && h->op->equal(item, cur))
    489                         return true;
     579                        return cur;
    490580               
    491581                cur = get_next(cur->link);
    492         } while (cur && node_hash(h, cur) == hash);
    493        
    494         return false;   
    495 }
    496 
     582        }
     583       
     584        return 0;       
     585}
    497586
    498587size_t cht_remove_key(cht_t *h, void *key)
    … …  
    544633                /* The table is resizing. Get the correct bucket head. */
    545634                if (resizing) {
    546                         upd_resizing_head(hash, &phead, &join_finishing, &walk_mode);
     635                        upd_resizing_head(h, hash, &phead, &join_finishing, &walk_mode);
    547636                }
    548637               
    … …  
    583672                }
    584673               
    585                 deleted = delete_at(wnd, walk_mode, &deleted_but_gc, &resizing);               
     674                deleted = delete_at(h, &wnd, walk_mode, &deleted_but_gc, &resizing);           
    586675        } while (!deleted || deleted_but_gc);
    587676       
    … …  
    608697        if (walk_mode == WM_LEAVE_JOIN && (N_JOIN & get_mark(wnd->cur->link)))
    609698                return true;
     699       
     700        cas_order_barrier();
    610701       
    611702        if (unlink_from_pred(wnd, walk_mode, resizing)) {
    … …  
    631722        if (walk_mode == WM_NORMAL) {
    632723                /* Only mark clean/normal nodes - JF/JN is used only during resize. */
    633                 marked_ptr_t normal_link = make_link(next, N_NORMAL);
    634                 marked_ptr_t del_link = make_link(next, N_DELETED);
    635                
    636                 marked_ptr_t ret = cas_link(&cur->link, normal_link, del_link);
    637                
    638                 if (normal_link != ret) {
    639                         *resizing = (N_JOIN | N_JOIN_FOLLOWS | N_INVALID) & get_mark(ret);
     724                marked_ptr_t ret = cas_link(&cur->link, next, N_NORMAL, next, N_DELETED);
     725               
     726                if (ret != make_link(next, N_NORMAL)) {
     727                        *resizing = (N_JOIN | N_JOIN_FOLLOWS) & get_mark(ret);
    640728                        return false;
    641729                }
    … …  
    646734                mark_t cur_mark = get_mark(cur->link) & N_JOIN_FOLLOWS;
    647735               
    648                 marked_ptr_t nondel_link = make_link(next, cur_mark);
    649                 marked_ptr_t del_link = make_link(next, cur_mark | N_DELETED);
    650                
    651                 if (nondel_link != cas_link(&cur->link, nondel_link, del_link))
     736                marked_ptr_t ret =
     737                        cas_link(&cur->link, next, cur_mark, next, cur_mark | N_DELETED);
     738               
     739                if (ret != make_link(next, cur_mark))
    652740                        return false;
    653741        }
    … …  
    667755
    668756                mark_t pred_mark = get_mark(*wnd->ppred);
    669                 /* Succeed only of the predecessor is clean/normal or a join node. */
     757                /* Succeed only if the predecessor is clean/normal or a join node. */
    670758                mark_t exp_pred_mark = (N_JOIN & pred_mark) ? pred_mark : N_NORMAL;
    671759               
    … …  
    673761                marked_ptr_t next_link = make_link(next, exp_pred_mark);
    674762               
    675                 if (pred_link != cas_link(wnd->ppred, pred_link, next_link))
     763                if (pred_link != _cas_link(wnd->ppred, pred_link, next_link))
    676764                        return false;
    677765        } else {
    … …  
    685773                marked_ptr_t next_link = make_link(next, cur_mark);             
    686774               
    687                 marked_ptr_t ret = cas_link(wnd->ppred, pred_link, next_link);
     775                marked_ptr_t ret = _cas_link(wnd->ppred, pred_link, next_link);
    688776               
    689777                if (pred_link != ret) {
    … …  
    834922
    835923static void upd_resizing_head(cht_t *h, size_t hash, marked_ptr_t **phead,
    836         bool *join_finishing,  bool *walk_mode)
     924        bool *join_finishing,  walk_mode_t *walk_mode)
    837925{
    838926        cht_buckets_t *b = rcu_access(h->b);
    … …  
    852940               
    853941                /* Complete moving the bucket from the old to the new table. */
    854                 move_head(pold_head, pmoved_head);
     942                help_head_move(pold_head, pmoved_head);
    855943               
    856944                /* The hash belongs to the moved bucket. */
    857945                if (move_dest_idx == new_idx) {
     946                        ASSERT(pmoved_head == pnew_head);
    858947                        /*
    859948                         * move_head() makes the new head of the moved bucket visible.
    860949                         * The new head may be marked with a JOIN_FOLLOWS
    861950                         */
    862                         ASSERT(!(N_CONST & get_mark(*pnew_head)));
     951                        ASSERT(!(N_CONST & get_mark(*pmoved_head)));
    863952                        *walk_mode = WM_MOVE_JOIN_FOLLOWS;
    864953                } else {
     954                        ASSERT(pmoved_head != pnew_head);
    865955                        /*
    866956                         * The hash belongs to the bucket that is the result of splitting
    … …  
    872962                        if (N_NORMAL != get_mark(*pnew_head)) {
    873963                                size_t split_hash = calc_split_hash(new_idx, h->new_b->order);
    874                                 split_bucket(pmoved_head, pnew_head, split_hash);
     964                                split_bucket(h, pmoved_head, pnew_head, split_hash);
    875965                                /*
    876966                                 * split_bucket() makes the new head visible. No
    … …  
    891981                 * Makes a valid pnew_head visible if already moved.
    892982                 */
    893                 move_head(&b->head[move_src_idx], pnew_head);
     983                help_head_move(&b->head[move_src_idx], pnew_head);
    894984               
    895985                /* Hash belongs to the bucket to be joined with the moved bucket. */
    … …  
    898988                        if (N_INVALID != get_mark(*pold_head)) {
    899989                                size_t split_hash = calc_split_hash(old_idx, b->order);
    900                                 join_buckets(pold_head, pnew_head, split_hash);
     990                                join_buckets(h, pold_head, pnew_head, split_hash);
    901991                        }
    902992                       
    … …  
    9261016{
    9271017        start_head_move(psrc_head);
     1018        cas_order_barrier();
    9281019        complete_head_move(psrc_head, pdest_head);
    9291020}
    … …  
    9411032                        /* Make the move visible on this cpu. */
    9421033                        read_barrier();
    943                         ASSERT(!(N_CONST & get_mark(*pdest_head)));
    9441034                }
    9451035        } else {
    9461036                complete_head_move(psrc_head, pdest_head);
    9471037        }
     1038       
     1039        ASSERT(!(N_CONST & get_mark(*pdest_head)));
    9481040}
    9491041
    … …  
    9741066       
    9751067        cht_link_t *next = get_next(*psrc_head);
    976         /* todo: cas order barrier */
    977         cas_link(pdest_head, 0, N_INVALID, next, N_NORMAL);
    978         /* todo: cas order barrier */
    979         cas_link(psrc_head, next, N_CONST, next, N_INVALID);   
     1068        marked_ptr_t ret;
     1069       
     1070        ret = cas_link(pdest_head, 0, N_INVALID, next, N_NORMAL);
     1071        ASSERT(ret == make_link(0, N_INVALID) || (N_NORMAL == get_mark(ret)));
     1072        cas_order_barrier();
     1073       
     1074        ret = cas_link(psrc_head, next, N_CONST, next, N_INVALID);     
     1075        ASSERT(ret == make_link(next, N_CONST) || (N_INVALID == get_mark(ret)));
     1076        cas_order_barrier();
    9801077}
    9811078
    … …  
    10311128         */
    10321129        wnd_t wnd;
    1033         bool done;
    10341130       
    10351131        rcu_read_lock();
    … …  
    10371133        /* Mark the last node of the first part of the split bucket as JF. */
    10381134        mark_join_follows(h, psrc_head, split_hash, &wnd);
    1039        
    1040         /* todo: cas order barrier */
     1135        cas_order_barrier();
    10411136       
    10421137        /* There are nodes in the dest bucket, ie the second part of the split. */
    … …  
    10471142                 */
    10481143                mark_join_node(wnd.cur);
     1144                cas_order_barrier();
    10491145        } else {
    10501146                /*
    … …  
    10551151       
    10561152        /* Link the dest head to the second part of the split. */
    1057         cas_link(pdest_head, 0, N_INVALID, wnd.cur, N_NORMAL);
     1153        marked_ptr_t ret = cas_link(pdest_head, 0, N_INVALID, wnd.cur, N_NORMAL);
     1154        ASSERT(ret == make_link(0, N_INVALID) || (N_NORMAL == get_mark(ret)));
     1155        cas_order_barrier();
    10581156       
    10591157        rcu_read_unlock();
    … …  
    10671165        bool done;
    10681166        do {
    1069                 bool dummy;
     1167                bool resizing = false;
    10701168                wnd->ppred = psrc_head;
    10711169                wnd->cur = get_next(*psrc_head);
    … …  
    10761174                 * the second part of the split bucket. Retry if GC failed.
    10771175                 */
    1078                 if (!find_wnd_and_gc(h, split_hash, WM_MOVE_JOIN_FOLLOWS, wnd, &dummy))
     1176                if (!find_wnd_and_gc(h, split_hash, WM_MOVE_JOIN_FOLLOWS, wnd, &resizing))
    10791177                        continue;
    10801178               
     1179                /* Must not report that the table is resizing if WM_MOVE_JOIN_FOLLOWS.*/
     1180                ASSERT(!resizing);
    10811181                /*
    10821182                 * Mark the last node of the first half of the split bucket
    … …  
    10861186                        = cas_link(wnd->ppred, wnd->cur, N_NORMAL, wnd->cur, N_JOIN_FOLLOWS);
    10871187
    1088                 /* Successfully marked as a JF node or already marked that way. */
     1188                /*
     1189                 * Successfully marked as a JF node or already marked that way (even
     1190                 * if also marked deleted - unlinking the node will move the JF mark).
     1191                 */
    10891192                done = (ret == make_link(wnd->cur, N_NORMAL))
    10901193                        || (N_JOIN_FOLLOWS & get_mark(ret));
    … …  
    10981201        bool done;
    10991202        do {
    1100                 cht_link_t *next = get_next(*join_node);
    1101                 mark_t mark = get_mark(*join_node);
     1203                cht_link_t *next = get_next(join_node->link);
     1204                mark_t mark = get_mark(join_node->link);
    11021205               
    11031206                /*
    … …  
    11891292        rcu_read_lock();
    11901293       
    1191         /* Mark src_head immutable - signals updaters bucket join started. */
     1294        /* Mark src_head immutable - signals updaters that bucket join started. */
    11921295        mark_const(psrc_head);
    1193         /* todo: cas order barrier*/
     1296        cas_order_barrier();
    11941297       
    11951298        cht_link_t *join_node = get_next(*psrc_head);
    … …  
    11971300        if (join_node) {
    11981301                mark_join_node(join_node);
    1199                 /* todo: cas order barrier*/
     1302                cas_order_barrier();
    12001303               
    12011304                link_to_join_node(h, pdest_head, join_node, split_hash);
    1202                 /* todo: cas order barrier*/
     1305                cas_order_barrier();
    12031306        }
    12041307       
    1205         cas_link(psrc_head, join_node, N_CONST, join_node, N_INVALID);
     1308        marked_ptr_t ret =
     1309                cas_link(psrc_head, join_node, N_CONST, join_node, N_INVALID);
     1310        ASSERT(ret == make_link(join_node, N_CONST) || (N_INVALID == get_mark(ret)));
     1311        cas_order_barrier();
    12061312       
    12071313        rcu_read_unlock();
    … …  
    12181324                };
    12191325               
    1220                 bool dummy;
    1221                
    1222                 if (!find_wnd_and_gc(h, split_hash, WM_LEAVE_JOIN, &wnd, &dummy))
     1326                bool resizing = false;
     1327               
     1328                if (!find_wnd_and_gc(h, split_hash, WM_LEAVE_JOIN, &wnd, &resizing))
    12231329                        continue;
    12241330
     1331                ASSERT(!resizing);
     1332               
    12251333                if (wnd.cur) {
    12261334                        /* Must be from the new appended bucket. */
    … …  
    12491357static void item_removed(cht_t *h)
    12501358{
    1251         /* todo: impl */
     1359        size_t items = (size_t) atomic_predec(&h->item_cnt);
     1360        size_t bucket_cnt = (1 << h->b->order);
     1361       
     1362        bool need_shrink = (items == CHT_MAX_LOAD * bucket_cnt / 4);
     1363        bool missed_shrink = (items == CHT_MAX_LOAD * bucket_cnt / 8);
     1364       
     1365        if ((need_shrink || missed_shrink) && h->b->order > h->min_order) {
     1366                atomic_count_t resize_reqs = atomic_preinc(&h->resize_reqs);
     1367                /* The first resize request. Start the resizer. */
     1368                if (1 == resize_reqs) {
     1369                        workq_global_enqueue_noblock(&h->resize_work, resize_table);
     1370                }
     1371        }
    12521372}
    12531373
    12541374static void item_inserted(cht_t *h)
    12551375{
    1256         /* todo: impl */
    1257 }
    1258 
    1259 static void resize_table(void *arg)
    1260 {
    1261         cht_t *h = (cht_t *)arg;
    1262        
     1376        size_t items = (size_t) atomic_preinc(&h->item_cnt);
     1377        size_t bucket_cnt = (1 << h->b->order);
     1378       
     1379        bool need_grow = (items == CHT_MAX_LOAD * bucket_cnt);
     1380        bool missed_grow = (items == 2 * CHT_MAX_LOAD * bucket_cnt);
     1381       
     1382        if ((need_grow || missed_grow) && h->b->order < CHT_MAX_ORDER) {
     1383                atomic_count_t resize_reqs = atomic_preinc(&h->resize_reqs);
     1384                /* The first resize request. Start the resizer. */
     1385                if (1 == resize_reqs) {
     1386                        workq_global_enqueue_noblock(&h->resize_work, resize_table);
     1387                }
     1388        }
     1389}
     1390
     1391static void resize_table(work_t *arg)
     1392{
     1393        cht_t *h = member_to_inst(arg, cht_t, resize_work);
     1394       
     1395#ifdef CONFIG_DEBUG
    12631396        ASSERT(h->b);
    1264         ASSERT(0 < (read_barrier(), atomic_get(&h->resize_reqs)));
    1265 
    1266         /* Load the most current h->item_cnt. */
     1397        /* Make resize_reqs visible. */
    12671398        read_barrier();
     1399        ASSERT(0 < atomic_get(&h->resize_reqs));
     1400#endif
     1401
     1402        bool done;
    12681403        do {
    1269                 size_t cur_items = h->item_cnt;
     1404                /* Load the most recent  h->item_cnt. */
     1405                read_barrier();
     1406                size_t cur_items = (size_t) atomic_get(&h->item_cnt);
    12701407                size_t bucket_cnt = (1 << h->b->order);
    1271 
    1272                 if (cur_items >= CHT_MAX_LOAD * bucket_cnt) {
     1408                size_t max_items = CHT_MAX_LOAD * bucket_cnt;
     1409
     1410                if (cur_items >= max_items && h->b->order < CHT_MAX_ORDER) {
    12731411                        grow_table(h);
    1274                 } else if (cur_items <= CHT_MAX_LOAD * bucket_cnt / 4) {
     1412                } else if (cur_items <= max_items / 4 && h->b->order > h->min_order) {
    12751413                        shrink_table(h);
    1276                 }
    1277                
    1278                 /* Load the most current h->item_cnt and h->resize_reqs. */
    1279                 read_barrier();
    1280         } while (0 < atomic_predec(&h->resize_reqs));
     1414                } else {
     1415                        /* Table is just the right size. */
     1416                        atomic_count_t reqs = atomic_predec(&h->resize_reqs);
     1417                        done = (reqs == 0);
     1418                }
     1419        } while (!done);
    12811420}
    12821421
    … …  
    12941433        /* Wait for all readers and updaters to see the initialized new table. */
    12951434        rcu_synchronize();
    1296        
    12971435        size_t old_bucket_cnt = (1 << h->b->order);
    12981436       
    … …  
    13051443        }
    13061444       
     1445        /* Order start_head_move() wrt complete_head_move(). */
     1446        cas_order_barrier();
     1447       
    13071448        /* Complete moving heads and split any buckets not yet split by updaters. */
    13081449        for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
    … …  
    13331474                size_t new_idx = grow_idx(old_idx);
    13341475               
    1335                 cleanup_join_follows(h, &h->new_b[new_idx]);
    1336         }
    1337        
     1476                cleanup_join_follows(h, &h->new_b->head[new_idx]);
     1477        }
     1478
    13381479        /*
    13391480         * Wait for everyone to notice that buckets were split, ie link connecting
    … …  
    13461487                size_t new_idx = grow_to_split_idx(old_idx);
    13471488               
    1348                 cleanup_join_node(h, &h->new_b[new_idx]);
    1349         }
    1350        
     1489                cleanup_join_node(h, &h->new_b->head[new_idx]);
     1490        }
     1491
    13511492        /* Wait for everyone to see that the table is clear of any resize marks. */
    13521493        rcu_synchronize();
    … …  
    13541495        cht_buckets_t *old_b = h->b;
    13551496        rcu_assign(h->b, h->new_b);
    1356        
     1497
    13571498        /* Wait for everyone to start using the new table. */
    13581499        rcu_synchronize();
    … …  
    13661507static void shrink_table(cht_t *h)
    13671508{
    1368         if (h->b->order <= CHT_MIN_ORDER)
     1509        if (h->b->order <= h->min_order)
    13691510                return;
    13701511       
    … …  
    13951536        }
    13961537       
     1538        /* Order start_head_move() wrt to complete_head_move(). */
     1539        cas_order_barrier();
     1540       
    13971541        /* Complete moving heads and join buckets with the moved buckets. */
    13981542        for (size_t old_idx = 0; old_idx < old_bucket_cnt; ++old_idx) {
    13991543                size_t new_idx = shrink_idx(old_idx);
     1544                size_t move_src_idx = grow_idx(new_idx);
    14001545               
    14011546                /* This bucket should be moved. */
    1402                 if (grow_idx(new_idx) == old_idx) {
     1547                if (move_src_idx == old_idx) {
    14031548                        /* Head move not yet completed. */
    14041549                        if (N_INVALID != get_mark(h->b->head[old_idx])) {
    … …  
    14261571                /* Set the invalid joinee head to NULL. */
    14271572                if (old_idx != move_src_idx) {
    1428                         ASSERT(N_INVALID == h->b->head[old_idx]);
     1573                        ASSERT(N_INVALID == get_mark(h->b->head[old_idx]));
    14291574                       
    14301575                        if (0 != get_next(h->b->head[old_idx]))
    … …  
    14401585        /* Clear the JOIN mark and GC any deleted join nodes. */
    14411586        for (size_t new_idx = 0; new_idx < new_bucket_cnt; ++new_idx) {
    1442                 cleanup_join_node(h, &h->new_b[new_idx]);
     1587                cleanup_join_node(h, &h->new_b->head[new_idx]);
    14431588        }
    14441589
    … …  
    14961641                /* Done if the mark was cleared. Retry if a new node was inserted. */
    14971642                done = (ret == jn_link);
     1643                ASSERT(ret == jn_link || (get_mark(ret) & N_JOIN));
    14981644        } while (!done);
    14991645       
    … …  
    15021648
    15031649        /* The join node had been marked as deleted - GC it. */
     1650
     1651        /* Clear the JOIN mark before trying to unlink the deleted join node.*/
     1652        cas_order_barrier();
    15041653       
    15051654        size_t jn_hash = node_hash(h, join_node);
    15061655        do {
    1507                 bool resizing;
     1656                bool resizing = false;
    15081657               
    15091658                wnd_t wnd = {
    … …  
    15651714                                marked_ptr_t ret
    15661715                                        = cas_link(cur_link, next, N_JOIN_FOLLOWS, 0, N_NORMAL);
     1716                               
     1717                                ASSERT(!next || ((N_JOIN | N_JOIN_FOLLOWS) & get_mark(ret)));
    15671718
    15681719                                /* Successfully cleared the JF mark of a non-deleted node. */
    … …  
    17031854         *   inserted in front of it, would require more than one grace period.
    17041855         */
    1705 }
     1856        void *ret = atomic_cas_ptr((void**)link, (void *)cur, (void *)new);
     1857        return (marked_ptr_t) ret;
     1858}
     1859
     1860static void cas_order_barrier(void)
     1861{
     1862        /* Make sure CAS to different memory locations are ordered. */
     1863        write_barrier();
     1864}
     1865
    17061866
    17071867/** @}
Note: See TracChangeset for help on using the changeset viewer.