| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.ignite.internal; |
| |
| import java.io.Externalizable; |
| import java.io.IOException; |
| import java.io.ObjectInput; |
| import java.io.ObjectOutput; |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashSet; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.UUID; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCluster; |
| import org.apache.ignite.cluster.ClusterGroup; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.compute.ComputeJob; |
| import org.apache.ignite.compute.ComputeJobAdapter; |
| import org.apache.ignite.compute.ComputeJobResult; |
| import org.apache.ignite.compute.ComputeTaskFuture; |
| import org.apache.ignite.compute.ComputeTaskSplitAdapter; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.events.EventType; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.G; |
| import org.apache.ignite.lang.IgniteBiTuple; |
| import org.apache.ignite.lang.IgniteCallable; |
| import org.apache.ignite.lang.IgniteClosure; |
| import org.apache.ignite.lang.IgniteFuture; |
| import org.apache.ignite.lang.IgnitePredicate; |
| import org.apache.ignite.lang.IgniteReducer; |
| import org.apache.ignite.lang.IgniteRunnable; |
| import org.apache.ignite.lang.IgniteUuid; |
| import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.jetbrains.annotations.Nullable; |
| import org.junit.Test; |
| |
| import static org.apache.ignite.events.EventType.EVT_JOB_STARTED; |
| |
| /** |
| * Abstract test for {@link org.apache.ignite.cluster.ClusterGroup} |
| */ |
| @SuppressWarnings("deprecation") |
| public abstract class ClusterGroupAbstractTest extends GridCommonAbstractTest implements Externalizable { |
| /** Waiting timeout. */ |
| private static final int WAIT_TIMEOUT = 30000; |
| |
| /** Utility static variable. */ |
| private static final AtomicInteger cnt = new AtomicInteger(0); |
| |
| /** Mutex. */ |
| private static final Object mux = new Object(); |
| |
| /** Projection. */ |
| private ClusterGroup prj; |
| |
| /** Runnable job. */ |
| private IgniteRunnable runJob = new TestRunnable(); |
| |
| /** Callable job. */ |
| private IgniteCallable<String> calJob = new TestCallable<>(); |
| |
| /** Closure job. */ |
| private IgniteClosure<String, String> clrJob = new IgniteClosure<String, String>() { |
| @Override public String apply(String s) { |
| return s; |
| } |
| |
| @Override public String toString() { |
| return "clrJob"; |
| } |
| }; |
| |
| /** Reducer. */ |
| private IgniteReducer<String, Object> rdc = new IgniteReducer<String, Object>() { |
| @Override public boolean collect(String e) { |
| return true; |
| } |
| |
| @Nullable @Override public Object reduce() { |
| return null; |
| } |
| |
| @Override public String toString() { |
| return "rdc"; |
| } |
| }; |
| |
| /** */ |
| protected ClusterGroupAbstractTest() { |
| // No-op. |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); |
| |
| cfg.setIncludeEventTypes(EventType.EVTS_ALL); |
| |
| ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true); |
| |
| return cfg; |
| } |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTest() throws Exception { |
| prj = projection(); |
| |
| cnt.set(0); |
| } |
| |
| /** |
| * @return Projection. |
| */ |
| protected abstract ClusterGroup projection(); |
| |
| /** |
| * @return Local node ID. |
| */ |
| @Nullable protected abstract UUID localNodeId(); |
| |
| /** |
| * @return Remote nodes IDs. |
| */ |
| protected Collection<UUID> remoteNodeIds() { |
| return F.nodeIds(projection().forRemotes().nodes()); |
| } |
| |
| /** |
| * @return Projection size. |
| */ |
| private int projectionSize() { |
| int size = localNodeId() != null ? 1 : 0; |
| |
| size += remoteNodeIds().size(); |
| |
| assert size > 0; |
| |
| return size; |
| } |
| |
| /** |
| * @return Collection of projection node IDs. |
| */ |
| private Collection<UUID> projectionNodeIds() { |
| Collection<UUID> ids = new LinkedList<>(); |
| |
| UUID id = localNodeId(); |
| |
| if (id != null) |
| ids.add(id); |
| |
| ids.addAll(remoteNodeIds()); |
| |
| assert !ids.isEmpty(); |
| |
| return ids; |
| } |
| |
| /** |
| * Test for projection on not existing node IDs. |
| */ |
| @Test |
| public void testInvalidProjection() { |
| Collection<UUID> ids = new HashSet<>(); |
| |
| ids.add(UUID.randomUUID()); |
| ids.add(UUID.randomUUID()); |
| |
| ClusterGroup invalidPrj = prj.forNodeIds(ids); |
| |
| assertEquals(0, invalidPrj.nodes().size()); |
| } |
| |
| /** |
| * @throws Exception If test failed. |
| */ |
| @Test |
| public void testProjection() throws Exception { |
| assert prj != null; |
| |
| assert prj.ignite() != null; |
| assert prj.predicate() != null; |
| |
| int size = projectionSize(); |
| |
| assert prj.nodes().size() == size; |
| |
| Collection<UUID> nodeIds = projectionNodeIds(); |
| |
| for (ClusterNode node : prj.nodes()) |
| assert nodeIds.contains(node.id()); |
| } |
| |
| /** |
| * @throws Exception If test failed. |
| */ |
| @Test |
| public void testRemoteNodes() throws Exception { |
| Collection<UUID> remoteNodeIds = remoteNodeIds(); |
| |
| UUID locNodeId = localNodeId(); |
| |
| int size = remoteNodeIds.size(); |
| |
| String name = "oneMoreGrid"; |
| |
| try { |
| Ignite g = startGrid(name); |
| |
| UUID excludedId = g.cluster().localNode().id(); |
| |
| assertEquals(size, prj.forRemotes().nodes().size()); |
| |
| for (ClusterNode node : prj.forRemotes().nodes()) { |
| UUID id = node.id(); |
| |
| assert !id.equals(locNodeId) && remoteNodeIds.contains(id) && !excludedId.equals(id); |
| } |
| } |
| finally { |
| stopGrid(name); |
| } |
| } |
| |
| /** |
| * @throws Exception If test failed. |
| */ |
| @Test |
| public void testRemoteProjection() throws Exception { |
| Collection<UUID> remoteNodeIds = remoteNodeIds(); |
| |
| ClusterGroup remotePrj = projection().forRemotes(); |
| |
| Collection<UUID> prjNodeIds = F.nodeIds(remotePrj.nodes()); |
| |
| assert prjNodeIds.size() == remoteNodeIds.size(); |
| |
| assert prjNodeIds.containsAll(remoteNodeIds()); |
| |
| assert !prjNodeIds.contains(localNodeId()); |
| |
| String name = "oneMoreGrid"; |
| |
| try { |
| Ignite g = startGrid(name); |
| |
| UUID excludedId = g.cluster().localNode().id(); |
| |
| assert !F.nodeIds(remotePrj.nodes()).contains(excludedId); |
| } |
| finally { |
| stopGrid(name); |
| } |
| } |
| |
| /** |
| * @throws Exception If test failed. |
| */ |
| @Test |
| public void testExecution() throws Exception { |
| String name = "oneMoreGrid"; |
| |
| Collection<IgniteBiTuple<Ignite, IgnitePredicate<Event>>> lsnrs = new LinkedList<>(); |
| |
| try { |
| final AtomicInteger cnt = new AtomicInteger(); |
| |
| Ignite g = startGrid(name); |
| |
| IgnitePredicate<Event> lsnr; |
| |
| if (!IgniteCluster.class.isAssignableFrom(projection().getClass())) { |
| g.events().localListen(lsnr = new IgnitePredicate<Event>() { |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_JOB_STARTED; |
| |
| assert false; |
| |
| return true; |
| } |
| }, EVT_JOB_STARTED); |
| |
| lsnrs.add(F.t(g, lsnr)); |
| } |
| |
| for (ClusterNode node : prj.nodes()) { |
| g = G.ignite(node.id()); |
| |
| g.events().localListen(lsnr = new IgnitePredicate<Event>() { |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_JOB_STARTED; |
| |
| synchronized (mux) { |
| cnt.incrementAndGet(); |
| |
| mux.notifyAll(); |
| } |
| |
| return true; |
| } |
| }, EVT_JOB_STARTED); |
| |
| lsnrs.add(F.t(g, lsnr)); |
| } |
| |
| run1(cnt); |
| run2(cnt); |
| |
| call1(cnt); |
| call2(cnt); |
| call3(cnt); |
| call4(cnt); |
| call5(cnt); |
| |
| forkjoin1(cnt); |
| forkjoin2(cnt); |
| |
| exec1(cnt); |
| exec2(cnt); |
| |
| executorService(cnt); |
| |
| checkActiveFutures(); |
| } |
| finally { |
| for (IgniteBiTuple<Ignite, IgnitePredicate<Event>> t : lsnrs) |
| t.get1().events().stopLocalListen(t.get2(), EVT_JOB_STARTED); |
| |
| stopGrid(name); |
| } |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void run1(AtomicInteger cnt) throws Exception { |
| IgniteFuture<Void> fut = compute(prj).broadcastAsync(runJob); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).broadcast(runJob); |
| |
| waitForValue(cnt, projectionSize()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void run2(AtomicInteger cnt) throws Exception { |
| Collection<IgniteRunnable> jobs = F.asList(runJob); |
| |
| IgniteFuture<Void> fut = compute(prj).runAsync(jobs); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).run(jobs); |
| |
| waitForValue(cnt, jobs.size()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void call1(AtomicInteger cnt) throws Exception { |
| IgniteFuture<Collection<String>> fut = compute(prj).broadcastAsync(calJob); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).broadcast(calJob); |
| |
| waitForValue(cnt, projectionSize()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void call2(AtomicInteger cnt) throws Exception { |
| Collection<IgniteCallable<String>> jobs = F.asList(calJob); |
| |
| IgniteFuture<Collection<String>> fut = compute(prj).callAsync(jobs); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).call(jobs); |
| |
| waitForValue(cnt, jobs.size()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void call3(AtomicInteger cnt) throws Exception { |
| IgniteFuture<String> fut = compute(prj).applyAsync(clrJob, (String)null); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).apply(clrJob, (String)null); |
| |
| waitForValue(cnt, 1); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void call4(AtomicInteger cnt) throws Exception { |
| Collection<String> args = F.asList("a", "b", "c"); |
| |
| IgniteFuture<Collection<String>> fut = compute(prj).applyAsync(clrJob, args); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).apply(clrJob, args); |
| |
| waitForValue(cnt, args.size()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void call5(AtomicInteger cnt) throws Exception { |
| IgniteFuture<Collection<String>> fut = compute(prj).broadcastAsync(new TestClosure(), "arg"); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| Collection<String> res = compute(prj).broadcast(new TestClosure(), "arg"); |
| |
| assertEquals(projectionSize(), res.size()); |
| |
| waitForValue(cnt, projectionSize()); |
| |
| for (String resStr : res) |
| assertEquals("arg", resStr); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void forkjoin1(AtomicInteger cnt) throws Exception { |
| Collection<String> args = F.asList("a", "b", "c"); |
| |
| IgniteFuture fut = compute(prj).applyAsync(clrJob, args, rdc); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).apply(clrJob, args, rdc); |
| |
| waitForValue(cnt, args.size()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void forkjoin2(AtomicInteger cnt) throws Exception { |
| Collection<IgniteCallable<String>> jobs = F.asList(calJob); |
| |
| IgniteFuture<Object> fut = compute(prj).callAsync(jobs, rdc); |
| |
| waitForExecution(fut); |
| |
| cnt.set(0); |
| |
| compute(prj).call(jobs, rdc); |
| |
| waitForValue(cnt, jobs.size()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void exec1(AtomicInteger cnt) throws Exception { |
| cnt.set(0); |
| |
| compute(prj).execute(TestTask.class.getName(), null); |
| |
| waitForValue(cnt, projectionSize()); |
| |
| cnt.set(0); |
| |
| compute(prj).execute(new TestTask(), null); |
| |
| waitForValue(cnt, projectionSize()); |
| |
| cnt.set(0); |
| |
| compute(prj).execute(TestTask.class, null); |
| |
| waitForValue(cnt, projectionSize()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void exec2(AtomicInteger cnt) throws Exception { |
| cnt.set(0); |
| |
| compute(prj).withTimeout(WAIT_TIMEOUT).execute(TestTask.class.getName(), null); |
| |
| waitForValue(cnt, projectionSize()); |
| |
| cnt.set(0); |
| |
| compute(prj).withTimeout(WAIT_TIMEOUT).execute(new TestTask(), null); |
| |
| waitForValue(cnt, projectionSize()); |
| |
| cnt.set(0); |
| |
| compute(prj).withTimeout(WAIT_TIMEOUT).execute(TestTask.class, null); |
| |
| waitForValue(cnt, projectionSize()); |
| } |
| |
| /** |
| * @param cnt Counter. |
| * @throws Exception If failed. |
| */ |
| private void executorService(AtomicInteger cnt) throws Exception { |
| cnt.set(0); |
| |
| ExecutorService execSrvc = prj.ignite().executorService(prj); |
| |
| Future<String> fut = execSrvc.submit(new TestCallable<String>() { |
| @Override public String call() throws Exception { |
| return "submit1"; |
| } |
| }); |
| |
| waitForValue(cnt, 1); |
| |
| assertEquals("submit1", fut.get()); |
| |
| cnt.set(0); |
| |
| fut = execSrvc.submit(new TestRunnable(), "submit2"); |
| |
| waitForValue(cnt, 1); |
| |
| assertEquals("submit2", fut.get()); |
| |
| cnt.set(0); |
| |
| Future<?> runFut = execSrvc.submit(new TestRunnable()); |
| |
| waitForValue(cnt, 1); |
| |
| runFut.get(); |
| } |
| |
| /** |
| * @param fut Execution future. |
| * @throws InterruptedException Thrown if wait was interrupted. |
| */ |
| @SuppressWarnings({"UnconditionalWait"}) |
| private void waitForExecution(IgniteFuture fut) throws InterruptedException { |
| long sleep = 250; |
| |
| long threshold = System.currentTimeMillis() + WAIT_TIMEOUT; |
| |
| do { |
| synchronized (mux) { |
| mux.wait(sleep); |
| } |
| } |
| while (fut != null && !fut.isDone() && !fut.isCancelled() && threshold > System.currentTimeMillis()); |
| |
| assert fut == null || fut.isDone(); |
| } |
| |
| /** |
| * @param cnt Counter to check. |
| * @param val Value to check. |
| * @throws InterruptedException Thrown if wait was interrupted. |
| */ |
| private void waitForValue(AtomicInteger cnt, int val) throws InterruptedException { |
| assert cnt != null; |
| assert val > 0; |
| |
| long threshold = System.currentTimeMillis() + WAIT_TIMEOUT; |
| |
| long time; |
| |
| while (threshold > (time = System.currentTimeMillis())) |
| synchronized (mux) { |
| if (cnt.get() == val) |
| break; |
| |
| mux.wait(threshold - time); |
| } |
| |
| assert cnt.get() == val; |
| } |
| |
| /** |
| * @throws Exception If test failed. |
| */ |
| private void checkActiveFutures() throws Exception { |
| assertEquals(0, compute(prj).activeTaskFutures().size()); |
| |
| cnt.set(0); |
| |
| Collection<IgniteFuture<Object>> futsList = new ArrayList<>(); |
| |
| for (int i = 0; i < 10; i++) { |
| IgniteFuture<Object> fut = compute(prj).callAsync(new TestWaitCallable<>()); |
| |
| assertFalse(fut.isDone()); |
| |
| Map<IgniteUuid, ComputeTaskFuture<Object>> futs = compute(prj).activeTaskFutures(); |
| |
| assertEquals(i + 1, futs.size()); |
| |
| assertTrue(futs.containsKey(((ComputeTaskFuture)fut).getTaskSession().getId())); |
| |
| futsList.add(fut); |
| } |
| |
| synchronized (mux) { |
| cnt.incrementAndGet(); |
| |
| mux.notifyAll(); |
| } |
| |
| for (IgniteFuture<Object> fut : futsList) |
| fut.get(); |
| |
| assertEquals(0, compute(prj).activeTaskFutures().size()); |
| } |
| |
| /** |
| * Test closure. |
| */ |
| private static class TestClosure implements IgniteClosure<String, String> { |
| /** {@inheritDoc} */ |
| @Override public String apply(String s) { |
| return s; |
| } |
| } |
| |
| /** |
| * Test runnable. |
| */ |
| private static class TestRunnable implements IgniteRunnable { |
| /** {@inheritDoc} */ |
| @Override public void run() { |
| // No-op. |
| } |
| } |
| |
| /** |
| * Test callable. |
| */ |
| private static class TestCallable<T> implements IgniteCallable<T> { |
| /** {@inheritDoc} */ |
| @Nullable @Override public T call() throws Exception { |
| return null; |
| } |
| } |
| |
| /** |
| * Test callable. |
| */ |
| private static class TestWaitCallable<T> implements IgniteCallable<T> { |
| /** {@inheritDoc} */ |
| @Nullable @Override public T call() throws Exception { |
| synchronized (mux) { |
| while (cnt.get() == 0) |
| mux.wait(); |
| } |
| |
| return null; |
| } |
| } |
| |
| /** |
| * Test task. |
| */ |
| @SuppressWarnings({"PublicInnerClass"}) |
| public static class TestTask extends ComputeTaskSplitAdapter<String, Void> { |
| /** {@inheritDoc} */ |
| @Override protected Collection<? extends ComputeJob> split(int gridSize, String arg) { |
| Collection<ComputeJob> jobs = new HashSet<>(); |
| |
| for (int i = 0; i < gridSize; i++) |
| jobs.add(new TestJob()); |
| |
| return jobs; |
| } |
| |
| /** {@inheritDoc} */ |
| @Nullable @Override public Void reduce(List<ComputeJobResult> results) { |
| return null; |
| } |
| } |
| |
| /** |
| * Test job. |
| */ |
| @SuppressWarnings({"PublicInnerClass"}) |
| public static class TestJob extends ComputeJobAdapter { |
| /** {@inheritDoc} */ |
| @Nullable @Override public Object execute() { |
| return null; |
| } |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void writeExternal(ObjectOutput out) throws IOException { |
| // No-op. |
| } |
| |
| /** {@inheritDoc} */ |
| @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { |
| // No-op. |
| } |
| } |
