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 getMapper().close(); 143 } finally { 144 serializationLock.unlock(); 145 } 146 } 147 148 @Override 149 public Lock getLock(final String id) { 150 serializationLock.lock(); 151 try { 152 Lock lock; 153 if (caching && (lock = lockCache.get(id)) != null) { 154 return lock == NULL_LOCK ? null : lock; 155 } 156 // no transaction needed, single operation 157 lock = getMapper().getLock(idFromString(id)); 158 if (caching) { 159 lockCache.put(id, lock == null ? NULL_LOCK : lock); 160 } 161 return lock; 162 } finally { 163 serializationLock.unlock(); 164 } 165 } 166 167 @Override 168 public Lock setLock(String id, Lock lock) { 169 // We don't call addSuppressed() on an existing exception 170 // because constructing it beforehand when it most likely 171 // won't be needed is expensive. 172 List<Throwable> suppressed = new ArrayList<>(0); 173 long sleepDelay = LOCK_SLEEP_DELAY; 174 for (int i = 0; i < LOCK_RETRIES; i++) { 175 if (i > 0) { 176 log.debug("Retrying lock on " + id + ": try " + (i + 1)); 177 } 178 try { 179 return setLockInternal(id, lock); 180 } catch (NuxeoException e) { 181 suppressed.add(e); 182 if (shouldRetry(e)) { 183 // cluster: two simultaneous inserts 184 // retry 185 try { 186 Thread.sleep(sleepDelay); 187 } catch (InterruptedException ie) { 188 // restore interrupted status 189 Thread.currentThread().interrupt(); 190 throw new RuntimeException(ie); 191 } 192 sleepDelay += LOCK_SLEEP_INCREMENT; 193 continue; 194 } 195 // not something to retry 196 NuxeoException exception = new NuxeoException(e); 197 for (Throwable t : suppressed) { 198 exception.addSuppressed(t); 199 } 200 throw exception; 201 } 202 } 203 LockException exception = new LockException("Failed to lock " + id + ", too much concurrency (tried " 204 + LOCK_RETRIES + " times)"); 205 for (Throwable t : suppressed) { 206 exception.addSuppressed(t); 207 } 208 throw exception; 209 } 210 211 /** 212 * Does the exception mean that we should retry the transaction? 213 */ 214 protected boolean shouldRetry(Exception e) { 215 if (e instanceof ConcurrentUpdateException) { 216 return true; 217 } 218 Throwable t = e.getCause(); 219 if (t instanceof BatchUpdateException && t.getCause() != null) { 220 t = t.getCause(); 221 } 222 return t instanceof SQLException && shouldRetry((SQLException) t); 223 } 224 225 protected boolean shouldRetry(SQLException e) { 226 String sqlState = e.getSQLState(); 227 if ("23000".equals(sqlState)) { 228 // MySQL: Duplicate entry ... for key ... 229 // Oracle: unique constraint ... violated 230 // SQL Server: Violation of PRIMARY KEY constraint 231 return true; 232 } 233 if ("23001".equals(sqlState)) { 234 // H2: Unique index or primary key violation 235 return true; 236 } 237 if ("23505".equals(sqlState)) { 238 // PostgreSQL: duplicate key value violates unique constraint 239 return true; 240 } 241 if ("S0003".equals(sqlState) || "S0005".equals(sqlState)) { 242 // SQL Server: Snapshot isolation transaction aborted due to update 243 // conflict 244 return true; 245 } 246 return false; 247 } 248 249 protected Lock setLockInternal(String id, Lock lock) { 250 serializationLock.lock(); 251 try { 252 Lock oldLock; 253 if (caching && (oldLock = lockCache.get(id)) != null && oldLock != NULL_LOCK) { 254 return oldLock; 255 } 256 oldLock = getMapper().setLock(idFromString(id), lock); 257 if (caching && oldLock == null) { 258 lockCache.put(id, lock == null ? NULL_LOCK : lock); 259 } 260 return oldLock; 261 } finally { 262 serializationLock.unlock(); 263 } 264 } 265 266 @Override 267 public Lock removeLock(final String id, final String owner) { 268 serializationLock.lock(); 269 try { 270 Lock oldLock = null; 271 if (caching && (oldLock = lockCache.get(id)) == NULL_LOCK) { 272 return null; 273 } 274 if (oldLock != null && !LockManager.canLockBeRemoved(oldLock.getOwner(), owner)) { 275 // existing mismatched lock, flag failure 276 oldLock = new Lock(oldLock, true); 277 } else { 278 if (oldLock == null) { 279 oldLock = getMapper().removeLock(idFromString(id), owner, false); 280 } else { 281 // we know the previous lock, we can force 282 // no transaction needed, single operation 283 getMapper().removeLock(idFromString(id), owner, true); 284 } 285 } 286 if (caching) { 287 if (oldLock != null && oldLock.getFailed()) { 288 // failed, but we now know the existing lock 289 lockCache.put(id, new Lock(oldLock, false)); 290 } else { 291 lockCache.put(id, NULL_LOCK); 292 } 293 } 294 return oldLock; 295 } finally { 296 serializationLock.unlock(); 297 } 298 } 299 300 @Override 301 public void clearLockManagerCaches() { 302 serializationLock.lock(); 303 try { 304 if (caching) { 305 lockCache.clear(); 306 } 307 } finally { 308 serializationLock.unlock(); 309 } 310 } 311 312 @Override 313 public String toString() { 314 return getClass().getSimpleName() + '(' + repository.getName() + ')'; 315 } 316 317}