ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [42/58] [abbrv] incubator-ignite git commit: Merge branch 'sprint-1' into ignite-188
Date Fri, 06 Feb 2015 20:48:04 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/b08492a5/modules/hadoop/src/test/java/org/apache/ignite/ignitefs/GridGgfsHadoopDualAbstractSelfTest.java
----------------------------------------------------------------------
diff --cc modules/hadoop/src/test/java/org/apache/ignite/ignitefs/GridGgfsHadoopDualAbstractSelfTest.java
index 62c2c95,0000000..d797346
mode 100644,000000..100644
--- a/modules/hadoop/src/test/java/org/apache/ignite/ignitefs/GridGgfsHadoopDualAbstractSelfTest.java
+++ b/modules/hadoop/src/test/java/org/apache/ignite/ignitefs/GridGgfsHadoopDualAbstractSelfTest.java
@@@ -1,304 -1,0 +1,304 @@@
 +/*
 + * 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.ignitefs;
 +
 +import org.apache.hadoop.conf.*;
 +import org.apache.hadoop.fs.*;
 +import org.apache.hadoop.fs.FileSystem;
 +import org.apache.ignite.*;
 +import org.apache.ignite.cache.*;
 +import org.apache.ignite.configuration.*;
 +import org.apache.ignite.internal.fs.hadoop.*;
 +import org.apache.ignite.internal.processors.fs.*;
 +import org.apache.ignite.internal.util.typedef.*;
 +import org.apache.ignite.internal.util.typedef.internal.*;
 +import org.apache.ignite.spi.discovery.tcp.*;
 +import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
 +import org.apache.ignite.testframework.*;
 +import org.jetbrains.annotations.*;
 +
 +import java.io.*;
 +import java.net.*;
 +import java.util.*;
 +import java.util.concurrent.*;
 +
 +import static org.apache.ignite.cache.CacheAtomicityMode.*;
 +import static org.apache.ignite.cache.CacheMode.*;
 +import static org.apache.ignite.ignitefs.IgniteFsMode.*;
 +import static org.apache.ignite.ignitefs.hadoop.GridGgfsHadoopParameters.*;
 +import static org.apache.ignite.internal.processors.fs.GridGgfsAbstractSelfTest.*;
 +
 +/**
 + * Tests for GGFS working in mode when remote file system exists: DUAL_SYNC, DUAL_ASYNC.
 + */
 +public abstract class GridGgfsHadoopDualAbstractSelfTest extends GridGgfsCommonAbstractTest
{
 +    /** GGFS block size. */
 +    protected static final int GGFS_BLOCK_SIZE = 512 * 1024;
 +
 +    /** Amount of blocks to prefetch. */
 +    protected static final int PREFETCH_BLOCKS = 1;
 +
 +    /** Amount of sequential block reads before prefetch is triggered. */
 +    protected static final int SEQ_READS_BEFORE_PREFETCH = 2;
 +
 +    /** Secondary file system URI. */
 +    protected static final String SECONDARY_URI = "ggfs://ggfs-secondary:grid-secondary@127.0.0.1:11500/";
 +
 +    /** Secondary file system configuration path. */
 +    protected static final String SECONDARY_CFG = "modules/core/src/test/config/hadoop/core-site-loopback-secondary.xml";
 +
 +    /** Primary file system URI. */
 +    protected static final String PRIMARY_URI = "ggfs://ggfs:grid@/";
 +
 +    /** Primary file system configuration path. */
 +    protected static final String PRIMARY_CFG = "modules/core/src/test/config/hadoop/core-site-loopback.xml";
 +
 +    /** Primary file system REST endpoint configuration map. */
 +    protected static final Map<String, String> PRIMARY_REST_CFG = new HashMap<String,
String>() {{
 +        put("type", "tcp");
 +        put("port", "10500");
 +    }};
 +
 +    /** Secondary file system REST endpoint configuration map. */
 +    protected static final Map<String, String> SECONDARY_REST_CFG = new HashMap<String,
String>() {{
 +        put("type", "tcp");
 +        put("port", "11500");
 +    }};
 +
 +    /** Directory. */
 +    protected static final IgniteFsPath DIR = new IgniteFsPath("/dir");
 +
 +    /** Sub-directory. */
 +    protected static final IgniteFsPath SUBDIR = new IgniteFsPath(DIR, "subdir");
 +
 +    /** File. */
 +    protected static final IgniteFsPath FILE = new IgniteFsPath(SUBDIR, "file");
 +
 +    /** Default data chunk (128 bytes). */
 +    protected static byte[] chunk;
 +
 +    /** Primary GGFS. */
 +    protected static GridGgfsImpl ggfs;
 +
 +    /** Secondary GGFS. */
 +    protected static GridGgfsImpl ggfsSecondary;
 +
 +    /** GGFS mode. */
 +    protected final IgniteFsMode mode;
 +
 +    /**
 +     * Constructor.
 +     *
 +     * @param mode GGFS mode.
 +     */
 +    protected GridGgfsHadoopDualAbstractSelfTest(IgniteFsMode mode) {
 +        this.mode = mode;
 +        assert mode == DUAL_SYNC || mode == DUAL_ASYNC;
 +    }
 +
 +    /**
 +     * Start grid with GGFS.
 +     *
 +     * @param gridName Grid name.
 +     * @param ggfsName GGFS name
 +     * @param mode GGFS mode.
 +     * @param secondaryFs Secondary file system (optional).
 +     * @param restCfg Rest configuration string (optional).
 +     * @return Started grid instance.
 +     * @throws Exception If failed.
 +     */
 +    protected Ignite startGridWithGgfs(String gridName, String ggfsName, IgniteFsMode mode,
 +        @Nullable IgniteFsFileSystem secondaryFs, @Nullable Map<String, String> restCfg)
throws Exception {
 +        IgniteFsConfiguration ggfsCfg = new IgniteFsConfiguration();
 +
 +        ggfsCfg.setDataCacheName("dataCache");
 +        ggfsCfg.setMetaCacheName("metaCache");
 +        ggfsCfg.setName(ggfsName);
 +        ggfsCfg.setBlockSize(GGFS_BLOCK_SIZE);
 +        ggfsCfg.setDefaultMode(mode);
 +        ggfsCfg.setIpcEndpointConfiguration(restCfg);
 +        ggfsCfg.setSecondaryFileSystem(secondaryFs);
 +        ggfsCfg.setPrefetchBlocks(PREFETCH_BLOCKS);
 +        ggfsCfg.setSequentialReadsBeforePrefetch(SEQ_READS_BEFORE_PREFETCH);
 +
 +        CacheConfiguration dataCacheCfg = defaultCacheConfiguration();
 +
 +        dataCacheCfg.setName("dataCache");
 +        dataCacheCfg.setCacheMode(PARTITIONED);
 +        dataCacheCfg.setDistributionMode(CacheDistributionMode.PARTITIONED_ONLY);
 +        dataCacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
 +        dataCacheCfg.setAffinityMapper(new IgniteFsGroupDataBlocksKeyMapper(2));
 +        dataCacheCfg.setBackups(0);
 +        dataCacheCfg.setQueryIndexEnabled(false);
 +        dataCacheCfg.setAtomicityMode(TRANSACTIONAL);
 +        dataCacheCfg.setOffHeapMaxMemory(0);
 +
 +        CacheConfiguration metaCacheCfg = defaultCacheConfiguration();
 +
 +        metaCacheCfg.setName("metaCache");
 +        metaCacheCfg.setCacheMode(REPLICATED);
 +        metaCacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
 +        metaCacheCfg.setQueryIndexEnabled(false);
 +        metaCacheCfg.setAtomicityMode(TRANSACTIONAL);
 +
 +        IgniteConfiguration cfg = new IgniteConfiguration();
 +
 +        cfg.setGridName(gridName);
 +
 +        TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
 +
 +        discoSpi.setIpFinder(new TcpDiscoveryVmIpFinder(true));
 +
 +        cfg.setDiscoverySpi(discoSpi);
 +        cfg.setCacheConfiguration(dataCacheCfg, metaCacheCfg);
 +        cfg.setGgfsConfiguration(ggfsCfg);
 +
 +        cfg.setLocalHost("127.0.0.1");
 +        cfg.setRestEnabled(false);
 +
 +        return G.start(cfg);
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void beforeTestsStarted() throws Exception {
 +        chunk = new byte[128];
 +
 +        for (int i = 0; i < chunk.length; i++)
 +            chunk[i] = (byte)i;
 +
 +        Ignite igniteSecondary = startGridWithGgfs("grid-secondary", "ggfs-secondary", PRIMARY,
null, SECONDARY_REST_CFG);
 +
 +        IgniteFsFileSystem hadoopFs = new GridGgfsHadoopFileSystemWrapper(SECONDARY_URI,
SECONDARY_CFG);
 +
 +        Ignite ignite = startGridWithGgfs("grid", "ggfs", mode, hadoopFs, PRIMARY_REST_CFG);
 +
 +        ggfsSecondary = (GridGgfsImpl) igniteSecondary.fileSystem("ggfs-secondary");
 +        ggfs = (GridGgfsImpl) ignite.fileSystem("ggfs");
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void afterTest() throws Exception {
 +        clear(ggfs);
 +        clear(ggfsSecondary);
 +    }
 +
 +    /** {@inheritDoc} */
 +    @Override protected void afterTestsStopped() throws Exception {
 +        G.stopAll(true);
 +    }
 +
 +    /**
 +     * Convenient method to group paths.
 +     *
 +     * @param paths Paths to group.
 +     * @return Paths as array.
 +     */
 +    protected IgniteFsPath[] paths(IgniteFsPath... paths) {
 +        return paths;
 +    }
 +
 +    /**
 +     * Check how prefetch override works.
 +     *
 +     * @throws Exception IF failed.
 +     */
 +    public void testOpenPrefetchOverride() throws Exception {
 +        create(ggfsSecondary, paths(DIR, SUBDIR), paths(FILE));
 +
 +        // Write enough data to the secondary file system.
 +        final int blockSize = GGFS_BLOCK_SIZE;
 +
 +        IgniteFsOutputStream out = ggfsSecondary.append(FILE, false);
 +
 +        int totalWritten = 0;
 +
 +        while (totalWritten < blockSize * 2 + chunk.length) {
 +            out.write(chunk);
 +
 +            totalWritten += chunk.length;
 +        }
 +
 +        out.close();
 +
 +        awaitFileClose(ggfsSecondary, FILE);
 +
 +        // Instantiate file system with overridden "seq reads before prefetch" property.
 +        Configuration cfg = new Configuration();
 +
-         cfg.addResource(U.resolveGridGainUrl(PRIMARY_CFG));
++        cfg.addResource(U.resolveIgniteUrl(PRIMARY_CFG));
 +
 +        int seqReads = SEQ_READS_BEFORE_PREFETCH + 1;
 +
 +        cfg.setInt(String.format(PARAM_GGFS_SEQ_READS_BEFORE_PREFETCH, "ggfs:grid@"), seqReads);
 +
 +        FileSystem fs = FileSystem.get(new URI(PRIMARY_URI), cfg);
 +
 +        // Read the first two blocks.
 +        Path fsHome = new Path(PRIMARY_URI);
 +        Path dir = new Path(fsHome, DIR.name());
 +        Path subdir = new Path(dir, SUBDIR.name());
 +        Path file = new Path(subdir, FILE.name());
 +
 +        FSDataInputStream fsIn = fs.open(file);
 +
 +        final byte[] readBuf = new byte[blockSize * 2];
 +
 +        fsIn.readFully(0, readBuf, 0, readBuf.length);
 +
 +        // Wait for a while for prefetch to finish (if any).
 +        GridGgfsMetaManager meta = ggfs.context().meta();
 +
 +        GridGgfsFileInfo info = meta.info(meta.fileId(FILE));
 +
 +        GridGgfsBlockKey key = new GridGgfsBlockKey(info.id(), info.affinityKey(), info.evictExclude(),
2);
 +
 +        GridCache<GridGgfsBlockKey, byte[]> dataCache = ggfs.context().kernalContext().cache().cache(
 +            ggfs.configuration().getDataCacheName());
 +
 +        for (int i = 0; i < 10; i++) {
 +            if (dataCache.containsKey(key))
 +                break;
 +            else
 +                U.sleep(100);
 +        }
 +
 +        fsIn.close();
 +
 +        // Remove the file from the secondary file system.
 +        ggfsSecondary.delete(FILE, false);
 +
 +        // Try reading the third block. Should fail.
 +        GridTestUtils.assertThrows(log, new Callable<Object>() {
 +            @Override public Object call() throws Exception {
 +                IgniteFsInputStream in0 = ggfs.open(FILE);
 +
 +                in0.seek(blockSize * 2);
 +
 +                try {
 +                    in0.read(readBuf);
 +                }
 +                finally {
 +                    U.closeQuiet(in0);
 +                }
 +
 +                return null;
 +            }
 +        }, IOException.class,
 +            "Failed to read data due to secondary file system exception: /dir/subdir/file");
 +    }
 +}


Mime
View raw message