#3783

Committed 20 Oct 2024 10:25AM CUT coverage: 34.033% (-0.01%) from 34.046%

Build # #3783

Build Type

push

github

Committed by

topframe

Commit Message

[maven-release-plugin] prepare release v8.1.5

Run Details

13336 of 39185 relevant lines covered (34.03%)

0.34 hits per line

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

62.79

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

/*
 * Copyright (c) 2008-2024 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.daemon.Daemon;
import com.aspectran.utils.ExceptionUtils;
import com.aspectran.utils.annotation.jsr305.NonNull;
import com.aspectran.utils.logging.Logger;
import com.aspectran.utils.logging.LoggerFactory;

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) {
        return execute(parameters, null);
    }

    public boolean execute(@NonNull 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 + "'");
            if (callback != null) {
                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);
                if (callback != null) {
                    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-2024 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.daemon.Daemon;
20	import com.aspectran.utils.ExceptionUtils;
21	import com.aspectran.utils.annotation.jsr305.NonNull;
22	import com.aspectran.utils.logging.Logger;
23	import com.aspectran.utils.logging.LoggerFactory;
24
25	import java.util.concurrent.BlockingQueue;
26	import java.util.concurrent.ExecutorService;
27	import java.util.concurrent.RejectedExecutionException;
28	import java.util.concurrent.SynchronousQueue;
29	import java.util.concurrent.ThreadPoolExecutor;
30	import java.util.concurrent.TimeUnit;
31	import java.util.concurrent.atomic.AtomicBoolean;
32	import java.util.concurrent.atomic.AtomicInteger;
33
34	public class CommandExecutor {
35
36	private static final Logger logger = LoggerFactory.getLogger(CommandExecutor.class);	1✔
37
38	private static final int DEFAULT_MAX_THREADS = Runtime.getRuntime().availableProcessors();	1✔
39
40	private final Daemon daemon;
41
42	private final int maxThreads;
43
44	private final ExecutorService executorService;
45
46	private final AtomicInteger queueSize = new AtomicInteger();	1✔
47
48	private final AtomicBoolean isolated = new AtomicBoolean();	1✔
49
50	public CommandExecutor(Daemon daemon, DaemonExecutorConfig executorConfig) {	1✔
51	if (daemon == null) {	1✔
52	throw new IllegalArgumentException("daemon must not be null");	×
53	}
54
55	this.daemon = daemon;	1✔
56
57	if (executorConfig != null) {	1✔
58	this.maxThreads = executorConfig.getMaxThreads(DEFAULT_MAX_THREADS);	1✔
59	} else {
60	this.maxThreads = DEFAULT_MAX_THREADS;	×
61	}
62
63	BlockingQueue<Runnable> workQueue = new SynchronousQueue<>();	1✔
64	this.executorService = new ThreadPoolExecutor(	1✔
65	1,
66	maxThreads,
67	180L,
68	TimeUnit.SECONDS,
69	workQueue
70	);
71	}	1✔
72
73	public int getAvailableThreads() {
74	return (maxThreads - queueSize.get());	1✔
75	}
76
77	public boolean execute(final CommandParameters parameters) {
78	return execute(parameters, null);	×
79	}
80
81	public boolean execute(@NonNull final CommandParameters parameters, final Callback callback) {
82	final String commandName = parameters.getCommandName();	1✔
83
84	if (isolated.get()) {	1✔
85	if (logger.isDebugEnabled()) {	×
86	logger.debug("Holds '" + commandName + "' command until the end of the command " +	×
87	"requiring a single execution guarantee.");
88	}
89	return false;	×
90	}
91
92	Command command = daemon.getCommandRegistry().getCommand(commandName);	1✔
93	if (command == null) {	1✔
94	parameters.setResult("No command mapped to '" + commandName + "'");	×
95	if (callback != null) {	×
96	try {
97	callback.failure();	×
98	} catch (Exception e) {	×
99	logger.error("Failed to execute callback", e);	×
100	}	×
101	}
102	return false;	×
103	}
104
105	if (command.isIsolated() && queueSize.get() > 0) {	1✔
106	if (logger.isDebugEnabled()) {	1✔
107	logger.debug("'" + commandName + "' command requires a single execution guarantee, " +	1✔
108	"so it is held until another command completes");
109	}
110	return false;	1✔
111	}
112
113	if (daemon.getDaemonService() != null) {	1✔
114	// DefaultActivity will always be specified here
115	parameters.setActivity(daemon.getDaemonService().getActivityContext().getAvailableActivity());	1✔
116	}
117
118	Runnable runnable = () -> {	1✔
119	Thread currentThread = Thread.currentThread();	1✔
120	String oldThreadName = currentThread.getName();	1✔
121	try {
122	String threadName = "cmd-" + commandName + "-" + queueSize;	1✔
123	currentThread.setName(threadName);	1✔
124
125	if (command.isIsolated()) {	1✔
126	isolated.set(true);	×
127	}
128
129	boolean success = execute(command, parameters);	1✔
130	if (callback != null) {	1✔
131	try {
132	if (success) {	1✔
133	callback.success();	1✔
134	} else {
135	callback.failure();	×
136	}
137	} catch (Exception e) {	×
138	logger.error("Failed to execute callback", e);	×
139	}	1✔
140	}
141	} finally {
142	currentThread.setName(oldThreadName);	1✔
143	isolated.compareAndSet(true, false);	1✔
144	queueSize.decrementAndGet();	1✔
145	}
146	};	1✔
147
148	queueSize.incrementAndGet();	1✔
149	try {
150	executorService.execute(runnable);	1✔
151	return true;	1✔
152	} catch (RejectedExecutionException e) {	×
153	logger.error("Failed to execute command", e);	×
154	queueSize.decrementAndGet();	×
155	return false;	×
156	}
157	}
158
159	private boolean execute(Command command, CommandParameters parameters) {
160	try {
161	CommandResult commandResult = command.execute(parameters);	1✔
162	if (commandResult.isSuccess()) {	1✔
163	parameters.setResult(commandResult.getResult());	1✔
164	return true;	1✔
165	} else {
166	parameters.setResult("[FAILED] " + commandResult.getResult());	×
167	return false;	×
168	}
169	} catch (Exception e) {	×
170	logger.error("Error executing daemon command " + command, e);	×
171	parameters.setResult("[FAILED] Error executing daemon command " + command +	×
172	System.lineSeparator() + ExceptionUtils.getStacktrace(e));	×
173	return false;	×
174	}
175	}
176
177	public void shutdown() {
178	if (logger.isDebugEnabled()) {	1✔
179	logger.debug("Shutting down executor...");	1✔
180	}
181
182	executorService.shutdown();	1✔
183	if (!executorService.isTerminated()) {	1✔
184	while (true) {
185	logger.info("Waiting for executor to terminate...");	1✔
186	if (executorService.isTerminated()) {	1✔
187	break;	×
188	}
189	try {
190	if (executorService.awaitTermination(3000L, TimeUnit.MILLISECONDS)) {	1✔
191	break;	1✔
192	}
193	} catch (InterruptedException ignored) {	×
194	break;	×
195	}	×
196	}
197	}
198	}	1✔
199
200	public interface Callback {
201
202	void success();
203
204	void failure();
205
206	}
207
208	}

aspectran / aspectran / #3783

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