Conseq (#24)

* + conseq for async

* + changed default front/back buffer size
+ added log event processing concurrency config

pom.xml

@@ -31,7 +31,7 @@
 
     <groupId>io.github.elf4j</groupId>
     <artifactId>elf4j-engine</artifactId>
-    <version>7.0.5</version>
+    <version>8.0.0</version>
     <packaging>jar</packaging>
     <name>elf4j-engine</name>
     <description>A stand-alone Java log engine implementing the ELF4J (Easy Logging Facade for Java) API</description>
@@ -69,6 +69,11 @@
             <artifactId>elf4j</artifactId>
             <version>2.2.4</version>
         </dependency>
+        <dependency>
+            <groupId>io.github.q3769</groupId>
+            <artifactId>conseq4j</artifactId>
+            <version>20230128.20230506.0</version>
+        </dependency>
         <dependency>
             <groupId>com.dslplatform</groupId>
             <artifactId>dsl-json-java8</artifactId>

src/main/java/elf4j/engine/NativeLogger.java

@@ -28,17 +28,9 @@
 import elf4j.Level;
 import elf4j.Logger;
 import elf4j.engine.service.LogService;
-import lombok.EqualsAndHashCode;
 import lombok.NonNull;
-import lombok.Value;
 
 import javax.annotation.concurrent.ThreadSafe;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Function;
-
-import static java.util.stream.Collectors.toMap;
 
 /**
  * Any instance of this class is thread-safe; it can be safely used as static, instance, or local variables. However,
@@ -47,11 +39,7 @@
  * of variables.
  */
 @ThreadSafe
-@Value
 public class NativeLogger implements Logger {
-    private static final Map<Level, Map<String, NativeLogger>> NATIVE_LOGGERS =
-            EnumSet.allOf(Level.class).stream().collect(toMap(Function.identity(), l -> new HashMap<>()));
-
     /**
      * Name of this logger's "owner class" - the logging service client class that first requested this logger instance
      * via the {@link Logger#instance()} service access method. The owner class is usually the same as the "caller
@@ -66,9 +54,9 @@ public class NativeLogger implements Logger {
      * only the caller class name, this field will be used in liu of checking the stack trace; i.e. the stack trace
      * walking is needed only when more caller details (e.g. method name, file name, line number) are required.
      */
-    @NonNull String ownerClassName;
-    @NonNull Level level;
-    @EqualsAndHashCode.Exclude @NonNull LogService logService;
+    private final @NonNull String ownerClassName;
+    private final @NonNull Level level;
+    private final @NonNull NativeLoggerFactory nativeLoggerFactory;
 
     /**
      * Constructor only meant to be used by {@link NativeLoggerFactory} and this class itself
@@ -77,24 +65,30 @@ public class NativeLogger implements Logger {
      *         name of the owner class that requested this instance via the {@link Logger#instance()} method
      * @param level
      *         severity level of this logger instance
-     * @param logService
-     *         service delegate to do the logging
+     * @param nativeLoggerFactory
+     *         log service access point from this instance, not reloadable
      */
-    public NativeLogger(@NonNull String ownerClassName, @NonNull Level level, @NonNull LogService logService) {
+    public NativeLogger(@NonNull String ownerClassName,
+            @NonNull Level level,
+            @NonNull NativeLoggerFactory nativeLoggerFactory) {
         this.ownerClassName = ownerClassName;
         this.level = level;
-        this.logService = logService;
+        this.nativeLoggerFactory = nativeLoggerFactory;
     }
 
     @Override
     public NativeLogger atLevel(Level level) {
-        return this.level == level ? this : NATIVE_LOGGERS.get(level)
-                .computeIfAbsent(this.ownerClassName, k -> new NativeLogger(k, level, this.logService));
+        return this.level == level ? this : this.nativeLoggerFactory.getLogger(level, this.ownerClassName);
+    }
+
+    @Override
+    public @NonNull Level getLevel() {
+        return this.level;
     }
 
     @Override
     public boolean isEnabled() {
-        return this.logService.isEnabled(this);
+        return getLogService().isEnabled(this);
     }
 
     @Override
@@ -122,7 +116,21 @@ public void log(Throwable throwable, String message, Object... arguments) {
         this.service(throwable, message, arguments);
     }
 
+    /**
+     * @return directly callable log service, useful for log APIs than elf4j to use this engine
+     */
+    public LogService getLogService() {
+        return this.nativeLoggerFactory.getLogService();
+    }
+
+    /**
+     * @return owner/caller class of this logger instance
+     */
+    public @NonNull String getOwnerClassName() {
+        return this.ownerClassName;
+    }
+
     private void service(Throwable throwable, Object message, Object[] arguments) {
-        this.logService.log(this, NativeLogger.class, throwable, message, arguments);
+        getLogService().log(this, NativeLogger.class, throwable, message, arguments);
     }
 }

src/main/java/elf4j/engine/NativeLoggerFactory.java

@@ -33,8 +33,12 @@
 import elf4j.spi.LoggerFactory;
 import lombok.NonNull;
 
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.Function;
+
+import static java.util.stream.Collectors.toMap;
 
 /**
  *
@@ -44,15 +48,14 @@ public class NativeLoggerFactory implements LoggerFactory {
      * Default to TRACE for this native implementation
      */
     private static final Level DEFAULT_LOGGER_SEVERITY_LEVEL = Level.INFO;
-    private static final Class<?> LOGGING_SERVICE_ACCESS_CLASS = Logger.class;
     /**
      * Made injectable for extensions other than this native ELF4J implementation
      */
     @NonNull private final Level defaultLoggerLevel;
-    private final Map<String, NativeLogger> nativeLoggers = new HashMap<>();
+    private final Map<Level, Map<String, NativeLogger>> nativeLoggers;
     /**
-     * The class that the API client calls first to get a logger instance. The client caller class of this class will be
-     * the "owner class" of the logger instances this factory produces.
+     * The class or interface that the API client calls first to get a logger instance. The client caller class of this
+     * class will be the "owner class" of the logger instances this factory produces.
      * <p></p>
      * For this native implementation, the service access class is the {@link Logger} interface itself as the client
      * calls the static factory method {@link Logger#instance()} first to get a logger instance. If this library is used
@@ -66,12 +69,12 @@ public class NativeLoggerFactory implements LoggerFactory {
      * Default constructor required by {@link java.util.ServiceLoader}
      */
     public NativeLoggerFactory() {
-        this(LOGGING_SERVICE_ACCESS_CLASS);
+        this(Logger.class);
     }
 
     /**
      * @param serviceAccessClass
-     *         the class that the API client uses to obtain access to a logger instance
+     *         the class or interface that the API client application calls first to a logger instance
      */
     public NativeLoggerFactory(@NonNull Class<?> serviceAccessClass) {
         this(DEFAULT_LOGGER_SEVERITY_LEVEL, serviceAccessClass, new DispatchingLogService());
@@ -82,6 +85,8 @@ public NativeLoggerFactory(@NonNull Class<?> serviceAccessClass) {
             @NonNull LogService logService) {
         this.defaultLoggerLevel = defaultLoggerLevel;
         this.serviceAccessClass = serviceAccessClass;
+        this.nativeLoggers =
+                EnumSet.allOf(Level.class).stream().collect(toMap(Function.identity(), l -> new HashMap<>()));
         this.logService = logService;
     }
 
@@ -93,7 +98,15 @@ public NativeLoggerFactory(@NonNull Class<?> serviceAccessClass) {
      */
     @Override
     public NativeLogger logger() {
-        return this.nativeLoggers.computeIfAbsent(StackTraceUtils.callerOf(this.serviceAccessClass).getClassName(),
-                ownerClassName -> new NativeLogger(ownerClassName, this.defaultLoggerLevel, this.logService));
+        return getLogger(this.defaultLoggerLevel, StackTraceUtils.callerOf(this.serviceAccessClass).getClassName());
+    }
+
+    @NonNull LogService getLogService() {
+        return logService;
+    }
+
+    NativeLogger getLogger(Level level, String ownerClassName) {
+        return this.nativeLoggers.get(level)
+                .computeIfAbsent(ownerClassName, ownerClass -> new NativeLogger(ownerClass, level, this));
     }
 }

src/main/java/elf4j/engine/service/ConseqLogEventProcessor.java

@@ -0,0 +1,109 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2023 Qingtian Wang
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ */
+
+package elf4j.engine.service;
+
+import conseq4j.execute.ConseqExecutor;
+import conseq4j.execute.SequentialExecutor;
+import elf4j.Level;
+import elf4j.engine.service.configuration.LogServiceConfiguration;
+import elf4j.engine.service.util.PropertiesUtils;
+import elf4j.engine.service.writer.LogWriter;
+import elf4j.util.InternalLogger;
+import lombok.NonNull;
+import org.awaitility.Awaitility;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Log events are asynchronously processed, optionally by multiple concurrent threads. However, events issued by the
+ * same caller application thread are processed sequentially with the {@link ConseqExecutor} API. Thus, logs by
+ * different caller threads may arrive at the final destination (e.g. system Console or a log file) in any order;
+ * meanwhile, logs from the same caller thread will arrive sequentially in the same order as they are called in the
+ * orginal thread.
+ */
+public class ConseqLogEventProcessor implements LogEventProcessor {
+    private static final int DEFAULT_FRONT_BUFFER_CAPACITY = 256;
+    private static final int DEFAULT_CONCURRENCY = Runtime.getRuntime().availableProcessors();
+    private final LogWriter logWriter;
+    private final SequentialExecutor conseqExecutor;
+
+    private ConseqLogEventProcessor(LogWriter logWriter, SequentialExecutor conseqExecutor) {
+        this.logWriter = logWriter;
+        this.conseqExecutor = conseqExecutor;
+        LogServiceManager.INSTANCE.registerStop(this);
+    }
+
+    /**
+     * @param logServiceConfiguration
+     *         entire configuration
+     * @return conseq executor
+     */
+    public static ConseqLogEventProcessor from(LogServiceConfiguration logServiceConfiguration) {
+        Properties properties = logServiceConfiguration.getProperties();
+        Integer workQueueCapacity = getWorkQueueCapacity(properties);
+        InternalLogger.INSTANCE.log(Level.INFO, "Log event work queue capacity: " + workQueueCapacity);
+        Integer concurrency = getConcurrency(properties);
+        InternalLogger.INSTANCE.log(Level.INFO, "Log process concurrency: " + concurrency);
+        SequentialExecutor conseqExecutor =
+                new ConseqExecutor.Builder().concurrency(concurrency).workQueueCapacity(workQueueCapacity).build();
+        return new ConseqLogEventProcessor(logServiceConfiguration.getLogServiceWriter(), conseqExecutor);
+    }
+
+    @NonNull
+    private static Integer getConcurrency(Properties properties) {
+        Integer concurrency = PropertiesUtils.getAsInteger("concurrency", properties);
+        concurrency = concurrency == null ? DEFAULT_CONCURRENCY : concurrency;
+        if (concurrency < 1) {
+            throw new IllegalArgumentException("Unexpected concurrency: " + concurrency + ", cannot be less than 1");
+        }
+        return concurrency;
+    }
+
+    @NonNull
+    private static Integer getWorkQueueCapacity(Properties properties) {
+        Integer workQueueCapacity = PropertiesUtils.getAsInteger("buffer.front", properties);
+        workQueueCapacity = workQueueCapacity == null ? DEFAULT_FRONT_BUFFER_CAPACITY : workQueueCapacity;
+        if (workQueueCapacity < 0) {
+            throw new IllegalArgumentException("Unexpected buffer.front: " + workQueueCapacity);
+        }
+        if (workQueueCapacity == 0) {
+            workQueueCapacity = 1;
+        }
+        return workQueueCapacity;
+    }
+
+    @Override
+    public void process(LogEvent logEvent) {
+        conseqExecutor.execute(() -> logWriter.write(logEvent), logEvent.getCallerThread().getId());
+    }
+
+    @Override
+    public void stop() {
+        this.conseqExecutor.shutdown();
+        Awaitility.with().timeout(30, TimeUnit.MINUTES).await().until(this.conseqExecutor::isTerminated);
+    }
+}

src/main/java/elf4j/engine/service/DispatchingLogService.java

@@ -41,7 +41,7 @@ public class DispatchingLogService implements LogService {
      *
      */
     public DispatchingLogService() {
-        this(Configuration.INSTANCE);
+        this(new RefreshableLogServiceConfiguration());
     }
 
     DispatchingLogService(LogServiceConfiguration logServiceConfiguration) {
@@ -80,16 +80,9 @@ public void log(@NonNull NativeLogger nativeLogger,
         if (this.includeCallerDetail()) {
             logEventBuilder.callerStack(new Throwable().getStackTrace()).serviceInterfaceClass(serviceInterfaceClass);
         }
-        if (this.includeCallerThread()) {
-            Thread callerThread = Thread.currentThread();
-            logEventBuilder.callerThread(new LogEvent.ThreadValue(callerThread.getName(), callerThread.getId()));
-        }
-        this.logServiceConfiguration.getLogEventIntakeThread()
-                .execute(() -> this.logServiceConfiguration.getLogServiceWriter().write(logEventBuilder.build()));
-    }
-
-    private static class Configuration {
-        private static final LogServiceConfiguration INSTANCE = new RefreshableLogServiceConfiguration();
+        Thread callerThread = Thread.currentThread();
+        logEventBuilder.callerThread(new LogEvent.ThreadValue(callerThread.getName(), callerThread.getId()));
+        this.logServiceConfiguration.getLogEventProcessor().process(logEventBuilder.build());
     }
 }