text:
package org.apache.kylin.job;

import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.job.dataGen.FactTableGenerator;
import org.apache.maven.model.Model;
import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
import org.apache.tools.ant.filters.StringInputStream;
import org.codehaus.plexus.util.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.ResourceTool;
import org.apache.kylin.common.util.AbstractKylinTestCase;
import org.apache.kylin.common.util.CliCommandExecutor;
import org.apache.kylin.common.util.HiveClient;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.job.hadoop.hive.SqlHiveDataTypeMapping;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.ColumnDesc;
import org.apache.kylin.metadata.model.TableDesc;

public class DeployUtil {
    @SuppressWarnings("unused")
    private static final Logger logger = LoggerFactory.getLogger(DeployUtil.class);

    public static void initCliWorkDir() throws IOException {
        execCliCommand("rm -rf " + getHadoopCliWorkingDir());
        execCliCommand("mkdir -p " + config().getKylinJobLogDir());
    }

    public static void deployMetadata() throws IOException {
        // install metadata to hbase
        ResourceTool.reset(config());
        ResourceTool.copy(KylinConfig.createInstanceFromUri(AbstractKylinTestCase.LOCALMETA_TEST_DATA), config());

        // update cube desc signature.
        for (CubeInstance cube : CubeManager.getInstance(config()).listAllCubes()) {
            cube.getDescriptor().setSignature(cube.getDescriptor().calculateSignature());
            CubeManager.getInstance(config()).updateCube(cube);
        }
    }

    public static void overrideJobJarLocations() {
        Pair<File, File> files = getJobJarFiles();
        File jobJar = files.getFirst();
        File coprocessorJar = files.getSecond();

        config().overrideKylinJobJarPath(jobJar.getAbsolutePath());
        config().overrideCoprocessorLocalJar(coprocessorJar.getAbsolutePath());
    }

    public static void deployJobJars() throws IOException {
        Pair<File, File> files = getJobJarFiles();
        File originalJobJar = files.getFirst();
        File originalCoprocessorJar = files.getSecond();

        String jobJarPath = config().getKylinJobJarPath();
        if (StringUtils.isEmpty(jobJarPath)) {
            throw new RuntimeException("deployJobJars cannot find job jar");
        }

        File targetJobJar = new File(jobJarPath);
        File jobJarRenamedAsTarget = new File(originalJobJar.getParentFile(), targetJobJar.getName());
        if (originalJobJar.equals(jobJarRenamedAsTarget) == false) {
            FileUtils.copyFile(originalJobJar, jobJarRenamedAsTarget);
        }

        File targetCoprocessorJar = new File(config().getCoprocessorLocalJar());
        File coprocessorJarRenamedAsTarget = new File(originalCoprocessorJar.getParentFile(), targetCoprocessorJar.getName());
        if (originalCoprocessorJar.equals(coprocessorJarRenamedAsTarget) == false) {
            FileUtils.copyFile(originalCoprocessorJar, coprocessorJarRenamedAsTarget);
        }

        CliCommandExecutor cmdExec = config().getCliCommandExecutor();
        cmdExec.copyFile(jobJarRenamedAsTarget.getAbsolutePath(), targetJobJar.getParent());
        cmdExec.copyFile(coprocessorJarRenamedAsTarget.getAbsolutePath(), targetCoprocessorJar.getParent());
    }

    private static Pair<File, File> getJobJarFiles() {
        String version;
        try {
            MavenXpp3Reader pomReader = new MavenXpp3Reader();
            Model model = pomReader.read(new FileReader("../pom.xml"));
            version = model.getVersion();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }

        File jobJar = new File("../job/target", "kylin-job-" + version + "-job.jar");
        File coprocessorJar = new File("../storage/target", "kylin-storage-" + version + "-coprocessor.jar");
        return new Pair<File, File>(jobJar, coprocessorJar);
    }

    private static void execCliCommand(String cmd) throws IOException {
        config().getCliCommandExecutor().execute(cmd);
    }

    private static String getHadoopCliWorkingDir() {
        return config().getCliWorkingDir();
    }

    private static KylinConfig config() {
        return KylinConfig.getInstanceFromEnv();
    }

    // ============================================================================

    static final String TABLE_CAL_DT = "edw.test_cal_dt";
    static final String TABLE_CATEGORY_GROUPINGS = "default.test_category_groupings";
    static final String TABLE_KYLIN_FACT = "default.test_kylin_fact";
    static final String TABLE_SELLER_TYPE_DIM = "edw.test_seller_type_dim";
    static final String TABLE_SITES = "edw.test_sites";

    static final String[] TABLE_NAMES = new String[] { TABLE_CAL_DT, TABLE_CATEGORY_GROUPINGS, TABLE_KYLIN_FACT, TABLE_SELLER_TYPE_DIM, TABLE_SITES };

    public static void prepareTestData(String joinType, String cubeName) throws Exception {

        String factTableName = TABLE_KYLIN_FACT.toUpperCase();
        String content = null;

        boolean buildCubeUsingProvidedData = Boolean.parseBoolean(System.getProperty("buildCubeUsingProvidedData"));
        if (!buildCubeUsingProvidedData) {
            System.out.println("build cube with random dataset");
            // data is generated according to cube descriptor and saved in resource store
            if (joinType.equalsIgnoreCase("inner")) {
                content = FactTableGenerator.generate(cubeName, "10000", "1", null, "inner");
            } else if (joinType.equalsIgnoreCase("left")) {
                content = FactTableGenerator.generate(cubeName, "10000", "0.6", null, "left");
            } else {
                throw new IllegalArgumentException("Unsupported join type : " + joinType);
            }

            assert content != null;
            overrideFactTableData(content, factTableName);
        } else {
            System.out.println("build cube with provided dataset");
        }

        duplicateFactTableData(factTableName, joinType);
        deployHiveTables();
    }

    public static void overrideFactTableData(String factTableContent, String factTableName) throws IOException {
        // Write to resource store
        ResourceStore store = ResourceStore.getStore(config());

        InputStream in = new StringInputStream(factTableContent);
        String factTablePath = "/data/" + factTableName + ".csv";
        store.deleteResource(factTablePath);
        store.putResource(factTablePath, in, System.currentTimeMillis());
        in.close();
    }

    public static void duplicateFactTableData(String factTableName, String joinType) throws IOException {
        // duplicate a copy of this fact table, with a naming convention with fact.csv.inner or fact.csv.left
        // so that later test cases can select different data files
        ResourceStore store = ResourceStore.getStore(config());
        InputStream in = store.getResource("/data/" + factTableName + ".csv");
        String factTablePathWithJoinType = "/data/" + factTableName + ".csv." + joinType.toLowerCase();
        store.deleteResource(factTablePathWithJoinType);
        store.putResource(factTablePathWithJoinType, in, System.currentTimeMillis());
        in.close();
    }

    private static void deployHiveTables() throws Exception {

        MetadataManager metaMgr = MetadataManager.getInstance(config());

        // scp data files, use the data from hbase, instead of local files
        File temp = File.createTempFile("temp", ".csv");
        temp.createNewFile();
        for (String tablename : TABLE_NAMES) {
            tablename = tablename.toUpperCase();

            File localBufferFile = new File(temp.getParent() + "/" + tablename + ".csv");
            localBufferFile.createNewFile();

            InputStream hbaseDataStream = metaMgr.getStore().getResource("/data/" + tablename + ".csv");
            FileOutputStream localFileStream = new FileOutputStream(localBufferFile);
            IOUtils.copy(hbaseDataStream, localFileStream);

            hbaseDataStream.close();
            localFileStream.close();

            localBufferFile.deleteOnExit();
        }
        String tableFileDir = temp.getParent();
        temp.delete();

        HiveClient hiveClient = new HiveClient();

        // create hive tables
        hiveClient.executeHQL("CREATE DATABASE IF NOT EXISTS EDW");
        hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CAL_DT.toUpperCase())));
        hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_CATEGORY_GROUPINGS.toUpperCase())));
        hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_KYLIN_FACT.toUpperCase())));
        hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SELLER_TYPE_DIM.toUpperCase())));
        hiveClient.executeHQL(generateCreateTableHql(metaMgr.getTableDesc(TABLE_SITES.toUpperCase())));

        // load data to hive tables
        // LOAD DATA LOCAL INPATH 'filepath' [OVERWRITE] INTO TABLE tablename
        hiveClient.executeHQL(generateLoadDataHql(TABLE_CAL_DT, tableFileDir));
        hiveClient.executeHQL(generateLoadDataHql(TABLE_CATEGORY_GROUPINGS, tableFileDir));
        hiveClient.executeHQL(generateLoadDataHql(TABLE_KYLIN_FACT, tableFileDir));
        hiveClient.executeHQL(generateLoadDataHql(TABLE_SELLER_TYPE_DIM, tableFileDir));
        hiveClient.executeHQL(generateLoadDataHql(TABLE_SITES, tableFileDir));
    }

    private static String generateLoadDataHql(String tableName, String tableFileDir) {
        return "LOAD DATA LOCAL INPATH '" + tableFileDir + "/" + tableName.toUpperCase() + ".csv' OVERWRITE INTO TABLE " + tableName.toUpperCase();
    }

    private static String[] generateCreateTableHql(TableDesc tableDesc) {

        String dropsql = "DROP TABLE IF EXISTS " + tableDesc.getIdentity();
        StringBuilder ddl = new StringBuilder();

        ddl.append("CREATE TABLE " + tableDesc.getIdentity() + "\n");
        ddl.append("(" + "\n");

        for (int i = 0; i < tableDesc.getColumns().length; i++) {
            ColumnDesc col = tableDesc.getColumns()[i];
            if (i > 0) {
                ddl.append(",");
            }
            ddl.append(col.getName() + " " + SqlHiveDataTypeMapping.getHiveDataType((col.getDatatype())) + "\n");
        }

        ddl.append(")" + "\n");
        ddl.append("ROW FORMAT DELIMITED FIELDS TERMINATED BY ','" + "\n");
        ddl.append("STORED AS TEXTFILE");

        return new String[] { dropsql, ddl.toString() };
    }

}


text:
package com.amazonaws.services.connectcases.model;

import java.io.Serializable;
import javax.annotation.Generated;
import com.amazonaws.protocol.StructuredPojo;
import com.amazonaws.protocol.ProtocolMarshaller;

/**
 *

 * A filter for related items of type Contact.
 *

 *
 * @see AWS API
 *      Documentation
 */
@Generated("com.amazonaws:aws-java-sdk-code-generator")
public class ContactFilter implements Serializable, Cloneable, StructuredPojo {

    /**
     *

     * A list of channels to filter on for related items of type Contact.
     *

     */
    private java.util.List<String> channel;
    /**
     *

     * A unique identifier of a contact in Amazon Connect.
     *

     */
    private String contactArn;

    /**
     *

     * A list of channels to filter on for related items of type Contact.
     *

     *
     * @return A list of channels to filter on for related items of type Contact.
     */

    public java.util.List<String> getChannel() {
        return channel;
    }

    /**
     *

     * A list of channels to filter on for related items of type Contact.
     *

     *
     * @param channel
     *        A list of channels to filter on for related items of type Contact.
     */

    public void setChannel(java.util.Collection<String> channel) {
        if (channel == null) {
            this.channel = null;
            return;
        }

        this.channel = new java.util.ArrayList<String>(channel);
    }

    /**
     *

     * A list of channels to filter on for related items of type Contact.
     *

     *

     * NOTE: This method appends the values to the existing list (if any). Use
     * {@link #setChannel(java.util.Collection)} or {@link #withChannel(java.util.Collection)} if you want to override
     * the existing values.
     *

     *
     * @param channel
     *        A list of channels to filter on for related items of type Contact.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public ContactFilter withChannel(String... channel) {
        if (this.channel == null) {
            setChannel(new java.util.ArrayList<String>(channel.length));
        }
        for (String ele : channel) {
            this.channel.add(ele);
        }
        return this;
    }

    /**
     *

     * A list of channels to filter on for related items of type Contact.
     *

     *
     * @param channel
     *        A list of channels to filter on for related items of type Contact.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public ContactFilter withChannel(java.util.Collection<String> channel) {
        setChannel(channel);
        return this;
    }

    /**
     *

     * A unique identifier of a contact in Amazon Connect.
     *

     *
     * @param contactArn
     *        A unique identifier of a contact in Amazon Connect.
     */

    public void setContactArn(String contactArn) {
        this.contactArn = contactArn;
    }

    /**
     *

     * A unique identifier of a contact in Amazon Connect.
     *

     *
     * @return A unique identifier of a contact in Amazon Connect.
     */

    public String getContactArn() {
        return this.contactArn;
    }

    /**
     *

     * A unique identifier of a contact in Amazon Connect.
     *

     *
     * @param contactArn
     *        A unique identifier of a contact in Amazon Connect.
     * @return Returns a reference to this object so that method calls can be chained together.
     */

    public ContactFilter withContactArn(String contactArn) {
        setContactArn(contactArn);
        return this;
    }

    /**
     * Returns a string representation of this object. This is useful for testing and debugging. Sensitive data will be
     * redacted from this string using a placeholder value.
     *
     * @return A string representation of this object.
     *
     * @see java.lang.Object#toString()
     */
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("{");
        if (getChannel() != null)
            sb.append("Channel: ").append(getChannel()).append(",");
        if (getContactArn() != null)
            sb.append("ContactArn: ").append(getContactArn());
        sb.append("}");
        return sb.toString();
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;

        if (obj instanceof ContactFilter == false)
            return false;
        ContactFilter other = (ContactFilter) obj;
        if (other.getChannel() == null ^ this.getChannel() == null)
            return false;
        if (other.getChannel() != null && other.getChannel().equals(this.getChannel()) == false)
            return false;
        if (other.getContactArn() == null ^ this.getContactArn() == null)
            return false;
        if (other.getContactArn() != null && other.getContactArn().equals(this.getContactArn()) == false)
            return false;
        return true;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int hashCode = 1;

        hashCode = prime * hashCode + ((getChannel() == null) ? 0 : getChannel().hashCode());
        hashCode = prime * hashCode + ((getContactArn() == null) ? 0 : getContactArn().hashCode());
        return hashCode;
    }

    @Override
    public ContactFilter clone() {
        try {
            return (ContactFilter) super.clone();
        } catch (CloneNotSupportedException e) {
            throw new IllegalStateException("Got a CloneNotSupportedException from Object.clone() " + "even though we're Cloneable!", e);
        }
    }

    @com.amazonaws.annotation.SdkInternalApi
    @Override
    public void marshall(ProtocolMarshaller protocolMarshaller) {
        com.amazonaws.services.connectcases.model.transform.ContactFilterMarshaller.getInstance().marshall(this, protocolMarshaller);
    }
}

redpajama_set_name: RedPajamaGithub


text:
package org.apache.reef.runtime.local.client;

import org.apache.reef.runtime.common.client.api.JobSubmissionEvent;
import org.apache.reef.runtime.common.files.FileResource;
import org.apache.reef.runtime.common.files.FileType;
import org.apache.reef.runtime.common.files.REEFFileNames;
import org.apache.reef.tang.formats.ConfigurationModule;
import org.apache.reef.tang.formats.OptionalParameter;

import java.io.File;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Represents the files added to a driver.
 *

 * This class is constructed via the from() method that instantiates it based on a JobSubmissionProto
 */
final class DriverFiles {

  private static final Logger LOG = Logger.getLogger(DriverFiles.class.getName());

  private final FileSet localFiles = new FileSet();
  private final FileSet localLibs = new FileSet();
  private final FileSet globalFiles = new FileSet();
  private final FileSet globalLibs = new FileSet();
  private final REEFFileNames fileNames;

  DriverFiles(final REEFFileNames fileNames) {
    this.fileNames = fileNames;
  }

  /**
   * Instantiates an instance based on the given JobSubmissionProto.
   *
   * @param jobSubmissionEvent the JobSubmissionProto to parse.
   * @return a DriverFiles instance pre-populated with the information from the given JobSubmissionProto.
   * @throws IOException
   */
  public static DriverFiles fromJobSubmission(
      final JobSubmissionEvent jobSubmissionEvent,
      final REEFFileNames fileNames) throws IOException {

    final DriverFiles driverFiles = new DriverFiles(fileNames);

    for (final FileResource frp : jobSubmissionEvent.getGlobalFileSet()) {
      final File f = new File(frp.getPath());
      if (frp.getType() == FileType.LIB) {
        driverFiles.addGlobalLib(f);
      } else {
        driverFiles.addGlobalFile(f);
      }
    }

    for (final FileResource frp : jobSubmissionEvent.getLocalFileSet()) {
      final File f = new File(frp.getPath());
      if (frp.getType() == FileType.LIB) {
        driverFiles.addLocalLib(f);
      } else {
        driverFiles.addLocalFile(f);
      }
    }

    return driverFiles;
  }

  private void addLocalLib(final File f) throws IOException {
    checkFile(f);
    this.localLibs.add(f);
  }

  private void addLocalFile(final File f) throws IOException {
    checkFile(f);
    this.localFiles.add(f);
  }

  private void addGlobalFile(final File f) throws IOException {
    checkFile(f);
    this.globalFiles.add(f);
  }

  private void addGlobalLib(final File f) throws IOException {
    checkFile(f);
    this.globalLibs.add(f);
  }

  private void checkFile(final File f) {
    if (this.globalLibs.containsFileWithName(f.getName())) {
      LOG.log(Level.FINEST, "Adding a file that is already part of the global libraries: {0}", f);
    }
    if (this.globalFiles.containsFileWithName(f.getName())) {
      LOG.log(Level.FINEST, "Adding a file that is already part of the global files: {0}", f);
    }
    if (this.localLibs.containsFileWithName(f.getName())) {
      LOG.log(Level.FINEST, "Adding a file that is already part of the local libraries: {0}", f);
    }
    if (this.localFiles.containsFileWithName(f.getName())) {
      LOG.log(Level.FINEST, "Adding a file that is already part of the local files: {0}", f);
    }
  }

  /**
   * Copies this set of files to the destination folder given.
   *

   * Will attempt to create symbolic links for the files to the destination
   * folder.  If the filesystem does not support symbolic links or the user
   * does not have appropriate permissions, the entire file will be copied instead.
   *
   * @param destinationFolder the folder the files shall be copied to.
   * @throws IOException if one or more of the copies fail.
   */
  public void copyTo(final File destinationFolder) throws IOException {
    if (!destinationFolder.exists() && !destinationFolder.mkdirs()) {
      LOG.log(Level.WARNING, "Failed to create [{0}]", destinationFolder.getAbsolutePath());
    }
    final File reefFolder = new File(destinationFolder, fileNames.getREEFFolderName());

    final File localFolder = new File(reefFolder, fileNames.getLocalFolderName());
    final File globalFolder = new File(reefFolder, fileNames.getGlobalFolderName());
    if (!localFolder.exists() && !localFolder.mkdirs()) {
      LOG.log(Level.WARNING, "Failed to create [{0}]", localFolder.getAbsolutePath());
    }
    if (!globalFolder.exists() && !globalFolder.mkdirs()) {
      LOG.log(Level.WARNING, "Failed to create [{0}]", globalFolder.getAbsolutePath());
    }

    try {
      this.localFiles.createSymbolicLinkTo(localFolder);
      this.localLibs.createSymbolicLinkTo(localFolder);
      this.globalLibs.createSymbolicLinkTo(globalFolder);
      this.globalFiles.createSymbolicLinkTo(globalFolder);
    } catch (final IOException e) {
      LOG.log(Level.FINE, "Can't symlink the files, copying them instead.", e);
      this.localFiles.copyTo(localFolder);
      this.localLibs.copyTo(localFolder);
      this.globalLibs.copyTo(globalFolder);
      this.globalFiles.copyTo(globalFolder);
    }
  }

  /**
   * Fills out a ConfigurationModule.
   *
   * @param input           The ConfigurationModule to start with.
   * @param globalFileField the field on which to set() the global files.
   * @param globalLibField  the field on which to set() the global libraries.
   * @param localFileField  the field on which to set() the local files.
   * @param localLibField   the field on which to set() the local libraries.
   * @return a copy of input with files and libraries added to the given fields.
   */
  public ConfigurationModule addNamesTo(final ConfigurationModule input,
                                        final OptionalParameter<String> globalFileField,
                                        final OptionalParameter<String> globalLibField,
                                        final OptionalParameter<String> localFileField,
                                        final OptionalParameter<String> localLibField) {
    ConfigurationModule result = input;
    result = this.globalFiles.addNamesTo(result, globalFileField);
    result = this.globalLibs.addNamesTo(result, globalLibField);
    result = this.localFiles.addNamesTo(result, localFileField);
    result = this.localLibs.addNamesTo(result, localLibField);
    return result;
  }
}


text:
require 'codeclimate-test-reporter'
CodeClimate::TestReporter.start

require 'coveralls'
Coveralls.wear_merged!('rails')

require 'devise'
require 'factory_girl_rails'
# spec/spec_helper.rb
# require 'rspec/retry'

RSpec.configure do |config|
  config.filter_run_including focus: true
  config.run_all_when_everything_filtered = true

  config.expect_with :rspec do |expectations|
    expectations.include_chain_clauses_in_custom_matcher_descriptions = true
  end

  config.include Devise::TestHelpers, type: :controller

  config.mock_with :rspec do |mocks|
    mocks.verify_partial_doubles = true
  end
end


text:
/* eslint no-var:0 */
var path = require('path');
var webpack = require('webpack');
var HtmlWebpackPlugin = require('html-webpack-plugin');

module.exports = {
  entry: [
    'webpack-hot-middleware/client',
    './src/index.js',
  ],
  output: {
    filename: 'bundle.js',
    path: path.join(__dirname, 'build'),
    publicPath: '/',
  },
  module: {
    loaders: [
      {
        test: /\.js$/,
        loaders: ['babel'],
        exclude: /node_modules/,
        include: path.join(__dirname, 'src'),
      },
      {
        test: /\.css$/,
        loaders: [
          'style?sourceMap',
          'css?modules&importLoaders=1&localIdentName=[path]___[name]__[local]___[hash:base64:5]',
        ],
      },
    ],
  },
  devtool: 'cheap-eval-source-map',
  plugins: [
    new HtmlWebpackPlugin({
      template: './src/index.html',
    }),
    new webpack.optimize.OccurenceOrderPlugin(),
    new webpack.HotModuleReplacementPlugin(),
  ],
};


text:
<?php

namespace Symfony\Component\Validator\Constraints;

use Symfony\Component\Validator\Constraint;

/**
 * @Annotation
 *
 * @api
 */
class Valid extends Constraint {
	public $traverse = true;
}


text:
package itercomplex64

import (
	"errors"
)

var (
	errNilReceiver = errors.New("Nil Receiver")
)


text:
package org.elasticsearch.index.store.cache;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.message.ParameterizedMessage;
import org.apache.lucene.store.AlreadyClosedException;
import org.apache.lucene.store.IOContext;
import org.apache.lucene.store.IndexInput;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.SuppressForbidden;
import org.elasticsearch.common.collect.Tuple;
import org.elasticsearch.common.io.Channels;
import org.elasticsearch.common.lease.Releasable;
import org.elasticsearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo;
import org.elasticsearch.index.store.BaseSearchableSnapshotIndexInput;
import org.elasticsearch.index.store.IndexInputStats;
import org.elasticsearch.index.store.SearchableSnapshotDirectory;
import org.elasticsearch.xpack.searchablesnapshots.SearchableSnapshotsConstants;

import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.IntStream;

public class CachedBlobContainerIndexInput extends BaseSearchableSnapshotIndexInput {

    /**
     * Specific IOContext used for prewarming the cache. This context allows to write
     * a complete part of the {@link #fileInfo} at once in the cache and should not be
     * used for anything else than what the {@link #prefetchPart(int)} method does.
     */
    public static final IOContext CACHE_WARMING_CONTEXT = new IOContext();

    private static final Logger logger = LogManager.getLogger(CachedBlobContainerIndexInput.class);
    private static final int COPY_BUFFER_SIZE = 8192;

    private final SearchableSnapshotDirectory directory;
    private final CacheFileReference cacheFileReference;
    private final int defaultRangeSize;

    // last read position is kept around in order to detect (non)contiguous reads for stats
    private long lastReadPosition;
    // last seek position is kept around in order to detect forward/backward seeks for stats
    private long lastSeekPosition;

    public CachedBlobContainerIndexInput(
        SearchableSnapshotDirectory directory,
        FileInfo fileInfo,
        IOContext context,
        IndexInputStats stats,
        int rangeSize
    ) {
        this(
            "CachedBlobContainerIndexInput(" + fileInfo.physicalName() + ")",
            directory,
            fileInfo,
            context,
            stats,
            0L,
            fileInfo.length(),
            new CacheFileReference(directory, fileInfo.physicalName(), fileInfo.length()),
            rangeSize
        );
        stats.incrementOpenCount();
    }

    private CachedBlobContainerIndexInput(
        String resourceDesc,
        SearchableSnapshotDirectory directory,
        FileInfo fileInfo,
        IOContext context,
        IndexInputStats stats,
        long offset,
        long length,
        CacheFileReference cacheFileReference,
        int rangeSize
    ) {
        super(resourceDesc, directory.blobContainer(), fileInfo, context, stats, offset, length);
        this.directory = directory;
        this.cacheFileReference = cacheFileReference;
        this.lastReadPosition = this.offset;
        this.lastSeekPosition = this.offset;
        this.defaultRangeSize = rangeSize;
    }

    @Override
    public void innerClose() {
        if (isClone == false) {
            cacheFileReference.releaseOnClose();
        }
    }

    private void ensureContext(Predicate<IOContext> predicate) throws IOException {
        if (predicate.test(context) == false) {
            assert false : "this method should not be used with this context " + context;
            throw new IOException("Cannot read the index input using context [context=" + context + ", input=" + this + ']');
        }
    }

    private long getDefaultRangeSize() {
        return (context != CACHE_WARMING_CONTEXT) ? defaultRangeSize : fileInfo.partSize().getBytes();
    }

    private Tuple<Long, Long> computeRange(long position) {
        final long rangeSize = getDefaultRangeSize();
        long start = (position / rangeSize) * rangeSize;
        long end = Math.min(start + rangeSize, fileInfo.length());
        return Tuple.tuple(start, end);
    }

    private CacheFile getCacheFileSafe() throws Exception {
        final CacheFile cacheFile = cacheFileReference.get();
        if (cacheFile == null) {
            throw new AlreadyClosedException("Failed to acquire a non-evicted cache file");
        }
        return cacheFile;
    }

    @Override
    protected void readInternal(ByteBuffer b) throws IOException {
        ensureContext(ctx -> ctx != CACHE_WARMING_CONTEXT);
        assert assertCurrentThreadIsNotCacheFetchAsync();
        final long position = getFilePointer() + this.offset;
        final int length = b.remaining();

        int totalBytesRead = 0;
        while (totalBytesRead < length) {
            final long pos = position + totalBytesRead;
            final int len = length - totalBytesRead;
            int bytesRead = 0;
            try {
                final CacheFile cacheFile = getCacheFileSafe();
                try (Releasable ignored = cacheFile.fileLock()) {
                    final Tuple<Long, Long> rangeToWrite = computeRange(pos);
                    final Tuple<Long, Long> rangeToRead = Tuple.tuple(pos, Math.min(pos + len, rangeToWrite.v2()));

                    bytesRead = cacheFile.fetchAsync(rangeToWrite, rangeToRead, (channel) -> {
                        final int read;
                        if ((rangeToRead.v2() - rangeToRead.v1()) < b.remaining()) {
                            final ByteBuffer duplicate = b.duplicate();
                            duplicate.limit(duplicate.position() + Math.toIntExact(rangeToRead.v2() - rangeToRead.v1()));
                            read = readCacheFile(channel, pos, duplicate);
                            assert duplicate.position() <= b.limit();
                            b.position(duplicate.position());
                        } else {
                            read = readCacheFile(channel, pos, b);
                        }
                        return read;
                    }, this::writeCacheFile, directory.cacheFetchAsyncExecutor()).get();
                }
            } catch (final Exception e) {
                if (e instanceof AlreadyClosedException || (e.getCause() != null && e.getCause() instanceof AlreadyClosedException)) {
                    try {
                        // cache file was evicted during the range fetching, read bytes directly from source
                        bytesRead = readDirectly(pos, pos + len, b);
                        continue;
                    } catch (Exception inner) {
                        e.addSuppressed(inner);
                    }
                }
                throw new IOException("Fail to read data from cache", e);

            } finally {
                totalBytesRead += bytesRead;
            }
        }
        assert totalBytesRead == length : "partial read operation, read [" + totalBytesRead + "] bytes of [" + length + "]";
        stats.incrementBytesRead(lastReadPosition, position, totalBytesRead);
        lastReadPosition = position + totalBytesRead;
        lastSeekPosition = lastReadPosition;
    }

    /**
     * Prefetches a complete part and writes it in cache. This method is used to prewarm the cache.
     */
    public void prefetchPart(final int part) throws IOException {
        ensureContext(ctx -> ctx == CACHE_WARMING_CONTEXT);
        if (part >= fileInfo.numberOfParts()) {
            throw new IllegalArgumentException("Unexpected part number [" + part + "]");
        }
        final Tuple<Long, Long> partRange = computeRange(IntStream.range(0, part).mapToLong(fileInfo::partBytes).sum());
        assert assertRangeIsAlignedWithPart(partRange);

        try {
            final CacheFile cacheFile = getCacheFileSafe();
            try (Releasable ignored = cacheFile.fileLock()) {

                final Tuple<Long, Long> range = cacheFile.getAbsentRangeWithin(partRange.v1(), partRange.v2());
                if (range == null) {
                    logger.trace(
                        "prefetchPart: part [{}] bytes [{}-{}] is already fully available for cache file [{}]",
                        part,
                        partRange.v1(),
                        partRange.v2(),
                        cacheFileReference
                    );
                    return;
                }

                final long rangeStart = range.v1();
                final long rangeEnd = range.v2();
                final long rangeLength = rangeEnd - rangeStart;

                logger.trace(
                    "prefetchPart: prewarming part [{}] bytes [{}-{}] by fetching bytes [{}-{}] for cache file [{}]",
                    part,
                    partRange.v1(),
                    partRange.v2(),
                    rangeStart,
                    rangeEnd,
                    cacheFileReference
                );

                final FileChannel fc = cacheFile.getChannel();
                assert assertFileChannelOpen(fc);
                final byte[] copyBuffer = new byte[Math.toIntExact(Math.min(COPY_BUFFER_SIZE, rangeLength))];

                long totalBytesRead = 0L;
                final AtomicLong totalBytesWritten = new AtomicLong();
                long remainingBytes = rangeEnd - rangeStart;
                final long startTimeNanos = stats.currentTimeNanos();
                try (InputStream input = openInputStream(rangeStart, rangeLength)) {
                    while (remainingBytes > 0L) {
                        assert totalBytesRead + remainingBytes == rangeLength;
                        final int bytesRead = readSafe(input, copyBuffer, rangeStart, rangeEnd, remainingBytes, cacheFileReference);

                        // The range to prewarm in cache
                        final long readStart = rangeStart + totalBytesRead;
                        final Tuple<Long, Long> rangeToWrite = Tuple.tuple(readStart, readStart + bytesRead);

                        cacheFile.fetchAsync(rangeToWrite, rangeToWrite, (channel) -> bytesRead, (channel, start, end, progressUpdater) -> {
                            final ByteBuffer byteBuffer = ByteBuffer.wrap(
                                copyBuffer,
                                Math.toIntExact(start - readStart),
                                Math.toIntExact(end - start)
                            );
                            final int writtenBytes = positionalWrite(channel, start, byteBuffer);
                            logger.trace(
                                "prefetchPart: writing range [{}-{}] of file [{}], [{}] bytes written",
                                start,
                                end,
                                fileInfo.physicalName(),
                                writtenBytes
                            );
                            totalBytesWritten.addAndGet(writtenBytes);
                            progressUpdater.accept(start + writtenBytes);
                        }, directory.cacheFetchAsyncExecutor()).get();
                        totalBytesRead += bytesRead;
                        remainingBytes -= bytesRead;
                    }
                    final long endTimeNanos = stats.currentTimeNanos();
                    stats.addCachedBytesWritten(totalBytesWritten.get(), endTimeNanos - startTimeNanos);
                }

                assert totalBytesRead == rangeLength;
            }
        } catch (final Exception e) {
            throw new IOException("Failed to prefetch file part in cache", e);
        }
    }

    @SuppressForbidden(reason = "Use positional writes on purpose")
    private static int positionalWrite(FileChannel fc, long start, ByteBuffer byteBuffer) throws IOException {
        assert assertCurrentThreadMayWriteCacheFile();
        return fc.write(byteBuffer, start);
    }

    /**
     * Perform a single {@code read()} from {@code inputStream} into {@code copyBuffer}, handling an EOF by throwing an {@link EOFException}
     * rather than returning {@code -1}. Returns the number of bytes read, which is always positive.
     *
     * Most of its arguments are there simply to make the message of the {@link EOFException} more informative.
     */
    private static int readSafe(
        InputStream inputStream,
        byte[] copyBuffer,
        long rangeStart,
        long rangeEnd,
        long remaining,
        CacheFileReference cacheFileReference
    ) throws IOException {
        final int len = (remaining < copyBuffer.length) ? Math.toIntExact(remaining) : copyBuffer.length;
        final int bytesRead = inputStream.read(copyBuffer, 0, len);
        if (bytesRead == -1) {
            throw new EOFException(
                String.format(
                    Locale.ROOT,
                    "unexpected EOF reading [%d-%d] ([%d] bytes remaining) from %s",
                    rangeStart,
                    rangeEnd,
                    remaining,
                    cacheFileReference
                )
            );
        }
        assert bytesRead > 0 : bytesRead;
        return bytesRead;
    }

    /**
     * Asserts that the range of bytes to warm in cache is aligned with {@link #fileInfo}'s part size.
     */
    private boolean assertRangeIsAlignedWithPart(Tuple<Long, Long> range) {
        if (fileInfo.numberOfParts() == 1L) {
            final long length = fileInfo.length();
            assert range.v1() == 0L : "start of range [" + range.v1() + "] is not aligned with zero";
            assert range.v2() == length : "end of range [" + range.v2() + "] is not aligned with file length [" + length + ']';
        } else {
            final long length = fileInfo.partSize().getBytes();
            assert range.v1() % length == 0L : "start of range [" + range.v1() + "] is not aligned with part start";
            assert range.v2() % length == 0L || (range.v2() == fileInfo.length()) : "end of range ["
                + range.v2()
                + "] is not aligned with part end or with file length";
        }
        return true;
    }

    private int readCacheFile(final FileChannel fc, final long position, final ByteBuffer buffer) throws IOException {
        assert assertFileChannelOpen(fc);
        final int bytesRead = Channels.readFromFileChannel(fc, position, buffer);
        if (bytesRead == -1) {
            throw new EOFException(
                String.format(
                    Locale.ROOT,
                    "unexpected EOF reading [%d-%d] from %s",
                    position,
                    position + buffer.remaining(),
                    cacheFileReference
                )
            );
        }
        stats.addCachedBytesRead(bytesRead);
        return bytesRead;
    }

    private void writeCacheFile(final FileChannel fc, final long start, final long end, final Consumer<Long> progressUpdater)
        throws IOException {
        assert assertFileChannelOpen(fc);
        assert assertCurrentThreadMayWriteCacheFile();
        final long length = end - start;
        final byte[] copyBuffer = new byte[Math.toIntExact(Math.min(COPY_BUFFER_SIZE, length))];
        logger.trace(() -> new ParameterizedMessage("writing range [{}-{}] to cache file [{}]", start, end, cacheFileReference));

        long bytesCopied = 0L;
        long remaining = end - start;
        final long startTimeNanos = stats.currentTimeNanos();
        try (InputStream input = openInputStream(start, length)) {
            while (remaining > 0L) {
                final int bytesRead = readSafe(input, copyBuffer, start, end, remaining, cacheFileReference);
                positionalWrite(fc, start + bytesCopied, ByteBuffer.wrap(copyBuffer, 0, bytesRead));
                bytesCopied += bytesRead;
                remaining -= bytesRead;
                progressUpdater.accept(start + bytesCopied);
            }
            final long endTimeNanos = stats.currentTimeNanos();
            stats.addCachedBytesWritten(bytesCopied, endTimeNanos - startTimeNanos);
        }
    }

    @Override
    protected void seekInternal(long pos) throws IOException {
        if (pos > length()) {
            throw new EOFException("Reading past end of file [position=" + pos + ", length=" + length() + "] for " + toString());
        } else if (pos < 0L) {
            throw new IOException("Seeking to negative position [" + pos + "] for " + toString());
        }
        final long position = pos + this.offset;
        stats.incrementSeeks(lastSeekPosition, position);
        lastSeekPosition = position;
    }

    @Override
    public CachedBlobContainerIndexInput clone() {
        return (CachedBlobContainerIndexInput) super.clone();
    }

    @Override
    public IndexInput slice(String sliceDescription, long offset, long length) {
        if (offset < 0 || length < 0 || offset + length > length()) {
            throw new IllegalArgumentException(
                "slice() "
                    + sliceDescription
                    + " out of bounds: offset="
                    + offset
                    + ",length="
                    + length
                    + ",fileLength="
                    + length()
                    + ": "
                    + this
            );
        }
        final CachedBlobContainerIndexInput slice = new CachedBlobContainerIndexInput(
            getFullSliceDescription(sliceDescription),
            directory,
            fileInfo,
            context,
            stats,
            this.offset + offset,
            length,
            cacheFileReference,
            defaultRangeSize
        );
        slice.isClone = true;
        return slice;
    }

    @Override
    public String toString() {
        return "CachedBlobContainerIndexInput{"
            + "cacheFileReference="
            + cacheFileReference
            + ", offset="
            + offset
            + ", length="
            + length()
            + ", position="
            + getFilePointer()
            + ", rangeSize="
            + getDefaultRangeSize()
            + '}';
    }

    private int readDirectly(long start, long end, ByteBuffer b) throws IOException {
        final long length = end - start;
        final byte[] copyBuffer = new byte[Math.toIntExact(Math.min(COPY_BUFFER_SIZE, length))];
        logger.trace(() -> new ParameterizedMessage("direct reading of range [{}-{}] for cache file [{}]", start, end, cacheFileReference));

        int bytesCopied = 0;
        final long startTimeNanos = stats.currentTimeNanos();
        try (InputStream input = openInputStream(start, length)) {
            long remaining = end - start;
            while (remaining > 0) {
                final int len = (remaining < copyBuffer.length) ? (int) remaining : copyBuffer.length;
                int bytesRead = input.read(copyBuffer, 0, len);
                if (bytesRead == -1) {
                    throw new EOFException(
                        String.format(
                            Locale.ROOT,
                            "unexpected EOF reading [%d-%d] ([%d] bytes remaining) from %s",
                            start,
                            end,
                            remaining,
                            cacheFileReference
                        )
                    );
                }
                b.put(copyBuffer, 0, bytesRead);
                bytesCopied += bytesRead;
                remaining -= bytesRead;
            }
            final long endTimeNanos = stats.currentTimeNanos();
            stats.addDirectBytesRead(bytesCopied, endTimeNanos - startTimeNanos);
        }
        return bytesCopied;
    }

    private static class CacheFileReference implements CacheFile.EvictionListener {

        private final long fileLength;
        private final CacheKey cacheKey;
        private final SearchableSnapshotDirectory directory;
        private final AtomicReference<CacheFile> cacheFile = new AtomicReference<>(); // null if evicted or not yet acquired

        private CacheFileReference(SearchableSnapshotDirectory directory, String fileName, long fileLength) {
            this.cacheKey = directory.createCacheKey(fileName);
            this.fileLength = fileLength;
            this.directory = directory;
        }

        @Nullable
        CacheFile get() throws Exception {
            CacheFile currentCacheFile = cacheFile.get();
            if (currentCacheFile != null) {
                return currentCacheFile;
            }

            final CacheFile newCacheFile = directory.getCacheFile(cacheKey, fileLength);
            synchronized (this) {
                currentCacheFile = cacheFile.get();
                if (currentCacheFile != null) {
                    return currentCacheFile;
                }
                if (newCacheFile.acquire(this)) {
                    final CacheFile previousCacheFile = cacheFile.getAndSet(newCacheFile);
                    assert previousCacheFile == null;
                    return newCacheFile;
                }
            }
            return null;
        }

        @Override
        public void onEviction(final CacheFile evictedCacheFile) {
            synchronized (this) {
                if (cacheFile.compareAndSet(evictedCacheFile, null)) {
                    evictedCacheFile.release(this);
                }
            }
        }

        void releaseOnClose() {
            synchronized (this) {
                final CacheFile currentCacheFile = cacheFile.getAndSet(null);
                if (currentCacheFile != null) {
                    currentCacheFile.release(this);
                }
            }
        }

        @Override
        public String toString() {
            return "CacheFileReference{"
                + "cacheKey='"
                + cacheKey
                + '\''
                + ", fileLength="
                + fileLength
                + ", acquired="
                + (cacheFile.get() != null)
                + '}';
        }
    }

    private static boolean assertFileChannelOpen(FileChannel fileChannel) {
        assert fileChannel != null;
        assert fileChannel.isOpen();
        return true;
    }

    private static boolean isCacheFetchAsyncThread(final String threadName) {
        return threadName.contains('[' + SearchableSnapshotsConstants.CACHE_FETCH_ASYNC_THREAD_POOL_NAME + ']');
    }

    private static boolean assertCurrentThreadMayWriteCacheFile() {
        final String threadName = Thread.currentThread().getName();
        assert isCacheFetchAsyncThread(threadName) : "expected the current thread ["
            + threadName
            + "] to belong to the cache fetch async thread pool";
        return true;
    }

    private static boolean assertCurrentThreadIsNotCacheFetchAsync() {
        final String threadName = Thread.currentThread().getName();
        assert false == isCacheFetchAsyncThread(threadName) : "expected the current thread ["
            + threadName
            + "] to belong to the cache fetch async thread pool";
        return true;
    }
}


text:
package mocks

import context "context"
import deployments "github.com/mendersoftware/deployments/resources/deployments"
import mock "github.com/stretchr/testify/mock"
import model "github.com/mendersoftware/deployments/resources/deployments/model"
import time "time"

// DeploymentsStorage is an autogenerated mock type for the DeploymentsStorage type
type DeploymentsStorage struct {
	mock.Mock
}

// Delete provides a mock function with given fields: ctx, id
func (_m *DeploymentsStorage) Delete(ctx context.Context, id string) error {
	ret := _m.Called(ctx, id)

	var r0 error
	if rf, ok := ret.Get(0).(func(context.Context, string) error); ok {
		r0 = rf(ctx, id)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}

// Find provides a mock function with given fields: ctx, query
func (_m *DeploymentsStorage) Find(ctx context.Context, query deployments.Query) ([]*deployments.Deployment, error) {
	ret := _m.Called(ctx, query)

	var r0 []*deployments.Deployment
	if rf, ok := ret.Get(0).(func(context.Context, deployments.Query) []*deployments.Deployment); ok {
		r0 = rf(ctx, query)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).([]*deployments.Deployment)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, deployments.Query) error); ok {
		r1 = rf(ctx, query)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// FindByID provides a mock function with given fields: ctx, id
func (_m *DeploymentsStorage) FindByID(ctx context.Context, id string) (*deployments.Deployment, error) {
	ret := _m.Called(ctx, id)

	var r0 *deployments.Deployment
	if rf, ok := ret.Get(0).(func(context.Context, string) *deployments.Deployment); ok {
		r0 = rf(ctx, id)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*deployments.Deployment)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
		r1 = rf(ctx, id)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// FindUnfinishedByID provides a mock function with given fields: ctx, id
func (_m *DeploymentsStorage) FindUnfinishedByID(ctx context.Context, id string) (*deployments.Deployment, error) {
	ret := _m.Called(ctx, id)

	var r0 *deployments.Deployment
	if rf, ok := ret.Get(0).(func(context.Context, string) *deployments.Deployment); ok {
		r0 = rf(ctx, id)
	} else {
		if ret.Get(0) != nil {
			r0 = ret.Get(0).(*deployments.Deployment)
		}
	}

	var r1 error
	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
		r1 = rf(ctx, id)
	} else {
		r1 = ret.Error(1)
	}

	return r0, r1
}

// Finish provides a mock function with given fields: ctx, id, when
func (_m *DeploymentsStorage) Finish(ctx context.Context, id string, when time.Time) error {
	ret := _m.Called(ctx, id, when)

	var r0 error
	if rf, ok := ret.Get(0).(func(context.Context, string, time.Time) error); ok {
		r0 = rf(ctx, id, when)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}

// Insert provides a mock function with given fields: ctx, deployment
func (_m *DeploymentsStorage) Insert(ctx context.Context, deployment *deployments.Deployment) error {
	ret := _m.Called(ctx, deployment)

	var r0 error
	if rf, ok := ret.Get(0).(func(context.Context, *deployments.Deployment) error); ok {
		r0 = rf(ctx, deployment)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}

// UpdateStats provides a mock function with given fields: ctx, id, state_from, state_to
func (_m *DeploymentsStorage) UpdateStats(ctx context.Context, id string, state_from string, state_to string) error {
	ret := _m.Called(ctx, id, state_from, state_to)

	var r0 error
	if rf, ok := ret.Get(0).(func(context.Context, string, string, string) error); ok {
		r0 = rf(ctx, id, state_from, state_to)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}

// UpdateStatsAndFinishDeployment provides a mock function with given fields: ctx, id, stats
func (_m *DeploymentsStorage) UpdateStatsAndFinishDeployment(ctx context.Context, id string, stats deployments.Stats) error {
	ret := _m.Called(ctx, id, stats)

	var r0 error
	if rf, ok := ret.Get(0).(func(context.Context, string, deployments.Stats) error); ok {
		r0 = rf(ctx, id, stats)
	} else {
		r0 = ret.Error(0)
	}

	return r0
}

var _ model.DeploymentsStorage = (*DeploymentsStorage)(nil)


text:
layout: article
title: "An awkward stare"
excerpt: "PaperFaces portrait of Matt S. drawn with Paper by 53 on an iPad."
image:
  feature: paperfaces-matt-s-lg.jpg
  thumb: paperfaces-matt-s-150.jpg
category: paperfaces
tags: [portrait, illustration, paper by 53, sktchy, black and white]
---

PaperFaces portrait of [Matt S.](http://sktchy.com/4SI2jH) inspired by Sktchy.

{% include paperfaces-boilerplate-2.html %}

Work in progress screen captures Made with Paper.


text:
const STATE_KEY = "STATE_KEY";
const PODCAST_KEY = "PODCAST_KEY";

export const loadState = () => {
    try {
        const serializedState = localStorage.getItem(STATE_KEY);
        if (serializedState === null) {
            return undefined;
        }
        return JSON.parse(serializedState);
    }
    catch (error) {
        console.error('could not deserialize state', error);
        return undefined;
    }
}

export const saveState = (state) => {
    try {
        const serializedState = JSON.stringify(state);
        localStorage.setItem(STATE_KEY, serializedState);
    }
    catch (error) {
        console.error('could not serialize state', error);
    }
}

export const savePodcast = (state) => {
    const position = state.audio.position;
    if(position === 0) {
        return;
    }
    const audioSrc = state.audio.audioSrc;
    const savedPC = getSavedPocastMap();
    savedPC[audioSrc] = { position: position };
    try {
        const serializedPC = JSON.stringify(savedPC);
        localStorage.setItem(PODCAST_KEY, serializedPC);
    }
    catch (error) {
        console.error('could not serialize pc', error);
    }
}

export const getSavedPodCast = (state) => {
    const audioSrc = state.audio.audioSrc;
    return getSavedPocastMap()[audioSrc];
}

const getSavedPocastMap = () => {
    try {
        const serializedPodcastList = localStorage.getItem(PODCAST_KEY);
        if (serializedPodcastList === null) {
            return {};
        }
        return JSON.parse(serializedPodcastList);
    }
    catch (error) {
        console.error('could not deserialize state', error);
        return undefined;
    }
}


text:
package org.anyline.web.tag;

import org.anyline.util.BasicUtil;
import org.anyline.util.CodeUtil;

import javax.servlet.jsp.JspWriter;
import java.io.IOException;

public class Decode extends BaseBodyTag {
	private static final long serialVersionUID = 1L;

	public int doEndTag() {
		String src = BasicUtil.nvl(value,body,"").toString().trim();
		if("".equals(src)){
			return EVAL_BODY_INCLUDE;
		}

		JspWriter writer = null;
		String result = CodeUtil.urlDecode(src);
		try {
			writer = pageContext.getOut();
			writer.print(result);
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			release();
		}
		return EVAL_PAGE;// 标签执行完毕之后继续执行下面的内容
	}

	@Override
	public void release() {
		super.release();
		value = null;
		body = null;
	}

}


text:
package org.onosproject.vtnweb.gui;

import static com.google.common.collect.ImmutableList.of;
import static org.onosproject.ui.UiView.Category.NETWORK;
import static org.slf4j.LoggerFactory.getLogger;

import java.util.List;

import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
import org.apache.felix.scr.annotations.Reference;
import org.apache.felix.scr.annotations.ReferenceCardinality;
import org.apache.felix.scr.annotations.Service;
import org.onosproject.ui.UiExtension;
import org.onosproject.ui.UiExtensionService;
import org.onosproject.ui.UiMessageHandlerFactory;
import org.onosproject.ui.UiView;
import org.slf4j.Logger;

import com.google.common.collect.ImmutableList;

/**
 * service function chain gui.
 */
@Component(immediate = true, enabled = true)
@Service(value = SfcUiExtensionManager.class)
public class SfcUiExtensionManager {
    private final Logger log = getLogger(getClass());

    private static final ClassLoader CL =
            SfcUiExtensionManager.class.getClassLoader();
    private static final String GUI = "gui";

    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
    protected UiExtensionService uiExtensionService;

    // service function chain extension
    private final UiExtension sfc = createSfcExtension();

    // Creates service function chain UI extension
    private UiExtension createSfcExtension() {
        List<UiView> coreViews = of(
                //TODO add a new type of icon for sfc
                new UiView(NETWORK, "sfc", "SFC", "nav_sfcs")
        );

        UiMessageHandlerFactory messageHandlerFactory =
                () -> ImmutableList.of(
                        new SfcViewMessageHandler()
                );

        return new UiExtension.Builder(CL, coreViews)
                .messageHandlerFactory(messageHandlerFactory)
                .resourcePath(GUI)
                .build();
    }

    @Activate
    public void activate() {
        uiExtensionService.register(sfc);
        log.info("Started");
    }

    @Deactivate
    public void deactivate() {
        uiExtensionService.unregister(sfc);
        log.info("Stopped");
    }
}


text:
<?php
namespace Pux;
use ReflectionClass;
use ReflectionObject;
use ReflectionMethod;
use Pux\Mux;

class Controller {

    public function getActionMethods() {
        $refObject = new ReflectionObject($this);
        return array_map(function($m){ return $m->getName(); }, array_filter($refObject->getMethods(), function($m) {
            return preg_match('#Action$#', $m->getName() );
        }));
    }

    public function getActionPaths() {
        $actionNames = array_map(function($method) {
            return preg_replace('#Action$#','', $method);
        }, $this->getActionMethods() );

        $pairs = array();
        foreach( $actionNames as $actionName ) {
            if ( $actionName == "index" ) {
                $path = '';
            } else {
                $path = '/' . preg_replace_callback('/[A-Z]/', function($regs) {
                    return '/' . strtolower($regs[0]);
                }, $actionName);
            }
            $pairs[] = array($path, $actionName);
        }
        return $pairs;
    }

    public function expand() {
        $mux = new Mux;
        $paths = $this->getActionPaths();
        foreach( $paths as $path ) {
            $mux->add($path[0], array(get_class($this), $path[1]) );
        }
        $mux->sort();
        return $mux;
    }

}


text:
const { assert } = require('chai');
const getLoader = require('../dependencies');
const components = require('../components.json');

describe('Dependency logic', function () {

	/** @type {import("../dependencies").Components} */
	const components = {
		languages: {
			'a': {
				alias: 'a2'
			},
			'b': {
				alias: 'b2'
			},
			'c': {
				require: 'a',
				optional: ['b', 'e']
			},
			'd': {
				require: ['c', 'b'],
				alias: 'xyz'
			},
		},
		pluginsOrSomething: {
			'e': {
				modify: 'a'
			},
		}
	};

	/**
	 * Returns the ids of `getLoader`.
	 *
	 * @param {string[]} load
	 * @param {string[]} [loaded]
	 * @returns {string[]}
	 */
	function getIds(load, loaded) {
		return getLoader(components, load, loaded).getIds();
	}

	describe('Returned ids', function () {

		it('- should load requirements', function () {
			assert.sameMembers(getIds(['d']), ['a', 'b', 'c', 'd']);
		});

		it('- should not load already loaded requirements if not necessary', function () {
			assert.sameMembers(getIds(['d'], ['a', 'b']), ['c', 'd']);
		});

		it('- should load already loaded requirements if requested', function () {
			assert.sameMembers(getIds(['a', 'd'], ['a', 'b']), ['a', 'c', 'd']);
		});

		it('- should reload modified components', function () {
			assert.sameMembers(getIds(['e'], ['a', 'b', 'c', 'd']), ['a', 'c', 'd', 'e']);
		});

		it('- should work with empty load', function () {
			assert.sameMembers(getIds([], ['a', 'b', 'c', 'd']), []);
		});

		it('- should return unknown ids as is', function () {
			assert.sameMembers(getIds(['c', 'foo'], ['bar']), ['foo', 'c', 'a']);
		});

		it('- should throw on unknown dependencies', function () {
			assert.throws(() => {
				/** @type {import("../dependencies").Components} */
				const circular = {
					languages: {
						a: {
							require: 'c'
						},
						b: 'B'
					}
				};
				getLoader(circular, ['a']).getIds();
			});
		});

	});

	describe('Load order', function () {

		// Note: The order of a and b isn't defined, so don't add any test with both of them being loaded here

		it('- should load components in the correct order (require)', function () {
			assert.deepStrictEqual(getIds(['c']), ['a', 'c']);
		});

		it('- should load components in the correct order (modify)', function () {
			assert.deepStrictEqual(getIds(['e', 'a']), ['a', 'e']);
		});

		it('- should load components in the correct order (optional)', function () {
			assert.deepStrictEqual(getIds(['c', 'b'], ['a']), ['b', 'c']);
		});

		it('- should load components in the correct order (require + optional)', function () {
			assert.deepStrictEqual(getIds(['d'], ['a']), ['b', 'c', 'd']);
		});

		it('- should load components in the correct order (require + modify + optional)', function () {
			assert.deepStrictEqual(getIds(['d', 'e'], ['b']), ['a', 'e', 'c', 'd']);
		});

	});

	describe('Aliases', function () {

		it('- should resolve aliases in the list of components to load', function () {
			assert.sameMembers(getIds(['xyz']), ['a', 'b', 'c', 'd']);
		});

		it('- should resolve aliases in the list of loaded components', function () {
			assert.sameMembers(getIds(['d'], ['a', 'a2', 'b2']), ['c', 'd']);
		});

		it('- should throw on duplicate aliases', function () {
			assert.throws(() => {
				/** @type {import("../dependencies").Components} */
				const circular = {
					languages: {
						a: {
							alias: 'c'
						},
						b: {
							alias: 'c'
						}
					}
				};
				getLoader(circular, ['a', 'foo' /* force the lazy alias resolver */]).getIds();
			});
		});

		it('- should throw on aliases which are components', function () {
			assert.throws(() => {
				/** @type {import("../dependencies").Components} */
				const circular = {
					languages: {
						a: {
							alias: 'b'
						},
						b: 'B'
					}
				};
				getLoader(circular, ['a', 'foo' /* force the lazy alias resolver */]).getIds();
			});
		});

	});

	describe('Circular dependencies', function () {

		it('- should throw on circular dependencies', function () {
			assert.throws(() => {
				/** @type {import("../dependencies").Components} */
				const circular = {
					languages: {
						a: {
							require: 'b'
						},
						b: {
							optional: 'a'
						}
					}
				};
				getLoader(circular, ['a']).getIds();
			});
		});

	});

	describe('Async loading', function () {

		it('- should load components in the correct order', async function () {

			/** @type {import("../dependencies").Components} */
			const localComponents = {
				languages: {
					'a': {},
					'b': {
						require: 'a'
					},
					'c': {
						require: 'b'
					}
				}
			};

			/** @type {string[]} */
			const actualLoadOrder = [];
			/** @type {string[]} */
			const actualResolveOrder = [];

			/**
			 *
			 * @param {string} id
			 * @returns {Promise<void>}
			 */
			function loadComp(id) {
				actualLoadOrder.push(id);

				// the idea is that the components which have to be loaded first, take the longest, so if all were to
				// start getting loaded at the same time, their order would be the reverse of the expected order.
				let delay;
				if (id === 'a') {
					delay = 30;
				} else if (id === 'b') {
					delay = 20;
				} else if (id === 'c') {
					delay = 10;
				}

				return new Promise((resolve) => {
					setTimeout(() => {
						actualResolveOrder.push(id);
						resolve();
					}, delay);
				});
			}

			const loader = getLoader(localComponents, ['c']);

			await loader.load(id => loadComp(id), {
				series: async (before, after) => {
					await before;
					await after();
				},
				parallel: async (values) => {
					await Promise.all(values);
				}
			});

			assert.deepStrictEqual(actualLoadOrder, ['a', 'b', 'c'], 'actualLoadOrder:');
			assert.deepStrictEqual(actualResolveOrder, ['a', 'b', 'c'], 'actualResolveOrder:');
		});

	});

});

describe('components.json', function () {

	/**
	 * @param {T | T[] | undefined | null} value
	 * @returns {T[]}
	 * @template T
	 */
	function toArray(value) {
		if (Array.isArray(value)) {
			return value;
		} else if (value == undefined) {
			return [];
		} else {
			return [value];
		}
	}

	/**
	 * @param {(entry: import("../dependencies").ComponentEntry, id: string, entries: Object<string, import("../dependencies").ComponentEntry>) => void} consumeFn
	 */
	function forEachEntry(consumeFn) {
		/** @type {Object<string, import("../dependencies").ComponentEntry>} */
		const entries = {};

		for (const category in components) {
			for (const id in components[category]) {
				const entry = components[category][id];
				if (id !== 'meta' && entry && typeof entry === 'object') {
					entries[id] = entry;
				}
			}
		}

		for (const id in entries) {
			consumeFn(entries[id], id, entries);
		}
	}

	const entryProperties = [
		'title',
		'description',
		'alias',
		'aliasTitles',
		'owner',

		'require',
		'optional',
		'modify',

		'noCSS',
		'option'
	];

	it('- should be valid', function () {
		try {
			const allIds = [];
			for (const category in components) {
				Object.keys(components[category]).forEach(id => allIds.push(id));
			}
			// and an alias, so we force the lazy alias resolver to check all aliases
			allIds.push('js');

			getLoader(components, allIds).getIds();
		} catch (error) {
			assert.fail(error.toString());
		}
	});

	it('- should not have redundant optional dependencies', function () {
		forEachEntry((entry, id) => {
			const optional = new Set(toArray(entry.optional));

			for (const modifyId of toArray(entry.modify)) {
				if (optional.has(modifyId)) {
					assert.fail(`The component "${id}" has declared "${modifyId}" as both optional and modify.`);
				}
			}
			for (const requireId of toArray(entry.require)) {
				if (optional.has(requireId)) {
					assert.fail(`The component "${id}" has declared "${requireId}" as both optional and require.`);
				}
			}
		});
	});

	it('- should have a sorted language list', function () {
		const ignore = new Set(['meta', 'markup', 'css', 'clike', 'javascript']);
		/** @type {{ id: string, title: string }[]} */
		const languages = Object.keys(components.languages).filter(key => !ignore.has(key)).map(key => {
			return {
				id: key,
				title: components.languages[key].title
			};
		});

		/**
		 * Transforms the given title into an intermediate representation to allowed for sensible comparisons
		 * between titles.
		 *
		 * @param {string} title
		 */
		function transformTitle(title) {
			return title.replace(/\W+/g, '').replace(/^\d+/, '').toLowerCase();
		}

		const sorted = [...languages].sort((a, b) => {
			const comp = transformTitle(a.title).localeCompare(transformTitle(b.title));
			if (comp !== 0) {
				return comp;
			}
			// a and b have the same intermediate form (e.g. "C" => "C", "C++" => "C", "C#" => "C").
			return a.title.toLowerCase().localeCompare(b.title.toLowerCase());
		});

		assert.sameOrderedMembers(languages, sorted);
	});

	it('- should not have single-element or empty arrays', function () {
		/** @type {keyof import("../dependencies").ComponentEntry} */
		const properties = ['alias', 'optional', 'require', 'modify'];

		forEachEntry((entry, id) => {
			for (const prop of properties) {
				const value = entry[prop];
				if (Array.isArray(value)) {
					if (value.length === 0) {
						assert.fail(
							`The component "${id}" defines an empty array for "${prop}".` +
							` Please remove the "${prop}" property.`
						);
					} else if (value.length === 1) {
						assert.fail(
							`The component "${id}" defines a single-empty array for "${prop}".` +
							` Please replace the array with its element.` +
							`\n\t${JSON.stringify(prop)}: ${JSON.stringify(value[0])}`
						);
					}
				}
			}
		});
	});

	it('- should only have alias titles for valid aliases', function () {
		forEachEntry((entry, id) => {
			const title = entry.title;
			const alias = toArray(entry.alias);
			const aliasTitles = entry.aliasTitles;

			for (const key in aliasTitles) {
				if (alias.indexOf(key) === -1) {
					assert.fail(
						`Component "${id}":` +
						` The alias ${JSON.stringify(key)} in "aliasTitles" is not defined in "alias".`
					);
				}
				if (aliasTitles[key] === title) {
					assert.fail(
						`Component "${id}":` +
						` The alias title for ${JSON.stringify(key)} is the same as the normal title.` +
						` Remove the alias title or choose a different alias title.`
					);
				}
			}
		});
	});

	it('- should not have unknown properties', function () {
		const knownProperties = new Set(entryProperties);

		forEachEntry((entry, id) => {
			for (const prop in entry) {
				if (!knownProperties.has(prop)) {
					assert.fail(
						`Component "${id}":` +
						` The property ${JSON.stringify(prop)} is not supported by Prism.`
					);
				}
			}
		});
	});

});


text:
<?php
namespace MvUser\Form;

use MvBase\Form\AbstractForm;
use Zend\InputFilter\InputFilterProviderInterface;

class Role
  extends AbstractForm
  implements InputFilterProviderInterface
{

  public function init()
  {
    $this->add([
      'name' => 'name',
      'type' => 'Text',
      'required' => true,
      'options' => [
        'label' => 'Função'
      ]
    ]);
  }

  public function getInputFilterSpecification()
  {
    return array(
      'name' => [
        'validators' => [
          ['name' => 'NotEmpty', 'options' => [
            'messages' => array(
              'isEmpty' => 'Você tem que insterir um nome'
            )
          ]],
        ],
      ]
    );
  }
}


text:
#ifndef _AKFXPARAMETERCHANGEHANDLER_H_
#define _AKFXPARAMETERCHANGEHANDLER_H_

#include <AK/SoundEngine/Common/AkTypes.h>
#include <AK/Tools/Common/AkAssert.h>
#include <AK/Tools/Common/AkPlatformFuncs.h>

/// Can be used to track individual parameter changes to avoid costly computations when they remain constant
/// This class is designed to use only the lower bit information of the parameter IDs
namespace AK
{

	template <AkUInt32 T_MAXNUMPARAMS>
	class AkFXParameterChangeHandler
	{
	public:

		inline AkFXParameterChangeHandler()
		{
			ResetAllParamChanges( );
		}

		inline void SetParamChange( AkPluginParamID in_ID )
		{
			AKASSERT( in_ID <= T_MAXNUMPARAMS );
			const AkUInt32 uByteIndex = in_ID/8;
			const AkUInt32 uBitMask = 1<<(in_ID-uByteIndex*8);
			m_uParamBitArray[uByteIndex] |= uBitMask;
		}

		inline bool HasChanged( AkPluginParamID in_ID )
		{
			AKASSERT( in_ID <= T_MAXNUMPARAMS );
			const AkUInt32 uByteIndex = in_ID/8;
			const AkUInt32 uBitMask = 1<<(in_ID-uByteIndex*8);
			return (m_uParamBitArray[uByteIndex] & uBitMask) > 0;
		}

		inline bool HasAnyChanged()
		{
			AkUInt32 uByteIndex = 0;
			do
			{
				if ( m_uParamBitArray[uByteIndex] > 0 )
					return true;
				++uByteIndex;
			} while (uByteIndex < (((T_MAXNUMPARAMS) + ((8)-1)) & ~((8)-1))/8 );
			return false;
		}

		inline void ResetParamChange( AkPluginParamID in_ID )
		{
			AKASSERT( in_ID <= T_MAXNUMPARAMS );
			const AkUInt32 uByteIndex = in_ID/8;
			const AkUInt32 uBitMask = 1<<(in_ID-uByteIndex*8);
			m_uParamBitArray[uByteIndex] &= ~uBitMask;
		}

		inline void ResetAllParamChanges( )
		{
			AkZeroMemSmall( m_uParamBitArray, sizeof(m_uParamBitArray) );
		}

		inline void SetAllParamChanges( )
		{
			AKPLATFORM::AkMemSet( m_uParamBitArray, 0xFF, sizeof(m_uParamBitArray) );
		}

	protected:

		// Minimum storage in bits aligned to next byte boundary
		AkUInt8 m_uParamBitArray[(((T_MAXNUMPARAMS) + ((8)-1)) & ~((8)-1))/8];

	};

} // namespace AK

#endif // _AKFXPARAMETERCHANGEHANDLER_H_


text:
<?xml version="1.0"?>

<!DOCTYPE document PUBLIC "-//APACHE//DTD Documentation V2.0//EN"
          "http://forrest.apache.org/dtd/document-v20.dtd">

<document>

    <title>
      HDFS Architecture Guide
    </title>
    <authors>
      <person name="Dhruba Borthakur" email="dhruba@yahoo-inc.com"/>
    </authors>

      <title> Introduction </title>

      The Hadoop Distributed File System (<acronym title="Hadoop Distributed File System">HDFS</acronym>) is a distributed file system
      designed to run on commodity hardware. It has many similarities with existing distributed file systems. However, the differences from
      other distributed file systems are significant. HDFS is highly fault-tolerant and is designed to be deployed on low-cost hardware.
      HDFS provides high throughput access to application data and is suitable for applications that have large data sets. HDFS relaxes
      a few POSIX requirements to enable streaming access to file system data.  HDFS was originally built as infrastructure for the
      Apache Nutch web search engine project. HDFS is now an Apache Hadoop subproject.
      The project URL is http://hadoop.apache.org/hdfs/.

      <title> Assumptions and Goals </title>

        <title> Hardware Failure </title>

        Hardware failure is the norm rather than the exception. An HDFS instance may consist of hundreds or thousands of server machines,
        each storing part of the file system’s data. The fact that there are a huge number of components and that each component has
        a non-trivial probability of failure means that some component of HDFS is always non-functional. Therefore, detection of faults and quick,
        automatic recovery from them is a core architectural goal of HDFS.

        <title> Streaming Data Access </title>

        Applications that run on HDFS need streaming access to their data sets. They are not general purpose applications that typically run
        on general purpose file systems. HDFS is designed more for batch processing rather than interactive use by users. The emphasis is on
        high throughput of data access rather than low latency of data access. POSIX imposes many hard requirements that are not needed for
        applications that are targeted for HDFS. POSIX semantics in a few key areas has been traded to increase data throughput rates.

        <title> Large Data Sets </title>

        Applications that run on HDFS have large data sets. A typical file in HDFS is gigabytes to terabytes in size. Thus, HDFS is tuned to
        support large files. It should provide high aggregate data bandwidth and scale to thousands of nodes in a single cluster. It should support
        tens of millions of files in a single instance.

        <title> Appending-Writes and File Syncs </title>

        Most HDFS applications need a write-once-read-many access model for files. HDFS provides two additional advanced features: hflush and
        append.  Hflush makes the last block of an unclosed file visible to readers while providing read consistency and data durability.  Append
        provides a mechanism for opening a closed file to add additional data.

        For complete details of the hflush and append design, see the
        Append/Hflush/Read Design document (PDF).

        <title> “Moving Computation is Cheaper than Moving Data” </title>

        A computation requested by an application is much more efficient if it is executed near the data it operates on. This is especially true
        when the size of the data set is huge. This minimizes network congestion and increases the overall throughput of the system. The
        assumption is that it is often better to migrate the computation closer to where the data is located rather than moving the data to where
        the application is running. HDFS provides interfaces for applications to move themselves closer to where the data is located.

        <title> Portability Across Heterogeneous Hardware and Software Platforms </title>

        HDFS has been designed to be easily portable from one platform to another. This facilitates widespread adoption of HDFS as a
        platform of choice for a large set of applications.

      <title> NameNode and DataNodes </title>

      HDFS has a master/slave architecture. An HDFS cluster consists of a single NameNode, a master server that manages the file
      system namespace and regulates access to files by clients. In addition, there are a number of DataNodes, usually one per node
      in the cluster, which manage storage attached to the nodes that they run on. HDFS exposes a file system namespace and allows
      user data to be stored in files. Internally, a file is split into one or more blocks and these blocks are stored in a set of DataNodes.
      The NameNode executes file system namespace operations like opening, closing, and renaming files and directories. It also
      determines the mapping of blocks to DataNodes. The DataNodes are responsible for serving read and write requests from the file
      system’s clients. The DataNodes also perform block creation, deletion, and replication upon instruction from the NameNode.

      The NameNode and DataNode are pieces of software designed to run on commodity machines. These machines typically run a
      GNU/Linux operating system (<acronym title="operating system">OS</acronym>). HDFS is built using the Java language; any
      machine that supports Java can run the NameNode or the DataNode software. Usage of the highly portable Java language means
      that HDFS can be deployed on a wide range of machines. A typical deployment has a dedicated machine that runs only the
      NameNode software. Each of the other machines in the cluster runs one instance of the DataNode software. The architecture
      does not preclude running multiple DataNodes on the same machine but in a real deployment that is rarely the case.

      The existence of a single NameNode in a cluster greatly simplifies the architecture of the system. The NameNode is the arbitrator
      and repository for all HDFS metadata. The system is designed in such a way that user data never flows through the NameNode.

      <title> The File System Namespace </title>

      HDFS supports a traditional hierarchical file organization. A user or an application can create directories and store files inside
      these directories. The file system namespace hierarchy is similar to most other existing file systems; one can create and
      remove files, move a file from one directory to another, or rename a file. HDFS implements user quotas for number of names and
      amount of data stored in a particular directory (See
      HDFS Quota Admin Guide). In addition, HDFS
      supports symbolic links.

      The NameNode maintains the file system namespace. Any change to the file system namespace or its properties is
      recorded by the NameNode. An application can specify the number of replicas of a file that should be maintained by
      HDFS. The number of copies of a file is called the replication factor of that file. This information is stored by the NameNode.

      <title> Data Replication </title>

      HDFS is designed to reliably store very large files across machines in a large cluster. It stores each file as a sequence
      of blocks; all blocks in a file except the last block are the same size. The blocks of a file are replicated for fault tolerance.
      The block size and replication factor are configurable per file. An application can specify the number of replicas of a file.
      The replication factor can be specified at file creation time and can be changed later. Files in HDFS are strictly one writer at any
      time.

      The NameNode makes all decisions regarding replication of blocks. It periodically receives a Heartbeat and a Blockreport
      from each of the DataNodes in the cluster. Receipt of a Heartbeat implies that the DataNode is functioning properly. A
      Blockreport contains a list of all blocks on a DataNode.

        <title> Replica Placement: The First Baby Steps </title>

        The placement of replicas is critical to HDFS reliability and performance. Optimizing replica placement distinguishes
        HDFS from most other distributed file systems. This is a feature that needs lots of tuning and experience. The purpose
        of a rack-aware replica placement policy is to improve data reliability, availability, and network bandwidth utilization.
        The current implementation for the replica placement policy is a first effort in this direction. The short-term goals of
        implementing this policy are to validate it on production systems, learn more about its behavior, and build a foundation
        to test and research more sophisticated policies.

        Large HDFS instances run on a cluster of computers that commonly spread across many racks. Communication
        between two nodes in different racks has to go through switches. In most cases, network bandwidth between machines
        in the same rack is greater than network bandwidth between machines in different racks.

        The NameNode determines the rack id each DataNode belongs to via the process outlined in
        Hadoop Rack Awareness.
        A simple but non-optimal policy is to place replicas on unique racks. This prevents losing data when an entire rack
        fails and allows use of bandwidth from multiple racks when reading data. This policy evenly distributes replicas in
        the cluster which makes it easy to balance load on component failure. However, this policy increases the cost of
        writes because a write needs to transfer blocks to multiple racks.

        For the common case, when the replication factor is three, HDFS’s placement policy is to put one replica
        on one node in the local rack, another on a node in a different (remote) rack, and the last on a different node in the
        same remote rack. This policy cuts the inter-rack write traffic which generally improves write performance. The
        chance of rack failure is far less than that of node failure; this policy does not impact data reliability and availability
        guarantees. However, it does reduce the aggregate network bandwidth used when reading data since a block is
        placed in only two unique racks rather than three. With this policy, the replicas of a file do not evenly distribute
        across the racks. One third of replicas are on one node, two thirds of replicas are on one rack, and the other third
        are evenly distributed across the remaining racks. This policy improves write performance without compromising
        data reliability or read performance.

        In addition to the default placement policy described above, HDFS also provides a pluggable interface for block placement. See
        BlockPlacementPolicy.

        <title> Replica Selection </title>

        To minimize global bandwidth consumption and read latency, HDFS tries to satisfy a read request from a replica
        that is closest to the reader. If there exists a replica on the same rack as the reader node, then that replica is
        preferred to satisfy the read request. If an HDFS cluster spans multiple data centers, then a replica that is
        resident in the local data center is preferred over any remote replica.

        <title> Safemode </title>

        On startup, the NameNode enters a special state called Safemode. Replication of data blocks does not occur
        when the NameNode is in the Safemode state. The NameNode receives Heartbeat and Blockreport messages
        from the DataNodes. A Blockreport contains the list of data blocks that a DataNode is hosting. Each block
        has a specified minimum number of replicas. A block is considered safely replicated when the minimum number
        of replicas of that data block has checked in with the NameNode. After a configurable percentage of safely
        replicated data blocks checks in with the NameNode (plus an additional 30 seconds), the NameNode exits
        the Safemode state. It then determines the list of data blocks (if any) that still have fewer than the specified
        number of replicas. The NameNode then replicates these blocks to other DataNodes.

      <title> The Persistence of File System Metadata </title>

        The HDFS namespace is stored by the NameNode. The NameNode uses a transaction log called the EditLog
        to persistently record every change that occurs to file system metadata. For example, creating a new file in
        HDFS causes the NameNode to insert a record into the EditLog indicating this. Similarly, changing the
        replication factor of a file causes a new record to be inserted into the EditLog. The NameNode uses a file
        in its local host OS file system to store the EditLog. The entire file system namespace, including the mapping
        of blocks to files and file system properties, is stored in a file called the FsImage. The FsImage is stored as
        a file in the NameNode’s local file system too.

        The NameNode keeps an image of the entire file system namespace and file Blockmap in memory. This key
        metadata item is designed to be compact, such that a NameNode with 4 GB of RAM is plenty to support a
        huge number of files and directories. When the NameNode starts up, it reads the FsImage and EditLog from
        disk, applies all the transactions from the EditLog to the in-memory representation of the FsImage, and flushes
        out this new version into a new FsImage on disk. It can then truncate the old EditLog because its transactions
        have been applied to the persistent FsImage. This process is called a checkpoint. The
        Checkpoint Node is a
        separate daemon that can be configured to periodically build checkpoints from the FsImage and EditLog which are
        uploaded to the NameNode.  The
        Backup Node builds
        checkpoints like the Checkpoint Node and also maintains an up-to-date copy of the FsImage in memory.

        The DataNode stores HDFS data in files in its local file system. The DataNode has no knowledge about HDFS files.
        It stores each block of HDFS data in a separate file in its local file system. The DataNode does not create all files
        in the same directory. Instead, it uses a heuristic to determine the optimal number of files per directory and creates
        subdirectories appropriately. It is not optimal to create all local files in the same directory because the local file
        system might not be able to efficiently support a huge number of files in a single directory. When a DataNode starts
        up, it scans through its local file system, generates a list of all HDFS data blocks that correspond to each of these
        local files and sends this report to the NameNode: this is the Blockreport.

      <title> The Communication Protocols </title>

      All HDFS communication protocols are layered on top of the TCP/IP protocol. A client establishes a connection to
      a configurable <acronym title="Transmission Control Protocol">TCP</acronym> port on the NameNode machine.
      It talks the ClientProtocol with the NameNode. The DataNodes talk to the NameNode using the DataNode Protocol.
      A Remote Procedure Call (<acronym title="Remote Procedure Call">RPC</acronym>) abstraction wraps both the
      Client Protocol and the DataNode Protocol. By design, the NameNode never initiates any RPCs. Instead, it only
      responds to RPC requests issued by DataNodes or clients.

      <title> Robustness </title>

      The primary objective of HDFS is to store data reliably even in the presence of failures. The three common types
      of failures are NameNode failures, DataNode failures and network partitions.

        <title> Data Disk Failure, Heartbeats and Re-Replication </title>

        Each DataNode sends a Heartbeat message to the NameNode periodically. A network partition can cause a
        subset of DataNodes to lose connectivity with the NameNode. The NameNode detects this condition by the
        absence of a Heartbeat message. The NameNode marks DataNodes without recent Heartbeats as dead and
        does not forward any new <acronym title="Input/Output">IO</acronym> requests to them. Any data that was
        registered to a dead DataNode is not available to HDFS any more. DataNode death may cause the replication
        factor of some blocks to fall below their specified value. The NameNode constantly tracks which blocks need
        to be replicated and initiates replication whenever necessary. The necessity for re-replication may arise due
        to many reasons: a DataNode may become unavailable, a replica may become corrupted, a hard disk on a
        DataNode may fail, or the replication factor of a file may be increased.

        <title> Cluster Rebalancing </title>

        The HDFS architecture is compatible with data rebalancing schemes. A scheme might automatically move
        data from one DataNode to another if the free space on a DataNode falls below a certain threshold. In the
        event of a sudden high demand for a particular file, a scheme might dynamically create additional replicas
        and rebalance other data in the cluster. These types of data rebalancing schemes are not yet implemented.

        <title> Data Integrity </title>

        It is possible that a block of data fetched from a DataNode arrives corrupted. This corruption can occur
        because of faults in a storage device, network faults, or buggy software. The HDFS client software
        implements checksum checking on the contents of HDFS files. When a client creates an HDFS file,
        it computes a checksum of each block of the file and stores these checksums in a separate hidden
        file in the same HDFS namespace. When a client retrieves file contents it verifies that the data it
        received from each DataNode matches the checksum stored in the associated checksum file. If not,
        then the client can opt to retrieve that block from another DataNode that has a replica of that block.

        <title> Metadata Disk Failure </title>

        The FsImage and the EditLog are central data structures of HDFS. A corruption of these files can
        cause the HDFS instance to be non-functional. For this reason, the NameNode can be configured
        to support maintaining multiple copies of the FsImage and EditLog. Any update to either the FsImage
        or EditLog causes each of the FsImages and EditLogs to get updated synchronously. This
        synchronous updating of multiple copies of the FsImage and EditLog may degrade the rate of
        namespace transactions per second that a NameNode can support. However, this degradation is
        acceptable because even though HDFS applications are very data intensive in nature, they are not
        metadata intensive. When a NameNode restarts, it selects the latest consistent FsImage and EditLog to use.

        The NameNode machine is a single point of failure for an HDFS cluster. If the NameNode machine fails,
        manual intervention is necessary. Currently, automatic restart and failover of the NameNode software to
        another machine is not supported.

        <title> Snapshots </title>

        Snapshots support storing a copy of data at a particular instant of time. One usage of the snapshot
        feature may be to roll back a corrupted HDFS instance to a previously known good point in time.
        HDFS does not currently support snapshots but will in a future release.

      <title> Data Organization </title>

        <title> Data Blocks </title>

        HDFS is designed to support very large files. Applications that are compatible with HDFS are those
        that deal with large data sets. These applications write their data only once but they read it one or
        more times and require these reads to be satisfied at streaming speeds. HDFS supports
        write-once-read-many semantics on files. A typical block size used by HDFS is 64 MB. Thus,
        an HDFS file is chopped up into 64 MB chunks, and if possible, each chunk will reside on a different DataNode.

        <title> Replication Pipelining </title>

        When a client is writing data to an HDFS file with a replication factor of 3, the NameNode retrieves a list of DataNodes using a replication target choosing algorithm.
        This list contains the DataNodes that will host a replica of that block. The client then writes to the first DataNode. The first DataNode starts receiving the data in small portions (4 KB),
        writes each portion to its local repository and transfers that portion to the second DataNode in the list.
        The second DataNode, in turn starts receiving each portion of the data block, writes that portion to its
        repository and then flushes that portion to the third DataNode. Finally, the third DataNode writes the
        data to its local repository. Thus, a DataNode can be receiving data from the previous one in the pipeline
        and at the same time forwarding data to the next one in the pipeline. Thus, the data is pipelined from
        one DataNode to the next.

      <title> Accessibility </title>

      HDFS can be accessed from applications in many different ways. Natively, HDFS provides a
      Java API for applications to
      use. A C language wrapper for this Java API is also available. In addition, an HTTP browser
      can also be used to browse the files of an HDFS instance. Work is in progress to expose
      HDFS through the <acronym title="Web-based Distributed Authoring and Versioning">WebDAV</acronym> protocol.

        <title> FS Shell </title>

        HDFS allows user data to be organized in the form of files and directories. It provides a commandline
        interface called  FS shell that lets a user interact with the data in HDFS. The syntax of this command
        set is similar to other shells (e.g. bash, csh) that users are already familiar with. Here are some sample
        action/command pairs:

             Action 	 Command

             Create a directory named /foodir
             bin/hadoop dfs -mkdir /foodir

             Remove a directory named /foodir
             bin/hadoop dfs -rmr /foodir

             View the contents of a file named /foodir/myfile.txt
             bin/hadoop dfs -cat /foodir/myfile.txt

        FS shell is targeted for applications that need a scripting language to interact with the stored data.

        <title> DFSAdmin </title>

        The DFSAdmin command set is used for administering an HDFS cluster. These are commands that are
        used only by an HDFS administrator. Here are some sample action/command pairs:

             Action 	 Command

             Put the cluster in Safemode 	  bin/hadoop dfsadmin -safemode enter

             Generate a list of DataNodes 	  bin/hadoop dfsadmin -report

             Recommission or decommission DataNode(s)
             bin/hadoop dfsadmin -refreshNodes

        <title> Browser Interface </title>

        A typical HDFS install configures a web server to expose the HDFS namespace through
        a configurable TCP port. This allows a user to navigate the HDFS namespace and view
        the contents of its files using a web browser.

      <title> Space Reclamation </title>

        <title> File Deletes and Undeletes </title>

        When a file is deleted by a user or an application, it is not immediately removed from HDFS.  Instead,
        HDFS first renames it to a file in the /trash directory. The file can be restored quickly
        as long as it remains in /trash. A file remains in /trash for a configurable
        amount of time. After the expiry of its life in /trash, the NameNode deletes the file from
        the HDFS namespace. The deletion of a file causes the blocks associated with the file to be freed.
        Note that there could be an appreciable time delay between the time a file is deleted by a user and
        the time of the corresponding increase in free space in HDFS.

        A user can Undelete a file after deleting it as long as it remains in the /trash directory.
        If a user wants to undelete a file that he/she has deleted, he/she can navigate the /trash
        directory and retrieve the file. The /trash directory contains only the latest copy of the file
        that was deleted. The /trash directory is just like any other directory with one special
        feature: HDFS applies specified policies to automatically delete files from this directory. The current
        default policy is to delete files from /trash that are more than 6 hours old. In the future,
        this policy will be configurable through a well defined interface.

        <title> Decrease Replication Factor </title>

        When the replication factor of a file is reduced, the NameNode selects excess replicas that can be deleted.
        The next Heartbeat transfers this information to the DataNode. The DataNode then removes the corresponding
        blocks and the corresponding free space appears in the cluster. Once again, there might be a time delay
        between the completion of the setReplication API call and the appearance of free space in the cluster.

      <title> References </title>

      HDFS Java API:

        http://hadoop.apache.org/core/docs/current/api/

      HDFS source code:

        http://hadoop.apache.org/hdfs/version_control.html

</document>