releases/2.6.32.19/dlm-send-reply-before-bast.patch

   1 From cf6620acc0f6fac57968aafef79ab372bdcf6157 Mon Sep 17 00:00:00 2001
   2 From: David Teigland <teigland@redhat.com>
   3 Date: Wed, 24 Feb 2010 11:59:23 -0600
   4 Subject: dlm: send reply before bast
   5
   6 From: David Teigland <teigland@redhat.com>
   7
   8 commit cf6620acc0f6fac57968aafef79ab372bdcf6157 upstream.
   9
  10 When the lock master processes a successful operation (request,
  11 convert, cancel, or unlock), it will process the effects of the
  12 change before sending the reply for the operation.  The "effects"
  13 of the operation are:
  14
  15 - blocking callbacks (basts) for any newly granted locks
  16 - waiting or converting locks that can now be granted
  17
  18 The cast is queued on the local node when the reply from the lock
  19 master is received.  This means that a lock holder can receive a
  20 bast for a lock mode that is doesn't yet know has been granted.
  21
  22 Signed-off-by: David Teigland <teigland@redhat.com>
  23 Signed-off-by: Greg Kroah-Hartman <gregkh@suse.de>
  24
  25 ---
  26  fs/dlm/lock.c |  110 ++++++++++++++++++++++++++++++++++++++++++++--------------
  27  1 file changed, 84 insertions(+), 26 deletions(-)
  28
  29 --- a/fs/dlm/lock.c
  30 +++ b/fs/dlm/lock.c
  31 @@ -2280,20 +2280,30 @@ static int do_request(struct dlm_rsb *r,
  32         if (can_be_queued(lkb)) {
  33                 error = -EINPROGRESS;
  34                 add_lkb(r, lkb, DLM_LKSTS_WAITING);
  35 -               send_blocking_asts(r, lkb);
  36                 add_timeout(lkb);
  37                 goto out;
  38         }
  39
  40         error = -EAGAIN;
  41 -       if (force_blocking_asts(lkb))
  42 -               send_blocking_asts_all(r, lkb);
  43         queue_cast(r, lkb, -EAGAIN);
  44 -
  45   out:
  46         return error;
  47  }
  48
  49 +static void do_request_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
  50 +                              int error)
  51 +{
  52 +       switch (error) {
  53 +       case -EAGAIN:
  54 +               if (force_blocking_asts(lkb))
  55 +                       send_blocking_asts_all(r, lkb);
  56 +               break;
  57 +       case -EINPROGRESS:
  58 +               send_blocking_asts(r, lkb);
  59 +               break;
  60 +       }
  61 +}
  62 +
  63  static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
  64  {
  65         int error = 0;
  66 @@ -2304,7 +2314,6 @@ static int do_convert(struct dlm_rsb *r,
  67         if (can_be_granted(r, lkb, 1, &deadlk)) {
  68                 grant_lock(r, lkb);
  69                 queue_cast(r, lkb, 0);
  70 -               grant_pending_locks(r);
  71                 goto out;
  72         }
  73
  74 @@ -2334,7 +2343,6 @@ static int do_convert(struct dlm_rsb *r,
  75                 if (_can_be_granted(r, lkb, 1)) {
  76                         grant_lock(r, lkb);
  77                         queue_cast(r, lkb, 0);
  78 -                       grant_pending_locks(r);
  79                         goto out;
  80                 }
  81                 /* else fall through and move to convert queue */
  82 @@ -2344,28 +2352,47 @@ static int do_convert(struct dlm_rsb *r,
  83                 error = -EINPROGRESS;
  84                 del_lkb(r, lkb);
  85                 add_lkb(r, lkb, DLM_LKSTS_CONVERT);
  86 -               send_blocking_asts(r, lkb);
  87                 add_timeout(lkb);
  88                 goto out;
  89         }
  90
  91         error = -EAGAIN;
  92 -       if (force_blocking_asts(lkb))
  93 -               send_blocking_asts_all(r, lkb);
  94         queue_cast(r, lkb, -EAGAIN);
  95 -
  96   out:
  97         return error;
  98  }
  99
 100 +static void do_convert_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
 101 +                              int error)
 102 +{
 103 +       switch (error) {
 104 +       case 0:
 105 +               grant_pending_locks(r);
 106 +               /* grant_pending_locks also sends basts */
 107 +               break;
 108 +       case -EAGAIN:
 109 +               if (force_blocking_asts(lkb))
 110 +                       send_blocking_asts_all(r, lkb);
 111 +               break;
 112 +       case -EINPROGRESS:
 113 +               send_blocking_asts(r, lkb);
 114 +               break;
 115 +       }
 116 +}
 117 +
 118  static int do_unlock(struct dlm_rsb *r, struct dlm_lkb *lkb)
 119  {
 120         remove_lock(r, lkb);
 121         queue_cast(r, lkb, -DLM_EUNLOCK);
 122 -       grant_pending_locks(r);
 123         return -DLM_EUNLOCK;
 124  }
 125
 126 +static void do_unlock_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
 127 +                             int error)
 128 +{
 129 +       grant_pending_locks(r);
 130 +}
 131 +
 132  /* returns: 0 did nothing, -DLM_ECANCEL canceled lock */
 133
 134  static int do_cancel(struct dlm_rsb *r, struct dlm_lkb *lkb)
 135 @@ -2375,12 +2402,18 @@ static int do_cancel(struct dlm_rsb *r,
 136         error = revert_lock(r, lkb);
 137         if (error) {
 138                 queue_cast(r, lkb, -DLM_ECANCEL);
 139 -               grant_pending_locks(r);
 140                 return -DLM_ECANCEL;
 141         }
 142         return 0;
 143  }
 144
 145 +static void do_cancel_effects(struct dlm_rsb *r, struct dlm_lkb *lkb,
 146 +                             int error)
 147 +{
 148 +       if (error)
 149 +               grant_pending_locks(r);
 150 +}
 151 +
 152  /*
 153   * Four stage 3 varieties:
 154   * _request_lock(), _convert_lock(), _unlock_lock(), _cancel_lock()
 155 @@ -2402,11 +2435,15 @@ static int _request_lock(struct dlm_rsb
 156                 goto out;
 157         }
 158
 159 -       if (is_remote(r))
 160 +       if (is_remote(r)) {
 161                 /* receive_request() calls do_request() on remote node */
 162                 error = send_request(r, lkb);
 163 -       else
 164 +       } else {
 165                 error = do_request(r, lkb);
 166 +               /* for remote locks the request_reply is sent
 167 +                  between do_request and do_request_effects */
 168 +               do_request_effects(r, lkb, error);
 169 +       }
 170   out:
 171         return error;
 172  }
 173 @@ -2417,11 +2454,15 @@ static int _convert_lock(struct dlm_rsb
 174  {
 175         int error;
 176
 177 -       if (is_remote(r))
 178 +       if (is_remote(r)) {
 179                 /* receive_convert() calls do_convert() on remote node */
 180                 error = send_convert(r, lkb);
 181 -       else
 182 +       } else {
 183                 error = do_convert(r, lkb);
 184 +               /* for remote locks the convert_reply is sent
 185 +                  between do_convert and do_convert_effects */
 186 +               do_convert_effects(r, lkb, error);
 187 +       }
 188
 189         return error;
 190  }
 191 @@ -2432,11 +2473,15 @@ static int _unlock_lock(struct dlm_rsb *
 192  {
 193         int error;
 194
 195 -       if (is_remote(r))
 196 +       if (is_remote(r)) {
 197                 /* receive_unlock() calls do_unlock() on remote node */
 198                 error = send_unlock(r, lkb);
 199 -       else
 200 +       } else {
 201                 error = do_unlock(r, lkb);
 202 +               /* for remote locks the unlock_reply is sent
 203 +                  between do_unlock and do_unlock_effects */
 204 +               do_unlock_effects(r, lkb, error);
 205 +       }
 206
 207         return error;
 208  }
 209 @@ -2447,11 +2492,15 @@ static int _cancel_lock(struct dlm_rsb *
 210  {
 211         int error;
 212
 213 -       if (is_remote(r))
 214 +       if (is_remote(r)) {
 215                 /* receive_cancel() calls do_cancel() on remote node */
 216                 error = send_cancel(r, lkb);
 217 -       else
 218 +       } else {
 219                 error = do_cancel(r, lkb);
 220 +               /* for remote locks the cancel_reply is sent
 221 +                  between do_cancel and do_cancel_effects */
 222 +               do_cancel_effects(r, lkb, error);
 223 +       }
 224
 225         return error;
 226  }
 227 @@ -3191,6 +3240,7 @@ static void receive_request(struct dlm_l
 228         attach_lkb(r, lkb);
 229         error = do_request(r, lkb);
 230         send_request_reply(r, lkb, error);
 231 +       do_request_effects(r, lkb, error);
 232
 233         unlock_rsb(r);
 234         put_rsb(r);
 235 @@ -3226,15 +3276,19 @@ static void receive_convert(struct dlm_l
 236                 goto out;
 237
 238         receive_flags(lkb, ms);
 239 +
 240         error = receive_convert_args(ls, lkb, ms);
 241 -       if (error)
 242 -               goto out_reply;
 243 +       if (error) {
 244 +               send_convert_reply(r, lkb, error);
 245 +               goto out;
 246 +       }
 247 +
 248         reply = !down_conversion(lkb);
 249
 250         error = do_convert(r, lkb);
 251 - out_reply:
 252         if (reply)
 253                 send_convert_reply(r, lkb, error);
 254 +       do_convert_effects(r, lkb, error);
 255   out:
 256         unlock_rsb(r);
 257         put_rsb(r);
 258 @@ -3266,13 +3320,16 @@ static void receive_unlock(struct dlm_ls
 259                 goto out;
 260
 261         receive_flags(lkb, ms);
 262 +
 263         error = receive_unlock_args(ls, lkb, ms);
 264 -       if (error)
 265 -               goto out_reply;
 266 +       if (error) {
 267 +               send_unlock_reply(r, lkb, error);
 268 +               goto out;
 269 +       }
 270
 271         error = do_unlock(r, lkb);
 272 - out_reply:
 273         send_unlock_reply(r, lkb, error);
 274 +       do_unlock_effects(r, lkb, error);
 275   out:
 276         unlock_rsb(r);
 277         put_rsb(r);
 278 @@ -3307,6 +3364,7 @@ static void receive_cancel(struct dlm_ls
 279
 280         error = do_cancel(r, lkb);
 281         send_cancel_reply(r, lkb, error);
 282 +       do_cancel_effects(r, lkb, error);
 283   out:
 284         unlock_rsb(r);
 285         put_rsb(r);