Skip to content

Commit

Permalink
Use custom Synchronizer class to manage ReentrantLock synchronization
Browse files Browse the repository at this point in the history
  • Loading branch information
@arturobernalg
arturobernalg committed Oct 11, 2023
1 parent eb77ea5 commit 026ba28
Show file tree
Hide file tree
Showing 37 changed files with 610 additions and 814 deletions.
5 changes: 5 additions & 0 deletions jspwiki-event/pom.xml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,5 +62,10 @@
<artifactId>junit-jupiter-engine</artifactId>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.apache.jspwiki</groupId>
<artifactId>jspwiki-util</artifactId>
</dependency>
</dependencies>
</project>
148 changes: 56 additions & 92 deletions jspwiki-event/src/main/java/org/apache/wiki/event/WikiEventManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ Licensed to the Apache Software Foundation (ASF) under one

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.wiki.util.Synchronizer;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
Expand All @@ -33,6 +34,7 @@ Licensed to the Apache Software Foundation (ASF) under one
import java.util.Set;
import java.util.TreeSet;
import java.util.Vector;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

/**
Expand Down Expand Up @@ -167,15 +169,12 @@ private WikiEventManager() {
*/
public static WikiEventManager getInstance() {
if (c_instance == null) {
lock.lock();
try {
Synchronizer.synchronize(lock, () -> {
if (c_instance == null) {
c_instance = new WikiEventManager();
// start up any post-instantiation services here
}
} finally {
lock.unlock();
}
});
}
return c_instance;
}
Expand Down Expand Up @@ -255,41 +254,28 @@ public static Set<WikiEventListener> getWikiEventListeners( final Object client
* @return true if the listener was found and removed.
*/
public static boolean removeWikiEventListener( final WikiEventListener listener ) {
boolean removed = false;
final AtomicBoolean removed = new AtomicBoolean(false);
// get the Map.entry object for the entire Map, then check match on entry (listener)
final WikiEventManager mgr = getInstance();
final Map< Object, WikiEventDelegate > sources = Collections.synchronizedMap( mgr.getDelegates() );
lock.lock();
try {
Synchronizer.synchronize(lock, () -> {
// get an iterator over the Map.Enty objects in the map
for( final Map.Entry< Object, WikiEventDelegate > entry : sources.entrySet() ) {
// the entry value is the delegate
final WikiEventDelegate delegate = entry.getValue();

// now see if we can remove the listener from the delegate (delegate may be null because this is a weak reference)
if( delegate != null && delegate.removeWikiEventListener( listener ) ) {
removed = true; // was removed
removed.set(true); // was removed
}
}
} finally {
lock.unlock();
}
return removed;
});
return removed.get();
}

private void removeDelegates() {
lock.lock();
try {
m_delegates.clear();
} finally {
lock.unlock();
}
lock.lock();
try {
m_preloadCache.clear();
} finally {
lock.unlock();
}
Synchronizer.synchronize(lock, m_delegates::clear);
Synchronizer.synchronize(lock, m_preloadCache::clear);
}

public static void shutdown() {
Expand Down Expand Up @@ -340,38 +326,35 @@ private Map< Object, WikiEventDelegate > getDelegates() {
* @param client the client Object, or alternately a Class reference
* @return the WikiEventDelegate.
*/
private WikiEventDelegate getDelegateFor( final Object client ) {
lock.lock();
try {
if( client == null || client instanceof Class ) { // then preload the cache
final WikiEventDelegate delegate = new WikiEventDelegate( client );
m_preloadCache.add( delegate );
m_delegates.put( client, delegate );
private WikiEventDelegate getDelegateFor(final Object client) {
return Synchronizer.synchronize(lock, () -> {
if (client == null || client instanceof Class) { // then preload the cache
final WikiEventDelegate delegate = new WikiEventDelegate(client);
m_preloadCache.add(delegate);
m_delegates.put(client, delegate);
return delegate;
} else if( !m_preloadCache.isEmpty() ) {
} else if (!m_preloadCache.isEmpty()) {
// then see if any of the cached delegates match the class of the incoming client
for( int i = m_preloadCache.size()-1 ; i >= 0 ; i-- ) { // start with most-recently added
final WikiEventDelegate delegate = m_preloadCache.elementAt( i );
if( delegate.getClientClass() == null || delegate.getClientClass().equals( client.getClass() ) ) {
for (int i = m_preloadCache.size() - 1; i >= 0; i--) { // start with most-recently added
final WikiEventDelegate delegate = m_preloadCache.elementAt(i);
if (delegate.getClientClass() == null || delegate.getClientClass().equals(client.getClass())) {
// we have a hit, so use it, but only on a client we haven't seen before
if( !m_delegates.containsKey( client ) ) {
m_preloadCache.remove( delegate );
m_delegates.put( client, delegate );
if (!m_delegates.containsKey(client)) {
m_preloadCache.remove(delegate);
m_delegates.put(client, delegate);
return delegate;
}
}
}
}
// otherwise treat normally...
WikiEventDelegate delegate = m_delegates.get( client );
if( delegate == null ) {
delegate = new WikiEventDelegate( client );
m_delegates.put( client, delegate );
WikiEventDelegate delegate = m_delegates.get(client);
if (delegate == null) {
delegate = new WikiEventDelegate(client);
m_delegates.put(client, delegate);
}
return delegate;
} finally {
lock.unlock();
}
});
}


Expand Down Expand Up @@ -415,8 +398,7 @@ private static final class WikiEventDelegate {
* @throws java.lang.UnsupportedOperationException if any attempt is made to modify the Set
*/
public Set< WikiEventListener > getWikiEventListeners() {
lock.lock();
try {
return Synchronizer.synchronize(lock, () -> {
final TreeSet< WikiEventListener > set = new TreeSet<>( new WikiEventListenerComparator() );
for( final WeakReference< WikiEventListener > wikiEventListenerWeakReference : m_listenerList ) {
final WikiEventListener l = wikiEventListenerWeakReference.get();
Expand All @@ -426,9 +408,7 @@ public Set< WikiEventListener > getWikiEventListeners() {
}

return Collections.unmodifiableSet( set );
} finally {
lock.unlock();
}
});
}

/**
Expand All @@ -438,18 +418,15 @@ public Set< WikiEventListener > getWikiEventListeners() {
* @return true if the listener was added (i.e., it was not already in the list and was added)
*/
public boolean addWikiEventListener( final WikiEventListener listener ) {
lock.lock();
try {
return Synchronizer.synchronize(lock, () -> {
final boolean listenerAlreadyContained = m_listenerList.stream()
.map( WeakReference::get )
.anyMatch( ref -> ref == listener );
if( !listenerAlreadyContained ) {
return m_listenerList.add( new WeakReference<>( listener ) );
}
} finally {
lock.unlock();
}
return false;
return false;
});
}

/**
Expand All @@ -459,67 +436,54 @@ public boolean addWikiEventListener( final WikiEventListener listener ) {
* @return true if the listener was removed (i.e., it was actually in the list and was removed)
*/
public boolean removeWikiEventListener( final WikiEventListener listener ) {
lock.lock();
try {
for( final Iterator< WeakReference< WikiEventListener > > i = m_listenerList.iterator(); i.hasNext(); ) {
return Synchronizer.synchronize(lock, () -> {
for (final Iterator<WeakReference<WikiEventListener>> i = m_listenerList.iterator(); i.hasNext(); ) {
final WikiEventListener l = i.next().get();
if( l == listener ) {
if (l == listener) {
i.remove();
return true;
}
}
} finally {
lock.unlock();
}

return false;
return false;
});
}

/**
* Returns true if there are one or more listeners registered with this instance.
*/
public boolean isListening() {
lock.lock();
try {
return !m_listenerList.isEmpty();
} finally {
lock.unlock();
}
return Synchronizer.synchronize(lock, () -> !m_listenerList.isEmpty());
}

/**
* Notify all listeners having a registered interest in change events of the supplied WikiEvent.
*/
public void fireEvent( final WikiEvent event ) {
boolean needsCleanup = false;
public void fireEvent(final WikiEvent event) {
final AtomicBoolean needsCleanup = new AtomicBoolean(false);
try {
lock.lock();
try {
for( final WeakReference< WikiEventListener > wikiEventListenerWeakReference : m_listenerList ) {
Synchronizer.synchronize(lock, () -> {
for (final WeakReference<WikiEventListener> wikiEventListenerWeakReference : m_listenerList) {
final WikiEventListener listener = wikiEventListenerWeakReference.get();
if( listener != null ) {
listener.actionPerformed( event );
if (listener != null) {
listener.actionPerformed(event);
} else {
needsCleanup = true;
needsCleanup.set(true);
}
}

// Remove all such listeners which have expired
if( needsCleanup ) {
for( int i = 0; i < m_listenerList.size(); i++ ) {
final WeakReference< WikiEventListener > w = m_listenerList.get( i );
if( w.get() == null ) {
// Remove all such listeners which have expired
if (needsCleanup.get()) {
for (int i = 0; i < m_listenerList.size(); i++) {
final WeakReference<WikiEventListener> w = m_listenerList.get(i);
if (w.get() == null) {
m_listenerList.remove(i--);
}
}
}

} finally {
lock.unlock();
}
} catch( final ConcurrentModificationException e ) {
// We don't die, we just don't do notifications in that case.
LOG.info( "Concurrent modification of event list; please report this.", e );
});
} catch (final ConcurrentModificationException e) {
// We don't die, we just don't do notifications in that case.
LOG.info("Concurrent modification of event list; please report this.", e);
}
}
}
Expand Down
17 changes: 8 additions & 9 deletions ...ndra-searchprovider/src/main/java/org/apache/wiki/search/kendra/KendraSearchProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@ Licensed to the Apache Software Foundation (ASF) under one
import org.apache.wiki.auth.permissions.PagePermission;
import org.apache.wiki.pages.PageManager;
import org.apache.wiki.search.SearchProvider;
import org.apache.wiki.util.Synchronizer;
import org.apache.wiki.util.TextUtil;

import java.io.IOException;
Expand Down Expand Up @@ -346,23 +347,21 @@ private void doFullReindex() throws IOException {
* index pages that have been modified
*/
private void doPartialReindex() {
if ( updates.isEmpty() ) {
if (updates.isEmpty()) {
return;
}
LOG.debug( "Indexing updated pages. Please wait ..." );
LOG.debug("Indexing updated pages. Please wait ...");
final String executionId = startExecution();
lock.lock();
try {

Synchronizer.synchronize(lock, () -> {
try {
while ( updates.size() > 0 ) {
indexOnePage( updates.remove( 0 ), executionId );
while (!updates.isEmpty()) {
indexOnePage(updates.remove(0), executionId);
}
} finally {
stopExecution();
}
} finally {
lock.unlock();
}
});
}

/**
Expand Down
Loading

0 comments on commit 026ba28

Please sign in to comment.