1/*
2 * CDDL HEADER START
3 *
4 * The contents of this file are subject to the terms of the
5 * Common Development and Distribution License (the "License").
6 * You may not use this file except in compliance with the License.
7 *
8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9 * or http://www.opensolaris.org/os/licensing.
10 * See the License for the specific language governing permissions
11 * and limitations under the License.
12 *
13 * When distributing Covered Code, include this CDDL HEADER in each
14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15 * If applicable, add the following below this CDDL HEADER, with the
16 * fields enclosed by brackets "[]" replaced with your own identifying
17 * information: Portions Copyright [yyyy] [name of copyright owner]
18 *
19 * CDDL HEADER END
20 */
21/*
22 * Copyright 2009 Sun Microsystems, Inc.  All rights reserved.
23 * Use is subject to license terms.
24 */
25/*
26 * Copyright (c) 2012 by Delphix. All rights reserved.
27 */
28
29#include <sys/refcount.h>
30#include <sys/rrwlock.h>
31
32/*
33 * This file contains the implementation of a re-entrant read
34 * reader/writer lock (aka "rrwlock").
35 *
36 * This is a normal reader/writer lock with the additional feature
37 * of allowing threads who have already obtained a read lock to
38 * re-enter another read lock (re-entrant read) - even if there are
39 * waiting writers.
40 *
41 * Callers who have not obtained a read lock give waiting writers priority.
42 *
43 * The rrwlock_t lock does not allow re-entrant writers, nor does it
44 * allow a re-entrant mix of reads and writes (that is, it does not
45 * allow a caller who has already obtained a read lock to be able to
46 * then grab a write lock without first dropping all read locks, and
47 * vice versa).
48 *
49 * The rrwlock_t uses tsd (thread specific data) to keep a list of
50 * nodes (rrw_node_t), where each node keeps track of which specific
51 * lock (rrw_node_t::rn_rrl) the thread has grabbed.  Since re-entering
52 * should be rare, a thread that grabs multiple reads on the same rrwlock_t
53 * will store multiple rrw_node_ts of the same 'rrn_rrl'. Nodes on the
54 * tsd list can represent a different rrwlock_t.  This allows a thread
55 * to enter multiple and unique rrwlock_ts for read locks at the same time.
56 *
57 * Since using tsd exposes some overhead, the rrwlock_t only needs to
58 * keep tsd data when writers are waiting.  If no writers are waiting, then
59 * a reader just bumps the anonymous read count (rr_anon_rcount) - no tsd
60 * is needed.  Once a writer attempts to grab the lock, readers then
61 * keep tsd data and bump the linked readers count (rr_linked_rcount).
62 *
63 * If there are waiting writers and there are anonymous readers, then a
64 * reader doesn't know if it is a re-entrant lock. But since it may be one,
65 * we allow the read to proceed (otherwise it could deadlock).  Since once
66 * waiting writers are active, readers no longer bump the anonymous count,
67 * the anonymous readers will eventually flush themselves out.  At this point,
68 * readers will be able to tell if they are a re-entrant lock (have a
69 * rrw_node_t entry for the lock) or not. If they are a re-entrant lock, then
70 * we must let the proceed.  If they are not, then the reader blocks for the
71 * waiting writers.  Hence, we do not starve writers.
72 */
73
74/* global key for TSD */
75uint_t rrw_tsd_key;
76
77typedef struct rrw_node {
78	struct rrw_node *rn_next;
79	rrwlock_t *rn_rrl;
80	void *rn_tag;
81} rrw_node_t;
82
83static rrw_node_t *
84rrn_find(rrwlock_t *rrl)
85{
86	rrw_node_t *rn;
87
88	if (refcount_count(&rrl->rr_linked_rcount) == 0)
89		return (NULL);
90
91	for (rn = tsd_get(rrw_tsd_key); rn != NULL; rn = rn->rn_next) {
92		if (rn->rn_rrl == rrl)
93			return (rn);
94	}
95	return (NULL);
96}
97
98/*
99 * Add a node to the head of the singly linked list.
100 */
101static void
102rrn_add(rrwlock_t *rrl, void *tag)
103{
104	rrw_node_t *rn;
105
106	rn = kmem_alloc(sizeof (*rn), KM_SLEEP);
107	rn->rn_rrl = rrl;
108	rn->rn_next = tsd_get(rrw_tsd_key);
109	rn->rn_tag = tag;
110	VERIFY(tsd_set(rrw_tsd_key, rn) == 0);
111}
112
113/*
114 * If a node is found for 'rrl', then remove the node from this
115 * thread's list and return TRUE; otherwise return FALSE.
116 */
117static boolean_t
118rrn_find_and_remove(rrwlock_t *rrl, void *tag)
119{
120	rrw_node_t *rn;
121	rrw_node_t *prev = NULL;
122
123	if (refcount_count(&rrl->rr_linked_rcount) == 0)
124		return (B_FALSE);
125
126	for (rn = tsd_get(rrw_tsd_key); rn != NULL; rn = rn->rn_next) {
127		if (rn->rn_rrl == rrl && rn->rn_tag == tag) {
128			if (prev)
129				prev->rn_next = rn->rn_next;
130			else
131				VERIFY(tsd_set(rrw_tsd_key, rn->rn_next) == 0);
132			kmem_free(rn, sizeof (*rn));
133			return (B_TRUE);
134		}
135		prev = rn;
136	}
137	return (B_FALSE);
138}
139
140void
141rrw_init(rrwlock_t *rrl, boolean_t track_all)
142{
143	mutex_init(&rrl->rr_lock, NULL, MUTEX_DEFAULT, NULL);
144	cv_init(&rrl->rr_cv, NULL, CV_DEFAULT, NULL);
145	rrl->rr_writer = NULL;
146	refcount_create(&rrl->rr_anon_rcount);
147	refcount_create(&rrl->rr_linked_rcount);
148	rrl->rr_writer_wanted = B_FALSE;
149	rrl->rr_track_all = track_all;
150}
151
152void
153rrw_destroy(rrwlock_t *rrl)
154{
155	mutex_destroy(&rrl->rr_lock);
156	cv_destroy(&rrl->rr_cv);
157	ASSERT(rrl->rr_writer == NULL);
158	refcount_destroy(&rrl->rr_anon_rcount);
159	refcount_destroy(&rrl->rr_linked_rcount);
160}
161
162void
163rrw_enter_read(rrwlock_t *rrl, void *tag)
164{
165	mutex_enter(&rrl->rr_lock);
166#if !defined(DEBUG) && defined(_KERNEL)
167	if (rrl->rr_writer == NULL && !rrl->rr_writer_wanted &&
168	    !rrl->rr_track_all) {
169		rrl->rr_anon_rcount.rc_count++;
170		mutex_exit(&rrl->rr_lock);
171		return;
172	}
173	DTRACE_PROBE(zfs__rrwfastpath__rdmiss);
174#endif
175	ASSERT(rrl->rr_writer != curthread);
176	ASSERT(refcount_count(&rrl->rr_anon_rcount) >= 0);
177
178	while (rrl->rr_writer != NULL || (rrl->rr_writer_wanted &&
179	    refcount_is_zero(&rrl->rr_anon_rcount) &&
180	    rrn_find(rrl) == NULL))
181		cv_wait(&rrl->rr_cv, &rrl->rr_lock);
182
183	if (rrl->rr_writer_wanted || rrl->rr_track_all) {
184		/* may or may not be a re-entrant enter */
185		rrn_add(rrl, tag);
186		(void) refcount_add(&rrl->rr_linked_rcount, tag);
187	} else {
188		(void) refcount_add(&rrl->rr_anon_rcount, tag);
189	}
190	ASSERT(rrl->rr_writer == NULL);
191	mutex_exit(&rrl->rr_lock);
192}
193
194void
195rrw_enter_write(rrwlock_t *rrl)
196{
197	mutex_enter(&rrl->rr_lock);
198	ASSERT(rrl->rr_writer != curthread);
199
200	while (refcount_count(&rrl->rr_anon_rcount) > 0 ||
201	    refcount_count(&rrl->rr_linked_rcount) > 0 ||
202	    rrl->rr_writer != NULL) {
203		rrl->rr_writer_wanted = B_TRUE;
204		cv_wait(&rrl->rr_cv, &rrl->rr_lock);
205	}
206	rrl->rr_writer_wanted = B_FALSE;
207	rrl->rr_writer = curthread;
208	mutex_exit(&rrl->rr_lock);
209}
210
211void
212rrw_enter(rrwlock_t *rrl, krw_t rw, void *tag)
213{
214	if (rw == RW_READER)
215		rrw_enter_read(rrl, tag);
216	else
217		rrw_enter_write(rrl);
218}
219
220void
221rrw_exit(rrwlock_t *rrl, void *tag)
222{
223	mutex_enter(&rrl->rr_lock);
224#if !defined(DEBUG) && defined(_KERNEL)
225	if (!rrl->rr_writer && rrl->rr_linked_rcount.rc_count == 0) {
226		rrl->rr_anon_rcount.rc_count--;
227		if (rrl->rr_anon_rcount.rc_count == 0)
228			cv_broadcast(&rrl->rr_cv);
229		mutex_exit(&rrl->rr_lock);
230		return;
231	}
232	DTRACE_PROBE(zfs__rrwfastpath__exitmiss);
233#endif
234	ASSERT(!refcount_is_zero(&rrl->rr_anon_rcount) ||
235	    !refcount_is_zero(&rrl->rr_linked_rcount) ||
236	    rrl->rr_writer != NULL);
237
238	if (rrl->rr_writer == NULL) {
239		int64_t count;
240		if (rrn_find_and_remove(rrl, tag)) {
241			count = refcount_remove(&rrl->rr_linked_rcount, tag);
242		} else {
243			ASSERT(!rrl->rr_track_all);
244			count = refcount_remove(&rrl->rr_anon_rcount, tag);
245		}
246		if (count == 0)
247			cv_broadcast(&rrl->rr_cv);
248	} else {
249		ASSERT(rrl->rr_writer == curthread);
250		ASSERT(refcount_is_zero(&rrl->rr_anon_rcount) &&
251		    refcount_is_zero(&rrl->rr_linked_rcount));
252		rrl->rr_writer = NULL;
253		cv_broadcast(&rrl->rr_cv);
254	}
255	mutex_exit(&rrl->rr_lock);
256}
257
258/*
259 * If the lock was created with track_all, rrw_held(RW_READER) will return
260 * B_TRUE iff the current thread has the lock for reader.  Otherwise it may
261 * return B_TRUE if any thread has the lock for reader.
262 */
263boolean_t
264rrw_held(rrwlock_t *rrl, krw_t rw)
265{
266	boolean_t held;
267
268	mutex_enter(&rrl->rr_lock);
269	if (rw == RW_WRITER) {
270		held = (rrl->rr_writer == curthread);
271	} else {
272		held = (!refcount_is_zero(&rrl->rr_anon_rcount) ||
273		    rrn_find(rrl) != NULL);
274	}
275	mutex_exit(&rrl->rr_lock);
276
277	return (held);
278}
279
280void
281rrw_tsd_destroy(void *arg)
282{
283	rrw_node_t *rn = arg;
284	if (rn != NULL) {
285		panic("thread %p terminating with rrw lock %p held",
286		    (void *)curthread, (void *)rn->rn_rrl);
287	}
288}
289