<commit_before>package com.firefly.utils.lang.pool;

import com.firefly.utils.concurrent.Atomics;
import com.firefly.utils.concurrent.Promise;
import com.firefly.utils.concurrent.ReentrantLocker;
import com.firefly.utils.exception.CommonRuntimeException;
import com.firefly.utils.lang.AbstractLifeCycle;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author <NAME>
 */
public class BoundedAsynchronousPool<T> extends AbstractLifeCycle implements AsynchronousPool<T> {

    protected final int maxSize;
    protected final AtomicInteger createdObjectSize = new AtomicInteger(0);
    protected final long timeout;
    protected final BlockingQueue<PooledObject<T>> queue;
    protected final ExecutorService service;
    protected final ObjectFactory<T> objectFactory;
    protected final Validator<T> validator;
    protected final Dispose<T> dispose;
    protected final ReentrantLocker locker = new ReentrantLocker();

    public BoundedAsynchronousPool(ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(32, objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(maxSize, 5000L, objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, long timeout,
                                   ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(maxSize, timeout,
                Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), r -> new Thread(r, "firefly bounded asynchronous pool")),
                objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, long timeout,
                                   ExecutorService service,
                                   ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this.maxSize = maxSize;
        this.timeout = timeout;
        this.service = service;
        this.objectFactory = objectFactory;
        this.validator = validator;
        this.dispose = dispose;
        this.queue = new ArrayBlockingQueue<>(maxSize);
        start();
    }

    protected void createObject(Promise.Completable<PooledObject<T>> completable) {
        try {
            Atomics.getAndIncrement(createdObjectSize, maxSize);
            CompletableFuture<PooledObject<T>> tmp = objectFactory.createNew();
            tmp.thenAccept(completable::succeeded)
               .exceptionally(e0 -> {
                   Atomics.getAndDecrement(createdObjectSize, 0);
                   completable.failed(e0);
                   return null;
               });
        } catch (Exception e) {
            System.err.println(e.getMessage());
            Atomics.getAndDecrement(createdObjectSize, 0);
        }
    }

    protected void destroyObject(PooledObject<T> t) {
        Atomics.getAndDecrement(createdObjectSize, 0);
        try {
            dispose.destroy(t);
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }

    @Override
    public CompletableFuture<PooledObject<T>> take() {
        Promise.Completable<PooledObject<T>> completable = new Promise.Completable<>();
        PooledObject<T> pooledObject = queue.poll();
        if (pooledObject != null) {
            checkObjectFromPool(pooledObject, completable);
            return completable;
        } else { // the queue is empty
            return locker.lock(() -> {
                int availableSize = maxSize - getCreatedObjectSize();
                if (availableSize > 0) {
                    createObject(completable);
                    return completable;
                } else {
                    service.execute(() -> {
                        try {
                            PooledObject<T> object = queue.poll(timeout, TimeUnit.MILLISECONDS);
                            if (object != null) {
                                checkObjectFromPool(object, completable);
                            } else {
                                completable.failed(new TimeoutException("take pooled object timeout"));
                            }
                        } catch (InterruptedException e) {
                            completable.failed(e);
                        }
                    });
                    return completable;
                }
            });
        }
    }

    private void checkObjectFromPool(PooledObject<T> pooledObject, Promise.Completable<PooledObject<T>> completable) {
        if (pooledObject.prepareTake()) {
            if (validator.isValid(pooledObject)) {
                completable.succeeded(pooledObject);
            } else {
                destroyObject(pooledObject);
                createObject(completable);
            }
        } else {
            completable.failed(new CommonRuntimeException("the pooled object has been used"));
        }
    }

    @Override
    public void release(PooledObject<T> pooledObject) {
        if (pooledObject == null) {
            return;
        }
        if (!pooledObject.prepareRelease()) { // Object is released
            return;
        }
        if (!queue.offer(pooledObject)) {
            // the queue is full
            service.execute(() -> {
                try {
                    if (!queue.offer(pooledObject, timeout, TimeUnit.MILLISECONDS)) {
                        destroyObject(pooledObject);
                    }
                } catch (InterruptedException e) {
                    destroyObject(pooledObject);
                }
            });
        }
    }

    @Override
    public PooledObject<T> get() {
        throw new CommonRuntimeException("the method is not implemented");
    }

    @Override
    public int size() {
        return queue.size();
    }

    @Override
    public int getCreatedObjectSize() {
        return createdObjectSize.get();
    }

    @Override
    public boolean isEmpty() {
        return queue.isEmpty();
    }<commit_msg>[feat]: implement get method of AsynchronousPool<commit_after>package com.firefly.utils.lang.pool;

import com.firefly.utils.concurrent.Atomics;
import com.firefly.utils.concurrent.Promise;
import com.firefly.utils.concurrent.ReentrantLocker;
import com.firefly.utils.exception.CommonRuntimeException;
import com.firefly.utils.lang.AbstractLifeCycle;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author <NAME>
 */
public class BoundedAsynchronousPool<T> extends AbstractLifeCycle implements AsynchronousPool<T> {

    protected final int maxSize;
    protected final AtomicInteger createdObjectSize = new AtomicInteger(0);
    protected final long timeout;
    protected final BlockingQueue<PooledObject<T>> queue;
    protected final ExecutorService service;
    protected final ObjectFactory<T> objectFactory;
    protected final Validator<T> validator;
    protected final Dispose<T> dispose;
    protected final ReentrantLocker locker = new ReentrantLocker();

    public BoundedAsynchronousPool(ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(32, objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(maxSize, 5000L, objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, long timeout,
                                   ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this(maxSize, timeout,
                Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), r -> new Thread(r, "firefly bounded asynchronous pool")),
                objectFactory, validator, dispose);
    }

    public BoundedAsynchronousPool(int maxSize, long timeout,
                                   ExecutorService service,
                                   ObjectFactory<T> objectFactory, Validator<T> validator, Dispose<T> dispose) {
        this.maxSize = maxSize;
        this.timeout = timeout;
        this.service = service;
        this.objectFactory = objectFactory;
        this.validator = validator;
        this.dispose = dispose;
        this.queue = new ArrayBlockingQueue<>(maxSize);
        start();
    }

    protected void createObject(Promise.Completable<PooledObject<T>> completable) {
        try {
            Atomics.getAndIncrement(createdObjectSize, maxSize);
            CompletableFuture<PooledObject<T>> tmp = objectFactory.createNew();
            tmp.thenAccept(completable::succeeded)
               .exceptionally(e0 -> {
                   Atomics.getAndDecrement(createdObjectSize, 0);
                   completable.failed(e0);
                   return null;
               });
        } catch (Exception e) {
            System.err.println(e.getMessage());
            Atomics.getAndDecrement(createdObjectSize, 0);
        }
    }

    protected void destroyObject(PooledObject<T> t) {
        Atomics.getAndDecrement(createdObjectSize, 0);
        try {
            dispose.destroy(t);
        } catch (Exception e) {
            System.err.println(e.getMessage());
        }
    }

    @Override
    public CompletableFuture<PooledObject<T>> take() {
        Promise.Completable<PooledObject<T>> completable = new Promise.Completable<>();
        PooledObject<T> pooledObject = queue.poll();
        if (pooledObject != null) {
            checkObjectFromPool(pooledObject, completable);
            return completable;
        } else { // the queue is empty
            return locker.lock(() -> {
                int availableSize = maxSize - getCreatedObjectSize();
                if (availableSize > 0) {
                    createObject(completable);
                    return completable;
                } else {
                    service.execute(() -> {
                        try {
                            PooledObject<T> object = queue.poll(timeout, TimeUnit.MILLISECONDS);
                            if (object != null) {
                                checkObjectFromPool(object, completable);
                            } else {
                                completable.failed(new TimeoutException("take pooled object timeout"));
                            }
                        } catch (InterruptedException e) {
                            completable.failed(e);
                        }
                    });
                    return completable;
                }
            });
        }
    }

    private void checkObjectFromPool(PooledObject<T> pooledObject, Promise.Completable<PooledObject<T>> completable) {
        if (pooledObject.prepareTake()) {
            if (validator.isValid(pooledObject)) {
                completable.succeeded(pooledObject);
            } else {
                destroyObject(pooledObject);
                createObject(completable);
            }
        } else {
            completable.failed(new CommonRuntimeException("the pooled object has been used"));
        }
    }

    @Override
    public void release(PooledObject<T> pooledObject) {
        if (pooledObject == null) {
            return;
        }
        if (!pooledObject.prepareRelease()) { // Object is released
            return;
        }
        if (!queue.offer(pooledObject)) {
            // the queue is full
            service.execute(() -> {
                try {
                    if (!queue.offer(pooledObject, timeout, TimeUnit.MILLISECONDS)) {
                        destroyObject(pooledObject);
                    }
                } catch (InterruptedException e) {
                    destroyObject(pooledObject);
                }
            });
        }
    }

    @Override
    public PooledObject<T> get() {
        try {
            return take().get();
        } catch (InterruptedException | ExecutionException e) {
            System.err.println(e.getMessage());
            return null;
        }
    }

    @Override
    public int size() {
        return queue.size();
    }

    @Override
    public int getCreatedObjectSize() {
        return createdObjectSize.get();
    }

    @Override
    public boolean isEmpty() {
        return queue.isEmpty();
    }
{'fraction_non_alphanumeric': 0.0935634779803194, 'fraction_numerical': 0.002097112437489918, 'mean_word_length': 1.9367598294647086, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5016164', 'n_tokens_mistral': 3275, 'n_tokens_neox': 3100, 'n_words': 648}