Skip to content

Commit

Permalink
Issue #5287 - rework CompressionPool to use the jetty-util pool
Browse files Browse the repository at this point in the history
Signed-off-by: Lachlan Roberts <lachlan@webtide.com>
  • Loading branch information
lachlan-roberts committed Oct 14, 2020
1 parent d78e1f8 commit b239fa0
Showing 1 changed file with 40 additions and 58 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -18,19 +18,14 @@

package org.eclipse.jetty.util.compression;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.eclipse.jetty.util.Pool;
import org.eclipse.jetty.util.component.AbstractLifeCycle;

public abstract class CompressionPool<T> extends AbstractLifeCycle
{
public static final int INFINITE_CAPACITY = -1;

private final Queue<T> _pool;
private final AtomicInteger _numObjects = new AtomicInteger(0);
private int _capacity;
private final Pool<T> _pool;

/**
* Create a Pool of {@link T} instances.
Expand All @@ -43,8 +38,7 @@ public abstract class CompressionPool<T> extends AbstractLifeCycle
*/
public CompressionPool(int capacity)
{
_capacity = capacity;
_pool = new ConcurrentLinkedQueue<>();
_pool = new Pool<T>(capacity, 1);
}

public int getCapacity()
Expand All @@ -66,73 +60,61 @@ public void setCapacity(int capacity)
/**
* @return Object taken from the pool if it is not empty or a newly created Object
*/
public T acquire()
public Entry acquire()
{
T object;

if (_capacity == 0)
object = newObject();
else
{
object = _pool.poll();
if (object == null)
object = newObject();
else if (_capacity > 0)
_numObjects.decrementAndGet();
}

return object;
return new Entry(_pool.acquire(e -> newObject()));
}

/**
* @param object returns this Object to the pool or calls {@link #end(Object)} if the pool is full.
* @param entry returns this Object to the pool or calls {@link #end(Object)} if the pool is full.
*/
public void release(T object)
public void release(Entry entry)
{
entry.release();
}

@Override
public void doStop()
{
// TODO: We can't use this because it will not end the entries after it removes them.
// _pool.close();
}

public class Entry
{
if (object == null)
return;
private T _value;
private Pool<T>.Entry _entry;

if (_capacity == 0 || !isRunning())
Entry(Pool<T>.Entry entry)
{
end(object);
_entry = entry;
_value = (_entry != null) ? _entry.getPooled() : newObject();
}
else if (_capacity < 0)

public T get()
{
reset(object);
_pool.add(object);
return _value;
}
else
{
while (true)
{
int d = _numObjects.get();

if (d >= _capacity)
{
end(object);
break;
}
public void release()
{
// Reset the value for the next usage.
reset(_value);

if (_numObjects.compareAndSet(d, d + 1))
if (_entry != null)
{
// If release return false, the entry should be removed and the object should be disposed.
if (!_pool.release(_entry))
{
reset(object);
_pool.add(object);
break;
// TODO: what does it mean if this returns false???
if (_pool.remove(_entry))
end(_value);
}
}
}
}

@Override
public void doStop()
{
T t = _pool.poll();
while (t != null)
{
end(t);
t = _pool.poll();
_value = null;
_entry = null;
}
_numObjects.set(0);
}

@Override
Expand Down

0 comments on commit b239fa0

Please sign in to comment.