r19401: make tdb_lockall() much more efficient, and add a tdb_lockall_read()
[ira/wip.git] / source / lib / tdb / common / lock.c
1  /* 
2    Unix SMB/CIFS implementation.
3
4    trivial database library
5
6    Copyright (C) Andrew Tridgell              1999-2005
7    Copyright (C) Paul `Rusty' Russell              2000
8    Copyright (C) Jeremy Allison                    2000-2003
9    
10      ** NOTE! The following LGPL license applies to the tdb
11      ** library. This does NOT imply that all of Samba is released
12      ** under the LGPL
13    
14    This library is free software; you can redistribute it and/or
15    modify it under the terms of the GNU Lesser General Public
16    License as published by the Free Software Foundation; either
17    version 2 of the License, or (at your option) any later version.
18
19    This library is distributed in the hope that it will be useful,
20    but WITHOUT ANY WARRANTY; without even the implied warranty of
21    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
22    Lesser General Public License for more details.
23
24    You should have received a copy of the GNU Lesser General Public
25    License along with this library; if not, write to the Free Software
26    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
27 */
28
29 #include "tdb_private.h"
30
31 /* a byte range locking function - return 0 on success
32    this functions locks/unlocks 1 byte at the specified offset.
33
34    On error, errno is also set so that errors are passed back properly
35    through tdb_open(). 
36
37    note that a len of zero means lock to end of file
38 */
39 int tdb_brlock(struct tdb_context *tdb, tdb_off_t offset, 
40                int rw_type, int lck_type, int probe, size_t len)
41 {
42         struct flock fl;
43         int ret;
44
45         if (tdb->flags & TDB_NOLOCK) {
46                 return 0;
47         }
48
49         if ((rw_type == F_WRLCK) && (tdb->read_only || tdb->traverse_read)) {
50                 tdb->ecode = TDB_ERR_RDONLY;
51                 return -1;
52         }
53
54         fl.l_type = rw_type;
55         fl.l_whence = SEEK_SET;
56         fl.l_start = offset;
57         fl.l_len = len;
58         fl.l_pid = 0;
59
60         do {
61                 ret = fcntl(tdb->fd,lck_type,&fl);
62         } while (ret == -1 && errno == EINTR);
63
64         if (ret == -1) {
65                 /* Generic lock error. errno set by fcntl.
66                  * EAGAIN is an expected return from non-blocking
67                  * locks. */
68                 if (!probe && lck_type != F_SETLK) {
69                         /* Ensure error code is set for log fun to examine. */
70                         tdb->ecode = TDB_ERR_LOCK;
71                         TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock failed (fd=%d) at offset %d rw_type=%d lck_type=%d len=%d\n", 
72                                  tdb->fd, offset, rw_type, lck_type, (int)len));
73                 }
74                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
75         }
76         return 0;
77 }
78
79
80 /*
81   upgrade a read lock to a write lock. This needs to be handled in a
82   special way as some OSes (such as solaris) have too conservative
83   deadlock detection and claim a deadlock when progress can be
84   made. For those OSes we may loop for a while.  
85 */
86 int tdb_brlock_upgrade(struct tdb_context *tdb, tdb_off_t offset, size_t len)
87 {
88         int count = 1000;
89         while (count--) {
90                 struct timeval tv;
91                 if (tdb_brlock(tdb, offset, F_WRLCK, F_SETLKW, 1, len) == 0) {
92                         return 0;
93                 }
94                 if (errno != EDEADLK) {
95                         break;
96                 }
97                 /* sleep for as short a time as we can - more portable than usleep() */
98                 tv.tv_sec = 0;
99                 tv.tv_usec = 1;
100                 select(0, NULL, NULL, NULL, &tv);
101         }
102         TDB_LOG((tdb, TDB_DEBUG_TRACE,"tdb_brlock_upgrade failed at offset %d\n", offset));
103         return -1;
104 }
105
106
107 /* lock a list in the database. list -1 is the alloc list */
108 int tdb_lock(struct tdb_context *tdb, int list, int ltype)
109 {
110         /* a global lock allows us to avoid per chain locks */
111         if (tdb->global_lock.count && 
112             (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
113                 return 0;
114         }
115
116         if (tdb->global_lock.count) {
117                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
118         }
119
120         if (list < -1 || list >= (int)tdb->header.hash_size) {
121                 TDB_LOG((tdb, TDB_DEBUG_ERROR,"tdb_lock: invalid list %d for ltype=%d\n", 
122                            list, ltype));
123                 return -1;
124         }
125         if (tdb->flags & TDB_NOLOCK)
126                 return 0;
127
128         /* Since fcntl locks don't nest, we do a lock for the first one,
129            and simply bump the count for future ones */
130         if (tdb->locked[list+1].count == 0) {
131                 if (tdb->methods->tdb_brlock(tdb,FREELIST_TOP+4*list,ltype,F_SETLKW, 0, 1)) {
132                         TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lock failed on list %d ltype=%d (%s)\n", 
133                                  list, ltype, strerror(errno)));
134                         return -1;
135                 }
136                 tdb->locked[list+1].ltype = ltype;
137                 tdb->num_locks++;
138         }
139         tdb->locked[list+1].count++;
140         return 0;
141 }
142
143 /* unlock the database: returns void because it's too late for errors. */
144         /* changed to return int it may be interesting to know there
145            has been an error  --simo */
146 int tdb_unlock(struct tdb_context *tdb, int list, int ltype)
147 {
148         int ret = -1;
149
150         /* a global lock allows us to avoid per chain locks */
151         if (tdb->global_lock.count && 
152             (ltype == tdb->global_lock.ltype || ltype == F_RDLCK)) {
153                 return 0;
154         }
155
156         if (tdb->global_lock.count) {
157                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
158         }
159
160         if (tdb->flags & TDB_NOLOCK)
161                 return 0;
162
163         /* Sanity checks */
164         if (list < -1 || list >= (int)tdb->header.hash_size) {
165                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: list %d invalid (%d)\n", list, tdb->header.hash_size));
166                 return ret;
167         }
168
169         if (tdb->locked[list+1].count==0) {
170                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: count is 0\n"));
171                 return ret;
172         }
173
174         if (tdb->locked[list+1].count == 1) {
175                 /* Down to last nested lock: unlock underneath */
176                 ret = tdb->methods->tdb_brlock(tdb, FREELIST_TOP+4*list, F_UNLCK, F_SETLKW, 0, 1);
177                 tdb->num_locks--;
178         } else {
179                 ret = 0;
180         }
181         tdb->locked[list+1].count--;
182
183         if (ret)
184                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlock: An error occurred unlocking!\n")); 
185         return ret;
186 }
187
188
189
190 /* lock/unlock entire database */
191 static int _tdb_lockall(struct tdb_context *tdb, int ltype)
192 {
193         /* There are no locks on read-only dbs */
194         if (tdb->read_only || tdb->traverse_read)
195                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
196
197         if (tdb->global_lock.count && tdb->global_lock.ltype == ltype) {
198                 tdb->global_lock.count++;
199                 return 0;
200         }
201
202         if (tdb->global_lock.count) {
203                 /* a global lock of a different type exists */
204                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
205         }
206         
207         if (tdb->num_locks != 0) {
208                 /* can't combine global and chain locks */
209                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
210         }
211
212         if (tdb->methods->tdb_brlock(tdb, FREELIST_TOP, ltype, F_SETLKW, 
213                                      0, 4*tdb->header.hash_size)) {
214                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_lockall failed (%s)\n", strerror(errno)));
215                 return -1;
216         }
217
218         tdb->global_lock.count = 1;
219         tdb->global_lock.ltype = ltype;
220
221         return 0;
222 }
223
224 /* unlock entire db */
225 static int _tdb_unlockall(struct tdb_context *tdb, int ltype)
226 {
227         /* There are no locks on read-only dbs */
228         if (tdb->read_only || tdb->traverse_read) {
229                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
230         }
231
232         if (tdb->global_lock.ltype != ltype || tdb->global_lock.count == 0) {
233                 return TDB_ERRCODE(TDB_ERR_LOCK, -1);
234         }
235
236         if (tdb->global_lock.count > 1) {
237                 tdb->global_lock.count--;
238                 return 0;
239         }
240
241         if (tdb->methods->tdb_brlock(tdb, FREELIST_TOP, F_UNLCK, F_SETLKW, 
242                                      0, 4*tdb->header.hash_size)) {
243                 TDB_LOG((tdb, TDB_DEBUG_ERROR, "tdb_unlockall failed (%s)\n", strerror(errno)));
244                 return -1;
245         }
246
247         tdb->global_lock.count = 0;
248         tdb->global_lock.ltype = 0;
249
250         return 0;
251 }
252
253 /* lock entire database with write lock */
254 int tdb_lockall(struct tdb_context *tdb)
255 {
256         return _tdb_lockall(tdb, F_WRLCK);
257 }
258
259 /* unlock entire database with write lock */
260 int tdb_unlockall(struct tdb_context *tdb)
261 {
262         return _tdb_unlockall(tdb, F_WRLCK);
263 }
264
265 /* lock entire database with read lock */
266 int tdb_lockall_read(struct tdb_context *tdb)
267 {
268         return _tdb_lockall(tdb, F_RDLCK);
269 }
270
271 /* unlock entire database with read lock */
272 int tdb_unlockall_read(struct tdb_context *tdb)
273 {
274         return _tdb_unlockall(tdb, F_RDLCK);
275 }
276
277 /* lock/unlock one hash chain. This is meant to be used to reduce
278    contention - it cannot guarantee how many records will be locked */
279 int tdb_chainlock(struct tdb_context *tdb, TDB_DATA key)
280 {
281         return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
282 }
283
284 int tdb_chainunlock(struct tdb_context *tdb, TDB_DATA key)
285 {
286         return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_WRLCK);
287 }
288
289 int tdb_chainlock_read(struct tdb_context *tdb, TDB_DATA key)
290 {
291         return tdb_lock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
292 }
293
294 int tdb_chainunlock_read(struct tdb_context *tdb, TDB_DATA key)
295 {
296         return tdb_unlock(tdb, BUCKET(tdb->hash_fn(&key)), F_RDLCK);
297 }
298
299
300
301 /* record lock stops delete underneath */
302 int tdb_lock_record(struct tdb_context *tdb, tdb_off_t off)
303 {
304         return off ? tdb->methods->tdb_brlock(tdb, off, F_RDLCK, F_SETLKW, 0, 1) : 0;
305 }
306
307 /*
308   Write locks override our own fcntl readlocks, so check it here.
309   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
310   an error to fail to get the lock here.
311 */
312 int tdb_write_lock_record(struct tdb_context *tdb, tdb_off_t off)
313 {
314         struct tdb_traverse_lock *i;
315         for (i = &tdb->travlocks; i; i = i->next)
316                 if (i->off == off)
317                         return -1;
318         return tdb->methods->tdb_brlock(tdb, off, F_WRLCK, F_SETLK, 1, 1);
319 }
320
321 /*
322   Note this is meant to be F_SETLK, *not* F_SETLKW, as it's not
323   an error to fail to get the lock here.
324 */
325 int tdb_write_unlock_record(struct tdb_context *tdb, tdb_off_t off)
326 {
327         return tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLK, 0, 1);
328 }
329
330 /* fcntl locks don't stack: avoid unlocking someone else's */
331 int tdb_unlock_record(struct tdb_context *tdb, tdb_off_t off)
332 {
333         struct tdb_traverse_lock *i;
334         u32 count = 0;
335
336         if (off == 0)
337                 return 0;
338         for (i = &tdb->travlocks; i; i = i->next)
339                 if (i->off == off)
340                         count++;
341         return (count == 1 ? tdb->methods->tdb_brlock(tdb, off, F_UNLCK, F_SETLKW, 0, 1) : 0);
342 }