"Fossies" - the Fresh Open Source Software Archive

Member "bind-9.11.23/lib/isc/rwlock.c" (7 Sep 2020, 25024 Bytes) of package /linux/misc/dns/bind9/9.11.23/bind-9.11.23.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) C and C++ source code syntax highlighting (style: standard) with prefixed line numbers and code folding option. Alternatively you can here view or download the uninterpreted source code file. For more information about "rwlock.c" see the Fossies "Dox" file reference documentation.

    1 /*
    2  * Copyright (C) Internet Systems Consortium, Inc. ("ISC")
    3  *
    4  * This Source Code Form is subject to the terms of the Mozilla Public
    5  * License, v. 2.0. If a copy of the MPL was not distributed with this
    6  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
    7  *
    8  * See the COPYRIGHT file distributed with this work for additional
    9  * information regarding copyright ownership.
   10  */
   11 
   12 
   13 /*! \file */
   14 
   15 #include <config.h>
   16 
   17 #include <stdbool.h>
   18 #include <stddef.h>
   19 #include <inttypes.h>
   20 
   21 #include <isc/atomic.h>
   22 #include <isc/magic.h>
   23 #include <isc/msgs.h>
   24 #include <isc/platform.h>
   25 #include <isc/print.h>
   26 #include <isc/rwlock.h>
   27 #include <isc/util.h>
   28 
   29 #define RWLOCK_MAGIC        ISC_MAGIC('R', 'W', 'L', 'k')
   30 #define VALID_RWLOCK(rwl)   ISC_MAGIC_VALID(rwl, RWLOCK_MAGIC)
   31 
   32 #ifdef ISC_PLATFORM_USETHREADS
   33 
   34 #if defined(ISC_RWLOCK_USESTDATOMIC)
   35 
   36 #define rwl_init(o, a) atomic_init(&rwl->o, a)
   37 #define rwl_load(o) atomic_load(&rwl->o)
   38 #define rwl_store(o, a) atomic_store(&rwl->o, (a))
   39 #define rwl_add(o, a) atomic_fetch_add(&rwl->o, (a))
   40 #define rwl_sub(o, a) atomic_fetch_sub(&rwl->o, (a))
   41 #define rwl_cmpxchg(o, e, d) atomic_compare_exchange_strong(&rwl->o, &(e), (d))
   42 
   43 #elif defined(ISC_RWLOCK_USEATOMIC)
   44 
   45 #define rwl_init(o, a) rwl->o = (a);
   46 #define rwl_load(o) isc_atomic_xadd(&rwl->o, 0)
   47 #define rwl_store(o, a) isc_atomic_store(&rwl->o, (a))
   48 #define rwl_add(o, a) isc_atomic_xadd(&rwl->o, (a))
   49 #define rwl_sub(o, a) isc_atomic_xadd(&rwl->o, -(a))
   50 #define rwl_cmpxchg(o, e, d) e = isc_atomic_cmpxchg(&rwl->o, e, d)
   51 
   52 #else
   53 
   54 #define rwl_init(o, a) rwl->o = (a)
   55 #define rwl_load(o) (rwl->o)
   56 #define rwl_store(o, a) rwl->o = (a)
   57 #define rwl_add(o, a) rwl->o += (a)
   58 #define rwl_sub(o, a) rwl->o -= (a)
   59 #define rwl_cmpxchg(o, e, d) \
   60     if (rwl->o == e) { e = rwl->o; rwl->o = d; } else { e = rwl->o; }
   61 
   62 #endif
   63 
   64 
   65 #ifndef RWLOCK_DEFAULT_READ_QUOTA
   66 #define RWLOCK_DEFAULT_READ_QUOTA 4
   67 #endif
   68 
   69 #ifndef RWLOCK_DEFAULT_WRITE_QUOTA
   70 #define RWLOCK_DEFAULT_WRITE_QUOTA 4
   71 #endif
   72 
   73 #ifndef RWLOCK_MAX_ADAPTIVE_COUNT
   74 #define RWLOCK_MAX_ADAPTIVE_COUNT 100
   75 #endif
   76 
   77 #if defined(ISC_RWLOCK_USEATOMIC)
   78 static isc_result_t
   79 isc__rwlock_lock(isc_rwlock_t *rwl, isc_rwlocktype_t type);
   80 #endif
   81 
   82 #ifdef ISC_RWLOCK_TRACE
   83 #include <stdio.h>      /* Required for fprintf/stderr. */
   84 #include <isc/thread.h>     /* Required for isc_thread_self(). */
   85 
   86 static void
   87 print_lock(const char *operation, isc_rwlock_t *rwl, isc_rwlocktype_t type) {
   88 #if defined(ISC_PLATFORM_HAVEXADD) && defined(ISC_PLATFORM_HAVECMPXCHG)
   89     fprintf(stderr,
   90         isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
   91                    ISC_MSG_PRINTLOCK2,
   92                    "rwlock %p thread %lu %s(%s): "
   93                    "write_requests=%u, write_completions=%u, "
   94                    "cnt_and_flag=0x%x, readers_waiting=%u, "
   95                    "write_granted=%u, write_quota=%u\n"),
   96         rwl, isc_thread_self(), operation,
   97         (type == isc_rwlocktype_read ?
   98          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
   99                 ISC_MSG_READ, "read") :
  100          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  101                 ISC_MSG_WRITE, "write")),
  102         rwl_load(write_requests), rwl_load(write_completions),
  103         rwl_load(cnt_and_flag), rwl->readers_waiting,
  104         rwl_load(write_granted), rwl->write_quota);
  105 #else
  106     fprintf(stderr,
  107         isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  108                    ISC_MSG_PRINTLOCK,
  109                    "rwlock %p thread %lu %s(%s): %s, %u active, "
  110                    "%u granted, %u rwaiting, %u wwaiting\n"),
  111         rwl, isc_thread_self(), operation,
  112         (type == isc_rwlocktype_read ?
  113          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  114                 ISC_MSG_READ, "read") :
  115          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  116                 ISC_MSG_WRITE, "write")),
  117         (rwl->type == isc_rwlocktype_read ?
  118          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  119                 ISC_MSG_READING, "reading") :
  120          isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  121                 ISC_MSG_WRITING, "writing")),
  122         rwl->active, rwl->granted,
  123         rwl->readers_waiting, rwl->writers_waiting);
  124 #endif
  125 }
  126 #endif /* ISC_RWLOCK_TRACE */
  127 
  128 isc_result_t
  129 isc_rwlock_init(isc_rwlock_t *rwl, unsigned int read_quota,
  130         unsigned int write_quota)
  131 {
  132     isc_result_t result;
  133 
  134     REQUIRE(rwl != NULL);
  135 
  136     /*
  137      * In case there's trouble initializing, we zero magic now.  If all
  138      * goes well, we'll set it to RWLOCK_MAGIC.
  139      */
  140     rwl->magic = 0;
  141 
  142     rwl_init(spins, 0);
  143 #if defined(ISC_RWLOCK_USEATOMIC)
  144     rwl_init(write_requests, 0);
  145     rwl_init(write_completions, 0);
  146     rwl_init(cnt_and_flag, 0);
  147     rwl_init(write_granted, 0);
  148 
  149     rwl->readers_waiting = 0;
  150     if (read_quota != 0) {
  151         UNEXPECTED_ERROR(__FILE__, __LINE__,
  152                  "read quota is not supported");
  153     }
  154     if (write_quota == 0) {
  155         write_quota = RWLOCK_DEFAULT_WRITE_QUOTA;
  156     }
  157     rwl->write_quota = write_quota;
  158 #else
  159     rwl->type = isc_rwlocktype_read;
  160     rwl->original = isc_rwlocktype_none;
  161     rwl->active = 0;
  162     rwl->granted = 0;
  163     rwl->readers_waiting = 0;
  164     rwl->writers_waiting = 0;
  165     if (read_quota == 0)
  166         read_quota = RWLOCK_DEFAULT_READ_QUOTA;
  167     rwl->read_quota = read_quota;
  168     if (write_quota == 0)
  169         write_quota = RWLOCK_DEFAULT_WRITE_QUOTA;
  170     rwl->write_quota = write_quota;
  171 #endif
  172 
  173     result = isc_mutex_init(&rwl->lock);
  174     if (result != ISC_R_SUCCESS)
  175         return (result);
  176 
  177     result = isc_condition_init(&rwl->readable);
  178     if (result != ISC_R_SUCCESS) {
  179         UNEXPECTED_ERROR(__FILE__, __LINE__,
  180                  "isc_condition_init(readable) %s: %s",
  181                  isc_msgcat_get(isc_msgcat, ISC_MSGSET_GENERAL,
  182                         ISC_MSG_FAILED, "failed"),
  183                  isc_result_totext(result));
  184         result = ISC_R_UNEXPECTED;
  185         goto destroy_lock;
  186     }
  187     result = isc_condition_init(&rwl->writeable);
  188     if (result != ISC_R_SUCCESS) {
  189         UNEXPECTED_ERROR(__FILE__, __LINE__,
  190                  "isc_condition_init(writeable) %s: %s",
  191                  isc_msgcat_get(isc_msgcat, ISC_MSGSET_GENERAL,
  192                         ISC_MSG_FAILED, "failed"),
  193                  isc_result_totext(result));
  194         result = ISC_R_UNEXPECTED;
  195         goto destroy_rcond;
  196     }
  197 
  198     rwl->magic = RWLOCK_MAGIC;
  199 
  200     return (ISC_R_SUCCESS);
  201 
  202   destroy_rcond:
  203     (void)isc_condition_destroy(&rwl->readable);
  204   destroy_lock:
  205     DESTROYLOCK(&rwl->lock);
  206 
  207     return (result);
  208 }
  209 
  210 void
  211 isc_rwlock_destroy(isc_rwlock_t *rwl) {
  212     REQUIRE(VALID_RWLOCK(rwl));
  213 
  214 #if defined(ISC_RWLOCK_USEATOMIC)
  215     REQUIRE(rwl_load(write_requests) == rwl_load(write_completions) &&
  216         rwl_load(cnt_and_flag) == 0 && rwl->readers_waiting == 0);
  217 #else
  218     LOCK(&rwl->lock);
  219     REQUIRE(rwl->active == 0 &&
  220         rwl->readers_waiting == 0 &&
  221         rwl->writers_waiting == 0);
  222     UNLOCK(&rwl->lock);
  223 #endif
  224 
  225     rwl->magic = 0;
  226     (void)isc_condition_destroy(&rwl->readable);
  227     (void)isc_condition_destroy(&rwl->writeable);
  228     DESTROYLOCK(&rwl->lock);
  229 }
  230 
  231 #if defined(ISC_RWLOCK_USEATOMIC)
  232 
  233 /*
  234  * When some architecture-dependent atomic operations are available,
  235  * rwlock can be more efficient than the generic algorithm defined below.
  236  * The basic algorithm is described in the following URL:
  237  *   http://www.cs.rochester.edu/u/scott/synchronization/pseudocode/rw.html
  238  *
  239  * The key is to use the following integer variables modified atomically:
  240  *   write_requests, write_completions, and cnt_and_flag.
  241  *
  242  * write_requests and write_completions act as a waiting queue for writers
  243  * in order to ensure the FIFO order.  Both variables begin with the initial
  244  * value of 0.  When a new writer tries to get a write lock, it increments
  245  * write_requests and gets the previous value of the variable as a "ticket".
  246  * When write_completions reaches the ticket number, the new writer can start
  247  * writing.  When the writer completes its work, it increments
  248  * write_completions so that another new writer can start working.  If the
  249  * write_requests is not equal to write_completions, it means a writer is now
  250  * working or waiting.  In this case, a new readers cannot start reading, or
  251  * in other words, this algorithm basically prefers writers.
  252  *
  253  * cnt_and_flag is a "lock" shared by all readers and writers.  This integer
  254  * variable is a kind of structure with two members: writer_flag (1 bit) and
  255  * reader_count (31 bits).  The writer_flag shows whether a writer is working,
  256  * and the reader_count shows the number of readers currently working or almost
  257  * ready for working.  A writer who has the current "ticket" tries to get the
  258  * lock by exclusively setting the writer_flag to 1, provided that the whole
  259  * 32-bit is 0 (meaning no readers or writers working).  On the other hand,
  260  * a new reader tries to increment the "reader_count" field provided that
  261  * the writer_flag is 0 (meaning there is no writer working).
  262  *
  263  * If some of the above operations fail, the reader or the writer sleeps
  264  * until the related condition changes.  When a working reader or writer
  265  * completes its work, some readers or writers are sleeping, and the condition
  266  * that suspended the reader or writer has changed, it wakes up the sleeping
  267  * readers or writers.
  268  *
  269  * As already noted, this algorithm basically prefers writers.  In order to
  270  * prevent readers from starving, however, the algorithm also introduces the
  271  * "writer quota" (Q).  When Q consecutive writers have completed their work,
  272  * suspending readers, the last writer will wake up the readers, even if a new
  273  * writer is waiting.
  274  *
  275  * Implementation specific note: due to the combination of atomic operations
  276  * and a mutex lock, ordering between the atomic operation and locks can be
  277  * very sensitive in some cases.  In particular, it is generally very important
  278  * to check the atomic variable that requires a reader or writer to sleep after
  279  * locking the mutex and before actually sleeping; otherwise, it could be very
  280  * likely to cause a deadlock.  For example, assume "var" is a variable
  281  * atomically modified, then the corresponding code would be:
  282  *  if (var == need_sleep) {
  283  *      LOCK(lock);
  284  *      if (var == need_sleep)
  285  *          WAIT(cond, lock);
  286  *      UNLOCK(lock);
  287  *  }
  288  * The second check is important, since "var" is protected by the atomic
  289  * operation, not by the mutex, and can be changed just before sleeping.
  290  * (The first "if" could be omitted, but this is also important in order to
  291  * make the code efficient by avoiding the use of the mutex unless it is
  292  * really necessary.)
  293  */
  294 
  295 #define WRITER_ACTIVE   0x1
  296 #define READER_INCR 0x2
  297 
  298 static isc_result_t
  299 isc__rwlock_lock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  300     int_fast32_t cntflag;
  301 
  302     REQUIRE(VALID_RWLOCK(rwl));
  303 
  304 #ifdef ISC_RWLOCK_TRACE
  305     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  306                   ISC_MSG_PRELOCK, "prelock"), rwl, type);
  307 #endif
  308 
  309     if (type == isc_rwlocktype_read) {
  310         if (rwl_load(write_requests) != rwl_load(write_completions)) {
  311             /* there is a waiting or active writer */
  312             LOCK(&rwl->lock);
  313             if (rwl_load(write_requests) !=
  314                 rwl_load(write_completions))
  315             {
  316                 rwl->readers_waiting++;
  317                 WAIT(&rwl->readable, &rwl->lock);
  318                 rwl->readers_waiting--;
  319             }
  320             UNLOCK(&rwl->lock);
  321         }
  322 
  323         cntflag = rwl_add(cnt_and_flag, READER_INCR);
  324 
  325         POST(cntflag);
  326         while (1) {
  327             if ((rwl_load(cnt_and_flag) & WRITER_ACTIVE) == 0) {
  328                 break;
  329             }
  330 
  331             /* A writer is still working */
  332             LOCK(&rwl->lock);
  333             rwl->readers_waiting++;
  334             if ((rwl_load(cnt_and_flag) & WRITER_ACTIVE) != 0) {
  335                 WAIT(&rwl->readable, &rwl->lock);
  336             }
  337             rwl->readers_waiting--;
  338             UNLOCK(&rwl->lock);
  339 
  340             /*
  341              * Typically, the reader should be able to get a lock
  342              * at this stage:
  343              *   (1) there should have been no pending writer when
  344              *       the reader was trying to increment the
  345              *       counter; otherwise, the writer should be in
  346              *       the waiting queue, preventing the reader from
  347              *       proceeding to this point.
  348              *   (2) once the reader increments the counter, no
  349              *       more writer can get a lock.
  350              * Still, it is possible another writer can work at
  351              * this point, e.g. in the following scenario:
  352              *   A previous writer unlocks the writer lock.
  353              *   This reader proceeds to point (1).
  354              *   A new writer appears, and gets a new lock before
  355              *   the reader increments the counter.
  356              *   The reader then increments the counter.
  357              *   The previous writer notices there is a waiting
  358              *   reader who is almost ready, and wakes it up.
  359              * So, the reader needs to confirm whether it can now
  360              * read explicitly (thus we loop).  Note that this is
  361              * not an infinite process, since the reader has
  362              * incremented the counter at this point.
  363              */
  364         }
  365 
  366         /*
  367          * If we are temporarily preferred to writers due to the writer
  368          * quota, reset the condition (race among readers doesn't
  369          * matter).
  370          */
  371         rwl_store(write_granted, 0);
  372     } else {
  373         int_fast32_t prev_writer;
  374 
  375         /* enter the waiting queue, and wait for our turn */
  376         prev_writer = rwl_add(write_requests, 1);
  377         while (rwl_load(write_completions) != prev_writer) {
  378             LOCK(&rwl->lock);
  379             if (rwl_load(write_completions) != prev_writer) {
  380                 WAIT(&rwl->writeable, &rwl->lock);
  381                 UNLOCK(&rwl->lock);
  382                 continue;
  383             }
  384             UNLOCK(&rwl->lock);
  385             break;
  386         }
  387 
  388         while (1) {
  389             int_fast32_t cntflag2 = 0;
  390             rwl_cmpxchg(cnt_and_flag, cntflag2, WRITER_ACTIVE);
  391 
  392             if (cntflag2 == 0)
  393                 break;
  394 
  395             /* Another active reader or writer is working. */
  396             LOCK(&rwl->lock);
  397             if (rwl_load(cnt_and_flag) != 0) {
  398                 WAIT(&rwl->writeable, &rwl->lock);
  399             }
  400             UNLOCK(&rwl->lock);
  401         }
  402 
  403         INSIST((rwl_load(cnt_and_flag) & WRITER_ACTIVE) != 0);
  404         rwl_add(write_granted, 1);
  405     }
  406 
  407 #ifdef ISC_RWLOCK_TRACE
  408     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  409                   ISC_MSG_POSTLOCK, "postlock"), rwl, type);
  410 #endif
  411 
  412     return (ISC_R_SUCCESS);
  413 }
  414 
  415 isc_result_t
  416 isc_rwlock_lock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  417     int_fast32_t cnt = 0;
  418     int_fast32_t max_cnt = rwl_load(spins) * 2 + 10;
  419     isc_result_t result = ISC_R_SUCCESS;
  420 
  421     if (max_cnt > RWLOCK_MAX_ADAPTIVE_COUNT)
  422         max_cnt = RWLOCK_MAX_ADAPTIVE_COUNT;
  423 
  424     do {
  425         if (cnt++ >= max_cnt) {
  426             result = isc__rwlock_lock(rwl, type);
  427             break;
  428         }
  429 #ifdef ISC_PLATFORM_BUSYWAITNOP
  430         ISC_PLATFORM_BUSYWAITNOP;
  431 #endif
  432     } while (isc_rwlock_trylock(rwl, type) != ISC_R_SUCCESS);
  433 
  434     rwl_add(spins, (cnt - rwl_load(spins)) / 8);
  435 
  436     return (result);
  437 }
  438 
  439 isc_result_t
  440 isc_rwlock_trylock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  441     int_fast32_t cntflag;
  442 
  443     REQUIRE(VALID_RWLOCK(rwl));
  444 
  445 #ifdef ISC_RWLOCK_TRACE
  446     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  447                   ISC_MSG_PRELOCK, "prelock"), rwl, type);
  448 #endif
  449 
  450     if (type == isc_rwlocktype_read) {
  451         /* If a writer is waiting or working, we fail. */
  452         if (rwl_load(write_requests) != rwl_load(write_completions)) {
  453             return (ISC_R_LOCKBUSY);
  454         }
  455 
  456         /* Otherwise, be ready for reading. */
  457         cntflag = rwl_add(cnt_and_flag, READER_INCR);
  458         if ((cntflag & WRITER_ACTIVE) != 0) {
  459             /*
  460              * A writer is working.  We lose, and cancel the read
  461              * request.
  462              */
  463             cntflag = rwl_sub(cnt_and_flag, READER_INCR);
  464             /*
  465              * If no other readers are waiting and we've suspended
  466              * new writers in this short period, wake them up.
  467              */
  468             if (cntflag == READER_INCR &&
  469                 rwl_load(write_completions) !=
  470                 rwl_load(write_requests))
  471             {
  472                 LOCK(&rwl->lock);
  473                 BROADCAST(&rwl->writeable);
  474                 UNLOCK(&rwl->lock);
  475             }
  476 
  477             return (ISC_R_LOCKBUSY);
  478         }
  479     } else {
  480         /* Try locking without entering the waiting queue. */
  481         int_fast32_t zero = 0;
  482         rwl_cmpxchg(cnt_and_flag, zero, WRITER_ACTIVE);
  483 
  484         if (zero != 0) {
  485             return (ISC_R_LOCKBUSY);
  486         }
  487 
  488         /*
  489          * XXXJT: jump into the queue, possibly breaking the writer
  490          * order.
  491          */
  492         rwl_sub(write_completions, 1);
  493         rwl_add(write_granted, 1);
  494     }
  495 
  496 #ifdef ISC_RWLOCK_TRACE
  497     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  498                   ISC_MSG_POSTLOCK, "postlock"), rwl, type);
  499 #endif
  500 
  501     return (ISC_R_SUCCESS);
  502 }
  503 
  504 isc_result_t
  505 isc_rwlock_tryupgrade(isc_rwlock_t *rwl) {
  506     int_fast32_t reader_incr = READER_INCR;
  507 
  508     REQUIRE(VALID_RWLOCK(rwl));
  509 
  510     /* Try to acquire write access. */
  511     rwl_cmpxchg(cnt_and_flag, reader_incr, WRITER_ACTIVE);
  512 
  513     /*
  514      * There must have been no writer, and there must have
  515      * been at least one reader.
  516      */
  517     INSIST((reader_incr & WRITER_ACTIVE) == 0 &&
  518            (reader_incr & ~WRITER_ACTIVE) != 0);
  519 
  520     if (reader_incr == READER_INCR) {
  521         /*
  522          * We are the only reader and have been upgraded.
  523          * Now jump into the head of the writer waiting queue.
  524          */
  525         rwl_sub(write_completions, 1);
  526     } else {
  527         return (ISC_R_LOCKBUSY);
  528     }
  529 
  530     return (ISC_R_SUCCESS);
  531 }
  532 
  533 void
  534 isc_rwlock_downgrade(isc_rwlock_t *rwl) {
  535     int_fast32_t prev_readers;
  536 
  537     REQUIRE(VALID_RWLOCK(rwl));
  538 
  539     /* Become an active reader. */
  540     prev_readers = rwl_add(cnt_and_flag, READER_INCR);
  541 
  542     /* We must have been a writer. */
  543     INSIST((prev_readers & WRITER_ACTIVE) != 0);
  544 
  545     /* Complete write */
  546     rwl_sub(cnt_and_flag, WRITER_ACTIVE);
  547     rwl_add(write_completions, 1);
  548 
  549     /* Resume other readers */
  550     LOCK(&rwl->lock);
  551     if (rwl->readers_waiting > 0) {
  552         BROADCAST(&rwl->readable);
  553     }
  554     UNLOCK(&rwl->lock);
  555 }
  556 
  557 isc_result_t
  558 isc_rwlock_unlock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  559     int_fast32_t prev_cnt;
  560 
  561     REQUIRE(VALID_RWLOCK(rwl));
  562 
  563 #ifdef ISC_RWLOCK_TRACE
  564     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  565                   ISC_MSG_PREUNLOCK, "preunlock"), rwl, type);
  566 #endif
  567 
  568     if (type == isc_rwlocktype_read) {
  569         prev_cnt = rwl_sub(cnt_and_flag, READER_INCR);
  570         /*
  571          * If we're the last reader and any writers are waiting, wake
  572          * them up.  We need to wake up all of them to ensure the
  573          * FIFO order.
  574          */
  575         if (prev_cnt == READER_INCR &&
  576             rwl_load(write_completions) !=
  577             rwl_load(write_requests))
  578         {
  579             LOCK(&rwl->lock);
  580             BROADCAST(&rwl->writeable);
  581             UNLOCK(&rwl->lock);
  582         }
  583     } else {
  584         bool wakeup_writers = true;
  585 
  586         /*
  587          * Reset the flag, and (implicitly) tell other writers
  588          * we are done.
  589          */
  590         rwl_sub(cnt_and_flag, WRITER_ACTIVE);
  591         rwl_add(write_completions, 1);
  592 
  593         if ((uint32_t)rwl_load(write_granted) >= rwl->write_quota ||
  594             rwl_load(write_requests) == rwl_load(write_completions) ||
  595             (rwl_load(cnt_and_flag) & ~WRITER_ACTIVE) != 0) {
  596             /*
  597              * We have passed the write quota, no writer is
  598              * waiting, or some readers are almost ready, pending
  599              * possible writers.  Note that the last case can
  600              * happen even if write_requests != write_completions
  601              * (which means a new writer in the queue), so we need
  602              * to catch the case explicitly.
  603              */
  604             LOCK(&rwl->lock);
  605             if (rwl->readers_waiting > 0) {
  606                 wakeup_writers = false;
  607                 BROADCAST(&rwl->readable);
  608             }
  609             UNLOCK(&rwl->lock);
  610         }
  611 
  612         if (rwl_load(write_requests) != rwl_load(write_completions) &&
  613             wakeup_writers) {
  614             LOCK(&rwl->lock);
  615             BROADCAST(&rwl->writeable);
  616             UNLOCK(&rwl->lock);
  617         }
  618     }
  619 
  620 #ifdef ISC_RWLOCK_TRACE
  621     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  622                   ISC_MSG_POSTUNLOCK, "postunlock"),
  623            rwl, type);
  624 #endif
  625 
  626     return (ISC_R_SUCCESS);
  627 }
  628 
  629 #else /* ISC_RWLOCK_USEATOMIC */
  630 
  631 static isc_result_t
  632 doit(isc_rwlock_t *rwl, isc_rwlocktype_t type, bool nonblock) {
  633     bool skip = false;
  634     bool done = false;
  635     isc_result_t result = ISC_R_SUCCESS;
  636 
  637     REQUIRE(VALID_RWLOCK(rwl));
  638 
  639     LOCK(&rwl->lock);
  640 
  641 #ifdef ISC_RWLOCK_TRACE
  642     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  643                   ISC_MSG_PRELOCK, "prelock"), rwl, type);
  644 #endif
  645 
  646     if (type == isc_rwlocktype_read) {
  647         if (rwl->readers_waiting != 0)
  648             skip = true;
  649         while (!done) {
  650             if (!skip &&
  651                 ((rwl->active == 0 ||
  652                   (rwl->type == isc_rwlocktype_read &&
  653                    (rwl->writers_waiting == 0 ||
  654                 rwl->granted < rwl->read_quota)))))
  655             {
  656                 rwl->type = isc_rwlocktype_read;
  657                 rwl->active++;
  658                 rwl->granted++;
  659                 done = true;
  660             } else if (nonblock) {
  661                 result = ISC_R_LOCKBUSY;
  662                 done = true;
  663             } else {
  664                 skip = false;
  665                 rwl->readers_waiting++;
  666                 WAIT(&rwl->readable, &rwl->lock);
  667                 rwl->readers_waiting--;
  668             }
  669         }
  670     } else {
  671         if (rwl->writers_waiting != 0)
  672             skip = true;
  673         while (!done) {
  674             if (!skip && rwl->active == 0) {
  675                 rwl->type = isc_rwlocktype_write;
  676                 rwl->active = 1;
  677                 rwl->granted++;
  678                 done = true;
  679             } else if (nonblock) {
  680                 result = ISC_R_LOCKBUSY;
  681                 done = true;
  682             } else {
  683                 skip = false;
  684                 rwl->writers_waiting++;
  685                 WAIT(&rwl->writeable, &rwl->lock);
  686                 rwl->writers_waiting--;
  687             }
  688         }
  689     }
  690 
  691 #ifdef ISC_RWLOCK_TRACE
  692     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  693                   ISC_MSG_POSTLOCK, "postlock"), rwl, type);
  694 #endif
  695 
  696     UNLOCK(&rwl->lock);
  697 
  698     return (result);
  699 }
  700 
  701 isc_result_t
  702 isc_rwlock_lock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  703     int_fast32_t cnt = 0;
  704     int_fast32_t max_cnt = rwl_load(spins) * 2 + 10;
  705     isc_result_t result = ISC_R_SUCCESS;
  706 
  707     if (max_cnt > RWLOCK_MAX_ADAPTIVE_COUNT)
  708         max_cnt = RWLOCK_MAX_ADAPTIVE_COUNT;
  709 
  710     do {
  711         if (cnt++ >= max_cnt) {
  712             result = doit(rwl, type, false);
  713             break;
  714         }
  715 #ifdef ISC_PLATFORM_BUSYWAITNOP
  716         ISC_PLATFORM_BUSYWAITNOP;
  717 #endif
  718     } while (doit(rwl, type, true) != ISC_R_SUCCESS);
  719 
  720     rwl_add(spins, (cnt - rwl_load(spins)) /8);
  721 
  722     return (result);
  723 }
  724 
  725 isc_result_t
  726 isc_rwlock_trylock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  727     return (doit(rwl, type, true));
  728 }
  729 
  730 isc_result_t
  731 isc_rwlock_tryupgrade(isc_rwlock_t *rwl) {
  732     isc_result_t result = ISC_R_SUCCESS;
  733 
  734     REQUIRE(VALID_RWLOCK(rwl));
  735     LOCK(&rwl->lock);
  736     REQUIRE(rwl->type == isc_rwlocktype_read);
  737     REQUIRE(rwl->active != 0);
  738 
  739     /* If we are the only reader then succeed. */
  740     if (rwl->active == 1) {
  741         rwl->original = (rwl->original == isc_rwlocktype_none) ?
  742                 isc_rwlocktype_read : isc_rwlocktype_none;
  743         rwl->type = isc_rwlocktype_write;
  744     } else
  745         result = ISC_R_LOCKBUSY;
  746 
  747     UNLOCK(&rwl->lock);
  748     return (result);
  749 }
  750 
  751 void
  752 isc_rwlock_downgrade(isc_rwlock_t *rwl) {
  753 
  754     REQUIRE(VALID_RWLOCK(rwl));
  755     LOCK(&rwl->lock);
  756     REQUIRE(rwl->type == isc_rwlocktype_write);
  757     REQUIRE(rwl->active == 1);
  758 
  759     rwl->type = isc_rwlocktype_read;
  760     rwl->original = (rwl->original == isc_rwlocktype_none) ?
  761             isc_rwlocktype_write : isc_rwlocktype_none;
  762     /*
  763      * Resume processing any read request that were blocked when
  764      * we upgraded.
  765      */
  766     if (rwl->original == isc_rwlocktype_none &&
  767         (rwl->writers_waiting == 0 || rwl->granted < rwl->read_quota) &&
  768         rwl->readers_waiting > 0)
  769         BROADCAST(&rwl->readable);
  770 
  771     UNLOCK(&rwl->lock);
  772 }
  773 
  774 isc_result_t
  775 isc_rwlock_unlock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  776 
  777     REQUIRE(VALID_RWLOCK(rwl));
  778     LOCK(&rwl->lock);
  779     REQUIRE(rwl->type == type);
  780 
  781     UNUSED(type);
  782 
  783 #ifdef ISC_RWLOCK_TRACE
  784     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  785                   ISC_MSG_PREUNLOCK, "preunlock"), rwl, type);
  786 #endif
  787 
  788     INSIST(rwl->active > 0);
  789     rwl->active--;
  790     if (rwl->active == 0) {
  791         if (rwl->original != isc_rwlocktype_none) {
  792             rwl->type = rwl->original;
  793             rwl->original = isc_rwlocktype_none;
  794         }
  795         if (rwl->type == isc_rwlocktype_read) {
  796             rwl->granted = 0;
  797             if (rwl->writers_waiting > 0) {
  798                 rwl->type = isc_rwlocktype_write;
  799                 SIGNAL(&rwl->writeable);
  800             } else if (rwl->readers_waiting > 0) {
  801                 /* Does this case ever happen? */
  802                 BROADCAST(&rwl->readable);
  803             }
  804         } else {
  805             if (rwl->readers_waiting > 0) {
  806                 if (rwl->writers_waiting > 0 &&
  807                     rwl->granted < rwl->write_quota) {
  808                     SIGNAL(&rwl->writeable);
  809                 } else {
  810                     rwl->granted = 0;
  811                     rwl->type = isc_rwlocktype_read;
  812                     BROADCAST(&rwl->readable);
  813                 }
  814             } else if (rwl->writers_waiting > 0) {
  815                 rwl->granted = 0;
  816                 SIGNAL(&rwl->writeable);
  817             } else {
  818                 rwl->granted = 0;
  819             }
  820         }
  821     }
  822     INSIST(rwl->original == isc_rwlocktype_none);
  823 
  824 #ifdef ISC_RWLOCK_TRACE
  825     print_lock(isc_msgcat_get(isc_msgcat, ISC_MSGSET_RWLOCK,
  826                   ISC_MSG_POSTUNLOCK, "postunlock"),
  827            rwl, type);
  828 #endif
  829 
  830     UNLOCK(&rwl->lock);
  831 
  832     return (ISC_R_SUCCESS);
  833 }
  834 
  835 #endif /* ISC_RWLOCK_USEATOMIC */
  836 #else /* ISC_PLATFORM_USETHREADS */
  837 
  838 isc_result_t
  839 isc_rwlock_init(isc_rwlock_t *rwl, unsigned int read_quota,
  840         unsigned int write_quota)
  841 {
  842     REQUIRE(rwl != NULL);
  843 
  844     UNUSED(read_quota);
  845     UNUSED(write_quota);
  846 
  847     rwl->type = isc_rwlocktype_read;
  848     rwl->active = 0;
  849     rwl->magic = RWLOCK_MAGIC;
  850 
  851     return (ISC_R_SUCCESS);
  852 }
  853 
  854 isc_result_t
  855 isc_rwlock_lock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  856     REQUIRE(VALID_RWLOCK(rwl));
  857 
  858     if (type == isc_rwlocktype_read) {
  859         if (rwl->type != isc_rwlocktype_read && rwl->active != 0)
  860             return (ISC_R_LOCKBUSY);
  861         rwl->type = isc_rwlocktype_read;
  862         rwl->active++;
  863     } else {
  864         if (rwl->active != 0)
  865             return (ISC_R_LOCKBUSY);
  866         rwl->type = isc_rwlocktype_write;
  867         rwl->active = 1;
  868     }
  869     return (ISC_R_SUCCESS);
  870 }
  871 
  872 isc_result_t
  873 isc_rwlock_trylock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  874     return (isc_rwlock_lock(rwl, type));
  875 }
  876 
  877 isc_result_t
  878 isc_rwlock_tryupgrade(isc_rwlock_t *rwl) {
  879     isc_result_t result = ISC_R_SUCCESS;
  880 
  881     REQUIRE(VALID_RWLOCK(rwl));
  882     REQUIRE(rwl->type == isc_rwlocktype_read);
  883     REQUIRE(rwl->active != 0);
  884 
  885     /* If we are the only reader then succeed. */
  886     if (rwl->active == 1)
  887         rwl->type = isc_rwlocktype_write;
  888     else
  889         result = ISC_R_LOCKBUSY;
  890     return (result);
  891 }
  892 
  893 void
  894 isc_rwlock_downgrade(isc_rwlock_t *rwl) {
  895 
  896     REQUIRE(VALID_RWLOCK(rwl));
  897     REQUIRE(rwl->type == isc_rwlocktype_write);
  898     REQUIRE(rwl->active == 1);
  899 
  900     rwl->type = isc_rwlocktype_read;
  901 }
  902 
  903 isc_result_t
  904 isc_rwlock_unlock(isc_rwlock_t *rwl, isc_rwlocktype_t type) {
  905     REQUIRE(VALID_RWLOCK(rwl));
  906     REQUIRE(rwl->type == type);
  907 
  908     UNUSED(type);
  909 
  910     INSIST(rwl->active > 0);
  911     rwl->active--;
  912 
  913     return (ISC_R_SUCCESS);
  914 }
  915 
  916 void
  917 isc_rwlock_destroy(isc_rwlock_t *rwl) {
  918     REQUIRE(rwl != NULL);
  919     REQUIRE(rwl->active == 0);
  920     rwl->magic = 0;
  921 }
  922 
  923 #endif /* ISC_PLATFORM_USETHREADS */