@ -82,6 +82,8 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
private final List < Future < ? > > running ;
private final List < Future < ? > > running ;
private final Object runningLock ;
private final Semaphore spaceAvailable ;
private final Semaphore spaceAvailable ;
private int queuedCount ;
private int queuedCount ;
@ -102,6 +104,7 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
this . executor = executor ;
this . executor = executor ;
this . bufferSize = bufferSize ;
this . bufferSize = bufferSize ;
this . running = new LinkedList < Future < ? > > ( ) ;
this . running = new LinkedList < Future < ? > > ( ) ;
this . runningLock = new Object ( ) ;
this . spaceAvailable = new Semaphore ( bufferSize ) ;
this . spaceAvailable = new Semaphore ( bufferSize ) ;
}
}
@ -189,15 +192,19 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
}
}
}
}
synchronized ( runningLock ) {
checkRunningTasks ( true ) ;
checkRunningTasks ( true ) ;
}
} finally {
} finally {
flushing = false ;
flushing = false ;
}
}
}
}
public void abort ( ) throws DhtException {
public void abort ( ) throws DhtException {
synchronized ( runningLock ) {
checkRunningTasks ( true ) ;
checkRunningTasks ( true ) ;
}
}
}
private void acquireSpace ( int sz ) throws DhtException {
private void acquireSpace ( int sz ) throws DhtException {
try {
try {
@ -259,10 +266,12 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
return ;
return ;
}
}
synchronized ( runningLock ) {
if ( ! flushing )
if ( ! flushing )
checkRunningTasks ( false ) ;
checkRunningTasks ( false ) ;
running . add ( executor . submit ( op ) ) ;
running . add ( executor . submit ( op ) ) ;
}
}
}
/ * *
/ * *
* Wrap a callback to update the buffer .
* Wrap a callback to update the buffer .
@ -284,8 +293,10 @@ public abstract class AbstractWriteBuffer implements WriteBuffer {
int size ) throws DhtException {
int size ) throws DhtException {
int permits = permitsForSize ( size ) ;
int permits = permitsForSize ( size ) ;
WrappedCallback < T > op = new WrappedCallback < T > ( callback , permits ) ;
WrappedCallback < T > op = new WrappedCallback < T > ( callback , permits ) ;
synchronized ( runningLock ) {
checkRunningTasks ( false ) ;
checkRunningTasks ( false ) ;
running . add ( op ) ;
running . add ( op ) ;
}
return op ;
return op ;
}
}