lockmanager/lockmanager-impl/src/main/java/org/opendaylight/lockmanager/LockManager.java

   1 /*
   2  * Copyright (c) 2015 Ericsson India Global Services Pvt Ltd. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.lockmanager;
  10
  11 import java.util.concurrent.ExecutionException;
  12 import java.util.concurrent.Future;
  13 import java.util.concurrent.TimeUnit;
  14
  15 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  16 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  17 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  18 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  19 import org.opendaylight.yang.gen.v1.urn.opendaylight.vpnservice.lockmanager.rev150819.LockInput;
  20 import org.opendaylight.yang.gen.v1.urn.opendaylight.vpnservice.lockmanager.rev150819.LockManagerService;
  21 import org.opendaylight.yang.gen.v1.urn.opendaylight.vpnservice.lockmanager.rev150819.TryLockInput;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.vpnservice.lockmanager.rev150819.UnlockInput;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.vpnservice.lockmanager.rev150819.locks.Lock;
  24 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  25 import org.opendaylight.yangtools.yang.common.RpcResult;
  26 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 import com.google.common.base.Optional;
  31 import com.google.common.util.concurrent.CheckedFuture;
  32 import com.google.common.util.concurrent.Futures;
  33
  34 public class LockManager implements LockManagerService {
  35     private static final Logger LOG = LoggerFactory.getLogger(LockManager.class);
  36
  37     private static final int DEFAULT_RETRY_COUNT = 3;
  38     private static final int DEFAULT_WAIT_TIME_IN_MILLIS = 1000;
  39
  40     private final DataBroker broker;
  41
  42     public LockManager(final DataBroker db) {
  43         broker = db;
  44     }
  45
  46     @Override
  47     public Future<RpcResult<Void>> lock(LockInput input) {
  48         String lockName = input.getLockName();
  49         LOG.info("Locking {}" , lockName);
  50         InstanceIdentifier<Lock> lockInstanceIdentifier = LockManagerUtils.getLockInstanceIdentifier(lockName);
  51         Lock lockData = LockManagerUtils.buildLockData(lockName);
  52         try {
  53             getLock(lockInstanceIdentifier, lockData);
  54             RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.success();
  55             LOG.info("Acquired lock {}" , lockName);
  56             return Futures.immediateFuture(lockRpcBuilder.build());
  57         } catch (InterruptedException e) {
  58             RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.failed();
  59             LOG.info("Failed to get lock {}" , lockName);
  60             return Futures.immediateFuture(lockRpcBuilder.build());
  61         }
  62     }
  63
  64     @Override
  65     public Future<RpcResult<Void>> tryLock(TryLockInput input) {
  66         String lockName = input.getLockName();
  67         LOG.info("Locking {}" , lockName);
  68         long waitTime = input.getTime() == null ? DEFAULT_WAIT_TIME_IN_MILLIS * DEFAULT_RETRY_COUNT : input.getTime();
  69         TimeUnit timeUnit = (TimeUnit) (input.getTimeUnit() == null ? TimeUnit.MILLISECONDS: LockManagerUtils.convertToTimeUnit(input.getTimeUnit()));
  70         waitTime = LockManagerUtils.convertToMillis(waitTime, timeUnit);
  71         long retryCount = waitTime / DEFAULT_WAIT_TIME_IN_MILLIS;
  72         InstanceIdentifier<Lock> lockInstanceIdentifier = LockManagerUtils.getLockInstanceIdentifier(lockName);
  73         Lock lockData = LockManagerUtils.buildLockData(lockName);
  74         try {
  75             if (getLock(lockInstanceIdentifier, lockData, retryCount)) {
  76                 RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.success();
  77                 LOG.info("Acquired lock {}" , lockName);
  78                 return Futures.immediateFuture(lockRpcBuilder.build());
  79             }
  80             RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.failed();
  81             LOG.info("Failed to get lock {}" , lockName);
  82             return Futures.immediateFuture(lockRpcBuilder.build());
  83         } catch (Exception e) {
  84             RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.failed();
  85             LOG.info("Failed to get lock {}" , lockName);
  86             return Futures.immediateFuture(lockRpcBuilder.build());
  87         }
  88     }
  89
  90     @Override
  91     public Future<RpcResult<Void>> unlock(UnlockInput input) {
  92         String lockName = input.getLockName();
  93         LOG.info("Unlocking {}" , lockName);
  94         InstanceIdentifier<Lock> lockInstanceIdentifier = LockManagerUtils.getLockInstanceIdentifier(lockName);
  95         unlock(lockName, lockInstanceIdentifier);
  96         RpcResultBuilder<Void> lockRpcBuilder = RpcResultBuilder.success();
  97         return Futures.immediateFuture(lockRpcBuilder.build());
  98     }
  99
 100     /**
 101      * Try to acquire lock indefinitely until it is successful.
 102      * @param lockInstanceIdentifier
 103      * @param lockData
 104      */
 105     private void getLock(final InstanceIdentifier<Lock> lockInstanceIdentifier, final Lock lockData) throws InterruptedException {
 106         try {
 107             if (!readWriteLock(lockInstanceIdentifier, lockData)) {
 108                 LOG.debug("Already locked trying after {}", DEFAULT_WAIT_TIME_IN_MILLIS);
 109                 LockManagerUtils.sleep(DEFAULT_WAIT_TIME_IN_MILLIS);
 110                 getLock(lockInstanceIdentifier, lockData);
 111             }
 112         } catch (ExecutionException e) {
 113             LOG.error("In getLock unable to get lock due to {}, trying again", e.getMessage());
 114             LockManagerUtils.sleep(DEFAULT_WAIT_TIME_IN_MILLIS);
 115             getLock(lockInstanceIdentifier, lockData);
 116         }
 117     }
 118
 119     /**
 120      * Try to acquire lock for mentioned retryCount. Returns true if successfully acquired lock.
 121      * @param lockInstanceIdentifier
 122      * @param lockData
 123      * @param retryCount
 124      * @return
 125      * @throws InterruptedException
 126      */
 127     private boolean getLock(InstanceIdentifier<Lock> lockInstanceIdentifier,
 128             Lock lockData, long retryCount) throws InterruptedException {
 129         if (retryCount < 0) {
 130             return false;
 131         }
 132         try {
 133             if (!readWriteLock(lockInstanceIdentifier, lockData)) {
 134                 LOG.debug("Already locked trying after {}, retry value {}", DEFAULT_WAIT_TIME_IN_MILLIS, retryCount);
 135                 LockManagerUtils.sleep(DEFAULT_WAIT_TIME_IN_MILLIS);
 136                 return getLock(lockInstanceIdentifier, lockData, retryCount - 1);
 137             }
 138         } catch (ExecutionException e) {
 139             LOG.error("In getLock unable to get lock due to {}, trying again, retry value {}", e.getMessage(), retryCount);
 140             LockManagerUtils.sleep(DEFAULT_WAIT_TIME_IN_MILLIS);
 141             return getLock(lockInstanceIdentifier, lockData, retryCount - 1);
 142         }
 143         return true;
 144     }
 145
 146     /**
 147      * Read and write the lock immediately if available. Returns true if successfully locked.
 148      * @param lockInstanceIdentifier
 149      * @param lockData
 150      * @return
 151      * @throws InterruptedException
 152      * @throws ExecutionException
 153      */
 154     private boolean readWriteLock (final InstanceIdentifier<Lock> lockInstanceIdentifier, final Lock lockData) throws InterruptedException, ExecutionException {
 155         ReadWriteTransaction tx = broker.newReadWriteTransaction();
 156         Optional<Lock> result = Optional.absent();
 157         result = tx.read(LogicalDatastoreType.OPERATIONAL, lockInstanceIdentifier).get();
 158         if (!result.isPresent()) {
 159             tx.put(LogicalDatastoreType.OPERATIONAL, lockInstanceIdentifier, lockData, true);
 160             CheckedFuture<Void, TransactionCommitFailedException> futures = tx.submit();
 161             futures.get();
 162             return true;
 163         }
 164         if (result.get().getLockOwner() == Thread.currentThread().getName()) {
 165             return true;
 166         }
 167         return false;
 168     }
 169
 170     private void unlock(final String lockName, final InstanceIdentifier<Lock> lockInstanceIdentifier) {
 171         ReadWriteTransaction tx = broker.newReadWriteTransaction();
 172         Optional<Lock> result = Optional.absent();
 173         try {
 174             result = tx.read(LogicalDatastoreType.OPERATIONAL, lockInstanceIdentifier).get();
 175             if (!result.isPresent()) {
 176                 LOG.info("{} is already unlocked", lockName);
 177                 return;
 178             }
 179             tx.delete(LogicalDatastoreType.OPERATIONAL, lockInstanceIdentifier);
 180             CheckedFuture<Void, TransactionCommitFailedException> futures = tx.submit();
 181             futures.get();
 182         } catch (Exception e) {
 183             LOG.error("In unlock unable to unlock due to {}", e.getMessage());
 184         }
 185     }
 186
 187
 188 }