001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.util;
019
020import java.io.IOException;
021import java.io.InterruptedIOException;
022import java.util.concurrent.ConcurrentHashMap;
023import java.util.concurrent.ConcurrentMap;
024
025import org.apache.yetus.audience.InterfaceAudience;
026
027import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
028import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
029
030/**
031 * Allows multiple concurrent clients to lock on a numeric id with a minimal
032 * memory overhead. The intended usage is as follows:
033 *
034 * <pre>
035 * IdLock.Entry lockEntry = idLock.getLockEntry(id);
036 * try {
037 *   // User code.
038 * } finally {
039 *   idLock.releaseLockEntry(lockEntry);
040 * }</pre>
041 */
042@InterfaceAudience.Private
043public class IdLock {
044
045  /** An entry returned to the client as a lock object */
046  public static final class Entry {
047    private final long id;
048    private int numWaiters;
049    private boolean locked = true;
050
051    private Entry(long id) {
052      this.id = id;
053    }
054
055    @Override
056    public String toString() {
057      return "id=" + id + ", numWaiter=" + numWaiters + ", isLocked="
058          + locked;
059    }
060  }
061
062  private ConcurrentMap<Long, Entry> map = new ConcurrentHashMap<>();
063
064  /**
065   * Blocks until the lock corresponding to the given id is acquired.
066   *
067   * @param id an arbitrary number to lock on
068   * @return an "entry" to pass to {@link #releaseLockEntry(Entry)} to release
069   *         the lock
070   * @throws IOException if interrupted
071   */
072  public Entry getLockEntry(long id) throws IOException {
073    Entry entry = new Entry(id);
074    Entry existing;
075    while ((existing = map.putIfAbsent(entry.id, entry)) != null) {
076      synchronized (existing) {
077        if (existing.locked) {
078          ++existing.numWaiters;  // Add ourselves to waiters.
079          while (existing.locked) {
080            try {
081              existing.wait();
082            } catch (InterruptedException e) {
083              --existing.numWaiters;  // Remove ourselves from waiters.
084              throw new InterruptedIOException(
085                  "Interrupted waiting to acquire sparse lock");
086            }
087          }
088
089          --existing.numWaiters;  // Remove ourselves from waiters.
090          existing.locked = true;
091          return existing;
092        }
093        // If the entry is not locked, it might already be deleted from the
094        // map, so we cannot return it. We need to get our entry into the map
095        // or get someone else's locked entry.
096      }
097    }
098    return entry;
099  }
100
101  /**
102   * Blocks until the lock corresponding to the given id is acquired.
103   *
104   * @param id an arbitrary number to lock on
105   * @param time time to wait in ms
106   * @return an "entry" to pass to {@link #releaseLockEntry(Entry)} to release
107   *         the lock
108   * @throws IOException if interrupted
109   */
110  public Entry tryLockEntry(long id, long time) throws IOException {
111    Preconditions.checkArgument(time >= 0);
112    Entry entry = new Entry(id);
113    Entry existing;
114    long waitUtilTS = System.currentTimeMillis() + time;
115    long remaining = time;
116    while ((existing = map.putIfAbsent(entry.id, entry)) != null) {
117      synchronized (existing) {
118        if (existing.locked) {
119          ++existing.numWaiters;  // Add ourselves to waiters.
120          try {
121            while (existing.locked) {
122              existing.wait(remaining);
123              if (existing.locked) {
124                long currentTS = System.currentTimeMillis();
125                if (currentTS >= waitUtilTS) {
126                  // time is up
127                  return null;
128                } else {
129                  // our wait is waken, but the lock is still taken, this can happen
130                  // due to JDK Object's wait/notify mechanism.
131                  // Calculate the new remaining time to wait
132                  remaining = waitUtilTS - currentTS;
133                }
134              }
135
136            }
137          } catch (InterruptedException e) {
138            throw new InterruptedIOException(
139                "Interrupted waiting to acquire sparse lock");
140          } finally {
141            --existing.numWaiters;  // Remove ourselves from waiters.
142          }
143          existing.locked = true;
144          return existing;
145        }
146        // If the entry is not locked, it might already be deleted from the
147        // map, so we cannot return it. We need to get our entry into the map
148        // or get someone else's locked entry.
149      }
150    }
151    return entry;
152  }
153
154  /**
155   * Must be called in a finally block to decrease the internal counter and
156   * remove the monitor object for the given id if the caller is the last
157   * client.
158   *
159   * @param entry the return value of {@link #getLockEntry(long)}
160   */
161  public void releaseLockEntry(Entry entry) {
162    synchronized (entry) {
163      entry.locked = false;
164      if (entry.numWaiters > 0) {
165        entry.notify();
166      } else {
167        map.remove(entry.id);
168      }
169    }
170  }
171
172  /** For testing */
173  void assertMapEmpty() {
174    assert map.isEmpty();
175  }
176
177  @VisibleForTesting
178  public void waitForWaiters(long id, int numWaiters) throws InterruptedException {
179    for (Entry entry;;) {
180      entry = map.get(id);
181      if (entry != null) {
182        synchronized (entry) {
183          if (entry.numWaiters >= numWaiters) {
184            return;
185          }
186        }
187      }
188      Thread.sleep(100);
189    }
190  }
191}