#3170

Committed 22 Dec 2023 02:10PM CUT coverage: 35.199% (-0.003%) from 35.202%

Build # #3170

Build Type

push

github

Committed by

topframe

Commit Message

[maven-release-plugin] prepare release v7.4.3

Run Details

13199 of 37498 relevant lines covered (35.2%)

0.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.86

/daemon/src/main/java/com/aspectran/daemon/command/CommandExecutor.java

/*
 * Copyright (c) 2008-2023 The Aspectran Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.aspectran.daemon.command;

import com.aspectran.core.context.config.DaemonExecutorConfig;
import com.aspectran.core.util.ExceptionUtils;
import com.aspectran.core.util.logging.Logger;
import com.aspectran.core.util.logging.LoggerFactory;
import com.aspectran.daemon.Daemon;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class CommandExecutor {

    private static final Logger logger = LoggerFactory.getLogger(CommandExecutor.class);

    private static final int DEFAULT_MAX_THREADS = Runtime.getRuntime().availableProcessors();

    private final Daemon daemon;

    private final int maxThreads;

    private final ExecutorService executorService;

    private final AtomicInteger queueSize = new AtomicInteger();

    private final AtomicBoolean isolated = new AtomicBoolean();

    public CommandExecutor(Daemon daemon, DaemonExecutorConfig executorConfig) {
        if (daemon == null) {
            throw new IllegalArgumentException("daemon must not be null");
        }

        this.daemon = daemon;

        if (executorConfig != null) {
            this.maxThreads = executorConfig.getMaxThreads(DEFAULT_MAX_THREADS);
        } else {
            this.maxThreads = DEFAULT_MAX_THREADS;
        }

        BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();
        this.executorService = new ThreadPoolExecutor(
                1,
                maxThreads,
                180L,
                TimeUnit.SECONDS,
                workQueue
        );
    }

    public int getAvailableThreads() {
        return maxThreads - queueSize.get();
    }

    public boolean execute(final CommandParameters parameters, final Callback callback) {
        final String commandName = parameters.getCommandName();

        if (isolated.get()) {
            if (logger.isDebugEnabled()) {
                logger.debug("Holds '" + commandName + "' command until the end of the command " +
                        "requiring a single execution guarantee.");
            }
            return false;
        }

        Command command = daemon.getCommandRegistry().getCommand(commandName);
        if (command == null) {
            parameters.setResult("No command mapped to '" + commandName + "'");
            try {
                callback.failure();
            } catch (Exception e) {
                logger.error("Failed to execute callback", e);
            }
            return false;
        }

        if (command.isIsolated() && queueSize.get() > 0) {
            if (logger.isDebugEnabled()) {
                logger.debug("'" + commandName + "' command requires a single execution guarantee, " +
                        "so it is held until another command completes");
            }
            return false;
        }

        if (daemon.getDaemonService() != null) {
            // DefaultActivity will always be specified here
            parameters.setActivity(daemon.getDaemonService().getActivityContext().getAvailableActivity());
        }

        Runnable runnable = () -> {
            Thread currentThread = Thread.currentThread();
            String oldThreadName = currentThread.getName();
            try {
                String threadName = "cmd-" + commandName + "-" + queueSize;
                currentThread.setName(threadName);

                if (command.isIsolated()) {
                    isolated.set(true);
                }

                boolean success = execute(command, parameters);
                try {
                    if (success) {
                        callback.success();
                    } else {
                        callback.failure();
                    }
                } catch (Exception e) {
                    logger.error("Failed to execute callback", e);
                }
            } finally {
                currentThread.setName(oldThreadName);
                isolated.compareAndSet(true, false);
                queueSize.decrementAndGet();
            }
        };

        queueSize.incrementAndGet();
        try {
            executorService.execute(runnable);
            return true;
        } catch (RejectedExecutionException e) {
            logger.error("Failed to execute command", e);
            queueSize.decrementAndGet();
            return false;
        }
    }

    private boolean execute(Command command, CommandParameters parameters) {
        try {
            CommandResult commandResult = command.execute(parameters);
            if (commandResult.isSuccess()) {
                parameters.setResult(commandResult.getResult());
                return true;
            } else {
                parameters.setResult("[FAILED] " + commandResult.getResult());
                return false;
            }
        } catch (Exception e) {
            logger.error("Error executing daemon command " + command, e);
            parameters.setResult("[FAILED] Error executing daemon command " + command +
                    System.lineSeparator() + ExceptionUtils.getStacktrace(e));
            return false;
        }
    }

    public void shutdown() {
        if (logger.isDebugEnabled()) {
            logger.debug("Shutting down executor...");
        }

        executorService.shutdown();
        if (!executorService.isTerminated()) {
            while (true) {
                logger.info("Waiting for executor to terminate...");
                if (executorService.isTerminated()) {
                    break;
                }
                try {
                    if (executorService.awaitTermination(3000L, TimeUnit.MILLISECONDS)) {
                        break;
                    }
                } catch (InterruptedException ignored) {
                    break;
                }
            }
        }
    }

    public interface Callback {

        void success();

        void failure();

    }

}

1	/*
2	* Copyright (c) 2008-2023 The Aspectran Project
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16	package com.aspectran.daemon.command;
17
18	import com.aspectran.core.context.config.DaemonExecutorConfig;
19	import com.aspectran.core.util.ExceptionUtils;
20	import com.aspectran.core.util.logging.Logger;
21	import com.aspectran.core.util.logging.LoggerFactory;
22	import com.aspectran.daemon.Daemon;
23
24	import java.util.concurrent.BlockingQueue;
25	import java.util.concurrent.ExecutorService;
26	import java.util.concurrent.RejectedExecutionException;
27	import java.util.concurrent.SynchronousQueue;
28	import java.util.concurrent.ThreadPoolExecutor;
29	import java.util.concurrent.TimeUnit;
30	import java.util.concurrent.atomic.AtomicBoolean;
31	import java.util.concurrent.atomic.AtomicInteger;
32
33	public class CommandExecutor {
34
35	private static final Logger logger = LoggerFactory.getLogger(CommandExecutor.class);	1✔
36
37	private static final int DEFAULT_MAX_THREADS = Runtime.getRuntime().availableProcessors();	1✔
38
39	private final Daemon daemon;
40
41	private final int maxThreads;
42
43	private final ExecutorService executorService;
44
45	private final AtomicInteger queueSize = new AtomicInteger();	1✔
46
47	private final AtomicBoolean isolated = new AtomicBoolean();	1✔
48
49	public CommandExecutor(Daemon daemon, DaemonExecutorConfig executorConfig) {	1✔
50	if (daemon == null) {	1✔
51	throw new IllegalArgumentException("daemon must not be null");	×
52	}
53
54	this.daemon = daemon;	1✔
55
56	if (executorConfig != null) {	1✔
57	this.maxThreads = executorConfig.getMaxThreads(DEFAULT_MAX_THREADS);	1✔
58	} else {
59	this.maxThreads = DEFAULT_MAX_THREADS;	×
60	}
61
62	BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();	1✔
63	this.executorService = new ThreadPoolExecutor(	1✔
64	1,
65	maxThreads,
66	180L,
67	TimeUnit.SECONDS,
68	workQueue
69	);
70	}	1✔
71
72	public int getAvailableThreads() {
73	return maxThreads - queueSize.get();	1✔
74	}
75
76	public boolean execute(final CommandParameters parameters, final Callback callback) {
77	final String commandName = parameters.getCommandName();	1✔
78
79	if (isolated.get()) {	1✔
80	if (logger.isDebugEnabled()) {	×
81	logger.debug("Holds '" + commandName + "' command until the end of the command " +	×
82	"requiring a single execution guarantee.");
83	}
84	return false;	×
85	}
86
87	Command command = daemon.getCommandRegistry().getCommand(commandName);	1✔
88	if (command == null) {	1✔
89	parameters.setResult("No command mapped to '" + commandName + "'");	×
90	try {
91	callback.failure();	×
92	} catch (Exception e) {	×
93	logger.error("Failed to execute callback", e);	×
94	}	×
95	return false;	×
96	}
97
98	if (command.isIsolated() && queueSize.get() > 0) {	1✔
99	if (logger.isDebugEnabled()) {	1✔
100	logger.debug("'" + commandName + "' command requires a single execution guarantee, " +	1✔
101	"so it is held until another command completes");
102	}
103	return false;	1✔
104	}
105
106	if (daemon.getDaemonService() != null) {	1✔
107	// DefaultActivity will always be specified here
108	parameters.setActivity(daemon.getDaemonService().getActivityContext().getAvailableActivity());	1✔
109	}
110
111	Runnable runnable = () -> {	1✔
112	Thread currentThread = Thread.currentThread();	1✔
113	String oldThreadName = currentThread.getName();	1✔
114	try {
115	String threadName = "cmd-" + commandName + "-" + queueSize;	1✔
116	currentThread.setName(threadName);	1✔
117
118	if (command.isIsolated()) {	1✔
119	isolated.set(true);	×
120	}
121
122	boolean success = execute(command, parameters);	1✔
123	try {
124	if (success) {	1✔
125	callback.success();	1✔
126	} else {
127	callback.failure();	×
128	}
129	} catch (Exception e) {	×
130	logger.error("Failed to execute callback", e);	×
131	}	1✔
132	} finally {
133	currentThread.setName(oldThreadName);	1✔
134	isolated.compareAndSet(true, false);	1✔
135	queueSize.decrementAndGet();	1✔
136	}
137	};	1✔
138
139	queueSize.incrementAndGet();	1✔
140	try {
141	executorService.execute(runnable);	1✔
142	return true;	1✔
143	} catch (RejectedExecutionException e) {	×
144	logger.error("Failed to execute command", e);	×
145	queueSize.decrementAndGet();	×
146	return false;	×
147	}
148	}
149
150	private boolean execute(Command command, CommandParameters parameters) {
151	try {
152	CommandResult commandResult = command.execute(parameters);	1✔
153	if (commandResult.isSuccess()) {	1✔
154	parameters.setResult(commandResult.getResult());	1✔
155	return true;	1✔
156	} else {
157	parameters.setResult("[FAILED] " + commandResult.getResult());	×
158	return false;	×
159	}
160	} catch (Exception e) {	×
161	logger.error("Error executing daemon command " + command, e);	×
162	parameters.setResult("[FAILED] Error executing daemon command " + command +	×
163	System.lineSeparator() + ExceptionUtils.getStacktrace(e));	×
164	return false;	×
165	}
166	}
167
168	public void shutdown() {
169	if (logger.isDebugEnabled()) {	1✔
170	logger.debug("Shutting down executor...");	1✔
171	}
172
173	executorService.shutdown();	1✔
174	if (!executorService.isTerminated()) {	1✔
175	while (true) {
176	logger.info("Waiting for executor to terminate...");	1✔
177	if (executorService.isTerminated()) {	1✔
178	break;	×
179	}
180	try {
181	if (executorService.awaitTermination(3000L, TimeUnit.MILLISECONDS)) {	1✔
182	break;	1✔
183	}
184	} catch (InterruptedException ignored) {	×
185	break;	×
186	}	×
187	}
188	}
189	}	1✔
190
191	public interface Callback {
192
193	void success();
194
195	void failure();
196
197	}
198
199	}

aspectran / aspectran / #3170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous