Merge "Add script files of test and spdk-ansible for SPDK"
[stor4nfv.git] / src / ceph / src / test / libcephfs / recordlock.cc
1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
3 /*
4  * Ceph - scalable distributed file system
5  *
6  * Copyright (C) 2011 New Dream Network
7  *               2016 Red Hat
8  *
9  * This is free software; you can redistribute it and/or
10  * modify it under the terms of the GNU Lesser General Public
11  * License version 2.1, as published by the Free Software
12  * Foundation.  See file COPYING.
13  *
14  */
15
16 #include <pthread.h>
17 #include "gtest/gtest.h"
18 #ifndef GTEST_IS_THREADSAFE
19 #error "!GTEST_IS_THREADSAFE"
20 #endif
21
22 #include "include/cephfs/libcephfs.h"
23 #include <errno.h>
24 #include <sys/fcntl.h>
25 #include <unistd.h>
26 #include <sys/file.h>
27 #include <sys/types.h>
28 #include <sys/stat.h>
29 #include <dirent.h>
30 #include <sys/xattr.h>
31
32 #include <stdlib.h>
33 #include <semaphore.h>
34 #include <time.h>
35 #include <sys/mman.h>
36
37 #ifdef __linux__
38 #include <limits.h>
39 #endif
40
41 // Startup common: create and mount ceph fs
42 #define STARTUP_CEPH() do {                             \
43     ASSERT_EQ(0, ceph_create(&cmount, NULL));           \
44     ASSERT_EQ(0, ceph_conf_read_file(cmount, NULL));    \
45     ASSERT_EQ(0, ceph_conf_parse_env(cmount, NULL));    \
46     ASSERT_EQ(0, ceph_mount(cmount, NULL));             \
47   } while(0)
48
49 // Cleanup common: unmount and release ceph fs
50 #define CLEANUP_CEPH() do {                     \
51     ASSERT_EQ(0, ceph_unmount(cmount));         \
52     ASSERT_EQ(0, ceph_release(cmount));         \
53   } while(0)
54
55 static const mode_t fileMode = S_IRWXU | S_IRWXG | S_IRWXO;
56
57 // Default wait time for normal and "slow" operations
58 // (5" should be enough in case of network congestion)
59 static const long waitMs = 10;
60 static const long waitSlowMs = 5000;
61
62 // Get the absolute struct timespec reference from now + 'ms' milliseconds
63 static const struct timespec* abstime(struct timespec &ts, long ms) {
64   if (clock_gettime(CLOCK_REALTIME, &ts) == -1) {
65     abort();
66   }
67   ts.tv_nsec += ms * 1000000;
68   ts.tv_sec += ts.tv_nsec / 1000000000;
69   ts.tv_nsec %= 1000000000;
70   return &ts;
71 }
72
73 /* Basic locking */
74
75 TEST(LibCephFS, BasicRecordLocking) {
76   struct ceph_mount_info *cmount = NULL;
77   STARTUP_CEPH();
78
79   char c_file[1024];
80   sprintf(c_file, "recordlock_test_%d", getpid());
81   Fh *fh = NULL;
82   Inode *root = NULL, *inode = NULL;
83   struct ceph_statx stx;
84   int rc;
85   struct flock lock1, lock2;
86   UserPerm *perms = ceph_mount_perms(cmount);
87
88   // Get the root inode
89   rc = ceph_ll_lookup_root(cmount, &root);
90   ASSERT_EQ(rc, 0); 
91
92   // Get the inode and Fh corresponding to c_file
93   rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
94                       &inode, &fh, &stx, 0, 0, perms);
95   ASSERT_EQ(rc, 0); 
96
97   // write lock twice
98   lock1.l_type = F_WRLCK;
99   lock1.l_whence = SEEK_SET;
100   lock1.l_start = 0;
101   lock1.l_len = 1024;
102   lock1.l_pid = getpid();
103   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
104
105   lock2.l_type = F_WRLCK;
106   lock2.l_whence = SEEK_SET;
107   lock2.l_start = 0;
108   lock2.l_len = 1024;
109   lock2.l_pid = getpid();
110   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock2, 43, false));
111
112   // Now try a conflicting read lock
113   lock2.l_type = F_RDLCK;
114   lock2.l_whence = SEEK_SET;
115   lock2.l_start = 100;
116   lock2.l_len = 100;
117   lock2.l_pid = getpid();
118   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock2, 43, false));
119
120   // Now do a getlk
121   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
122   ASSERT_EQ(lock2.l_type, F_WRLCK);
123   ASSERT_EQ(lock2.l_start, 0);
124   ASSERT_EQ(lock2.l_len, 1024);
125   ASSERT_EQ(lock2.l_pid, getpid());
126
127   // Extend the range of the write lock
128   lock1.l_type = F_WRLCK;
129   lock1.l_whence = SEEK_SET;
130   lock1.l_start = 1024;
131   lock1.l_len = 1024;
132   lock1.l_pid = getpid();
133   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
134
135   // Now do a getlk
136   lock2.l_type = F_RDLCK;
137   lock2.l_whence = SEEK_SET;
138   lock2.l_start = 100;
139   lock2.l_len = 100;
140   lock2.l_pid = getpid();
141   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
142   ASSERT_EQ(lock2.l_type, F_WRLCK);
143   ASSERT_EQ(lock2.l_start, 0);
144   ASSERT_EQ(lock2.l_len, 2048);
145   ASSERT_EQ(lock2.l_pid, getpid());
146
147   // Now release part of the range
148   lock1.l_type = F_UNLCK;
149   lock1.l_whence = SEEK_SET;
150   lock1.l_start = 512;
151   lock1.l_len = 1024;
152   lock1.l_pid = getpid();
153   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
154
155   // Now do a getlk to check 1st part
156   lock2.l_type = F_RDLCK;
157   lock2.l_whence = SEEK_SET;
158   lock2.l_start = 100;
159   lock2.l_len = 100;
160   lock2.l_pid = getpid();
161   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
162   ASSERT_EQ(lock2.l_type, F_WRLCK);
163   ASSERT_EQ(lock2.l_start, 0);
164   ASSERT_EQ(lock2.l_len, 512);
165   ASSERT_EQ(lock2.l_pid, getpid());
166
167   // Now do a getlk to check 2nd part
168   lock2.l_type = F_RDLCK;
169   lock2.l_whence = SEEK_SET;
170   lock2.l_start = 2000;
171   lock2.l_len = 100;
172   lock2.l_pid = getpid();
173   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
174   ASSERT_EQ(lock2.l_type, F_WRLCK);
175   ASSERT_EQ(lock2.l_start, 1536);
176   ASSERT_EQ(lock2.l_len, 512);
177   ASSERT_EQ(lock2.l_pid, getpid());
178
179   // Now do a getlk to check released part
180   lock2.l_type = F_RDLCK;
181   lock2.l_whence = SEEK_SET;
182   lock2.l_start = 512;
183   lock2.l_len = 1024;
184   lock2.l_pid = getpid();
185   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
186   ASSERT_EQ(lock2.l_type, F_UNLCK);
187   ASSERT_EQ(lock2.l_start, 512);
188   ASSERT_EQ(lock2.l_len, 1024);
189   ASSERT_EQ(lock2.l_pid, getpid());
190
191   // Now downgrade the 1st part of the lock
192   lock1.l_type = F_RDLCK;
193   lock1.l_whence = SEEK_SET;
194   lock1.l_start = 0;
195   lock1.l_len = 512;
196   lock1.l_pid = getpid();
197   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
198
199   // Now do a getlk to check 1st part
200   lock2.l_type = F_WRLCK;
201   lock2.l_whence = SEEK_SET;
202   lock2.l_start = 100;
203   lock2.l_len = 100;
204   lock2.l_pid = getpid();
205   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
206   ASSERT_EQ(lock2.l_type, F_RDLCK);
207   ASSERT_EQ(lock2.l_start, 0);
208   ASSERT_EQ(lock2.l_len, 512);
209   ASSERT_EQ(lock2.l_pid, getpid());
210
211   // Now upgrade the 1st part of the lock
212   lock1.l_type = F_WRLCK;
213   lock1.l_whence = SEEK_SET;
214   lock1.l_start = 0;
215   lock1.l_len = 512;
216   lock1.l_pid = getpid();
217   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, 42, false));
218
219   // Now do a getlk to check 1st part
220   lock2.l_type = F_WRLCK;
221   lock2.l_whence = SEEK_SET;
222   lock2.l_start = 100;
223   lock2.l_len = 100;
224   lock2.l_pid = getpid();
225   ASSERT_EQ(0, ceph_ll_getlk(cmount, fh, &lock2, 43));
226   ASSERT_EQ(lock2.l_type, F_WRLCK);
227   ASSERT_EQ(lock2.l_start, 0);
228   ASSERT_EQ(lock2.l_len, 512);
229   ASSERT_EQ(lock2.l_pid, getpid());
230
231   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
232   ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
233   CLEANUP_CEPH();
234 }
235
236 /* Locking in different threads */
237
238 // Used by ConcurrentLocking test
239 struct str_ConcurrentRecordLocking {
240   const char *file;
241   struct ceph_mount_info *cmount;  // !NULL if shared
242   sem_t sem[2];
243   sem_t semReply[2];
244   void sem_init(int pshared) {
245     ASSERT_EQ(0, ::sem_init(&sem[0], pshared, 0));
246     ASSERT_EQ(0, ::sem_init(&sem[1], pshared, 0));
247     ASSERT_EQ(0, ::sem_init(&semReply[0], pshared, 0));
248     ASSERT_EQ(0, ::sem_init(&semReply[1], pshared, 0));
249   }
250   void sem_destroy() {
251     ASSERT_EQ(0, ::sem_destroy(&sem[0]));
252     ASSERT_EQ(0, ::sem_destroy(&sem[1]));
253     ASSERT_EQ(0, ::sem_destroy(&semReply[0]));
254     ASSERT_EQ(0, ::sem_destroy(&semReply[1]));
255   }
256 };
257
258 // Wakeup main (for (N) steps)
259 #define PING_MAIN(n) ASSERT_EQ(0, sem_post(&s.sem[n%2]))
260 // Wait for main to wake us up (for (RN) steps)
261 #define WAIT_MAIN(n) \
262   ASSERT_EQ(0, sem_timedwait(&s.semReply[n%2], abstime(ts, waitSlowMs)))
263
264 // Wakeup worker (for (RN) steps)
265 #define PING_WORKER(n) ASSERT_EQ(0, sem_post(&s.semReply[n%2]))
266 // Wait for worker to wake us up (for (N) steps)
267 #define WAIT_WORKER(n) \
268   ASSERT_EQ(0, sem_timedwait(&s.sem[n%2], abstime(ts, waitSlowMs)))
269 // Worker shall not wake us up (for (N) steps)
270 #define NOT_WAIT_WORKER(n) \
271   ASSERT_EQ(-1, sem_timedwait(&s.sem[n%2], abstime(ts, waitMs)))
272
273 // Do twice an operation
274 #define TWICE(EXPR) do {                        \
275     EXPR;                                       \
276     EXPR;                                       \
277   } while(0)
278
279 /* Locking in different threads */
280
281 // Used by ConcurrentLocking test
282 static void thread_ConcurrentRecordLocking(str_ConcurrentRecordLocking& s) {
283   struct ceph_mount_info *const cmount = s.cmount;
284   Fh *fh = NULL;
285   Inode *root = NULL, *inode = NULL;
286   struct ceph_statx stx;
287   struct flock lock1;
288   int rc;
289   struct timespec ts;
290
291   // Get the root inode
292   rc = ceph_ll_lookup_root(cmount, &root);
293   ASSERT_EQ(rc, 0); 
294
295   // Get the inode and Fh corresponding to c_file
296   rc = ceph_ll_create(cmount, root, s.file, fileMode, O_RDWR | O_CREAT,
297                       &inode, &fh, &stx, 0, 0, ceph_mount_perms(cmount));
298   ASSERT_EQ(rc, 0); 
299
300   lock1.l_type = F_WRLCK;
301   lock1.l_whence = SEEK_SET;
302   lock1.l_start = 0;
303   lock1.l_len = 1024;
304   lock1.l_pid = getpid();
305   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
306
307   PING_MAIN(1); // (1)
308   lock1.l_type = F_WRLCK;
309   lock1.l_whence = SEEK_SET;
310   lock1.l_start = 0;
311   lock1.l_len = 1024;
312   lock1.l_pid = getpid();
313   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
314   PING_MAIN(2); // (2)
315
316   lock1.l_type = F_UNLCK;
317   lock1.l_whence = SEEK_SET;
318   lock1.l_start = 0;
319   lock1.l_len = 1024;
320   lock1.l_pid = getpid();
321   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
322   PING_MAIN(3); // (3)
323
324   lock1.l_type = F_RDLCK;
325   lock1.l_whence = SEEK_SET;
326   lock1.l_start = 0;
327   lock1.l_len = 1024;
328   lock1.l_pid = getpid();
329   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
330   PING_MAIN(4); // (4)
331
332   WAIT_MAIN(1); // (R1)
333   lock1.l_type = F_UNLCK;
334   lock1.l_whence = SEEK_SET;
335   lock1.l_start = 0;
336   lock1.l_len = 1024;
337   lock1.l_pid = getpid();
338   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
339   PING_MAIN(5); // (5)
340
341   WAIT_MAIN(2); // (R2)
342   lock1.l_type = F_WRLCK;
343   lock1.l_whence = SEEK_SET;
344   lock1.l_start = 0;
345   lock1.l_len = 1024;
346   lock1.l_pid = getpid();
347   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
348   PING_MAIN(6); // (6)
349
350   WAIT_MAIN(3); // (R3)
351   lock1.l_type = F_UNLCK;
352   lock1.l_whence = SEEK_SET;
353   lock1.l_start = 0;
354   lock1.l_len = 1024;
355   lock1.l_pid = getpid();
356   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
357   PING_MAIN(7); // (7)
358
359   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
360 }
361
362 // Used by ConcurrentRecordLocking test
363 static void* thread_ConcurrentRecordLocking_(void *arg) {
364   str_ConcurrentRecordLocking *const s =
365     reinterpret_cast<str_ConcurrentRecordLocking*>(arg);
366   thread_ConcurrentRecordLocking(*s);
367   return NULL;
368 }
369
370 TEST(LibCephFS, ConcurrentRecordLocking) {
371   const pid_t mypid = getpid();
372   struct ceph_mount_info *cmount;
373   STARTUP_CEPH();
374
375   char c_file[1024];
376   sprintf(c_file, "recordlock_test_%d", mypid);
377   Fh *fh = NULL;
378   Inode *root = NULL, *inode = NULL;
379   struct ceph_statx stx;
380   struct flock lock1;
381   int rc;
382   UserPerm *perms = ceph_mount_perms(cmount);
383
384   // Get the root inode
385   rc = ceph_ll_lookup_root(cmount, &root);
386   ASSERT_EQ(rc, 0); 
387
388   // Get the inode and Fh corresponding to c_file
389   rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
390                       &inode, &fh, &stx, 0, 0, perms);
391   ASSERT_EQ(rc, 0); 
392
393   // Lock
394   lock1.l_type = F_WRLCK;
395   lock1.l_whence = SEEK_SET;
396   lock1.l_start = 0;
397   lock1.l_len = 1024;
398   lock1.l_pid = getpid();
399   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
400
401   // Start locker thread
402   pthread_t thread;
403   struct timespec ts;
404   str_ConcurrentRecordLocking s = { c_file, cmount };
405   s.sem_init(0);
406   ASSERT_EQ(0, pthread_create(&thread, NULL, thread_ConcurrentRecordLocking_, &s));
407   // Synchronization point with thread (failure: thread is dead)
408   WAIT_WORKER(1); // (1)
409
410   // Shall not have lock immediately
411   NOT_WAIT_WORKER(2); // (2)
412
413   // Unlock
414   lock1.l_type = F_UNLCK;
415   lock1.l_whence = SEEK_SET;
416   lock1.l_start = 0;
417   lock1.l_len = 1024;
418   lock1.l_pid = getpid();
419   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
420
421   // Shall have lock
422   // Synchronization point with thread (failure: thread is dead)
423   WAIT_WORKER(2); // (2)
424
425   // Synchronization point with thread (failure: thread is dead)
426   WAIT_WORKER(3); // (3)
427
428   // Wait for thread to share lock
429   WAIT_WORKER(4); // (4)
430   lock1.l_type = F_WRLCK;
431   lock1.l_whence = SEEK_SET;
432   lock1.l_start = 0;
433   lock1.l_len = 1024;
434   lock1.l_pid = getpid();
435   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
436   lock1.l_type = F_RDLCK;
437   lock1.l_whence = SEEK_SET;
438   lock1.l_start = 0;
439   lock1.l_len = 1024;
440   lock1.l_pid = getpid();
441   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
442
443   // Wake up thread to unlock shared lock
444   PING_WORKER(1); // (R1)
445   WAIT_WORKER(5); // (5)
446
447   // Now we can lock exclusively
448   // Upgrade to exclusive lock (as per POSIX)
449   lock1.l_type = F_WRLCK;
450   lock1.l_whence = SEEK_SET;
451   lock1.l_start = 0;
452   lock1.l_len = 1024;
453   lock1.l_pid = getpid();
454   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
455
456   // Wake up thread to lock shared lock
457   PING_WORKER(2); // (R2)
458
459   // Shall not have lock immediately
460   NOT_WAIT_WORKER(6); // (6)
461
462   // Release lock ; thread will get it
463   lock1.l_type = F_UNLCK;
464   lock1.l_whence = SEEK_SET;
465   lock1.l_start = 0;
466   lock1.l_len = 1024;
467   lock1.l_pid = getpid();
468   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
469   WAIT_WORKER(6); // (6)
470
471   // We no longer have the lock
472   lock1.l_type = F_WRLCK;
473   lock1.l_whence = SEEK_SET;
474   lock1.l_start = 0;
475   lock1.l_len = 1024;
476   lock1.l_pid = getpid();
477   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
478   lock1.l_type = F_RDLCK;
479   lock1.l_whence = SEEK_SET;
480   lock1.l_start = 0;
481   lock1.l_len = 1024;
482   lock1.l_pid = getpid();
483   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
484
485   // Wake up thread to unlock exclusive lock
486   PING_WORKER(3); // (R3)
487   WAIT_WORKER(7); // (7)
488
489   // We can lock it again
490   lock1.l_type = F_WRLCK;
491   lock1.l_whence = SEEK_SET;
492   lock1.l_start = 0;
493   lock1.l_len = 1024;
494   lock1.l_pid = getpid();
495   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
496   lock1.l_type = F_UNLCK;
497   lock1.l_whence = SEEK_SET;
498   lock1.l_start = 0;
499   lock1.l_len = 1024;
500   lock1.l_pid = getpid();
501   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
502
503   // Cleanup
504   void *retval = (void*) (uintptr_t) -1;
505   ASSERT_EQ(0, pthread_join(thread, &retval));
506   ASSERT_EQ(NULL, retval);
507   s.sem_destroy();
508   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
509   ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
510   CLEANUP_CEPH();
511 }
512
513 TEST(LibCephFS, ThreesomeRecordLocking) {
514   const pid_t mypid = getpid();
515   struct ceph_mount_info *cmount;
516   STARTUP_CEPH();
517
518   char c_file[1024];
519   sprintf(c_file, "recordlock_test_%d", mypid);
520   Fh *fh = NULL;
521   Inode *root = NULL, *inode = NULL;
522   struct ceph_statx stx;
523   struct flock lock1;
524   int rc;
525   UserPerm *perms = ceph_mount_perms(cmount);
526
527   // Get the root inode
528   rc = ceph_ll_lookup_root(cmount, &root);
529   ASSERT_EQ(rc, 0); 
530
531   // Get the inode and Fh corresponding to c_file
532   rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
533                       &inode, &fh, &stx, 0, 0, perms);
534   ASSERT_EQ(rc, 0); 
535
536   // Lock
537   lock1.l_type = F_WRLCK;
538   lock1.l_whence = SEEK_SET;
539   lock1.l_start = 0;
540   lock1.l_len = 1024;
541   lock1.l_pid = getpid();
542   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
543
544   // Start locker thread
545   pthread_t thread[2];
546   struct timespec ts;
547   str_ConcurrentRecordLocking s = { c_file, cmount };
548   s.sem_init(0);
549   ASSERT_EQ(0, pthread_create(&thread[0], NULL, thread_ConcurrentRecordLocking_, &s));
550   ASSERT_EQ(0, pthread_create(&thread[1], NULL, thread_ConcurrentRecordLocking_, &s));
551   // Synchronization point with thread (failure: thread is dead)
552   TWICE(WAIT_WORKER(1)); // (1)
553
554   // Shall not have lock immediately
555   NOT_WAIT_WORKER(2); // (2)
556
557   // Unlock
558   lock1.l_type = F_UNLCK;
559   lock1.l_whence = SEEK_SET;
560   lock1.l_start = 0;
561   lock1.l_len = 1024;
562   lock1.l_pid = getpid();
563   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
564
565   // Shall have lock
566   TWICE(// Synchronization point with thread (failure: thread is dead)
567         WAIT_WORKER(2); // (2)
568         
569         // Synchronization point with thread (failure: thread is dead)
570         WAIT_WORKER(3)); // (3)
571   
572   // Wait for thread to share lock
573   TWICE(WAIT_WORKER(4)); // (4)
574   lock1.l_type = F_WRLCK;
575   lock1.l_whence = SEEK_SET;
576   lock1.l_start = 0;
577   lock1.l_len = 1024;
578   lock1.l_pid = getpid();
579   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
580   lock1.l_type = F_RDLCK;
581   lock1.l_whence = SEEK_SET;
582   lock1.l_start = 0;
583   lock1.l_len = 1024;
584   lock1.l_pid = getpid();
585   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
586
587   // Wake up thread to unlock shared lock
588   TWICE(PING_WORKER(1); // (R1)
589         WAIT_WORKER(5)); // (5)
590
591   // Now we can lock exclusively
592   // Upgrade to exclusive lock (as per POSIX)
593   lock1.l_type = F_WRLCK;
594   lock1.l_whence = SEEK_SET;
595   lock1.l_start = 0;
596   lock1.l_len = 1024;
597   lock1.l_pid = getpid();
598   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), true));
599
600   TWICE(  // Wake up thread to lock shared lock
601         PING_WORKER(2); // (R2)
602         
603         // Shall not have lock immediately
604         NOT_WAIT_WORKER(6)); // (6)
605   
606   // Release lock ; thread will get it
607   lock1.l_type = F_UNLCK;
608   lock1.l_whence = SEEK_SET;
609   lock1.l_start = 0;
610   lock1.l_len = 1024;
611   lock1.l_pid = getpid();
612   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
613   TWICE(WAIT_WORKER(6); // (6)
614         
615         // We no longer have the lock
616         lock1.l_type = F_WRLCK;
617         lock1.l_whence = SEEK_SET;
618         lock1.l_start = 0;
619         lock1.l_len = 1024;
620         lock1.l_pid = getpid();
621         ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
622         lock1.l_type = F_RDLCK;
623         lock1.l_whence = SEEK_SET;
624         lock1.l_start = 0;
625         lock1.l_len = 1024;
626         lock1.l_pid = getpid();
627         ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
628         
629         // Wake up thread to unlock exclusive lock
630         PING_WORKER(3); // (R3)
631         WAIT_WORKER(7); // (7)
632         );
633   
634   // We can lock it again
635   lock1.l_type = F_WRLCK;
636   lock1.l_whence = SEEK_SET;
637   lock1.l_start = 0;
638   lock1.l_len = 1024;
639   lock1.l_pid = getpid();
640   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
641   lock1.l_type = F_UNLCK;
642   lock1.l_whence = SEEK_SET;
643   lock1.l_start = 0;
644   lock1.l_len = 1024;
645   lock1.l_pid = getpid();
646   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
647
648   // Cleanup
649   void *retval = (void*) (uintptr_t) -1;
650   ASSERT_EQ(0, pthread_join(thread[0], &retval));
651   ASSERT_EQ(NULL, retval);
652   ASSERT_EQ(0, pthread_join(thread[1], &retval));
653   ASSERT_EQ(NULL, retval);
654   s.sem_destroy();
655   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
656   ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
657   CLEANUP_CEPH();
658 }
659
660 /* Locking in different processes */
661
662 #define PROCESS_SLOW_MS() \
663   static const long waitMs = 100; \
664   (void) waitMs
665
666 // Used by ConcurrentLocking test
667 static void process_ConcurrentRecordLocking(str_ConcurrentRecordLocking& s) {
668   const pid_t mypid = getpid();
669   PROCESS_SLOW_MS();
670
671   struct ceph_mount_info *cmount = NULL;
672   struct timespec ts;
673   Fh *fh = NULL;
674   Inode *root = NULL, *inode = NULL;
675   struct ceph_statx stx;
676   int rc;
677   struct flock lock1;
678
679   STARTUP_CEPH();
680   s.cmount = cmount;
681
682   // Get the root inode
683   rc = ceph_ll_lookup_root(cmount, &root);
684   ASSERT_EQ(rc, 0); 
685
686   // Get the inode and Fh corresponding to c_file
687   rc = ceph_ll_create(cmount, root, s.file, fileMode, O_RDWR | O_CREAT,
688                       &inode, &fh, &stx, 0, 0, ceph_mount_perms(cmount));
689   ASSERT_EQ(rc, 0); 
690
691   WAIT_MAIN(1); // (R1)
692
693   lock1.l_type = F_WRLCK;
694   lock1.l_whence = SEEK_SET;
695   lock1.l_start = 0;
696   lock1.l_len = 1024;
697   lock1.l_pid = getpid();
698   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
699   PING_MAIN(1); // (1)
700   lock1.l_type = F_WRLCK;
701   lock1.l_whence = SEEK_SET;
702   lock1.l_start = 0;
703   lock1.l_len = 1024;
704   lock1.l_pid = getpid();
705   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
706   PING_MAIN(2); // (2)
707
708   lock1.l_type = F_UNLCK;
709   lock1.l_whence = SEEK_SET;
710   lock1.l_start = 0;
711   lock1.l_len = 1024;
712   lock1.l_pid = getpid();
713   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
714   PING_MAIN(3); // (3)
715
716   lock1.l_type = F_RDLCK;
717   lock1.l_whence = SEEK_SET;
718   lock1.l_start = 0;
719   lock1.l_len = 1024;
720   lock1.l_pid = getpid();
721   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
722   PING_MAIN(4); // (4)
723
724   WAIT_MAIN(2); // (R2)
725   lock1.l_type = F_UNLCK;
726   lock1.l_whence = SEEK_SET;
727   lock1.l_start = 0;
728   lock1.l_len = 1024;
729   lock1.l_pid = getpid();
730   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
731   PING_MAIN(5); // (5)
732
733   WAIT_MAIN(3); // (R3)
734   lock1.l_type = F_WRLCK;
735   lock1.l_whence = SEEK_SET;
736   lock1.l_start = 0;
737   lock1.l_len = 1024;
738   lock1.l_pid = getpid();
739   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
740   PING_MAIN(6); // (6)
741
742   WAIT_MAIN(4); // (R4)
743   lock1.l_type = F_UNLCK;
744   lock1.l_whence = SEEK_SET;
745   lock1.l_start = 0;
746   lock1.l_len = 1024;
747   lock1.l_pid = getpid();
748   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
749   PING_MAIN(7); // (7)
750
751   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
752   CLEANUP_CEPH();
753
754   s.sem_destroy();
755   exit(EXIT_SUCCESS);
756 }
757
758 // Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
759 TEST(LibCephFS, DISABLED_InterProcessRecordLocking) {
760   PROCESS_SLOW_MS();
761   // Process synchronization
762   char c_file[1024];
763   const pid_t mypid = getpid();
764   sprintf(c_file, "recordlock_test_%d", mypid);
765   Fh *fh = NULL;
766   Inode *root = NULL, *inode = NULL;
767   struct ceph_statx stx;
768   struct flock lock1;
769   int rc;
770
771   // Note: the semaphores MUST be on a shared memory segment
772   str_ConcurrentRecordLocking *const shs =
773     reinterpret_cast<str_ConcurrentRecordLocking*>
774     (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
775           -1, 0));
776   str_ConcurrentRecordLocking &s = *shs;
777   s.file = c_file;
778   s.sem_init(1);
779
780   // Start locker process
781   const pid_t pid = fork();
782   ASSERT_GE(pid, 0);
783   if (pid == 0) {
784     process_ConcurrentRecordLocking(s);
785     exit(EXIT_FAILURE);
786   }
787
788   struct timespec ts;
789   struct ceph_mount_info *cmount;
790   STARTUP_CEPH();
791   UserPerm *perms = ceph_mount_perms(cmount);
792
793   // Get the root inode
794   rc = ceph_ll_lookup_root(cmount, &root);
795   ASSERT_EQ(rc, 0); 
796
797   // Get the inode and Fh corresponding to c_file
798   rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
799                       &inode, &fh, &stx, 0, 0, perms);
800   ASSERT_EQ(rc, 0); 
801
802   // Lock
803   lock1.l_type = F_WRLCK;
804   lock1.l_whence = SEEK_SET;
805   lock1.l_start = 0;
806   lock1.l_len = 1024;
807   lock1.l_pid = getpid();
808   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
809
810   // Synchronization point with process (failure: process is dead)
811   PING_WORKER(1); // (R1)
812   WAIT_WORKER(1); // (1)
813
814   // Shall not have lock immediately
815   NOT_WAIT_WORKER(2); // (2)
816
817   // Unlock
818   lock1.l_type = F_UNLCK;
819   lock1.l_whence = SEEK_SET;
820   lock1.l_start = 0;
821   lock1.l_len = 1024;
822   lock1.l_pid = getpid();
823   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
824
825   // Shall have lock
826   // Synchronization point with process (failure: process is dead)
827   WAIT_WORKER(2); // (2)
828
829   // Synchronization point with process (failure: process is dead)
830   WAIT_WORKER(3); // (3)
831
832   // Wait for process to share lock
833   WAIT_WORKER(4); // (4)
834   lock1.l_type = F_WRLCK;
835   lock1.l_whence = SEEK_SET;
836   lock1.l_start = 0;
837   lock1.l_len = 1024;
838   lock1.l_pid = getpid();
839   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
840   lock1.l_type = F_RDLCK;
841   lock1.l_whence = SEEK_SET;
842   lock1.l_start = 0;
843   lock1.l_len = 1024;
844   lock1.l_pid = getpid();
845   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
846
847   // Wake up process to unlock shared lock
848   PING_WORKER(2); // (R2)
849   WAIT_WORKER(5); // (5)
850
851   // Now we can lock exclusively
852   // Upgrade to exclusive lock (as per POSIX)
853   lock1.l_type = F_WRLCK;
854   lock1.l_whence = SEEK_SET;
855   lock1.l_start = 0;
856   lock1.l_len = 1024;
857   lock1.l_pid = getpid();
858   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
859
860   // Wake up process to lock shared lock
861   PING_WORKER(3); // (R3)
862
863   // Shall not have lock immediately
864   NOT_WAIT_WORKER(6); // (6)
865
866   // Release lock ; process will get it
867   lock1.l_type = F_UNLCK;
868   lock1.l_whence = SEEK_SET;
869   lock1.l_start = 0;
870   lock1.l_len = 1024;
871   lock1.l_pid = getpid();
872   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
873   WAIT_WORKER(6); // (6)
874
875   // We no longer have the lock
876   lock1.l_type = F_WRLCK;
877   lock1.l_whence = SEEK_SET;
878   lock1.l_start = 0;
879   lock1.l_len = 1024;
880   lock1.l_pid = getpid();
881   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
882   lock1.l_type = F_RDLCK;
883   lock1.l_whence = SEEK_SET;
884   lock1.l_start = 0;
885   lock1.l_len = 1024;
886   lock1.l_pid = getpid();
887   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
888
889   // Wake up process to unlock exclusive lock
890   PING_WORKER(4); // (R4)
891   WAIT_WORKER(7); // (7)
892
893   // We can lock it again
894   lock1.l_type = F_WRLCK;
895   lock1.l_whence = SEEK_SET;
896   lock1.l_start = 0;
897   lock1.l_len = 1024;
898   lock1.l_pid = getpid();
899   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
900   lock1.l_type = F_UNLCK;
901   lock1.l_whence = SEEK_SET;
902   lock1.l_start = 0;
903   lock1.l_len = 1024;
904   lock1.l_pid = getpid();
905   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
906
907   // Wait pid
908   int status;
909   ASSERT_EQ(pid, waitpid(pid, &status, 0));
910   ASSERT_EQ(EXIT_SUCCESS, status);
911
912   // Cleanup
913   s.sem_destroy();
914   ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
915   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
916   ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
917   CLEANUP_CEPH();
918 }
919
920 // Disabled because of fork() issues (http://tracker.ceph.com/issues/16556)
921 TEST(LibCephFS, DISABLED_ThreesomeInterProcessRecordLocking) {
922   PROCESS_SLOW_MS();
923   // Process synchronization
924   char c_file[1024];
925   const pid_t mypid = getpid();
926   sprintf(c_file, "recordlock_test_%d", mypid);
927   Fh *fh = NULL;
928   Inode *root = NULL, *inode = NULL;
929   struct ceph_statx stx;
930   struct flock lock1;
931   int rc;
932
933   // Note: the semaphores MUST be on a shared memory segment
934   str_ConcurrentRecordLocking *const shs =
935     reinterpret_cast<str_ConcurrentRecordLocking*>
936     (mmap(0, sizeof(*shs), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS,
937           -1, 0));
938   str_ConcurrentRecordLocking &s = *shs;
939   s.file = c_file;
940   s.sem_init(1);
941
942   // Start locker processes
943   pid_t pid[2];
944   pid[0] = fork();
945   ASSERT_GE(pid[0], 0);
946   if (pid[0] == 0) {
947     process_ConcurrentRecordLocking(s);
948     exit(EXIT_FAILURE);
949   }
950   pid[1] = fork();
951   ASSERT_GE(pid[1], 0);
952   if (pid[1] == 0) {
953     process_ConcurrentRecordLocking(s);
954     exit(EXIT_FAILURE);
955   }
956
957   struct timespec ts;
958   struct ceph_mount_info *cmount;
959   STARTUP_CEPH();
960
961   // Get the root inode
962   rc = ceph_ll_lookup_root(cmount, &root);
963   ASSERT_EQ(rc, 0); 
964
965   // Get the inode and Fh corresponding to c_file
966   UserPerm *perms = ceph_mount_perms(cmount);
967   rc = ceph_ll_create(cmount, root, c_file, fileMode, O_RDWR | O_CREAT,
968                       &inode, &fh, &stx, 0, 0, perms);
969   ASSERT_EQ(rc, 0); 
970
971   // Lock
972   lock1.l_type = F_WRLCK;
973   lock1.l_whence = SEEK_SET;
974   lock1.l_start = 0;
975   lock1.l_len = 1024;
976   lock1.l_pid = getpid();
977   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
978
979   // Synchronization point with process (failure: process is dead)
980   TWICE(PING_WORKER(1)); // (R1)
981   TWICE(WAIT_WORKER(1)); // (1)
982
983   // Shall not have lock immediately
984   NOT_WAIT_WORKER(2); // (2)
985
986   // Unlock
987   lock1.l_type = F_UNLCK;
988   lock1.l_whence = SEEK_SET;
989   lock1.l_start = 0;
990   lock1.l_len = 1024;
991   lock1.l_pid = getpid();
992   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
993
994   // Shall have lock
995   TWICE(// Synchronization point with process (failure: process is dead)
996         WAIT_WORKER(2); // (2)
997         
998         // Synchronization point with process (failure: process is dead)
999         WAIT_WORKER(3)); // (3)
1000   
1001   // Wait for process to share lock
1002   TWICE(WAIT_WORKER(4)); // (4)
1003   lock1.l_type = F_WRLCK;
1004   lock1.l_whence = SEEK_SET;
1005   lock1.l_start = 0;
1006   lock1.l_len = 1024;
1007   lock1.l_pid = getpid();
1008   ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
1009   lock1.l_type = F_RDLCK;
1010   lock1.l_whence = SEEK_SET;
1011   lock1.l_start = 0;
1012   lock1.l_len = 1024;
1013   lock1.l_pid = getpid();
1014   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
1015
1016   // Wake up process to unlock shared lock
1017   TWICE(PING_WORKER(2); // (R2)
1018         WAIT_WORKER(5)); // (5)
1019
1020   // Now we can lock exclusively
1021   // Upgrade to exclusive lock (as per POSIX)
1022   lock1.l_type = F_WRLCK;
1023   lock1.l_whence = SEEK_SET;
1024   lock1.l_start = 0;
1025   lock1.l_len = 1024;
1026   lock1.l_pid = getpid();
1027   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, true));
1028
1029   TWICE(  // Wake up process to lock shared lock
1030         PING_WORKER(3); // (R3)
1031         
1032         // Shall not have lock immediately
1033         NOT_WAIT_WORKER(6)); // (6)
1034   
1035   // Release lock ; process will get it
1036   lock1.l_type = F_UNLCK;
1037   lock1.l_whence = SEEK_SET;
1038   lock1.l_start = 0;
1039   lock1.l_len = 1024;
1040   lock1.l_pid = getpid();
1041   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
1042   TWICE(WAIT_WORKER(6); // (6)
1043         
1044         // We no longer have the lock
1045         lock1.l_type = F_WRLCK;
1046         lock1.l_whence = SEEK_SET;
1047         lock1.l_start = 0;
1048         lock1.l_len = 1024;
1049         lock1.l_pid = getpid();
1050         ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
1051         lock1.l_type = F_RDLCK;
1052         lock1.l_whence = SEEK_SET;
1053         lock1.l_start = 0;
1054         lock1.l_len = 1024;
1055         lock1.l_pid = getpid();
1056         ASSERT_EQ(-EAGAIN, ceph_ll_setlk(cmount, fh, &lock1, pthread_self(), false));
1057         
1058         // Wake up process to unlock exclusive lock
1059         PING_WORKER(4); // (R4)
1060         WAIT_WORKER(7); // (7)
1061         );
1062   
1063   // We can lock it again
1064   lock1.l_type = F_WRLCK;
1065   lock1.l_whence = SEEK_SET;
1066   lock1.l_start = 0;
1067   lock1.l_len = 1024;
1068   lock1.l_pid = getpid();
1069   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
1070   lock1.l_type = F_UNLCK;
1071   lock1.l_whence = SEEK_SET;
1072   lock1.l_start = 0;
1073   lock1.l_len = 1024;
1074   lock1.l_pid = getpid();
1075   ASSERT_EQ(0, ceph_ll_setlk(cmount, fh, &lock1, mypid, false));
1076
1077   // Wait pids
1078   int status;
1079   ASSERT_EQ(pid[0], waitpid(pid[0], &status, 0));
1080   ASSERT_EQ(EXIT_SUCCESS, status);
1081   ASSERT_EQ(pid[1], waitpid(pid[1], &status, 0));
1082   ASSERT_EQ(EXIT_SUCCESS, status);
1083
1084   // Cleanup
1085   s.sem_destroy();
1086   ASSERT_EQ(0, munmap(shs, sizeof(*shs)));
1087   ASSERT_EQ(0, ceph_ll_close(cmount, fh));
1088   ASSERT_EQ(0, ceph_ll_unlink(cmount, root, c_file, perms));
1089   CLEANUP_CEPH();
1090 }