<commit_before>package com.emc.mongoose.storage.driver.base;

import com.emc.mongoose.api.common.collection.OptLockArrayBuffer;
import com.emc.mongoose.api.common.collection.OptLockBuffer;
import com.emc.mongoose.api.common.concurrent.Coroutine;
import com.emc.mongoose.api.common.concurrent.CoroutineBase;
import com.emc.mongoose.api.common.concurrent.StoppableTaskBase;
import com.emc.mongoose.api.common.exception.UserShootHisFootException;
import com.emc.mongoose.api.model.DaemonBase;
import static com.emc.mongoose.api.common.Constants.KEY_CLASS_NAME;
import static com.emc.mongoose.api.common.Constants.KEY_TEST_STEP_ID;
import com.emc.mongoose.api.common.io.Input;
import com.emc.mongoose.api.model.data.DataInput;
import com.emc.mongoose.api.model.io.task.composite.CompositeIoTask;
import com.emc.mongoose.api.model.io.task.IoTask;
import com.emc.mongoose.api.model.io.task.data.DataIoTask;
import com.emc.mongoose.api.model.io.task.partial.PartialIoTask;
import com.emc.mongoose.api.model.item.Item;
import com.emc.mongoose.api.model.storage.Credential;
import com.emc.mongoose.api.model.storage.StorageDriver;
import com.emc.mongoose.ui.config.load.LoadConfig;
import com.emc.mongoose.ui.config.storage.StorageConfig;
import com.emc.mongoose.ui.config.storage.auth.AuthConfig;
import com.emc.mongoose.ui.config.storage.driver.queue.QueueConfig;
import com.emc.mongoose.ui.log.LogUtil;
import com.emc.mongoose.ui.log.Loggers;

import static org.apache.logging.log4j.CloseableThreadContext.Instance;
import org.apache.logging.log4j.CloseableThreadContext;
import org.apache.logging.log4j.Level;

import java.io.EOFException;
import java.io.IOException;
import java.rmi.ServerException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;

/**
 Created by kurila on 11.07.16.
 */
public abstract class StorageDriverBase<I extends Item, O extends IoTask<I>>
extends DaemonBase
implements StorageDriver<I, O> {

	private final static String CLS_NAME = StorageDriverBase.class.getSimpleName();

	private final DataInput contentSrc;
	private final int batchSize;
	private final int outputQueueCapacity;
	protected final BlockingQueue<O> childTasksQueue;
	private final BlockingQueue<O> inTasksQueue;
	private final BlockingQueue<O> ioResultsQueue;
	protected final String stepId;
	protected final int concurrencyLevel;
	protected final Semaphore concurrencyThrottle;
	protected final Credential credential;
	protected final boolean verifyFlag;
	private final LongAdder scheduledTaskCount = new LongAdder();
	private final LongAdder completedTaskCount = new LongAdder();
	
	private final ConcurrentMap<String, String> pathMap = new ConcurrentHashMap<>(1);
	protected abstract String requestNewPath(final String path);
	protected Function<String, String> requestNewPathFunc = this::requestNewPath;
	
	protected final ConcurrentMap<Credential, String> authTokens = new ConcurrentHashMap<>(1);
	protected abstract String requestNewAuthToken(final Credential credential);
	protected Function<Credential, String> requestAuthTokenFunc = this::requestNewAuthToken;
	private final IoTasksDispatchCoroutine ioTasksDispatchCoroutine;
	
	protected StorageDriverBase(
		final String stepId, final DataInput contentSrc, final LoadConfig loadConfig,
		final StorageConfig storageConfig, final boolean verifyFlag
	) throws UserShootHisFootException {
		this.contentSrc = contentSrc;
		this.batchSize = loadConfig.getBatchConfig().getSize();
		final QueueConfig queueConfig = storageConfig.getDriverConfig().getQueueConfig();
		this.outputQueueCapacity = queueConfig.getOutput();
		this.childTasksQueue = new ArrayBlockingQueue<>(queueConfig.getInput());
		this.inTasksQueue = new ArrayBlockingQueue<>(queueConfig.getInput());
		this.ioResultsQueue = new ArrayBlockingQueue<>(outputQueueCapacity);
		this.stepId = stepId;
		final AuthConfig authConfig = storageConfig.getAuthConfig();
		this.credential = Credential.getInstance(authConfig.getUid(), authConfig.getSecret());
		final String authToken = authConfig.getToken();
		if(authToken != null) {
			if(this.credential == null) {
				this.authTokens.put(Credential.NONE, authToken);
			} else {
				this.authTokens.put(credential, authToken);
			}
		}
		this.concurrencyLevel = storageConfig.getDriverConfig().getConcurrency();
		this.concurrencyThrottle = new Semaphore(concurrencyLevel, true);
		this.verifyFlag = verifyFlag;
		this.ioTasksDispatchCoroutine = new IoTasksDispatchCoroutine(svcCoroutines);
		svcCoroutines.add(ioTasksDispatchCoroutine);
	}

	private final class IoTasksDispatchCoroutine
	extends CoroutineBase {

		private final OptLockBuffer<O> buff = new OptLockArrayBuffer<>(batchSize);
		private int n = 0;
		private int m;
		private long t;

		public IoTasksDispatchCoroutine(final List<Coroutine> svcCoroutines) {
			super(svcCoroutines);
		}

		@Override
		protected final void invoke() {
			t = System.currentTimeMillis();
			if(buff.tryLock()) {
				try(<commit_msg>+ remote debugging jvm args for the tests<commit_after>package com.emc.mongoose.storage.driver.base;

import com.emc.mongoose.api.common.collection.OptLockArrayBuffer;
import com.emc.mongoose.api.common.collection.OptLockBuffer;
import com.emc.mongoose.api.common.concurrent.Coroutine;
import com.emc.mongoose.api.common.concurrent.CoroutineBase;
import com.emc.mongoose.api.common.concurrent.StoppableTaskBase;
import com.emc.mongoose.api.common.exception.UserShootHisFootException;
import com.emc.mongoose.api.model.DaemonBase;
import static com.emc.mongoose.api.common.Constants.KEY_CLASS_NAME;
import static com.emc.mongoose.api.common.Constants.KEY_TEST_STEP_ID;
import com.emc.mongoose.api.common.io.Input;
import com.emc.mongoose.api.model.data.DataInput;
import com.emc.mongoose.api.model.io.task.composite.CompositeIoTask;
import com.emc.mongoose.api.model.io.task.IoTask;
import com.emc.mongoose.api.model.io.task.data.DataIoTask;
import com.emc.mongoose.api.model.io.task.partial.PartialIoTask;
import com.emc.mongoose.api.model.item.Item;
import com.emc.mongoose.api.model.storage.Credential;
import com.emc.mongoose.api.model.storage.StorageDriver;
import com.emc.mongoose.ui.config.load.LoadConfig;
import com.emc.mongoose.ui.config.storage.StorageConfig;
import com.emc.mongoose.ui.config.storage.auth.AuthConfig;
import com.emc.mongoose.ui.config.storage.driver.queue.QueueConfig;
import com.emc.mongoose.ui.log.LogUtil;
import com.emc.mongoose.ui.log.Loggers;

import static org.apache.logging.log4j.CloseableThreadContext.Instance;
import org.apache.logging.log4j.CloseableThreadContext;
import org.apache.logging.log4j.Level;

import java.io.EOFException;
import java.io.IOException;
import java.rmi.ServerException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Function;

/**
 Created by kurila on 11.07.16.
 */
public abstract class StorageDriverBase<I extends Item, O extends IoTask<I>>
extends DaemonBase
implements StorageDriver<I, O> {

	private final static String CLS_NAME = StorageDriverBase.class.getSimpleName();

	private final DataInput contentSrc;
	private final int batchSize;
	private final int outputQueueCapacity;
	protected final BlockingQueue<O> childTasksQueue;
	private final BlockingQueue<O> inTasksQueue;
	private final BlockingQueue<O> ioResultsQueue;
	protected final String stepId;
	protected final int concurrencyLevel;
	protected final Semaphore concurrencyThrottle;
	protected final Credential credential;
	protected final boolean verifyFlag;
	private final LongAdder scheduledTaskCount = new LongAdder();
	private final LongAdder completedTaskCount = new LongAdder();
	
	private final ConcurrentMap<String, String> pathMap = new ConcurrentHashMap<>(1);
	protected abstract String requestNewPath(final String path);
	protected Function<String, String> requestNewPathFunc = this::requestNewPath;
	
	protected final ConcurrentMap<Credential, String> authTokens = new ConcurrentHashMap<>(1);
	protected abstract String requestNewAuthToken(final Credential credential);
	protected Function<Credential, String> requestAuthTokenFunc = this::requestNewAuthToken;
	private final IoTasksDispatchCoroutine ioTasksDispatchCoroutine;
	
	protected StorageDriverBase(
		final String stepId, final DataInput contentSrc, final LoadConfig loadConfig,
		final StorageConfig storageConfig, final boolean verifyFlag
	) throws UserShootHisFootException {
		this.contentSrc = contentSrc;
		this.batchSize = loadConfig.getBatchConfig().getSize();
		final QueueConfig queueConfig = storageConfig.getDriverConfig().getQueueConfig();
		this.outputQueueCapacity = queueConfig.getOutput();
		this.childTasksQueue = new ArrayBlockingQueue<>(queueConfig.getInput());
		this.inTasksQueue = new ArrayBlockingQueue<>(queueConfig.getInput());
		this.ioResultsQueue = new ArrayBlockingQueue<>(outputQueueCapacity);
		this.stepId = stepId;
		final AuthConfig authConfig = storageConfig.getAuthConfig();
		this.credential = Credential.getInstance(authConfig.getUid(), authConfig.getSecret());
		final String authToken = authConfig.getToken();
		if(authToken != null) {
			if(this.credential == null) {
				this.authTokens.put(Credential.NONE, authToken);
			} else {
				this.authTokens.put(credential, authToken);
			}
		}
		this.concurrencyLevel = storageConfig.getDriverConfig().getConcurrency();
		this.concurrencyThrottle = new Semaphore(concurrencyLevel, true);
		this.verifyFlag = verifyFlag;
		this.ioTasksDispatchCoroutine = new IoTasksDispatchCoroutine(svcCoroutines);
		svcCoroutines.add(ioTasksDispatchCoroutine);
	}

	private final class IoTasksDispatchCoroutine
	extends CoroutineBase {

		private final OptLockBuffer<O> buff = new OptLockArrayBuffer<>(batchSize);
		private int n = 0;
		private int m;

		public IoTasksDispatchCoroutine(final List<Coroutine> svcCoroutines) {
			super(svcCoroutines);
		}

		@Override
		protected final void invoke() {
			if(buff.tryLock()) {
				long t = System.currentTimeMillis();
				try(
{'fraction_non_alphanumeric': 0.0975517890772128, 'fraction_numerical': 0.0022598870056497176, 'mean_word_length': 6.1812035158891145, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1224114', 'n_tokens_mistral': 3323, 'n_tokens_neox': 3284, 'n_words': 636}