src/mcas/rb_lock_serialisedwriters.c

   1 /******************************************************************************
   2  * rb_lock_serialisedwriters.c
   3  *
   4  * Lock-based red-black trees, using multi-reader locks.
   5  *
   6  * Updates are serialised on a global mutual-exclusion spinlock.
   7  *
   8  * Updates never need to read-lock, as updates are serialised. Must write-lock
   9  * for all node changes except colour changes and parent-link updates.
  10  *
  11  * Searches must read-lock down the tree, as they do not serialise.
  12  *
  13  * Copyright (c) 2002-2003, K A Fraser
  14
  15 Redistribution and use in source and binary forms, with or without
  16 modification, are permitted provided that the following conditions are
  17 met:
  18
  19     * Redistributions of source code must retain the above copyright
  20     * notice, this list of conditions and the following disclaimer.
  21     * Redistributions in binary form must reproduce the above
  22     * copyright notice, this list of conditions and the following
  23     * disclaimer in the documentation and/or other materials provided
  24     * with the distribution.  Neither the name of the Keir Fraser
  25     * nor the names of its contributors may be used to endorse or
  26     * promote products derived from this software without specific
  27     * prior written permission.
  28
  29 THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  30 "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  31 LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  32 A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  33 OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  34 SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  35 LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  36 DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  37 THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  38 (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  39 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  40  */
  41
  42 #define __SET_IMPLEMENTATION__
  43
  44 #include <stdio.h>
  45 #include <assert.h>
  46 #include <stdlib.h>
  47 #include <unistd.h>
  48 #include "portable_defns.h"
  49 #include "gc.h"
  50 #include "set.h"
  51
  52 #define IS_BLACK(_v)   ((int_addr_t)(_v)&1)
  53 #define IS_RED(_v)     (!IS_BLACK(_v))
  54 #define MK_BLACK(_v)   ((setval_t)((int_addr_t)(_v)|1))
  55 #define MK_RED(_v)     ((setval_t)((int_addr_t)(_v)&~1))
  56 #define GET_VALUE(_v)  (MK_RED(_v))
  57 #define GET_COLOUR(_v) (IS_BLACK(_v))
  58 #define SET_COLOUR(_v,_c) ((setval_t)((unsigned long)(_v)|(unsigned long)(_c)))
  59
  60 typedef struct node_st node_t;
  61 typedef struct set_st set_t;
  62
  63 struct node_st
  64 {
  65     setkey_t k;
  66     setval_t v;
  67     node_t *l, *r, *p;
  68     mrsw_lock_t lock;
  69 };
  70
  71 struct set_st
  72 {
  73     node_t root;
  74     CACHE_PAD(0);
  75     mcs_lock_t writer_lock;
  76 };
  77
  78 static node_t null;
  79 static int gc_id;
  80
  81 static void left_rotate(node_t *x)
  82 {
  83     mrsw_qnode_t p_qn, x_qn, y_qn;
  84     node_t *y = x->r, *p = x->p;
  85
  86     wr_lock(&p->lock, &p_qn);
  87     wr_lock(&x->lock, &x_qn);
  88     wr_lock(&y->lock, &y_qn);
  89
  90     /* No need to write-lock to update parent link. */
  91     if ( (x->r = y->l) != &null ) x->r->p = x;
  92
  93     x->p = y;
  94     y->l = x;
  95     y->p = p;
  96     if ( x == p->l ) p->l = y; else p->r = y;
  97
  98     wr_unlock(&y->lock, &y_qn);
  99     wr_unlock(&x->lock, &x_qn);
 100     wr_unlock(&p->lock, &p_qn);
 101 }
 102
 103
 104 static void right_rotate(node_t *x)
 105 {
 106     mrsw_qnode_t p_qn, x_qn, y_qn;
 107     node_t *y = x->l, *p = x->p;
 108
 109     wr_lock(&p->lock, &p_qn);
 110     wr_lock(&x->lock, &x_qn);
 111     wr_lock(&y->lock, &y_qn);
 112
 113     /* No need to write-lock to update parent link. */
 114     if ( (x->l = y->r) != &null ) x->l->p = x;
 115
 116     x->p = y;
 117     y->r = x;
 118     y->p = p;
 119     if ( x == p->l ) p->l = y; else p->r = y;
 120
 121     wr_unlock(&y->lock, &y_qn);
 122     wr_unlock(&x->lock, &x_qn);
 123     wr_unlock(&p->lock, &p_qn);
 124 }
 125
 126
 127 /* No locks held on entry/exit. Colour changes safe. Rotations lock for us. */
 128 static void delete_fixup(ptst_t *ptst, set_t *s, node_t *x)
 129 {
 130     node_t *p, *w;
 131
 132     while ( (x->p != &s->root) && IS_BLACK(x->v) )
 133     {
 134         p = x->p;
 135
 136         if ( x == p->l )
 137         {
 138             w = p->r;
 139             if ( IS_RED(w->v) )
 140             {
 141                 w->v = MK_BLACK(w->v);
 142                 p->v = MK_RED(p->v);
 143                 /* Node W will be new parent of P. */
 144                 left_rotate(p);
 145                 /* Get new sibling W. */
 146                 w = p->r;
 147             }
 148
 149             if ( IS_BLACK(w->l->v) && IS_BLACK(w->r->v) )
 150             {
 151                 w->v = MK_RED(w->v);
 152                 x = p;
 153             }
 154             else
 155             {
 156                 if ( IS_BLACK(w->r->v) )
 157                 {
 158                     /* w->l is red => it cannot be null node. */
 159                     w->l->v = MK_BLACK(w->l->v);
 160                     w->v  = MK_RED(w->v);
 161                     right_rotate(w);
 162                     /* Old w is new w->r. Old w->l is new w.*/
 163                     w = p->r;
 164                 }
 165
 166                 w->v = SET_COLOUR(GET_VALUE(w->v), GET_COLOUR(p->v));
 167                 p->v = MK_BLACK(p->v);
 168                 w->r->v = MK_BLACK(w->r->v);
 169                 left_rotate(p);
 170                 break;
 171             }
 172         }
 173         else /* SYMMETRIC CASE */
 174         {
 175             w = p->l;
 176             if ( IS_RED(w->v) )
 177             {
 178                 w->v = MK_BLACK(w->v);
 179                 p->v = MK_RED(p->v);
 180                 /* Node W will be new parent of P. */
 181                 right_rotate(p);
 182                 /* Get new sibling W. */
 183                 w = p->l;
 184             }
 185
 186             if ( IS_BLACK(w->l->v) && IS_BLACK(w->r->v) )
 187             {
 188                 w->v = MK_RED(w->v);
 189                 x = p;
 190             }
 191             else
 192             {
 193                 if ( IS_BLACK(w->l->v) )
 194                 {
 195                     /* w->r is red => it cannot be the null node. */
 196                     w->r->v = MK_BLACK(w->r->v);
 197                     w->v  = MK_RED(w->v);
 198                     left_rotate(w);
 199                     /* Old w is new w->l. Old w->r is new w.*/
 200                     w = p->l;
 201                 }
 202
 203                 w->v = SET_COLOUR(GET_VALUE(w->v), GET_COLOUR(p->v));
 204                 p->v = MK_BLACK(p->v);
 205                 w->l->v = MK_BLACK(w->l->v);
 206                 right_rotate(p);
 207                 break;
 208             }
 209         }
 210     }
 211
 212     x->v = MK_BLACK(x->v);
 213 }
 214
 215
 216 set_t *set_alloc(void)
 217 {
 218     ptst_t *ptst;
 219     set_t  *set;
 220     node_t *root;
 221
 222     ptst = critical_enter();
 223
 224     set = (set_t *)malloc(sizeof(*set));
 225
 226     root = &set->root;
 227     root->k = SENTINEL_KEYMIN;
 228     root->v = MK_RED(NULL);
 229     root->l = &null;
 230     root->r = &null;
 231     root->p = NULL;
 232     mrsw_init(&root->lock);
 233
 234     mcs_init(&set->writer_lock);
 235
 236     critical_exit(ptst);
 237
 238     return set;
 239 }
 240
 241
 242 setval_t set_update(set_t *s, setkey_t k, setval_t v, int overwrite)
 243 {
 244     ptst_t  *ptst;
 245     node_t  *x, *p, *g, *y, *new;
 246     mrsw_qnode_t x_qn;
 247     qnode_t writer_qn;
 248     setval_t ov;
 249
 250     k = CALLER_TO_INTERNAL_KEY(k);
 251
 252     ptst = critical_enter();
 253
 254     mcs_lock(&s->writer_lock, &writer_qn);
 255
 256     x = &s->root;
 257     while ( (y = (k < x->k) ? x->l : x->r) != &null )
 258     {
 259         x = y;
 260         if ( k == x->k ) break;
 261     }
 262
 263     if ( k == x->k )
 264     {
 265         ov = x->v;
 266         /* Lock X to change mapping. */
 267         wr_lock(&x->lock, &x_qn);
 268         if ( overwrite ) x->v = SET_COLOUR(v, GET_COLOUR(ov));
 269         wr_unlock(&x->lock, &x_qn);
 270         ov = GET_VALUE(ov);
 271     }
 272     else
 273     {
 274         ov = NULL;
 275
 276         new = (node_t *)gc_alloc(ptst, gc_id);
 277         new->k = k;
 278         new->v = MK_RED(v);
 279         new->l = &null;
 280         new->r = &null;
 281         new->p = x;
 282         mrsw_init(&new->lock);
 283
 284         /* Lock X to change a child. */
 285         wr_lock(&x->lock, &x_qn);
 286         if ( k < x->k ) x->l = new; else x->r = new;
 287         wr_unlock(&x->lock, &x_qn);
 288
 289         x = new;
 290
 291         /* No locks held here. Colour changes safe. Rotations lock for us. */
 292         for ( ; ; )
 293         {
 294             if ( (p = x->p) == &s->root )
 295             {
 296                 x->v = MK_BLACK(x->v);
 297                 break;
 298             }
 299
 300             if ( IS_BLACK(p->v) ) break;
 301
 302             g = p->p;
 303             if ( p == g->l )
 304             {
 305                 y = g->r;
 306                 if ( IS_RED(y->v) )
 307                 {
 308                     p->v = MK_BLACK(p->v);
 309                     y->v = MK_BLACK(y->v);
 310                     g->v = MK_RED(g->v);
 311                     x = g;
 312                 }
 313                 else
 314                 {
 315                     if ( x == p->r )
 316                     {
 317                         x = p;
 318                         left_rotate(x);
 319                         /* X and P switched round. */
 320                         p = x->p;
 321                     }
 322                     p->v = MK_BLACK(p->v);
 323                     g->v = MK_RED(g->v);
 324                     right_rotate(g);
 325                     /* G no longer on the path. */
 326                 }
 327             }
 328             else /* SYMMETRIC CASE */
 329             {
 330                 y = g->l;
 331                 if ( IS_RED(y->v) )
 332                 {
 333                     p->v = MK_BLACK(p->v);
 334                     y->v = MK_BLACK(y->v);
 335                     g->v = MK_RED(g->v);
 336                     x = g;
 337                 }
 338                 else
 339                 {
 340                     if ( x == p->l )
 341                     {
 342                         x = p;
 343                         right_rotate(x);
 344                         /* X and P switched round. */
 345                         p = x->p;
 346                     }
 347                     p->v = MK_BLACK(p->v);
 348                     g->v = MK_RED(g->v);
 349                     left_rotate(g);
 350                     /* G no longer on the path. */
 351                 }
 352             }
 353         }
 354     }
 355
 356     mcs_unlock(&s->writer_lock, &writer_qn);
 357
 358     critical_exit(ptst);
 359
 360     return ov;
 361 }
 362
 363
 364 setval_t set_remove(set_t *s, setkey_t k)
 365 {
 366     ptst_t  *ptst;
 367     node_t  *x, *y, *z;
 368     mrsw_qnode_t qn[2], *y_pqn=qn+0, *yp_pqn=qn+1, *t_pqn;
 369     mrsw_qnode_t z_qn, zp_qn;
 370     qnode_t writer_qn;
 371     setval_t ov = NULL;
 372
 373     k = CALLER_TO_INTERNAL_KEY(k);
 374
 375     ptst = critical_enter();
 376
 377     mcs_lock(&s->writer_lock, &writer_qn);
 378
 379     z = &s->root;
 380     while ( (z = (k < z->k) ? z->l : z->r) != &null )
 381     {
 382         if ( k == z->k ) break;
 383     }
 384
 385     if ( k == z->k )
 386     {
 387         ov = GET_VALUE(z->v);
 388
 389         if ( (z->l != &null) && (z->r != &null) )
 390         {
 391             /* Lock Z. It will get new key copied in. */
 392             wr_lock(&z->lock, &z_qn);
 393             y = z->r;
 394             /*
 395              * Write-lock from Z to Y. We end up with (YP,Y) locked.
 396              * Write-coupling is needed so we don't overtake searches for Y.
 397              */
 398             wr_lock(&y->lock, y_pqn);
 399             while ( y->l != &null )
 400             {
 401                 if ( y->p != z ) wr_unlock(&y->p->lock, yp_pqn);
 402                 y = y->l;
 403                 t_pqn  = yp_pqn;
 404                 yp_pqn = y_pqn;
 405                 y_pqn  = t_pqn;
 406                 wr_lock(&y->lock, y_pqn);
 407             }
 408         }
 409         else
 410         {
 411             y = z;
 412             /* Lock ZP. It will get new child. */
 413             wr_lock(&z->p->lock, &zp_qn);
 414             /* Lock Z. It will be deleted. */
 415             wr_lock(&z->lock, &z_qn);
 416         }
 417
 418         /* No need to lock X. Only parent link is modified. */
 419         x = (y->l != &null) ? y->l : y->r;
 420         x->p = y->p;
 421
 422         if ( y == y->p->l ) y->p->l = x; else y->p->r = x;
 423
 424         if ( y != z )
 425         {
 426             z->k = y->k;
 427             z->v = SET_COLOUR(GET_VALUE(y->v), GET_COLOUR(z->v));
 428             if ( y->p != z ) wr_unlock(&y->p->lock, yp_pqn);
 429             wr_unlock(&y->lock, y_pqn);
 430         }
 431         else
 432         {
 433             wr_unlock(&z->p->lock, &zp_qn);
 434         }
 435
 436         wr_unlock(&z->lock, &z_qn);
 437
 438         gc_free(ptst, y, gc_id);
 439
 440         if ( IS_BLACK(y->v) ) delete_fixup(ptst, s, x);
 441     }
 442
 443     mcs_unlock(&s->writer_lock, &writer_qn);
 444
 445     critical_exit(ptst);
 446
 447     return ov;
 448 }
 449
 450
 451 setval_t set_lookup(set_t *s, setkey_t k)
 452 {
 453     ptst_t  *ptst;
 454     node_t  *m, *n;
 455     mrsw_qnode_t qn[2], *m_pqn=&qn[0], *n_pqn=&qn[1], *t_pqn;
 456     setval_t v = NULL;
 457
 458     k = CALLER_TO_INTERNAL_KEY(k);
 459
 460     ptst = critical_enter();
 461
 462     n = &s->root;
 463     rd_lock(&n->lock, n_pqn);
 464
 465     while ( (m = (k < n->k) ? n->l : n->r) != &null )
 466     {
 467         rd_lock(&m->lock, m_pqn);
 468         rd_unlock(&n->lock, n_pqn);
 469         n = m;
 470         t_pqn = m_pqn;
 471         m_pqn = n_pqn;
 472         n_pqn = t_pqn;
 473         if ( k == n->k )
 474         {
 475             v = GET_VALUE(n->v);
 476             break;
 477         }
 478     }
 479
 480     rd_unlock(&n->lock, n_pqn);
 481
 482     critical_exit(ptst);
 483
 484     return v;
 485 }
 486
 487
 488 void _init_set_subsystem(void)
 489 {
 490     gc_id = gc_add_allocator(sizeof(node_t));
 491
 492     null.k = 0;
 493     null.v = MK_BLACK(NULL);
 494     null.l = NULL;
 495     null.r = NULL;
 496     null.p = NULL;
 497     mrsw_init(&null.lock);
 498 }