1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17 package org.apache.hadoop.hbase.util;
18
19 import static org.apache.hadoop.hbase.util.test.LoadTestDataGenerator.INCREMENT;
20 import static org.apache.hadoop.hbase.util.test.LoadTestDataGenerator.MUTATE_INFO;
21
22 import java.io.IOException;
23 import java.util.Arrays;
24 import java.util.Collection;
25 import java.util.HashMap;
26 import java.util.Map;
27 import java.util.Random;
28 import java.util.Set;
29 import java.util.concurrent.atomic.AtomicInteger;
30 import java.util.concurrent.atomic.AtomicLong;
31
32 import org.apache.commons.logging.Log;
33 import org.apache.commons.logging.LogFactory;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.hbase.HBaseTestingUtility;
36 import org.apache.hadoop.hbase.TableName;
37 import org.apache.hadoop.hbase.client.Result;
38 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
39 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
40 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType;
41 import org.apache.hadoop.hbase.util.test.LoadTestDataGenerator;
42 import org.apache.hadoop.hbase.util.test.LoadTestKVGenerator;
43 import org.apache.hadoop.util.StringUtils;
44
45 import com.google.common.base.Preconditions;
46
47
48
49
50
51 public abstract class MultiThreadedAction {
52 private static final Log LOG = LogFactory.getLog(MultiThreadedAction.class);
53
54 protected final TableName tableName;
55 protected final Configuration conf;
56
57 protected int numThreads = 1;
58
59
60 protected long startKey = 0;
61
62
63 protected long endKey = 1;
64
65 protected AtomicInteger numThreadsWorking = new AtomicInteger();
66 protected AtomicLong numKeys = new AtomicLong();
67 protected AtomicLong numCols = new AtomicLong();
68 protected AtomicLong totalOpTimeMs = new AtomicLong();
69 protected boolean verbose = false;
70
71 protected LoadTestDataGenerator dataGenerator = null;
72
73
74
75
76
77
78
79
80 public static class DefaultDataGenerator extends LoadTestDataGenerator {
81 private byte[][] columnFamilies = null;
82 private int minColumnsPerKey;
83 private int maxColumnsPerKey;
84 private final Random random = new Random();
85
86 public DefaultDataGenerator(int minValueSize, int maxValueSize,
87 int minColumnsPerKey, int maxColumnsPerKey, byte[]... columnFamilies) {
88 super(minValueSize, maxValueSize);
89 this.columnFamilies = columnFamilies;
90 this.minColumnsPerKey = minColumnsPerKey;
91 this.maxColumnsPerKey = maxColumnsPerKey;
92 }
93
94 public DefaultDataGenerator(byte[]... columnFamilies) {
95
96 this(256, 1024, 1, 10, columnFamilies);
97 }
98
99 @Override
100 public byte[] getDeterministicUniqueKey(long keyBase) {
101 return LoadTestKVGenerator.md5PrefixedKey(keyBase).getBytes();
102 }
103
104 @Override
105 public byte[][] getColumnFamilies() {
106 return columnFamilies;
107 }
108
109 @Override
110 public byte[][] generateColumnsForCf(byte[] rowKey, byte[] cf) {
111 int numColumns = minColumnsPerKey + random.nextInt(maxColumnsPerKey - minColumnsPerKey + 1);
112 byte[][] columns = new byte[numColumns][];
113 for (int i = 0; i < numColumns; ++i) {
114 columns[i] = Integer.toString(i).getBytes();
115 }
116 return columns;
117 }
118
119 @Override
120 public byte[] generateValue(byte[] rowKey, byte[] cf, byte[] column) {
121 return kvGenerator.generateRandomSizeValue(rowKey, cf, column);
122 }
123
124 @Override
125 public boolean verify(byte[] rowKey, byte[] cf, byte[] column, byte[] value) {
126 return LoadTestKVGenerator.verify(value, rowKey, cf, column);
127 }
128
129 @Override
130 public boolean verify(byte[] rowKey, byte[] cf, Set<byte[]> columnSet) {
131 return (columnSet.size() >= minColumnsPerKey) && (columnSet.size() <= maxColumnsPerKey);
132 }
133 }
134
135
136 private String actionLetter;
137
138
139 private boolean streamingCounters;
140
141 public static final int REPORTING_INTERVAL_MS = 5000;
142
143 public MultiThreadedAction(LoadTestDataGenerator dataGen, Configuration conf,
144 TableName tableName,
145 String actionLetter) {
146 this.conf = conf;
147 this.dataGenerator = dataGen;
148 this.tableName = tableName;
149 this.actionLetter = actionLetter;
150 }
151
152 public void start(long startKey, long endKey, int numThreads) throws IOException {
153 this.startKey = startKey;
154 this.endKey = endKey;
155 this.numThreads = numThreads;
156 (new Thread(new ProgressReporter(actionLetter))).start();
157 }
158
159 private static String formatTime(long elapsedTime) {
160 String format = String.format("%%0%dd", 2);
161 elapsedTime = elapsedTime / 1000;
162 String seconds = String.format(format, elapsedTime % 60);
163 String minutes = String.format(format, (elapsedTime % 3600) / 60);
164 String hours = String.format(format, elapsedTime / 3600);
165 String time = hours + ":" + minutes + ":" + seconds;
166 return time;
167 }
168
169
170 private class ProgressReporter implements Runnable {
171
172 private String reporterId = "";
173
174 public ProgressReporter(String id) {
175 this.reporterId = id;
176 }
177
178 @Override
179 public void run() {
180 long startTime = System.currentTimeMillis();
181 long priorNumKeys = 0;
182 long priorCumulativeOpTime = 0;
183 int priorAverageKeysPerSecond = 0;
184
185
186 Threads.sleep(REPORTING_INTERVAL_MS);
187
188 while (numThreadsWorking.get() != 0) {
189 String threadsLeft =
190 "[" + reporterId + ":" + numThreadsWorking.get() + "] ";
191 if (numKeys.get() == 0) {
192 LOG.info(threadsLeft + "Number of keys = 0");
193 } else {
194 long numKeys = MultiThreadedAction.this.numKeys.get();
195 long time = System.currentTimeMillis() - startTime;
196 long totalOpTime = totalOpTimeMs.get();
197
198 long numKeysDelta = numKeys - priorNumKeys;
199 long totalOpTimeDelta = totalOpTime - priorCumulativeOpTime;
200
201 double averageKeysPerSecond =
202 (time > 0) ? (numKeys * 1000 / time) : 0;
203
204 LOG.info(threadsLeft
205 + "Keys="
206 + numKeys
207 + ", cols="
208 + StringUtils.humanReadableInt(numCols.get())
209 + ", time="
210 + formatTime(time)
211 + ((numKeys > 0 && time > 0) ? (" Overall: [" + "keys/s= "
212 + numKeys * 1000 / time + ", latency=" + totalOpTime
213 / numKeys + " ms]") : "")
214 + ((numKeysDelta > 0) ? (" Current: [" + "keys/s="
215 + numKeysDelta * 1000 / REPORTING_INTERVAL_MS + ", latency="
216 + totalOpTimeDelta / numKeysDelta + " ms]") : "")
217 + progressInfo());
218
219 if (streamingCounters) {
220 printStreamingCounters(numKeysDelta,
221 averageKeysPerSecond - priorAverageKeysPerSecond);
222 }
223
224 priorNumKeys = numKeys;
225 priorCumulativeOpTime = totalOpTime;
226 priorAverageKeysPerSecond = (int) averageKeysPerSecond;
227 }
228
229 Threads.sleep(REPORTING_INTERVAL_MS);
230 }
231 }
232
233 private void printStreamingCounters(long numKeysDelta,
234 double avgKeysPerSecondDelta) {
235
236
237 System.err.println("reporter:counter:numKeys," + reporterId + ","
238 + numKeysDelta);
239 System.err.println("reporter:counter:numCols," + reporterId + ","
240 + numCols.get());
241 System.err.println("reporter:counter:avgKeysPerSecond," + reporterId
242 + "," + (long) (avgKeysPerSecondDelta));
243 }
244 }
245
246 public void waitForFinish() {
247 while (numThreadsWorking.get() != 0) {
248 Threads.sleepWithoutInterrupt(1000);
249 }
250 }
251
252 public boolean isDone() {
253 return (numThreadsWorking.get() == 0);
254 }
255
256 protected void startThreads(Collection<? extends Thread> threads) {
257 numThreadsWorking.addAndGet(threads.size());
258 for (Thread thread : threads) {
259 thread.start();
260 }
261 }
262
263
264 public long getEndKey() {
265 return endKey;
266 }
267
268
269 protected abstract String progressInfo();
270
271 protected static void appendToStatus(StringBuilder sb, String desc,
272 long v) {
273 if (v == 0) {
274 return;
275 }
276 sb.append(", ");
277 sb.append(desc);
278 sb.append("=");
279 sb.append(v);
280 }
281
282 protected static void appendToStatus(StringBuilder sb, String desc,
283 String v) {
284 sb.append(", ");
285 sb.append(desc);
286 sb.append("=");
287 sb.append(v);
288 }
289
290
291
292
293
294 public boolean verifyResultAgainstDataGenerator(Result result, boolean verifyValues) {
295 return verifyResultAgainstDataGenerator(result, verifyValues, false);
296 }
297
298
299
300
301
302
303
304
305
306
307 public boolean verifyResultAgainstDataGenerator(Result result, boolean verifyValues,
308 boolean verifyCfAndColumnIntegrity) {
309 String rowKeyStr = Bytes.toString(result.getRow());
310
311
312 if (result.isEmpty()) {
313 LOG.error("Error checking data for key [" + rowKeyStr + "], no data returned");
314 return false;
315 }
316
317 if (!verifyValues && !verifyCfAndColumnIntegrity) {
318 return true;
319 }
320
321
322 byte[][] expectedCfs = dataGenerator.getColumnFamilies();
323 if (verifyCfAndColumnIntegrity && (expectedCfs.length != result.getMap().size())) {
324 LOG.error("Error checking data for key [" + rowKeyStr
325 + "], bad family count: " + result.getMap().size());
326 return false;
327 }
328
329
330 for (byte[] cf : result.getMap().keySet()) {
331 String cfStr = Bytes.toString(cf);
332 Map<byte[], byte[]> columnValues = result.getFamilyMap(cf);
333 if (columnValues == null) {
334 LOG.error("Error checking data for key [" + rowKeyStr
335 + "], no data for family [" + cfStr + "]]");
336 return false;
337 }
338
339 Map<String, MutationType> mutateInfo = null;
340 if (verifyCfAndColumnIntegrity || verifyValues) {
341 if (!columnValues.containsKey(MUTATE_INFO)) {
342 LOG.error("Error checking data for key [" + rowKeyStr + "], column family ["
343 + cfStr + "], column [" + Bytes.toString(MUTATE_INFO) + "]; value is not found");
344 return false;
345 }
346
347 long cfHash = Arrays.hashCode(cf);
348
349 byte[] mutateInfoValue = columnValues.remove(MUTATE_INFO);
350 mutateInfo = parseMutateInfo(mutateInfoValue);
351 for (Map.Entry<String, MutationType> mutate: mutateInfo.entrySet()) {
352 if (mutate.getValue() == MutationType.DELETE) {
353 byte[] column = Bytes.toBytes(mutate.getKey());
354 long columnHash = Arrays.hashCode(column);
355 long hashCode = cfHash + columnHash;
356 if (hashCode % 2 == 0) {
357 if (columnValues.containsKey(column)) {
358 LOG.error("Error checking data for key [" + rowKeyStr + "], column family ["
359 + cfStr + "], column [" + mutate.getKey() + "]; should be deleted");
360 return false;
361 }
362 byte[] hashCodeBytes = Bytes.toBytes(hashCode);
363 columnValues.put(column, hashCodeBytes);
364 }
365 }
366 }
367
368
369 if (!columnValues.containsKey(INCREMENT)) {
370 LOG.error("Error checking data for key [" + rowKeyStr + "], column family ["
371 + cfStr + "], column [" + Bytes.toString(INCREMENT) + "]; value is not found");
372 return false;
373 }
374 long currentValue = Bytes.toLong(columnValues.remove(INCREMENT));
375 if (verifyValues) {
376 long amount = mutateInfo.isEmpty() ? 0 : cfHash;
377 long originalValue = Arrays.hashCode(result.getRow());
378 long extra = currentValue - originalValue;
379 if (extra != 0 && (amount == 0 || extra % amount != 0)) {
380 LOG.error("Error checking data for key [" + rowKeyStr + "], column family ["
381 + cfStr + "], column [increment], extra [" + extra + "], amount [" + amount + "]");
382 return false;
383 }
384 if (amount != 0 && extra != amount) {
385 LOG.warn("Warning checking data for key [" + rowKeyStr + "], column family ["
386 + cfStr + "], column [increment], incremented [" + (extra / amount) + "] times");
387 }
388 }
389
390
391 if (verifyCfAndColumnIntegrity
392 && !dataGenerator.verify(result.getRow(), cf, columnValues.keySet())) {
393 String colsStr = "";
394 for (byte[] col : columnValues.keySet()) {
395 if (colsStr.length() > 0) {
396 colsStr += ", ";
397 }
398 colsStr += "[" + Bytes.toString(col) + "]";
399 }
400 LOG.error("Error checking data for key [" + rowKeyStr
401 + "], bad columns for family [" + cfStr + "]: " + colsStr);
402 return false;
403 }
404
405 if (verifyValues) {
406 for (Map.Entry<byte[], byte[]> kv : columnValues.entrySet()) {
407 String column = Bytes.toString(kv.getKey());
408 MutationType mutation = mutateInfo.get(column);
409 boolean verificationNeeded = true;
410 byte[] bytes = kv.getValue();
411 if (mutation != null) {
412 boolean mutationVerified = true;
413 long columnHash = Arrays.hashCode(kv.getKey());
414 long hashCode = cfHash + columnHash;
415 byte[] hashCodeBytes = Bytes.toBytes(hashCode);
416 if (mutation == MutationType.APPEND) {
417 int offset = bytes.length - hashCodeBytes.length;
418 mutationVerified = offset > 0 && Bytes.equals(hashCodeBytes,
419 0, hashCodeBytes.length, bytes, offset, hashCodeBytes.length);
420 if (mutationVerified) {
421 int n = 1;
422 while (true) {
423 int newOffset = offset - hashCodeBytes.length;
424 if (newOffset < 0 || !Bytes.equals(hashCodeBytes, 0,
425 hashCodeBytes.length, bytes, newOffset, hashCodeBytes.length)) {
426 break;
427 }
428 offset = newOffset;
429 n++;
430 }
431 if (n > 1) {
432 LOG.warn("Warning checking data for key [" + rowKeyStr + "], column family ["
433 + cfStr + "], column [" + column + "], appended [" + n + "] times");
434 }
435 byte[] dest = new byte[offset];
436 System.arraycopy(bytes, 0, dest, 0, offset);
437 bytes = dest;
438 }
439 } else if (hashCode % 2 == 0) {
440 mutationVerified = Bytes.equals(bytes, hashCodeBytes);
441 verificationNeeded = false;
442 }
443 if (!mutationVerified) {
444 LOG.error("Error checking data for key [" + rowKeyStr
445 + "], mutation checking failed for column family [" + cfStr + "], column ["
446 + column + "]; mutation [" + mutation + "], hashCode ["
447 + hashCode + "], verificationNeeded ["
448 + verificationNeeded + "]");
449 return false;
450 }
451 }
452 if (verificationNeeded &&
453 !dataGenerator.verify(result.getRow(), cf, kv.getKey(), bytes)) {
454 LOG.error("Error checking data for key [" + rowKeyStr + "], column family ["
455 + cfStr + "], column [" + column + "], mutation [" + mutation
456 + "]; value of length " + bytes.length);
457 return false;
458 }
459 }
460 }
461 }
462 }
463 return true;
464 }
465
466
467 private Map<String, MutationType> parseMutateInfo(byte[] mutateInfo) {
468 Map<String, MutationType> mi = new HashMap<String, MutationType>();
469 if (mutateInfo != null) {
470 String mutateInfoStr = Bytes.toString(mutateInfo);
471 String[] mutations = mutateInfoStr.split("#");
472 for (String mutation: mutations) {
473 if (mutation.isEmpty()) continue;
474 Preconditions.checkArgument(mutation.contains(":"),
475 "Invalid mutation info " + mutation);
476 int p = mutation.indexOf(":");
477 String column = mutation.substring(0, p);
478 MutationType type = MutationType.valueOf(
479 Integer.parseInt(mutation.substring(p+1)));
480 mi.put(column, type);
481 }
482 }
483 return mi;
484 }
485 }