1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.procedure2;
20
21 import java.io.IOException;
22 import java.io.InputStream;
23 import java.io.OutputStream;
24 import java.util.ArrayList;
25 import java.util.concurrent.atomic.AtomicLong;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.fs.FileSystem;
30 import org.apache.hadoop.fs.Path;
31 import org.apache.hadoop.hbase.HBaseCommonTestingUtility;
32 import org.apache.hadoop.hbase.io.util.StreamUtils;
33 import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
34 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
35 import org.apache.hadoop.hbase.testclassification.LargeTests;
36
37 import org.junit.After;
38 import org.junit.Before;
39 import org.junit.Test;
40 import org.junit.experimental.categories.Category;
41
42 import static org.junit.Assert.assertEquals;
43 import static org.junit.Assert.assertTrue;
44 import static org.junit.Assert.fail;
45
46 @Category(LargeTests.class)
47 public class TestProcedureReplayOrder {
48 private static final Log LOG = LogFactory.getLog(TestProcedureReplayOrder.class);
49
50 private static final int NUM_THREADS = 16;
51
52 private ProcedureExecutor<Void> procExecutor;
53 private TestProcedureEnv procEnv;
54 private ProcedureStore procStore;
55
56 private HBaseCommonTestingUtility htu;
57 private FileSystem fs;
58 private Path testDir;
59 private Path logDir;
60
61 @Before
62 public void setUp() throws IOException {
63 htu = new HBaseCommonTestingUtility();
64 htu.getConfiguration().setInt(WALProcedureStore.SYNC_WAIT_MSEC_CONF_KEY, 25);
65
66 testDir = htu.getDataTestDir();
67 fs = testDir.getFileSystem(htu.getConfiguration());
68 assertTrue(testDir.depth() > 1);
69
70 logDir = new Path(testDir, "proc-logs");
71 procEnv = new TestProcedureEnv();
72 procStore = ProcedureTestingUtility.createWalStore(htu.getConfiguration(), fs, logDir);
73 procExecutor = new ProcedureExecutor(htu.getConfiguration(), procEnv, procStore);
74 procStore.start(NUM_THREADS);
75 procExecutor.start(1, true);
76 }
77
78 @After
79 public void tearDown() throws IOException {
80 procExecutor.stop();
81 procStore.stop(false);
82 fs.delete(logDir, true);
83 }
84
85 @Test(timeout=90000)
86 public void testSingleStepReplayOrder() throws Exception {
87 final int NUM_PROC_XTHREAD = 32;
88 final int NUM_PROCS = NUM_THREADS * NUM_PROC_XTHREAD;
89
90
91 submitProcedures(NUM_THREADS, NUM_PROC_XTHREAD, TestSingleStepProcedure.class);
92
93 while (procEnv.getExecId() < NUM_PROCS) {
94 Thread.sleep(100);
95 }
96
97
98 ProcedureTestingUtility.restart(procExecutor);
99
100
101
102 ProcedureTestingUtility.waitNoProcedureRunning(procExecutor);
103 procEnv.assertSortedExecList(NUM_PROCS);
104 }
105
106 @Test(timeout=90000)
107 public void testMultiStepReplayOrder() throws Exception {
108 final int NUM_PROC_XTHREAD = 24;
109 final int NUM_PROCS = NUM_THREADS * (NUM_PROC_XTHREAD * 2);
110
111
112 submitProcedures(NUM_THREADS, NUM_PROC_XTHREAD, TestTwoStepProcedure.class);
113
114 while (procEnv.getExecId() < NUM_PROCS) {
115 Thread.sleep(100);
116 }
117
118
119 ProcedureTestingUtility.restart(procExecutor);
120
121
122
123 ProcedureTestingUtility.waitNoProcedureRunning(procExecutor);
124 procEnv.assertSortedExecList(NUM_PROCS);
125 }
126
127 private void submitProcedures(final int nthreads, final int nprocPerThread,
128 final Class<?> procClazz) throws Exception {
129 Thread[] submitThreads = new Thread[nthreads];
130 for (int i = 0; i < submitThreads.length; ++i) {
131 submitThreads[i] = new Thread() {
132 @Override
133 public void run() {
134 for (int i = 0; i < nprocPerThread; ++i) {
135 try {
136 procExecutor.submitProcedure((Procedure)procClazz.newInstance());
137 } catch (InstantiationException|IllegalAccessException e) {
138 LOG.error("unable to instantiate the procedure", e);
139 fail("failure during the proc.newInstance(): " + e.getMessage());
140 }
141 }
142 }
143 };
144 }
145
146 for (int i = 0; i < submitThreads.length; ++i) {
147 submitThreads[i].start();
148 }
149
150 for (int i = 0; i < submitThreads.length; ++i) {
151 submitThreads[i].join();
152 }
153 }
154
155 private static class TestProcedureEnv {
156 private ArrayList<TestProcedure> execList = new ArrayList<TestProcedure>();
157 private AtomicLong execTimestamp = new AtomicLong(0);
158
159 public long getExecId() {
160 return execTimestamp.get();
161 }
162
163 public long nextExecId() {
164 return execTimestamp.incrementAndGet();
165 }
166
167 public void addToExecList(final TestProcedure proc) {
168 execList.add(proc);
169 }
170
171 public void assertSortedExecList(int numProcs) {
172 assertEquals(numProcs, execList.size());
173 LOG.debug("EXEC LIST: " + execList);
174 for (int i = 0; i < execList.size() - 1; ++i) {
175 TestProcedure a = execList.get(i);
176 TestProcedure b = execList.get(i + 1);
177 assertTrue("exec list not sorted: " + a + " < " + b, a.getExecId() > b.getExecId());
178 }
179 }
180 }
181
182 public static abstract class TestProcedure extends Procedure<TestProcedureEnv> {
183 protected long execId = 0;
184 protected int step = 0;
185
186 public long getExecId() {
187 return execId;
188 }
189
190 @Override
191 protected void rollback(TestProcedureEnv env) { }
192
193 @Override
194 protected boolean abort(TestProcedureEnv env) { return true; }
195
196 @Override
197 protected void serializeStateData(final OutputStream stream) throws IOException {
198 StreamUtils.writeLong(stream, execId);
199 }
200
201 @Override
202 protected void deserializeStateData(final InputStream stream) throws IOException {
203 execId = StreamUtils.readLong(stream);
204 step = 2;
205 }
206 }
207
208 public static class TestSingleStepProcedure extends TestProcedure {
209 public TestSingleStepProcedure() { }
210
211 @Override
212 protected Procedure[] execute(TestProcedureEnv env) throws ProcedureYieldException {
213 LOG.trace("execute procedure step=" + step + ": " + this);
214 if (step == 0) {
215 step = 1;
216 execId = env.nextExecId();
217 return new Procedure[] { this };
218 } else if (step == 2) {
219 env.addToExecList(this);
220 return null;
221 }
222 throw new ProcedureYieldException();
223 }
224
225 @Override
226 public String toString() {
227 return "SingleStep(procId=" + getProcId() + " execId=" + execId + ")";
228 }
229 }
230
231 public static class TestTwoStepProcedure extends TestProcedure {
232 public TestTwoStepProcedure() { }
233
234 @Override
235 protected Procedure[] execute(TestProcedureEnv env) throws ProcedureYieldException {
236 LOG.trace("execute procedure step=" + step + ": " + this);
237 if (step == 0) {
238 step = 1;
239 execId = env.nextExecId();
240 return new Procedure[] { new TestSingleStepProcedure() };
241 } else if (step == 2) {
242 env.addToExecList(this);
243 return null;
244 }
245 throw new ProcedureYieldException();
246 }
247
248 @Override
249 public String toString() {
250 return "TwoStep(procId=" + getProcId() + " execId=" + execId + ")";
251 }
252 }
253 }