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}