<commit_before>/*
 * Copyright (C) 2015 SoftIndex LLC.
 *
 * 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.
 */

package io.datakernel.bytebuf;

import io.datakernel.util.ConcurrentStack;

import java.util.ArrayList;
import java.util.List;

import static java.lang.Integer.numberOfLeadingZeros;

public final class ByteBufPool {
	private static final int NUMBER_SLABS = 33;

	private static int minSize = 32;
	private static int maxSize = 1 << 30;

	private static final ConcurrentStack<ByteBuf>[] slabs = createSlabs(NUMBER_SLABS);
	private static final int[] created = new int[NUMBER_SLABS];

	private ByteBufPool() {}

	public static ByteBuf allocate(int size) {
		if (size < minSize || size >= maxSize) {
			// not willing to register in pool
			return ByteBuf.wrapForWriting(new byte[size]);
		}
		int index = 32 - numberOfLeadingZeros(size - 1); // index==32 for size==0
		ConcurrentStack<ByteBuf> queue = slabs[index];
		ByteBuf buf = queue.pop();
		if (buf != null) {
			buf.reset();
		} else {
			buf = ByteBuf.wrapForWriting(new byte[1 << index]);
			buf.refs++;
			created[index]++;
		}
		return buf;
	}

	public static void recycle(ByteBuf buf) {
		assert buf.array.length >= minSize && buf.array.length <= maxSize;
		ConcurrentStack<ByteBuf> queue = slabs[32 - numberOfLeadingZeros(buf.array.length - 1)];
		assert !queue.contains(buf) : "duplicate recycle array";
		queue.push(buf);
	}

	public static ByteBuf recycleIfEmpty(ByteBuf buf) {
		if (buf.canRead())
			return buf;
		buf.recycle();
		return ByteBuf.empty();
	}

	public static ConcurrentStack<ByteBuf>[] getPool() {
		return slabs;
	}

	public static void clear() {
		for (int i = 0; i < ByteBufPool.NUMBER_SLABS; i++) {
			slabs[i].clear();
			created[i] = 0;
		}
	}

	private static ConcurrentStack<ByteBuf>[] createSlabs(int numberOfSlabs) {
		//noinspection unchecked
		ConcurrentStack<ByteBuf>[] slabs = new ConcurrentStack[numberOfSlabs];
		for (int i = 0; i < slabs.length; i++) {
			slabs[i] = new ConcurrentStack<>();
		}
		return slabs;
	}

	public static ByteBuf ensureTailRemaining(ByteBuf buf, int newTailRemaining) {
		assert !(buf instanceof ByteBuf.ByteBufSlice);
		if (buf.writeRemaining() >= newTailRemaining) {
			return buf;
		} else {
			ByteBuf newBuf = allocate(newTailRemaining + buf.readRemaining());
			newBuf.put(buf);
			buf.recycle();
			return newBuf;
		}
	}

	public static ByteBuf append(ByteBuf to, ByteBuf from) {
		assert !to.isRecycled() && !from.isRecycled();
		if (to.readRemaining() == 0) {
			to.recycle();
			return from;
		}
		to = ensureTailRemaining(to, from.readRemaining());
		to.put(from);
		from.recycle();
		return to;
	}

	public static ByteBuf append(ByteBuf to, byte[] from, int offset, int length) {
		assert !to.isRecycled();
		to = ensureTailRemaining(to, length);
		to.put(from, offset, length);
		return to;
	}

	public static ByteBuf append(ByteBuf to, byte[] from) {
		return append(to, from, 0, from.length);
	}

	private static final ByteBufPoolStats stats = new ByteBufPoolStats();

	public static ByteBufPoolStats getStats() {
		return stats;
	}

	public static int getCreatedItems() {
		int items = 0;
		for (int n : created) {
			items += n;
		}
		return items;
	}

	public static int getCreatedItems(int slab) {
		assert slab >= 0 && slab < slabs.length;
		return created[slab];
	}

	public static int getPoolItems(int slab) {
		assert slab >= 0 && slab < slabs.length;
		return slabs[slab].size();
	}

	public static int getPoolItems() {
		int result = 0;
		for (ConcurrentStack<ByteBuf> slab : slabs) {<commit_msg>fix concurrency issue in ByteBufPool jmx stats<commit_after>/*
 * Copyright (C) 2015 SoftIndex LLC.
 *
 * 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.
 */

package io.datakernel.bytebuf;

import io.datakernel.util.ConcurrentStack;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import static java.lang.Integer.numberOfLeadingZeros;

public final class ByteBufPool {
	private static final int NUMBER_SLABS = 33;

	private static int minSize = 32;
	private static int maxSize = 1 << 30;

	private static final ConcurrentStack<ByteBuf>[] slabs = createSlabs(NUMBER_SLABS);
	private static final AtomicInteger[] created = createCounters(NUMBER_SLABS);

	private ByteBufPool() {}

	public static ByteBuf allocate(int size) {
		if (size < minSize || size >= maxSize) {
			// not willing to register in pool
			return ByteBuf.wrapForWriting(new byte[size]);
		}
		int index = 32 - numberOfLeadingZeros(size - 1); // index==32 for size==0
		ConcurrentStack<ByteBuf> queue = slabs[index];
		ByteBuf buf = queue.pop();
		if (buf != null) {
			buf.reset();
		} else {
			buf = ByteBuf.wrapForWriting(new byte[1 << index]);
			buf.refs++;
			assert created[index].incrementAndGet() != 0;
		}
		return buf;
	}

	public static void recycle(ByteBuf buf) {
		assert buf.array.length >= minSize && buf.array.length <= maxSize;
		ConcurrentStack<ByteBuf> queue = slabs[32 - numberOfLeadingZeros(buf.array.length - 1)];
		assert !queue.contains(buf) : "duplicate recycle array";
		queue.push(buf);
	}

	public static ByteBuf recycleIfEmpty(ByteBuf buf) {
		if (buf.canRead())
			return buf;
		buf.recycle();
		return ByteBuf.empty();
	}

	public static ConcurrentStack<ByteBuf>[] getPool() {
		return slabs;
	}

	public static void clear() {
		for (int i = 0; i < ByteBufPool.NUMBER_SLABS; i++) {
			slabs[i].clear();
			created[i].set(0);
		}
	}

	private static ConcurrentStack<ByteBuf>[] createSlabs(int numberOfSlabs) {
		//noinspection unchecked
		ConcurrentStack<ByteBuf>[] slabs = new ConcurrentStack[numberOfSlabs];
		for (int i = 0; i < slabs.length; i++) {
			slabs[i] = new ConcurrentStack<>();
		}
		return slabs;
	}

	private static AtomicInteger[] createCounters(int amount) {
		AtomicInteger[] counters = new AtomicInteger[amount];
		for (int i = 0; i < counters.length; i++) {
			counters[i] = new AtomicInteger(0);
		}
		return counters;
	}

	public static ByteBuf ensureTailRemaining(ByteBuf buf, int newTailRemaining) {
		assert !(buf instanceof ByteBuf.ByteBufSlice);
		if (buf.writeRemaining() >= newTailRemaining) {
			return buf;
		} else {
			ByteBuf newBuf = allocate(newTailRemaining + buf.readRemaining());
			newBuf.put(buf);
			buf.recycle();
			return newBuf;
		}
	}

	public static ByteBuf append(ByteBuf to, ByteBuf from) {
		assert !to.isRecycled() && !from.isRecycled();
		if (to.readRemaining() == 0) {
			to.recycle();
			return from;
		}
		to = ensureTailRemaining(to, from.readRemaining());
		to.put(from);
		from.recycle();
		return to;
	}

	public static ByteBuf append(ByteBuf to, byte[] from, int offset, int length) {
		assert !to.isRecycled();
		to = ensureTailRemaining(to, length);
		to.put(from, offset, length);
		return to;
	}

	public static ByteBuf append(ByteBuf to, byte[] from) {
		return append(to, from, 0, from.length);
	}

	private static final ByteBufPoolStats stats = new ByteBufPoolStats();

	public static ByteBufPoolStats getStats() {
		return stats;
	}

	public static int getCreatedItems() {
		int items = 0;
		for (AtomicInteger counter : created) {
			items += counter.get();
		}
		return items;
	}

	public static int getCreatedItems(int slab) {
		assert slab >= 0 && slab < slabs.length;
		return created[slab].get();
	}

	public static int getPoolItems(int slab) {
		assert slab >= 0 && slab < slabs.length;
		return slabs[slab].size();
	}

	public static int getPoolItems() {
		int result = 0;
		for (ConcurrentStack<ByteBuf> slab : slabs) {
{'fraction_non_alphanumeric': 0.11985898942420682, 'fraction_numerical': 0.008343125734430083, 'mean_word_length': 4.027170702894271, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 29, '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': '6814327', 'n_tokens_mistral': 3055, 'n_tokens_neox': 2751, 'n_words': 823}