<commit_before>package org.voovan.tools;

import org.voovan.Global;
import org.voovan.tools.exception.MemoryReleasedException;
import org.voovan.tools.log.Logger;
import sun.misc.Unsafe;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ByteBuffer双向通道
 *
 * @author helyho
 *
 * Voovan Framework.
 * WebSite: https://github.com/helyho/Voovan
 * Licence: Apache v2 License
 */
public class ByteBufferChannel {
	private final ByteBuffer emptyByteBuffer = ByteBuffer.allocateDirect(0);

	private volatile AtomicLong address = new AtomicLong(0);
	private Unsafe unsafe = TUnsafe.getUnsafe();
	private ByteBuffer byteBuffer;
	private int size;
	private ReentrantLock lock;
	private AtomicBoolean borrowed = new AtomicBoolean(false);

	/**
	 * 构造函数
	 * @param capacity 分配的容量
	 */
	public ByteBufferChannel(int capacity) {
		init(capacity);
	}

	/**
	 * 构造函数
	 * @param byteBuffer 分配的容量
	 */
	public ByteBufferChannel(ByteBuffer byteBuffer) {
		lock = new ReentrantLock(true);
		this.byteBuffer = byteBuffer;
		resetAddress();
		this.size = byteBuffer.remaining();
	}

	/**
	 * 构造函数
	 */
	public ByteBufferChannel() {
		init(1024);
	}

	/**
	 * 初始化函数
	 * @param capacity 分配的容量
	 */
	private void init(int capacity){
		lock = new ReentrantLock(true);
		this.byteBuffer = newByteBuffer(capacity);
		byteBuffer.limit(0);
		resetAddress();
		this.size = 0;
	}

	/**
	 * 构造一个ByteBuffer
	 * @param capacity 分配的容量
	 * @return ByteBuffer 对象
	 */
	private ByteBuffer newByteBuffer(int capacity){
		try {

			ByteBuffer instance = TByteBuffer.allocateManualReleaseBuffer(capacity, false);
			address.set(TByteBuffer.getAddress(instance));

			return instance;

		}catch(Exception e){
			Logger.error("Create ByteBufferChannel error. ", e);
			return null;
		}
	}

	/**
	 * 是否已经释放
	 * @return true 已释放, false: 未释放
	 */
	public boolean isReleased(){
		if(address.get() == 0 || byteBuffer == null){
			return true;
		}else{
			return false;
		}
	}

	/**
	 * 测试是否被释放
	 */
	private void checkRelease(){
		if(isReleased()){
			throw new MemoryReleasedException("ByteBufferChannel is released.");
		}
	}

	/**
	 * 立刻释放内存
	 */
	public synchronized void release(){
		if(byteBuffer==null){
			return;
		}

		//是否手工释放
		if(!Global.NO_HEAP_MANUAL_RELEASE || byteBuffer.getClass() != TByteBuffer.DIRECT_BYTE_BUFFER_CLASS) {
			return;
		}

		synchronized (byteBuffer) {
			while(lock.isLocked()){

				if(borrowed.compareAndSet(true, false)){
					lock.unlock();
				}

				//如果加锁成功说明是自锁, 解锁并继续
				if(lock.isHeldByCurrentThread()){
					while(true){
						if(lock.getHoldCount()!=0) {
							lock.unlock();
						} else {
							break;
						}
					}
				}

				TEnv.sleep(1);
			}

			lock.lock();
			try {
				if (address.get() != 0) {
					TByteBuffer.release(byteBuffer);
					address.set(0);
					byteBuffer = null;
					size = -1;
				}
			} finally {
				lock.unlock();
			}
		}
	}

	/**
	 * 重新设置当前内存地址
	 */
	private void resetAddress(){<commit_msg>imp: 优化 ByteBufferChannel 的细节<commit_after>package org.voovan.tools;

import org.voovan.Global;
import org.voovan.tools.exception.MemoryReleasedException;
import org.voovan.tools.log.Logger;
import sun.misc.Unsafe;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * ByteBuffer双向通道
 *
 * @author helyho
 *
 * Voovan Framework.
 * WebSite: https://github.com/helyho/Voovan
 * Licence: Apache v2 License
 */
public class ByteBufferChannel {
	private final ByteBuffer emptyByteBuffer = ByteBuffer.allocateDirect(0);

	private volatile AtomicLong address = new AtomicLong(0);
	private Unsafe unsafe = TUnsafe.getUnsafe();
	private ByteBuffer byteBuffer;
	private int size;
	private ReentrantLock lock;
	private AtomicBoolean borrowed = new AtomicBoolean(false);

	/**
	 * 构造函数
	 * @param capacity 分配的容量
	 */
	public ByteBufferChannel(int capacity) {
		init(capacity);
	}

	/**
	 * 构造函数
	 * @param byteBuffer 分配的容量
	 */
	public ByteBufferChannel(ByteBuffer byteBuffer) {
		lock = new ReentrantLock(true);
		this.byteBuffer = byteBuffer;
		resetAddress();
		this.size = byteBuffer.remaining();
	}

	/**
	 * 构造函数
	 */
	public ByteBufferChannel() {
		init(1024);
	}

	/**
	 * 初始化函数
	 * @param capacity 分配的容量
	 */
	private void init(int capacity){
		lock = new ReentrantLock(true);
		this.byteBuffer = newByteBuffer(capacity);
		byteBuffer.limit(0);
		resetAddress();
		this.size = 0;
	}

	/**
	 * 构造一个ByteBuffer
	 * @param capacity 分配的容量
	 * @return ByteBuffer 对象
	 */
	private ByteBuffer newByteBuffer(int capacity){
		try {

			ByteBuffer instance = TByteBuffer.allocateManualReleaseBuffer(capacity, false);
			address.set(TByteBuffer.getAddress(instance));

			return instance;

		}catch(Exception e){
			Logger.error("Create ByteBufferChannel error. ", e);
			return null;
		}
	}

	/**
	 * 是否已经释放
	 * @return true 已释放, false: 未释放
	 */
	public boolean isReleased(){
		if(address.get() == 0 || byteBuffer == null){
			return true;
		}else{
			return false;
		}
	}

	/**
	 * 测试是否被释放
	 */
	private void checkRelease(){
		if(isReleased()){
			throw new MemoryReleasedException("ByteBufferChannel is released.");
		}
	}

	/**
	 * 立刻释放内存
	 */
	public synchronized void release(){
		if(byteBuffer==null){
			return;
		}

		//是否手工释放
		if(!Global.NO_HEAP_MANUAL_RELEASE || byteBuffer.getClass() != TByteBuffer.DIRECT_BYTE_BUFFER_CLASS) {
			return;
		}

		synchronized (byteBuffer) {
			while(lock.isLocked()){


				//如果加锁成功说明是自锁, 解锁并继续
				if(lock.isHeldByCurrentThread()){

					if(borrowed.compareAndSet(true, false)){
						lock.unlock();
					}

					while(true){
						if(lock.getHoldCount()!=0) {
							lock.unlock();
						} else {
							break;
						}
					}
				}

				TEnv.sleep(1);
			}

			lock.lock();
			try {
				if (address.get() != 0) {
					TByteBuffer.release(byteBuffer);
					address.set(0);
					byteBuffer = null;
					size = -1;
				}
			} finally {
				lock.unlock();
			}
		}
	}

	/**
	 * 重新设置当前内存地址
	 */
	private void resetAddress(){
{'fraction_non_alphanumeric': 0.13013375295043272, 'fraction_numerical': 0.004720692368214005, 'mean_word_length': 3.3925362819626814, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, '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': '3631021', 'n_tokens_mistral': 2633, 'n_tokens_neox': 2301, 'n_words': 426}