DistributedLockManager


1   package org.jgroups.blocks;
2   
3   import org.apache.commons.logging.Log;
4   import org.apache.commons.logging.LogFactory;
5   import org.jgroups.ChannelException;
6   
7   import java.io.Serializable  ;
8   import java.util.HashMap  ;
9   
10  /**
11   * Distributed lock manager is responsible for maintaining the lock information
12   * consistent on all participating nodes.
13   * 
14   * @author Roman Rokytskyy (rrokytskyy@acm.org)
15   */
16  public class DistributedLockManager implements TwoPhaseVotingListener, LockManager {
17  
18      /**
19       * This parameter means that lock acquisition expires after 5 seconds.
20       * If there were no "commit" operation on prepared lock, then it
21       * is treated as expired and is removed from the prepared locks table.
22       */
23      private static final long ACQUIRE_EXPIRATION = 5000;
24      
25      /**
26       * This parameter is used during lock releasing. If group fails to release
27       * the lock during the specified period of time, unlocking fails.
28       */
29      private static final long VOTE_TIMEOUT = 10000;
30  
31      // list of all prepared locks
32      private final HashMap   preparedLocks = new HashMap  ();
33  
34      // list of all prepared releases
35      private final HashMap   preparedReleases = new HashMap  ();
36  
37      // list of locks on the node
38      private final HashMap   heldLocks = new HashMap  ();
39  
40      private final TwoPhaseVotingAdapter votingAdapter;
41  
42      private final Object   id;
43  
44      protected final Log log=LogFactory.getLog(getClass());
45  
46  
47      /**
48       * Create instance of this class.
49       * 
50       * @param voteChannel instance of {@link VotingAdapter} that will be used 
51       * for voting purposes on the lock decrees. <tt>voteChannel()</tt> will
52       * be wrapped by the instance of the {@link TwoPhaseVotingAdapter}.
53       * 
54       * @param id the unique identifier of this lock manager.
55       * 
56       * @todo check if the node with the same id is already in the group.
57       */
58      public DistributedLockManager(VotingAdapter voteChannel, Object   id) {
59          this(new TwoPhaseVotingAdapter(voteChannel), id);
60      }
61  
62      /**
63       *  Constructor for the DistributedLockManager_cl object.
64       * 
65       *  @param channel instance of {@link TwoPhaseVotingAdapter}
66       *  that will be used for voting purposes on the lock decrees.
67       * 
68       *  @param id the unique identifier of this lock manager.
69       * 
70       *  @todo check if the node with the same id is already in the group.
71       */
72      public DistributedLockManager(TwoPhaseVotingAdapter channel, Object   id) {
73          this.id = id;
74          this.votingAdapter = channel;
75          this.votingAdapter.addListener(this);
76      }
77  
78      /**
79       * Performs local lock. This method also performs the clean-up of the lock
80       * table, all expired locks are removed.
81       */
82      private boolean localLock(LockDecree lockDecree) {
83          // remove expired locks
84          removeExpired(lockDecree);
85  
86          LockDecree localLock =
87              (LockDecree) heldLocks.get(lockDecree.getKey());
88  
89          if (localLock == null) {
90  
91              // promote lock into commited state
92              lockDecree.commit();
93  
94              // no lock exist, perform local lock, note:
95              // we do not store locks that were requested by other manager.
96              if (lockDecree.managerId.equals(id))
97                  heldLocks.put(lockDecree.getKey(), lockDecree);
98  
99              // everything is fine :)
100             return true;
101         } else
102         if (localLock.requester.equals(lockDecree.requester))
103             // requester already owns the lock
104             return true;
105         else
106             // lock does not belong to requester
107             return false;
108 
109     }
110 
111     /**
112      * Returns <code>true</code> if the requested lock can be granted by the
113      * current node.
114      * 
115      * @param decree instance of <code>LockDecree</code> containing information
116      * about the lock.
117      */
118     private boolean canLock(LockDecree decree) {
119         // clean expired locks
120         removeExpired(decree);
121 
122         LockDecree lock = (LockDecree)heldLocks.get(decree.getKey());
123         if (lock == null)
124             return true;
125         else
126             return lock.requester.equals(decree.requester);
127     }
128 
129     /**
130      * Returns <code>true</code> if the requested lock can be released by the
131      * current node.
132      * 
133      * @param decree instance of {@link LockDecree} containing information
134      * about the lock.
135      */
136     private boolean canRelease(LockDecree decree) {
137         // clean expired locks
138         removeExpired(decree);
139 
140         // we need to check only hold locks, because
141         // prepared locks cannot contain the lock
142         LockDecree lock = (LockDecree)heldLocks.get(decree.getKey());
143         if (lock == null)
144             // check if this holds...
145             return true;
146         else
147             return lock.requester.equals(decree.requester);
148     }
149 
150     /**
151      * Removes expired locks.
152      * 
153      * @param decree instance of {@link LockDecree} describing the lock.
154      */
155     private void removeExpired(LockDecree decree) {
156         // remove the invalid (expired) lock
157         LockDecree localLock = (LockDecree)heldLocks.get(decree.getKey());
158         if (localLock != null && !localLock.isValid())
159             heldLocks.remove(localLock.getKey());
160     }
161 
162     /**
163      * Releases lock locally.
164      * 
165      * @param lockDecree instance of {@link LockDecree} describing the lock.
166      */
167     private boolean localRelease(LockDecree lockDecree) {
168         // remove expired locks
169         removeExpired(lockDecree);
170 
171         LockDecree localLock=
172                 (LockDecree) heldLocks.get(lockDecree.getKey());
173 
174         if(localLock == null) {
175             // no lock exist
176             return true;
177         }
178         else if(localLock.requester.equals(lockDecree.requester)) {
179             // requester owns the lock, release the lock
180             heldLocks.remove(lockDecree.getKey());
181             return true;
182         }
183         else
184         // lock does not belong to requester
185             return false;
186     }
187 
188     /**
189      * Locks an object with <code>lockId</code> on behalf of the specified
190      * <code>owner</code>.
191      * 
192      * @param lockId <code>Object</code> representing the object to be locked.
193      * @param owner object that requests the lock.
194      * @param timeout time during which group members should decide
195      * whether to grant a lock or not.
196      *
197      * @throws LockNotGrantedException when the lock cannot be granted.
198      * 
199      * @throws ClassCastException if lockId or owner are not serializable.
200      * 
201      * @throws ChannelException if something bad happened to underlying channel.
202      */
203     public void lock(Object   lockId, Object   owner, int timeout)
204         throws LockNotGrantedException, ChannelException
205     {
206         if (!(lockId instanceof Serializable  ) || !(owner instanceof Serializable  ))
207             throw new ClassCastException  ("DistributedLockManager " +
208                 "works only with serializable objects.");
209 
210         boolean acquired = votingAdapter.vote(
211             new AcquireLockDecree(lockId, owner, id), timeout);
212 
213         if (!acquired)
214             throw new LockNotGrantedException("Lock cannot be granted.");
215     }
216 
217     /**
218      * Unlocks an object with <code>lockId</code> on behalf of the specified
219      * <code>owner</code>.
220      * @param lockId <code>long</code> representing the object to be unlocked.
221      * @param owner object that releases the lock.
222      *
223      * @throws LockNotReleasedException when the lock cannot be released.
224      * @throws ClassCastException if lockId or owner are not serializable.
225      */
226     public void unlock(Object   lockId, Object   owner)
227         throws LockNotReleasedException, ChannelException
228     {
229 
230         if (!(lockId instanceof Serializable  ) || !(owner instanceof Serializable  ))
231             throw new ClassCastException  ("DistributedLockManager " +
232                 "works only with serializable objects.");
233 
234             
235         boolean released = votingAdapter.vote(
236             new ReleaseLockDecree(lockId, owner, id), VOTE_TIMEOUT);
237         
238         if (!released)
239             throw new LockNotReleasedException("Lock cannot be unlocked.");
240     }
241 
242     /**
243      * Checks the list of prepared locks/unlocks to determine if we are in the
244      * middle of the two-phase commit process for the lock acqusition/release.
245      * Here we do not tolerate if the request comes from the same node on behalf
246      * of the same owner.
247      * 
248      * @param preparedContainer either <code>preparedLocks</code> or
249      * <code>preparedReleases</code> depending on the situation.
250      * 
251      * @param requestedDecree instance of <code>LockDecree</code> representing
252      * the lock.
253      */
254     private boolean checkPrepared(HashMap   preparedContainer, 
255         LockDecree requestedDecree)
256     {
257         LockDecree preparedDecree =
258             (LockDecree)preparedContainer.get(requestedDecree.getKey());
259 
260         // if prepared lock is not valid, remove it from the list
261         if ((preparedDecree != null) && !preparedDecree.isValid()) {
262             preparedContainer.remove(preparedDecree.getKey());
263 
264             preparedDecree = null;
265         }
266 
267         if (preparedDecree != null) {
268             if (requestedDecree.requester.equals(preparedDecree.requester))
269                 return true;
270             else
271                 return false;
272         } else
273             // it was not prepared... sorry...
274             return true;
275     }
276 
277     /**
278      * Prepare phase for the lock acquisition or release.
279      * 
280      * @param decree should be an instance <code>LockDecree</code>, if not,
281      * we throw <code>VoteException</code> to be ignored by the
282      * <code>VoteChannel</code>.
283      * 
284      * @return <code>true</code> when preparing the lock operation succeeds.
285      * 
286      * @throws VoteException if we should be ignored during voting.
287      */
288     public synchronized boolean prepare(Object   decree) throws VoteException {
289         if (!(decree instanceof LockDecree))
290             throw new VoteException("Uknown decree type. Ignore me.");
291             
292         if (decree instanceof AcquireLockDecree) {
293             AcquireLockDecree acquireDecree = (AcquireLockDecree)decree;
294             if(log.isDebugEnabled()) log.debug("Preparing to acquire decree " + acquireDecree.lockId);
295             
296             if (!checkPrepared(preparedLocks, acquireDecree))
297                 // there is a prepared lock owned by third party
298                 return false;
299 
300             if (canLock(acquireDecree)) {
301                 preparedLocks.put(acquireDecree.getKey(), acquireDecree);
302                 return true;
303             } else
304                 // we are unable to aquire local lock
305                 return false;
306         } else
307         if (decree instanceof ReleaseLockDecree) {
308             ReleaseLockDecree releaseDecree = (ReleaseLockDecree)decree;
309             
310 
311                 if(log.isDebugEnabled()) log.debug("Preparing to release decree " + releaseDecree.lockId);
312 
313             if (!checkPrepared(preparedReleases, releaseDecree))
314                 // there is a prepared release owned by third party
315                 return false;
316 
317             if (canRelease(releaseDecree)) {
318                 preparedReleases.put(releaseDecree.getKey(), releaseDecree);
319                 // we have local lock and the prepared lock
320                 return true;
321             } else
322                 // we were unable to aquire local lock
323                 return false;
324         }
325 
326         // we should not be here
327         return false;
328     }
329 
330     /**
331      * Commit phase for the lock acquisition or release.
332      * 
333      * @param decree should be an instance <code>LockDecree</code>, if not,
334      * we throw <code>VoteException</code> to be ignored by the
335      * <code>VoteChannel</code>.
336      * 
337      * @return <code>true</code> when commiting the lock operation succeeds.
338      * 
339      * @throws VoteException if we should be ignored during voting.
340      */
341     public synchronized boolean commit(Object   decree) throws VoteException {
342         if (!(decree instanceof LockDecree))
343             throw new VoteException("Uknown decree type. Ignore me.");
344 
345         if (decree instanceof AcquireLockDecree) {
346             
347 
348                 if(log.isDebugEnabled()) log.debug("Committing decree acquisition " + ((LockDecree)decree).lockId);
349             
350             if (!checkPrepared(preparedLocks, (LockDecree)decree))
351                 // there is a prepared lock owned by third party
352                 return false;
353 
354             if (localLock((LockDecree)decree)) {
355                 preparedLocks.remove(((LockDecree)decree).getKey());
356                 return true;
357             } else
358                 return false;
359         } else
360         if (decree instanceof ReleaseLockDecree) {
361             
362 
363                 if(log.isDebugEnabled()) log.debug("Committing decree release " + ((LockDecree)decree).lockId);
364             
365             if (!checkPrepared(preparedReleases, (LockDecree)decree))
366                 // there is a prepared release owned by third party
367                 return false;
368 
369             if (localRelease((LockDecree)decree)) {
370                 preparedReleases.remove(((LockDecree)decree).getKey());
371                 return true;
372             } else
373                 return false;
374         }
375 
376         // we should not be here
377         return false;
378     }
379 
380     /**
381      * Abort phase for the lock acquisition or release.
382      * 
383      * @param decree should be an instance <code>LockDecree</code>, if not,
384      * we throw <code>VoteException</code> to be ignored by the
385      * <code>VoteChannel</code>.
386      * 
387      * @throws VoteException if we should be ignored during voting.
388      */
389     public synchronized void abort(Object   decree) throws VoteException {
390         if (!(decree instanceof LockDecree))
391             throw new VoteException("Uknown decree type. Ignore me.");
392 
393         if (decree instanceof AcquireLockDecree) {
394             
395 
396                 if(log.isDebugEnabled()) log.debug("Aborting decree acquisition " + ((LockDecree)decree).lockId);
397             
398             if (!checkPrepared(preparedLocks, (LockDecree)decree))
399                 // there is a prepared lock owned by third party
400                 return;
401 
402             preparedLocks.remove(((LockDecree)decree).getKey());
403         } else
404         if (decree instanceof ReleaseLockDecree) {
405             
406 
407                 if(log.isDebugEnabled()) log.debug("Aborting decree release " + ((LockDecree)decree).lockId);
408             
409             if (!checkPrepared(preparedReleases, (LockDecree)decree))
410                 // there is a prepared release owned by third party
411                 return;
412 
413             preparedReleases.remove(((LockDecree)decree).getKey());
414         }
415 
416     }
417 
418     /**
419      * This class represents the lock
420      */
421     public static class LockDecree implements Serializable   {
422 
423         protected final Object   lockId;
424         protected final Object   requester;
425         protected final Object   managerId;
426 
427         protected boolean commited;
428 
429         private LockDecree(Object   lockId, Object   requester, Object   managerId) {
430             this.lockId = lockId;
431             this.requester = requester;
432             this.managerId = managerId;
433         }
434 
435         /**
436          * Returns the key that should be used for Map lookup.
437          */
438         public Object   getKey() { return lockId; }
439 
440         /**
441          * This is a place-holder for future lock expiration code.
442          */
443         public boolean isValid() { return true; }
444 
445         public void commit() { this.commited = true; }
446 
447 
448         /**
449          * This is hashcode from the java.lang.Long class.
450          */
451         public int hashCode() {
452             return lockId.hashCode();
453         }
454 
455         public boolean equals(Object   other) {
456 
457             if (other instanceof LockDecree) {
458                 return ((LockDecree)other).lockId.equals(this.lockId);
459             } else {
460                 return false;
461             }
462         }
463     }
464 
465 
466     /**
467      * This class represents the lock to be released.
468      */
469     public static class AcquireLockDecree extends LockDecree {
470         private final long creationTime;
471 
472         private AcquireLockDecree(LockDecree lockDecree) {
473             this(lockDecree.lockId, lockDecree.requester, lockDecree.managerId);
474         }
475 
476         private AcquireLockDecree(Object   lockId, Object   requester, Object   managerId) {
477             super(lockId, requester, managerId);
478             this.creationTime = System.currentTimeMillis();
479         }
480 
481         /**
482          * Lock aquire decree is valid for a <code>ACQUIRE_EXPIRATION</code>
483          * time after creation and if the lock is still valid (in the
484          * future locks will be leased for a predefined period of time).
485          */
486         public boolean isValid() {
487             boolean result =  super.isValid();
488 
489             if (!commited && result)
490                 result = ((creationTime + ACQUIRE_EXPIRATION) > System.currentTimeMillis());
491 
492             return result;
493         }
494 
495     }
496 
497     /**
498      * This class represents the lock to be released.
499      */
500     public static class ReleaseLockDecree extends LockDecree {
501         ReleaseLockDecree(Object   lockId, Object   requester, Object   managerId) {
502             super(lockId, requester, managerId);
503         }
504     }
505 }
506
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags