View Javadoc

1   /**
2    * Copyright 2010 The Apache Software Foundation
3    *
4    * Licensed to the Apache Software Foundation (ASF) under one
5    * or more contributor license agreements.  See the NOTICE file
6    * distributed with this work for additional information
7    * regarding copyright ownership.  The ASF licenses this file
8    * to you under the Apache License, Version 2.0 (the
9    * "License"); you may not use this file except in compliance
10   * with the License.  You may obtain a copy of the License at
11   *
12   *     http://www.apache.org/licenses/LICENSE-2.0
13   *
14   * Unless required by applicable law or agreed to in writing, software
15   * distributed under the License is distributed on an "AS IS" BASIS,
16   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17   * See the License for the specific language governing permissions and
18   * limitations under the License.
19   */
20  package org.apache.hadoop.hbase.client;
21  
22  import java.io.Closeable;
23  import java.io.IOException;
24  import java.util.Collection;
25  import java.util.List;
26  import java.util.Map;
27  
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.hbase.HBaseConfiguration;
30  import org.apache.hadoop.hbase.HTableDescriptor;
31  import org.apache.hadoop.hbase.client.coprocessor.Batch;
32  import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
33  import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
34  import org.apache.hadoop.hbase.util.Bytes;
35  import org.apache.hadoop.hbase.util.PoolMap;
36  import org.apache.hadoop.hbase.util.PoolMap.PoolType;
37  
38  /**
39   * A simple pool of HTable instances.
40   * 
41   * Each HTablePool acts as a pool for all tables. To use, instantiate an
42   * HTablePool and use {@link #getTable(String)} to get an HTable from the pool.
43   *
44     * This method is not needed anymore, clients should call
45     * HTableInterface.close() rather than returning the tables to the pool
46     *
47   * Once you are done with it, close your instance of {@link HTableInterface}
48   * by calling {@link HTableInterface#close()} rather than returning the tables
49   * to the pool with (deprecated) {@link #putTable(HTableInterface)}.
50   * 
51   * <p>
52   * A pool can be created with a <i>maxSize</i> which defines the most HTable
53   * references that will ever be retained for each table. Otherwise the default
54   * is {@link Integer#MAX_VALUE}.
55   * 
56   * <p>
57   * Pool will manage its own connections to the cluster. See
58   * {@link HConnectionManager}.
59   * @deprecated Use {@link HConnection#getTable(String)} instead.
60   */
61  public class HTablePool implements Closeable {
62    private final PoolMap<String, HTableInterface> tables;
63    private final int maxSize;
64    private final PoolType poolType;
65    private final Configuration config;
66    private final HTableInterfaceFactory tableFactory;
67  
68    /**
69     * Default Constructor. Default HBaseConfiguration and no limit on pool size.
70     */
71    public HTablePool() {
72      this(HBaseConfiguration.create(), Integer.MAX_VALUE);
73    }
74  
75    /**
76     * Constructor to set maximum versions and use the specified configuration.
77     * 
78     * @param config
79     *          configuration
80     * @param maxSize
81     *          maximum number of references to keep for each table
82     */
83    public HTablePool(final Configuration config, final int maxSize) {
84      this(config, maxSize, null, null);
85    }
86  
87    /**
88     * Constructor to set maximum versions and use the specified configuration and
89     * table factory.
90     * 
91     * @param config
92     *          configuration
93     * @param maxSize
94     *          maximum number of references to keep for each table
95     * @param tableFactory
96     *          table factory
97     */
98    public HTablePool(final Configuration config, final int maxSize,
99        final HTableInterfaceFactory tableFactory) {
100     this(config, maxSize, tableFactory, PoolType.Reusable);
101   }
102 
103   /**
104    * Constructor to set maximum versions and use the specified configuration and
105    * pool type.
106    * 
107    * @param config
108    *          configuration
109    * @param maxSize
110    *          maximum number of references to keep for each table
111    * @param poolType
112    *          pool type which is one of {@link PoolType#Reusable} or
113    *          {@link PoolType#ThreadLocal}
114    */
115   public HTablePool(final Configuration config, final int maxSize,
116       final PoolType poolType) {
117     this(config, maxSize, null, poolType);
118   }
119 
120   /**
121    * Constructor to set maximum versions and use the specified configuration,
122    * table factory and pool type. The HTablePool supports the
123    * {@link PoolType#Reusable} and {@link PoolType#ThreadLocal}. If the pool
124    * type is null or not one of those two values, then it will default to
125    * {@link PoolType#Reusable}.
126    * 
127    * @param config
128    *          configuration
129    * @param maxSize
130    *          maximum number of references to keep for each table
131    * @param tableFactory
132    *          table factory
133    * @param poolType
134    *          pool type which is one of {@link PoolType#Reusable} or
135    *          {@link PoolType#ThreadLocal}
136    */
137   public HTablePool(final Configuration config, final int maxSize,
138       final HTableInterfaceFactory tableFactory, PoolType poolType) {
139     // Make a new configuration instance so I can safely cleanup when
140     // done with the pool.
141     this.config = config == null ? HBaseConfiguration.create() : config;
142     this.maxSize = maxSize;
143     this.tableFactory = tableFactory == null ? new HTableFactory()
144         : tableFactory;
145     if (poolType == null) {
146       this.poolType = PoolType.Reusable;
147     } else {
148       switch (poolType) {
149       case Reusable:
150       case ThreadLocal:
151         this.poolType = poolType;
152         break;
153       default:
154         this.poolType = PoolType.Reusable;
155         break;
156       }
157     }
158     this.tables = new PoolMap<String, HTableInterface>(this.poolType,
159         this.maxSize);
160   }
161 
162   /**
163    * Get a reference to the specified table from the pool.
164    * <p>
165    * <p/>
166    * 
167    * @param tableName
168    *          table name
169    * @return a reference to the specified table
170    * @throws RuntimeException
171    *           if there is a problem instantiating the HTable
172    */
173   public HTableInterface getTable(String tableName) {
174     // call the old getTable implementation renamed to findOrCreateTable
175     HTableInterface table = findOrCreateTable(tableName);
176     // return a proxy table so when user closes the proxy, the actual table
177     // will be returned to the pool
178     return new PooledHTable(table);
179   }
180 
181   /**
182    * Get a reference to the specified table from the pool.
183    * <p>
184    * 
185    * Create a new one if one is not available.
186    * 
187    * @param tableName
188    *          table name
189    * @return a reference to the specified table
190    * @throws RuntimeException
191    *           if there is a problem instantiating the HTable
192    */
193   private HTableInterface findOrCreateTable(String tableName) {
194     HTableInterface table = tables.get(tableName);
195     if (table == null) {
196       table = createHTable(tableName);
197     }
198     return table;
199   }
200 
201   /**
202    * Get a reference to the specified table from the pool.
203    * <p>
204    * 
205    * Create a new one if one is not available.
206    * 
207    * @param tableName
208    *          table name
209    * @return a reference to the specified table
210    * @throws RuntimeException
211    *           if there is a problem instantiating the HTable
212    */
213   public HTableInterface getTable(byte[] tableName) {
214     return getTable(Bytes.toString(tableName));
215   }
216 
217   /**
218    * This method is not needed anymore, clients should call
219    * HTableInterface.close() rather than returning the tables to the pool
220    * 
221    * @param table
222    *          the proxy table user got from pool
223    * @deprecated
224    */
225   public void putTable(HTableInterface table) throws IOException {
226     // we need to be sure nobody puts a proxy implementation in the pool
227     // but if the client code is not updated
228     // and it will continue to call putTable() instead of calling close()
229     // then we need to return the wrapped table to the pool instead of the
230     // proxy
231     // table
232     if (table instanceof PooledHTable) {
233       returnTable(((PooledHTable) table).getWrappedTable());
234     } else {
235       // normally this should not happen if clients pass back the same
236       // table
237       // object they got from the pool
238       // but if it happens then it's better to reject it
239       throw new IllegalArgumentException("not a pooled table: " + table);
240     }
241   }
242 
243   /**
244    * Puts the specified HTable back into the pool.
245    * <p>
246    * 
247    * If the pool already contains <i>maxSize</i> references to the table, then
248    * the table instance gets closed after flushing buffered edits.
249    * 
250    * @param table
251    *          table
252    */
253   private void returnTable(HTableInterface table) throws IOException {
254     // this is the old putTable method renamed and made private
255     String tableName = Bytes.toString(table.getTableName());
256     if (tables.size(tableName) >= maxSize) {
257       // release table instance since we're not reusing it
258       this.tables.remove(tableName, table);
259       this.tableFactory.releaseHTableInterface(table);
260       return;
261     }
262     tables.put(tableName, table);
263   }
264 
265   protected HTableInterface createHTable(String tableName) {
266     return this.tableFactory.createHTableInterface(config,
267         Bytes.toBytes(tableName));
268   }
269 
270   /**
271    * Closes all the HTable instances , belonging to the given table, in the
272    * table pool.
273    * <p>
274    * Note: this is a 'shutdown' of the given table pool and different from
275    * {@link #putTable(HTableInterface)}, that is used to return the table
276    * instance to the pool for future re-use.
277    * 
278    * @param tableName
279    */
280   public void closeTablePool(final String tableName) throws IOException {
281     Collection<HTableInterface> tables = this.tables.values(tableName);
282     if (tables != null) {
283       for (HTableInterface table : tables) {
284         this.tableFactory.releaseHTableInterface(table);
285       }
286     }
287     this.tables.remove(tableName);
288   }
289 
290   /**
291    * See {@link #closeTablePool(String)}.
292    * 
293    * @param tableName
294    */
295   public void closeTablePool(final byte[] tableName) throws IOException {
296     closeTablePool(Bytes.toString(tableName));
297   }
298 
299   /**
300    * Closes all the HTable instances , belonging to all tables in the table
301    * pool.
302    * <p>
303    * Note: this is a 'shutdown' of all the table pools.
304    */
305   public void close() throws IOException {
306     for (String tableName : tables.keySet()) {
307       closeTablePool(tableName);
308     }
309     this.tables.clear();
310   }
311 
312   public int getCurrentPoolSize(String tableName) {
313     return tables.size(tableName);
314   }
315 
316   /**
317    * A proxy class that implements HTableInterface.close method to return the
318    * wrapped table back to the table pool
319    *
320    */
321   class PooledHTable implements HTableInterface {
322 
323     private HTableInterface table; // actual table implementation
324 
325     public PooledHTable(HTableInterface table) {
326       this.table = table;
327     }
328 
329     @Override
330     public byte[] getTableName() {
331       return table.getTableName();
332     }
333 
334     @Override
335     public Configuration getConfiguration() {
336       return table.getConfiguration();
337     }
338 
339     @Override
340     public HTableDescriptor getTableDescriptor() throws IOException {
341       return table.getTableDescriptor();
342     }
343 
344     @Override
345     public boolean exists(Get get) throws IOException {
346       return table.exists(get);
347     }
348 
349     @Override
350     public void batch(List<? extends Row> actions, Object[] results) throws IOException,
351         InterruptedException {
352       table.batch(actions, results);
353     }
354 
355     @Override
356     public Object[] batch(List<? extends Row> actions) throws IOException,
357         InterruptedException {
358       return table.batch(actions);
359     }
360 
361     @Override
362     public Result get(Get get) throws IOException {
363       return table.get(get);
364     }
365 
366     @Override
367     public Result[] get(List<Get> gets) throws IOException {
368       return table.get(gets);
369     }
370 
371     @Override
372     @SuppressWarnings("deprecation")
373     public Result getRowOrBefore(byte[] row, byte[] family) throws IOException {
374       return table.getRowOrBefore(row, family);
375     }
376 
377     @Override
378     public ResultScanner getScanner(Scan scan) throws IOException {
379       return table.getScanner(scan);
380     }
381 
382     @Override
383     public ResultScanner getScanner(byte[] family) throws IOException {
384       return table.getScanner(family);
385     }
386 
387     @Override
388     public ResultScanner getScanner(byte[] family, byte[] qualifier)
389         throws IOException {
390       return table.getScanner(family, qualifier);
391     }
392 
393     @Override
394     public void put(Put put) throws IOException {
395       table.put(put);
396     }
397 
398     @Override
399     public void put(List<Put> puts) throws IOException {
400       table.put(puts);
401     }
402 
403     @Override
404     public boolean checkAndPut(byte[] row, byte[] family, byte[] qualifier,
405         byte[] value, Put put) throws IOException {
406       return table.checkAndPut(row, family, qualifier, value, put);
407     }
408 
409     @Override
410     public void delete(Delete delete) throws IOException {
411       table.delete(delete);
412     }
413 
414     @Override
415     public void delete(List<Delete> deletes) throws IOException {
416       table.delete(deletes);
417     }
418 
419     @Override
420     public boolean checkAndDelete(byte[] row, byte[] family, byte[] qualifier,
421         byte[] value, Delete delete) throws IOException {
422       return table.checkAndDelete(row, family, qualifier, value, delete);
423     }
424 
425     @Override
426     public Result increment(Increment increment) throws IOException {
427       return table.increment(increment);
428     }
429 
430     @Override
431     public long incrementColumnValue(byte[] row, byte[] family,
432         byte[] qualifier, long amount) throws IOException {
433       return table.incrementColumnValue(row, family, qualifier, amount);
434     }
435 
436     @Override
437     public long incrementColumnValue(byte[] row, byte[] family,
438         byte[] qualifier, long amount, boolean writeToWAL) throws IOException {
439       return table.incrementColumnValue(row, family, qualifier, amount,
440           writeToWAL);
441     }
442 
443     @Override
444     public boolean isAutoFlush() {
445       return table.isAutoFlush();
446     }
447 
448     @Override
449     public void flushCommits() throws IOException {
450       table.flushCommits();
451     }
452 
453     /**
454      * Returns the actual table back to the pool
455      * 
456      * @throws IOException
457      */
458     public void close() throws IOException {
459       returnTable(table);
460     }
461 
462     /**
463      * @deprecated {@link RowLock} and associated operations are deprecated
464      */
465     @Override
466     public RowLock lockRow(byte[] row) throws IOException {
467       return table.lockRow(row);
468     }
469 
470     /**
471      * @deprecated {@link RowLock} and associated operations are deprecated
472      */
473     @Override
474     public void unlockRow(RowLock rl) throws IOException {
475       table.unlockRow(rl);
476     }
477 
478     @Override
479     public <T extends CoprocessorProtocol> T coprocessorProxy(
480         Class<T> protocol, byte[] row) {
481       return table.coprocessorProxy(protocol, row);
482     }
483 
484     @Override
485     public <T extends CoprocessorProtocol, R> Map<byte[], R> coprocessorExec(
486         Class<T> protocol, byte[] startKey, byte[] endKey,
487         Batch.Call<T, R> callable) throws IOException, Throwable {
488       return table.coprocessorExec(protocol, startKey, endKey, callable);
489     }
490 
491     @Override
492     public <T extends CoprocessorProtocol, R> void coprocessorExec(
493         Class<T> protocol, byte[] startKey, byte[] endKey,
494         Batch.Call<T, R> callable, Batch.Callback<R> callback)
495         throws IOException, Throwable {
496       table.coprocessorExec(protocol, startKey, endKey, callable, callback);
497     }
498 
499     @Override
500     public String toString() {
501       return "PooledHTable{" + ", table=" + table + '}';
502     }
503 
504     /**
505      * Expose the wrapped HTable to tests in the same package
506      * 
507      * @return wrapped htable
508      */
509     HTableInterface getWrappedTable() {
510       return table;
511     }
512 
513     @Override
514     public void mutateRow(RowMutations rm) throws IOException {
515       table.mutateRow(rm);
516     }
517 
518     @Override
519     public Result append(Append append) throws IOException {
520       return table.append(append);
521     }
522 
523     @Override
524     public void setAutoFlush(boolean autoFlush) {
525       table.setAutoFlush(autoFlush);
526     }
527 
528     @Override
529     public void setAutoFlush(boolean autoFlush, boolean clearBufferOnFail) {
530       table.setAutoFlush(autoFlush, clearBufferOnFail);
531     }
532 
533     @Override
534     public long getWriteBufferSize() {
535       return table.getWriteBufferSize();
536     }
537 
538     @Override
539     public void setWriteBufferSize(long writeBufferSize) throws IOException {
540       table.setWriteBufferSize(writeBufferSize);
541     }
542   }
543 }