<commit_before>package org.rapidoid.inmem;

/*
 * #%L
 * rapidoid-inmem
 * %%
 * Copyright (C) 2014 <NAME>
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.rapidoid.json.JSON;
import org.rapidoid.lambda.Operation;
import org.rapidoid.lambda.Predicate;
import org.rapidoid.util.U;

class Rec {
	final Class<?> type;
	final String json;

	public Rec(Class<?> type, String json) {
		this.type = type;
		this.json = json;
	}
}

public class InMem {

	private static final String SUFFIX_B = "b";

	private static final String SUFFIX_A = "a";

	private static final byte[] CR_LF = { 13, 10 };

	private static final Object INSERTION = new Object();

	private final String filename;

	private final AtomicLong ids = new AtomicLong();

	private final AtomicBoolean active = new AtomicBoolean(true);

	private final AtomicBoolean aOrB = new AtomicBoolean(true);

	private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

	private final Thread persistor;

	private final AtomicBoolean insideTx = new AtomicBoolean(false);

	private final ConcurrentHashMap<Long, Rec> txChanges = new ConcurrentHashMap<Long, Rec>();

	private final ConcurrentHashMap<Long, Object> txInsertions = new ConcurrentHashMap<Long, Object>();

	private ConcurrentHashMap<Long, Rec> prevData = new ConcurrentHashMap<Long, Rec>();

	private ConcurrentHashMap<Long, Rec> data = new ConcurrentHashMap<Long, Rec>();

	public InMem() {
		this(null);
	}

	public InMem(String filename) {
		this.filename = filename;

		if (filename != null && !filename.isEmpty()) {
			persistor = new Thread(new Runnable() {
				@Override
				public void run() {
					persist();
				}
			});
			persistor.start();
		} else {
			persistor = null;
		}
	}

	public long insert(Object record) {
		sharedLock();
		try {
			long id = ids.incrementAndGet();
			setId(record, id);

			if (txInsertions.putIfAbsent(id, INSERTION) != null) {
				throw new IllegalStateException("Cannot insert changelog record with existing ID: " + id);
			}

			if (data.putIfAbsent(id, rec(record, id)) != null) {
				throw new IllegalStateException("Cannot insert record with existing ID: " + id);
			}

			return id;
		} finally {
			sharedUnlock();
		}
	}

	public void delete(long id) {
		sharedLock();
		try {
			validateId(id);

			Rec removed = data.remove(id);
			txChanges.putIfAbsent(id, removed);

		} finally {
			sharedUnlock();
		}
	}

	@SuppressWarnings("unchecked")
	public <E> E get(long id) {
		sharedLock();
		try {
			validateId(id);
			Rec rec = data.get(id);
			return (E) (rec != null ? setId(obj(rec), id) : null);
		} finally {
			sharedUnlock();
		}
	}

	public <E> E get(long id, Class<E> clazz) {
		sharedLock();
		try {
			validateId(id);
			return get_(id, clazz);
		} finally {
			sharedUnlock();
		}
	}

	public void update(long id, Object record) {
		sharedLock();
		try {
			validateId(id);

			Rec removed = data.put(id, rec(record, id));
			txChanges.putIfAbsent(id, removed);

			if (removed == null) {
				throw new IllegalStateException("Cannot update non-existing record with ID=" + id);
			}

			setId(record, id);
		} finally {
			sharedUnlock();
		}
	}
<commit_msg>Improved InMem changelog operations (executing only inside transactions).<commit_after>package org.rapidoid.inmem;

/*
 * #%L
 * rapidoid-inmem
 * %%
 * Copyright (C) 2014 <NAME>
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.rapidoid.json.JSON;
import org.rapidoid.lambda.Operation;
import org.rapidoid.lambda.Predicate;
import org.rapidoid.util.U;

class Rec {
	final Class<?> type;
	final String json;

	public Rec(Class<?> type, String json) {
		this.type = type;
		this.json = json;
	}
}

public class InMem {

	private static final String SUFFIX_B = "b";

	private static final String SUFFIX_A = "a";

	private static final byte[] CR_LF = { 13, 10 };

	private static final Object INSERTION = new Object();

	private final String filename;

	private final AtomicLong ids = new AtomicLong();

	private final AtomicBoolean active = new AtomicBoolean(true);

	private final AtomicBoolean aOrB = new AtomicBoolean(true);

	private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

	private final Thread persistor;

	private final AtomicBoolean insideTx = new AtomicBoolean(false);

	private final ConcurrentHashMap<Long, Rec> txChanges = new ConcurrentHashMap<Long, Rec>();

	private final ConcurrentHashMap<Long, Object> txInsertions = new ConcurrentHashMap<Long, Object>();

	private ConcurrentHashMap<Long, Rec> prevData = new ConcurrentHashMap<Long, Rec>();

	private ConcurrentHashMap<Long, Rec> data = new ConcurrentHashMap<Long, Rec>();

	public InMem() {
		this(null);
	}

	public InMem(String filename) {
		this.filename = filename;

		if (filename != null && !filename.isEmpty()) {
			persistor = new Thread(new Runnable() {
				@Override
				public void run() {
					persist();
				}
			});
			persistor.start();
		} else {
			persistor = null;
		}
	}

	public long insert(Object record) {
		sharedLock();
		try {
			long id = ids.incrementAndGet();
			setId(record, id);

			if (insideTx.get()) {
				if (txInsertions.putIfAbsent(id, INSERTION) != null) {
					throw new IllegalStateException("Cannot insert changelog record with existing ID: " + id);
				}
			}

			if (data.putIfAbsent(id, rec(record, id)) != null) {
				throw new IllegalStateException("Cannot insert record with existing ID: " + id);
			}

			return id;
		} finally {
			sharedUnlock();
		}
	}

	public void delete(long id) {
		sharedLock();
		try {
			validateId(id);

			Rec removed = data.remove(id);

			if (insideTx.get()) {
				txChanges.putIfAbsent(id, removed);
			}

		} finally {
			sharedUnlock();
		}
	}

	@SuppressWarnings("unchecked")
	public <E> E get(long id) {
		sharedLock();
		try {
			validateId(id);
			Rec rec = data.get(id);
			return (E) (rec != null ? setId(obj(rec), id) : null);
		} finally {
			sharedUnlock();
		}
	}

	public <E> E get(long id, Class<E> clazz) {
		sharedLock();
		try {
			validateId(id);
			return get_(id, clazz);
		} finally {
			sharedUnlock();
		}
	}

	public void update(long id, Object record) {
		sharedLock();
		try {
			validateId(id);

			Rec removed = data.replace(id, rec(record, id));

			if (insideTx.get()) {
				txChanges.putIfAbsent(id, removed);
			}

			if (removed == null) {
				throw new IllegalStateException("Cannot update non-existing record with ID=" + id);
			}

			setId(record, id);
		} finally {
			sharedUnlock();
		}
	}

{'fraction_non_alphanumeric': 0.11823577330385197, 'fraction_numerical': 0.002792971022925637, 'mean_word_length': 3.8912919749573134, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1690633', 'n_tokens_mistral': 3015, 'n_tokens_neox': 2673, 'n_words': 795}