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}