在我的代码中,我有许多执行器服务作为管道运行,因为第一个执行器服务可以向任何后来的执行器服务提交任务,但从不反过来.
services.add(songLoaderService);
services.add(AcoustIdMatcher.getExecutorService());
services.add(SongPrematcherMatcher.getExecutorService());
services.add(MusicBrainzMetadataMatcher.getExecutorService());
//Start Loading Songs
songLoaderService.submit(loader);
我们只为第一次服务提交一个任务,然后我可以请求关闭.在此任务完成之前,这将不会成功,到那时它将把一些任务放到第二个服务上,依此类推.
所以这段代码总是工作多年,在所有要提交的任务完成之前,从不调用shutdown(),并且awaitTermination()方法直到所有提交的任务都完成为止.
int count = 0;
for (ExecutorService service : services)
{
MainWindow.logger.severe("Requested Shutdown Task:" + count + ":"+((SongKongThreadFactory)((TimeoutThreadPoolExecutor) service).getThreadFactory()).getName());
//Request Shutdown
service.shutdown();
//Now wait for service to terminate
service.awaitTermination(10, TimeUnit.DAYS);
MainWindow.logger.severe("Completed Shutdown Task:" + count);
if(count==2)
{
MainWindow.logger.severe("Report:"+currentReportId+":SongPreMatcher:" + SongPrematcherMatcher.getPipelineQueuedCount()+":"+ SongPrematcherMatcher.getPipelineCallCount()+":"+ SongPrematcherMatcher.getPipelineCompletedCount()+":"+SongPrematcherMatcher.getPipelineFileCount());
}
count++;
}
但我现在看到一个ExecutorService无法以这种方式工作的问题.
在先前(AcoustIdMatcher)服务添加到此服务的所有任务已经提交并启动之前但在其中一个任务完成之前,关闭SongPrematcherMatcher服务的请求正在成功,如下面的调试行所示
Report:353:SongPreMatcher:init:57:started:57:Finished:56
丢失的任务没有失败,因为我们可以看到它在日志输出结束时完成,但关键是它在运行它的服务成功终止后完成.
这会产生重大影响,因为这意味着此任务要提交给MusicBrainzMetadataMatcher服务的所有任务都会失败,因为自上次服务(PrematcherMatched)已关闭以来已经为此设置了关闭请求.
PrematcherMatcher最近刚加入,所以我的假设是它有问题,但我看不出它可能是什么.
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:0
analyser.AcoustIdMatcher:<init>:SEVERE: GROUP 115:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:0
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:1:analyser.AcoustIdMatcher
analyser.SongPrematcherMatcher:<init>:SEVERE: Queue:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.SongPrematcherMatcher:call:SEVERE: Start:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:1
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:2:analyser.SongPrematcherMatcher
toplevelanalyzer.FixSongsController:start:SEVERE: Completed Shutdown Task:2
toplevelanalyzer.FixSongsController:start:SEVERE: Report:353:SongPreMatcher:init:57:started:57:Finished:56
toplevelanalyzer.FixSongsController:start:SEVERE: Requested Shutdown Task:3:analyser.MusicBrainzMetadataMatcher
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 795:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 797:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 799:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 821:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.MusicBrainzMetadataMatcher:<init>:SEVERE: Queue:GROUP 823:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
analyser.SongPrematcherMatcher:call:SEVERE: Finish:GROUP 791:C:\Users\Paul\Desktop\FlatRandomFolder:true:false:true:false
如果我在循环中人为地添加了一个延迟,那么至少对于这个特定的测试用例它是有效的,但这不是一个解决方案,因为当有一系列执行器服务时,它会引入延迟,现在一个接一个地关闭它们是有效的.它也不清楚为什么这个修复工作,如果它总是有效.
for (ExecutorService service : services)
{
Thread.sleep(5000);
//Request Shutdown
service.shutdown();
......
}
每个任务都有一个executorservice,只有一种类型的任务可以添加到特定的executorservice中. executorService确实有特殊处理,允许用户取消任务并防止长期运行任务,但这不是问题.
我在PreMatcherMatcher代码中看不到任何与其他任务不同的内容.
package com.jthink.songkong.analyse.analyser;
import com.jthink.songkong.analyse.general.Errors;
import com.jthink.songkong.cmdline.SongKong;
import com.jthink.songkong.ui.MainWindow;
import com.jthink.songkong.util.SongKongThreadFactory;
import java.util.List;
import java.util.concurrent.*;
/**
* From https://*.com/questions/2758612/executorservice-that-interrupts-tasks-after-a-timeout
* With additional support for caller running task when bounded queue is full
*/
public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor {
private final long timeout;
private final TimeUnit timeoutUnit;
private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
private final ConcurrentMap<Runnable, ScheduledFuture> runningTasks = new ConcurrentHashMap<Runnable, ScheduledFuture>();
private final static int WAIT_BEFORE_STOP = 10000;
public long getTimeout()
{
return timeout;
}
public TimeUnit getTimeoutUnit()
{
return timeoutUnit;
}
public TimeoutThreadPoolExecutor(int workerSize, ThreadFactory threadFactory, LinkedBlockingQueue<Runnable> queue,long timeout, TimeUnit timeoutUnit)
{
super(workerSize, workerSize, 0L, TimeUnit.MILLISECONDS, queue, threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
}
@Override
public List<Runnable> shutdownNow() {
timeoutExecutor.shutdownNow();
return super.shutdownNow();
}
@Override
public <T> FutureCallable<T> newTaskFor(Callable<T> callable) {
return new FutureCallable<T>(callable);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
MainWindow.logger.warning("beforeExecute:"+t.getName()+":"+r.toString());
SongKong.checkIn();
if(timeout > 0) {
final ScheduledFuture<?> scheduled = timeoutExecutor.schedule(new TimeoutTask(t,r), timeout, timeoutUnit);
runningTasks.put(r, scheduled);
}
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
MainWindow.logger.warning("afterExecute:"+r.toString());
//AfterExecute will be called after the task has completed, either of its own accord or because it
//took too long and was interrupted by corresponding timeout task
//Remove mapping and cancel timeout task
ScheduledFuture timeoutTask = runningTasks.remove(r);
if(timeoutTask != null) {
timeoutTask.cancel(false);
}
}
@Override
protected void terminated()
{
//All tasks have completed either naturally or via being cancelled by timeout task so close the timeout task
MainWindow.logger.warning("---Terminated:"+((SongKongThreadFactory)getThreadFactory()).getName());
timeoutExecutor.shutdown();
}
class TimeoutTask implements Runnable {
private final Thread thread;
private Callable c;
public TimeoutTask(Thread thread, Runnable c) {
this.thread = thread;
if(c instanceof FutureCallable)
{
this.c = ((FutureCallable) c).getCallable();
}
}
@Override
public void run()
{
String msg = "";
if (c != null)
{
if (c instanceof AcoustIdMatcher)
{
msg = c.getClass() + ":" + ((AcoustIdMatcher) c).getSongGroup().getKey();
}
else if (c instanceof SongPrematcherMatcher)
{
msg = c.getClass() + ":" + ((SongPrematcherMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzSongGroupMatcher)
{
msg = c.getClass() + ":" + ((MusicBrainzSongGroupMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzMetadataMatcher)
{
msg = c.getClass() + ":" + ((MusicBrainzMetadataMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzUpdateSongOnly)
{
msg = c.getClass() + ":" + ((MusicBrainzUpdateSongOnly) c).getSongGroup().getKey();
}
else if (c instanceof DiscogsSongGroupMatcher)
{
msg = c.getClass() + ":" + ((DiscogsSongGroupMatcher) c).getSongGroup().getKey();
}
else if (c instanceof MusicBrainzSongMatcher)
{
msg = c.getClass() + ":" + String.valueOf(((MusicBrainzSongMatcher) c).getSongId());
}
else if (c instanceof SongSaver)
{
msg = c.getClass() + ":" + String.valueOf(((SongSaver) c).getSongId());
}
else
{
msg = c.getClass().getName();
}
}
if (c != null && c instanceof CancelableTask)
{
MainWindow.logger.warning("+++Cancelling " + msg + " task because taking too long");
((CancelableTask) c).setCancelTask(true);
StackTraceElement[] stackTrace = thread.getStackTrace();
Errors.addError("Cancelled " + msg + " because taken too long", stackTrace);
Counters.getErrors().getCounter().incrementAndGet();
if(stackTrace.length>0)
{
boolean isKnownProblem = false;
for(int i=0;i<stackTrace.length;i++)
{
if(
(stackTrace[i].getClassName().contains("CosineSimilarity")) ||
(stackTrace[i].getClassName().contains("com.jthink.songkong.fileloader.FileFilters"))
)
{
isKnownProblem=true;
break;
}
}
if(isKnownProblem)
{
MainWindow.logger.warning("+++Interrupting " + msg + " task because taking too long");
thread.interrupt();
try
{
Thread.sleep(WAIT_BEFORE_STOP);
}
catch (InterruptedException ie)
{
MainWindow.logger.warning("+++Interrupted TimeoutTask " + msg + " task because taking too long");
}
if(thread.isAlive())
{
MainWindow.logger.warning("+++Stopping CosineSimailarity task");
thread.stop();
}
}
}
}
}
}
}
public class AnalyserService
{
protected static final int BOUNDED_QUEUE_SIZE = 500;
protected String threadGroup;
public AnalyserService(String threadGroup)
{
this.threadGroup=threadGroup;
}
protected ExecutorService executorService;
protected void initExecutorService()
{
int workerSize = Runtime.getRuntime().availableProcessors();
executorService = new PausableExecutor(workerSize, workerSize,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(BOUNDED_QUEUE_SIZE),new SongKongThreadFactory(threadGroup));
}
public ExecutorService getExecutorService()
{
if (executorService == null || executorService.isShutdown())
{
initExecutorService();
}
return executorService;
}
/** Submit and return immediately
*
* @param task
*/
public void submit(Callable<Boolean> task) //throws Exception
{
executorService.submit(task);
}
}
public class AnalyserServiceWithTimeout extends AnalyserService
{
private static final int TIMEOUT_PER_TASK = 30;
public AnalyserServiceWithTimeout(String threadGroup)
{
super(threadGroup);
}
@Override
protected void initExecutorService()
{
int workerSize = Runtime.getRuntime().availableProcessors();
executorService = new TimeoutThreadPoolExecutor(workerSize,
new SongKongThreadFactory(threadGroup),
new LinkedBlockingQueue<Runnable>(BOUNDED_QUEUE_SIZE),
TIMEOUT_PER_TASK,
TimeUnit.MINUTES);
}
}
package com.jthink.songkong.analyse.analyser;
import com.google.common.base.Strings;
import com.jthink.songkong.analyse.general.Errors;
import com.jthink.songkong.cmdline.SongKong;
import com.jthink.songkong.db.SongCache;
import com.jthink.songkong.match.MetadataGatherer;
import com.jthink.songkong.preferences.UserPreferences;
import com.jthink.songkong.ui.MainWindow;
import com.jthink.songkong.util.SongKongThreadGroup;
import com.jthink.songlayer.Song;
import com.jthink.songlayer.hibernate.HibernateUtil;
import org.hibernate.Session;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.logging.Level;
/**
* Try and match songs to acoustid only first as a starting point
*
* Use when we have no or little metadata
*/
public class SongPrematcherMatcher extends CancelableTask implements Callable<Boolean> {
private static PipelineCount pipelineCount = new PipelineCount();
public static int getPipelineQueuedCount()
{
return pipelineCount.getQueuedCount();
}
public static int getPipelineCallCount()
{
return pipelineCount.getCallCount();
}
public static void resetPipelineCount()
{
pipelineCount.resetCounts();
}
public static int getPipelineFileCount()
{
return pipelineCount.getFileCount();
}
public static int getPipelineCompletedCount()
{
return pipelineCount.getCompletedCount();
}
private static AnalyserService analyserService = new AnalyserServiceWithTimeout(SongKongThreadGroup.THREAD_PREMATCHER_WORKER);
private Session session;
private SongGroup songGroup;
public SongGroup getSongGroup()
{
return songGroup;
}
public SongPrematcherMatcher(SongGroup songGroup)
{
SongKong.logger.severe("Queue:"+ songGroup.getKey());
pipelineCount.incQueuedCount();
pipelineCount.incFileCount(songGroup.getSongIds().size());
this.songGroup = songGroup;
}
public static ExecutorService getExecutorService()
{
return analyserService.getExecutorService();
}
public static AnalyserService getService()
{
return analyserService;
}
public Boolean call()
{
try
{
SongKong.logger.severe("Start:" + songGroup.getKey());
if (SongKong.isStopTask() || isCancelTask())
{
return false;
}
SongKong.checkIn();
pipelineCount.incCallCount();
session = HibernateUtil.beginTransaction();
AnalysisStats stats = new AnalysisStats();
List<Song> songs = SongCache.loadSongsFromDatabase(session, songGroup.getSongIds());
//Try to match acoustid this should allow more to be grouped and matched by metadata on first pass
try
{
new RecordingOnlyMatcher().matchRecordingsOnlyByAcoustid(session, songGroup, songs, stats);
}
catch(Exception ex)
{
MainWindow.logger.log(Level.SEVERE, Strings.nullToEmpty(ex.getMessage()), ex);
Errors.addError(Strings.nullToEmpty(ex.getMessage()));
}
session.getTransaction().commit();
HibernateUtil.closeSession(session);
processSongsWithNewMetadata(songGroup, songs);
pipelineCount.incCompletedCount();
SongKong.logger.severe("Finish:" + songGroup.getKey());
return true;
}
catch (Exception e)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + e.getMessage(), e);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
catch (Error e)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + e.getMessage(), e);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
catch (Throwable t)
{
SongKong.logger.severe("FinishFail:" + songGroup.getKey());
MainWindow.logger.log(Level.SEVERE, "SongPrematcherMatcher:" + t.getMessage(), t);
if (session.getTransaction() != null)
{
session.getTransaction().rollback();
}
return false;
}
finally
{
if(session.isOpen())
{
session.getTransaction().commit();
HibernateUtil.closeSession(session);
}
}
}
private boolean processSongsWithNewMetadata(SongGroup songGroup, List<Song> songs)
{
MainWindow.logger.info("Prematcher:" + songGroup.getKey() + ":totalcount:" + songs.size());
int count = 0;
//Group based on actual metadata only
MetadataGatherer mg = new MetadataGatherer(songs);
for (String album : mg.getAlbums().keySet())
{
List<Song> songsInGrouping = mg.getAlbums().get(album);
count+=songsInGrouping.size();
MainWindow.logger.warning("Prematcher:" + songGroup.getKey() + ":" + album + ":count:" + songsInGrouping.size());
SongGroup sg = SongGroup.createSongGroupForSongs(songGroup, songsInGrouping);
sg.setRandomFolderNoMetadata(false);
sg.setRandomFolder(false);
processRandomFolder(sg, songsInGrouping);
}
List<Song> songsWithNoInfo = new ArrayList<>(mg.getSongsWithNoRelease());
if(songsWithNoInfo.size()>0)
{
count+=songsWithNoInfo.size();
SongGroup sgWithNoInfo = SongGroup.createSongGroupForSongs(songGroup, songsWithNoInfo);
MainWindow.logger.warning("Prematcher:" + songGroup.getKey() + ":NoMetadata:" + ":count:" + songsWithNoInfo.size());
processRandomFolderNoMetadata(sgWithNoInfo, songsWithNoInfo);
}
if(count<songs.size())
{
MainWindow.logger.warning(songGroup.getKey()+":Not all songs have been processed"+songs.size());
Errors.addErrorWithoutStackTrace(songGroup.getKey()+":Not all songs have been processed:"+songs.size());
}
return true;
}
private boolean processRandomFolder(SongGroup songGroup, List<Song> songs)
{
if(UserPreferences.getInstance().isSearchMusicBrainz())
{
MusicBrainzMetadataMatcher.getService().submit(new MusicBrainzMetadataMatcher(songGroup));
}
else if(UserPreferences.getInstance().isSearchDiscogs())
{
if(songGroup.getSubSongGroups().size() > 1)
{
DiscogsMultiFolderSongGroupMatcher.getService().submit(new DiscogsMultiFolderSongGroupMatcher(songGroup));
}
else if(songGroup.getSongIds().size()==1)
{
DiscogsSongMatcher.getService().submit(new DiscogsSongMatcher(songGroup, songGroup.getSongIds().get(0)));
}
else
{
DiscogsSongGroupMatcher.getService().submit(new DiscogsSongGroupMatcher(songGroup));
}
}
else
{
for (Integer songId : songGroup.getSongIds())
{
SongSaver.getService().submit(new SongSaver(songId));
}
}
return true;
}
/**
* Process a group of files that are in a Random folder and dont seem to have anything in common so should not be grouped
* together.
*
* @param songGroup
* @param songs
* @return
*/
private boolean processRandomFolderNoMetadata(SongGroup songGroup, List<Song> songs)
{
if(UserPreferences.getInstance().isSearchMusicBrainz())
{
for (Song song : songs)
{
MusicBrainzSongMatcher.getService().submit(new MusicBrainzSongMatcher(songGroup, song.getRecNo()));
}
}
else if(UserPreferences.getInstance().isSearchDiscogs())
{
for (Song song : songs)
{
DiscogsSongMatcher.getService().submit(new DiscogsSongMatcher(songGroup, song.getRecNo()));
}
}
else
{
for (Integer songId : songGroup.getSongIds())
{
SongSaver.getService().submit(new SongSaver(songId));
}
}
return true;
}
}
解决方法:
我认为这是愚蠢但无法找到它,但问题是我将任务提交给不正确的ExecutorService
在AcoustidMatcher我有
private boolean processFolderWithPoorMetadata(SongGroup songGroup)
{
MusicBrainzMetadataMatcher.getService().submit(new SongPrematcherMatcher(songGroup));
return true;
}
什么时候我应该
private boolean processFolderWithPoorMetadata(SongGroup songGroup)
{
SongPrematcherMatcher.getService().submit(new SongPrematcherMatcher(songGroup));
return true;
}
所以这意味着所有的SongPrematcher任务都被提交给了错误的ExecutorService,因此当我请求关闭SongPrematcher ExecutorService时,它可以立即关闭！