001/* 002 * (C) Copyright 2006-2011 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * Florent Guillaume 018 */ 019package org.nuxeo.ecm.core.storage.sql; 020 021import java.io.Serializable; 022import java.sql.BatchUpdateException; 023import java.sql.SQLException; 024import java.util.ArrayList; 025import java.util.LinkedHashMap; 026import java.util.List; 027import java.util.Map.Entry; 028import java.util.concurrent.locks.ReentrantLock; 029 030import org.apache.commons.logging.Log; 031import org.apache.commons.logging.LogFactory; 032import org.nuxeo.ecm.core.api.ConcurrentUpdateException; 033import org.nuxeo.ecm.core.api.Lock; 034import org.nuxeo.ecm.core.api.LockException; 035import org.nuxeo.ecm.core.api.NuxeoException; 036import org.nuxeo.ecm.core.model.LockManager; 037import org.nuxeo.ecm.core.storage.sql.coremodel.SQLRepositoryService; 038import org.nuxeo.runtime.api.Framework; 039 040/** 041 * Manager of locks that serializes access to them. 042 * <p> 043 * The public methods called by the session are {@link #setLock}, {@link #removeLock} and {@link #getLock}. Method 044 * {@link #shutdown} must be called when done with the lock manager. 045 * <p> 046 * In cluster mode, changes are executed in a begin/commit so that tests/updates can be atomic. 047 * <p> 048 * Transaction management can be done by hand because we're dealing with a low-level {@link Mapper} and not something 049 * wrapped by a JCA pool. 050 */ 051public class VCSLockManager implements LockManager { 052 053 private static final Log log = LogFactory.getLog(VCSLockManager.class); 054 055 public static final int LOCK_RETRIES = 10; 056 057 public static final long LOCK_SLEEP_DELAY = 1; // 1 ms 058 059 public static final long LOCK_SLEEP_INCREMENT = 50; // add 50 ms each time 060 061 protected final RepositoryImpl repository; 062 063 /** 064 * The mapper to use. In this mapper we only ever touch the lock table, so no need to deal with fulltext and complex 065 * saves, and we don't do prefetch. 066 */ 067 protected Mapper mapper; 068 069 /** 070 * If clustering is enabled then we have to wrap test/set and test/remove in a transaction. 071 */ 072 protected final boolean clusteringEnabled; 073 074 /** 075 * Lock serializing access to the mapper. 076 */ 077 protected final ReentrantLock serializationLock; 078 079 protected static final Lock NULL_LOCK = new Lock(null, null); 080 081 protected final boolean caching; 082 083 /** 084 * A cache of locks, used only in non-cluster mode, when this lock manager is the only one dealing with locks. 085 * <p> 086 * Used under {@link #serializationLock}. 087 */ 088 protected final LRUCache<Serializable, Lock> lockCache; 089 090 protected static final int CACHE_SIZE = 100; 091 092 protected static class LRUCache<K, V> extends LinkedHashMap<K, V> { 093 private static final long serialVersionUID = 1L; 094 095 private final int max; 096 097 public LRUCache(int max) { 098 super(max, 1.0f, true); 099 this.max = max; 100 } 101 102 @Override 103 protected boolean removeEldestEntry(Entry<K, V> eldest) { 104 return size() > max; 105 } 106 } 107 108 /** 109 * Creates a lock manager for the given repository. 110 * <p> 111 * The mapper will from then on be only used and closed by the lock manager. 112 * <p> 113 * {@link #close} must be called when done with the lock manager. 114 */ 115 public VCSLockManager(String repositoryName) { 116 SQLRepositoryService repositoryService = Framework.getService(SQLRepositoryService.class); 117 repository = repositoryService.getRepositoryImpl(repositoryName); 118 clusteringEnabled = repository.getRepositoryDescriptor().getClusteringEnabled(); 119 serializationLock = new ReentrantLock(); 120 caching = !clusteringEnabled; 121 lockCache = caching ? new LRUCache<Serializable, Lock>(CACHE_SIZE) : null; 122 } 123 124 /** 125 * Delay mapper acquisition until the repository has been fully initialized. 126 */ 127 protected Mapper getMapper() { 128 if (mapper == null) { 129 mapper = repository.newMapper(null, false); 130 } 131 return mapper; 132 } 133 134 protected Serializable idFromString(String id) { 135 return repository.getModel().idFromString(id); 136 } 137 138 @Override 139 public void closeLockManager() { 140 serializationLock.lock(); 141 try { 142 if (mapper != null) { 143 getMapper().close(); 144 } 145 } finally { 146 serializationLock.unlock(); 147 } 148 } 149 150 @Override 151 public Lock getLock(final String id) { 152 serializationLock.lock(); 153 try { 154 Lock lock; 155 if (caching && (lock = lockCache.get(id)) != null) { 156 return lock == NULL_LOCK ? null : lock; 157 } 158 // no transaction needed, single operation 159 lock = getMapper().getLock(idFromString(id)); 160 if (caching) { 161 lockCache.put(id, lock == null ? NULL_LOCK : lock); 162 } 163 return lock; 164 } finally { 165 serializationLock.unlock(); 166 } 167 } 168 169 @Override 170 public Lock setLock(String id, Lock lock) { 171 // We don't call addSuppressed() on an existing exception 172 // because constructing it beforehand when it most likely 173 // won't be needed is expensive. 174 List<Throwable> suppressed = new ArrayList<>(0); 175 long sleepDelay = LOCK_SLEEP_DELAY; 176 for (int i = 0; i < LOCK_RETRIES; i++) { 177 if (i > 0) { 178 log.debug("Retrying lock on " + id + ": try " + (i + 1)); 179 } 180 try { 181 return setLockInternal(id, lock); 182 } catch (NuxeoException e) { 183 suppressed.add(e); 184 if (shouldRetry(e)) { 185 // cluster: two simultaneous inserts 186 // retry 187 try { 188 Thread.sleep(sleepDelay); 189 } catch (InterruptedException ie) { 190 // restore interrupted status 191 Thread.currentThread().interrupt(); 192 throw new RuntimeException(ie); 193 } 194 sleepDelay += LOCK_SLEEP_INCREMENT; 195 continue; 196 } 197 // not something to retry 198 NuxeoException exception = new NuxeoException(e); 199 for (Throwable t : suppressed) { 200 exception.addSuppressed(t); 201 } 202 throw exception; 203 } 204 } 205 LockException exception = new LockException("Failed to lock " + id + ", too much concurrency (tried " 206 + LOCK_RETRIES + " times)"); 207 for (Throwable t : suppressed) { 208 exception.addSuppressed(t); 209 } 210 throw exception; 211 } 212 213 /** 214 * Does the exception mean that we should retry the transaction? 215 */ 216 protected boolean shouldRetry(Exception e) { 217 if (e instanceof ConcurrentUpdateException) { 218 return true; 219 } 220 Throwable t = e.getCause(); 221 if (t instanceof BatchUpdateException && t.getCause() != null) { 222 t = t.getCause(); 223 } 224 return t instanceof SQLException && shouldRetry((SQLException) t); 225 } 226 227 protected boolean shouldRetry(SQLException e) { 228 String sqlState = e.getSQLState(); 229 if ("23000".equals(sqlState)) { 230 // MySQL: Duplicate entry ... for key ... 231 // Oracle: unique constraint ... violated 232 // SQL Server: Violation of PRIMARY KEY constraint 233 return true; 234 } 235 if ("23001".equals(sqlState)) { 236 // H2: Unique index or primary key violation 237 return true; 238 } 239 if ("23505".equals(sqlState)) { 240 // PostgreSQL: duplicate key value violates unique constraint 241 return true; 242 } 243 if ("S0003".equals(sqlState) || "S0005".equals(sqlState)) { 244 // SQL Server: Snapshot isolation transaction aborted due to update 245 // conflict 246 return true; 247 } 248 return false; 249 } 250 251 protected Lock setLockInternal(String id, Lock lock) { 252 serializationLock.lock(); 253 try { 254 Lock oldLock; 255 if (caching && (oldLock = lockCache.get(id)) != null && oldLock != NULL_LOCK) { 256 return oldLock; 257 } 258 oldLock = getMapper().setLock(idFromString(id), lock); 259 if (caching && oldLock == null) { 260 lockCache.put(id, lock == null ? NULL_LOCK : lock); 261 } 262 return oldLock; 263 } finally { 264 serializationLock.unlock(); 265 } 266 } 267 268 @Override 269 public Lock removeLock(final String id, final String owner) { 270 serializationLock.lock(); 271 try { 272 Lock oldLock = null; 273 if (caching && (oldLock = lockCache.get(id)) == NULL_LOCK) { 274 return null; 275 } 276 if (oldLock != null && !LockManager.canLockBeRemoved(oldLock.getOwner(), owner)) { 277 // existing mismatched lock, flag failure 278 oldLock = new Lock(oldLock, true); 279 } else { 280 if (oldLock == null) { 281 oldLock = getMapper().removeLock(idFromString(id), owner, false); 282 } else { 283 // we know the previous lock, we can force 284 // no transaction needed, single operation 285 getMapper().removeLock(idFromString(id), owner, true); 286 } 287 } 288 if (caching) { 289 if (oldLock != null && oldLock.getFailed()) { 290 // failed, but we now know the existing lock 291 lockCache.put(id, new Lock(oldLock, false)); 292 } else { 293 lockCache.put(id, NULL_LOCK); 294 } 295 } 296 return oldLock; 297 } finally { 298 serializationLock.unlock(); 299 } 300 } 301 302 @Override 303 public void clearLockManagerCaches() { 304 serializationLock.lock(); 305 try { 306 if (caching) { 307 lockCache.clear(); 308 } 309 } finally { 310 serializationLock.unlock(); 311 } 312 } 313 314 @Override 315 public String toString() { 316 return getClass().getSimpleName() + '(' + repository.getName() + ')'; 317 } 318 319}