Set leader-follower strategy in grizzly if selected in oncrpc4j.

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.

Signed-off-by: John Calcote <[email protected]>
(cherry picked from commit e50ceda)
Signed-off-by: Tigran Mkrtchyan <[email protected]>

Author: John Calcote

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/grizzly/GrizzlyUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2018 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2023 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -19,19 +19,21 @@
  */
 package org.dcache.oncrpc4j.grizzly;
 
+import org.dcache.oncrpc4j.rpc.IoStrategy;
 import org.dcache.oncrpc4j.rpc.RpcMessageParserTCP;
 import org.dcache.oncrpc4j.rpc.RpcMessageParserUDP;
 import org.dcache.oncrpc4j.rpc.net.IpProtocolType;
-import org.dcache.oncrpc4j.rpc.IoStrategy;
 import org.glassfish.grizzly.IOStrategy;
 import org.glassfish.grizzly.Transport;
 import org.glassfish.grizzly.filterchain.Filter;
 import org.glassfish.grizzly.nio.transport.TCPNIOTransport;
 import org.glassfish.grizzly.nio.transport.UDPNIOTransport;
+import org.glassfish.grizzly.strategies.LeaderFollowerNIOStrategy;
+import org.glassfish.grizzly.strategies.SameThreadIOStrategy;
 import org.glassfish.grizzly.threadpool.ThreadPoolConfig;
 
-import static org.dcache.oncrpc4j.rpc.IoStrategy.*;
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.dcache.oncrpc4j.rpc.IoStrategy.WORKER_THREAD;
 
 /**
  * Class with utility methods for Grizzly
@@ -116,4 +118,23 @@ public static ThreadPoolConfig getSelectorPoolCfg(IoStrategy ioStrategy, String
 
         return poolCfg;
     }
+
+    /**
+     * Convert an oncrpc4j IoStrategy enum value into a grizzly NIOStrategy instance. Note that the only two that matter
+     * here are single-thread and leader-follower. Worker threads should not be used in the grizzly layer as they would
+     * just inject more context switching overhead to no benefit.
+     *
+     * @param ioStrategy the oncrpc4j IoStategy to map to an NIOStrategy instance.
+     * @return the matching NIOStrategy instance for the specified IoStrategy value.
+     */
+    public static IOStrategy getNIOStrategy(IoStrategy ioStrategy) {
+        switch (ioStrategy) {
+            case LEADER_FOLLOWER:
+                return LeaderFollowerNIOStrategy.getInstance();
+            case WORKER_THREAD:
+            case SAME_THREAD:
+            default:
+                return SameThreadIOStrategy.getInstance();
+        }
+    }
 }

oncrpc4j-core/src/main/java/org/dcache/oncrpc4j/rpc/OncRpcSvc.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2009 - 2019 Deutsches Elektronen-Synchroton,
+ * Copyright (c) 2009 - 2023 Deutsches Elektronen-Synchroton,
  * Member of the Helmholtz Association, (DESY), HAMBURG, GERMANY
  *
  * This library is free software; you can redistribute it and/or modify
@@ -19,14 +19,15 @@
  */
 package org.dcache.oncrpc4j.rpc;
 
+import org.dcache.oncrpc4j.grizzly.GrizzlyRpcTransport;
 import org.dcache.oncrpc4j.grizzly.StartTlsFilter;
-import org.dcache.oncrpc4j.rpc.net.IpProtocolType;
-import org.dcache.oncrpc4j.rpc.net.InetSocketAddresses;
-import org.dcache.oncrpc4j.rpc.gss.GssProtocolFilter;
-import org.dcache.oncrpc4j.rpc.gss.GssSessionManager;
 import org.dcache.oncrpc4j.portmap.GenericPortmapClient;
 import org.dcache.oncrpc4j.portmap.OncPortmapClient;
 import org.dcache.oncrpc4j.portmap.OncRpcPortmap;
+import org.dcache.oncrpc4j.rpc.gss.GssProtocolFilter;
+import org.dcache.oncrpc4j.rpc.gss.GssSessionManager;
+import org.dcache.oncrpc4j.rpc.net.InetSocketAddresses;
+import org.dcache.oncrpc4j.rpc.net.IpProtocolType;
 import org.glassfish.grizzly.CloseType;
 import org.glassfish.grizzly.Connection;
 import org.glassfish.grizzly.ConnectionProbe;
@@ -45,15 +46,17 @@
 import org.glassfish.grizzly.nio.transport.UDPNIOTransportBuilder;
 import org.glassfish.grizzly.ssl.SSLEngineConfigurator;
 import org.glassfish.grizzly.ssl.SSLFilter;
-import org.glassfish.grizzly.strategies.SameThreadIOStrategy;
 import org.glassfish.grizzly.threadpool.ThreadPoolConfig;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLParameters;
 import java.io.IOException;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -66,14 +69,11 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import javax.net.ssl.SSLContext;
+import java.util.stream.Collectors;
 
 import static com.google.common.base.Throwables.getRootCause;
 import static com.google.common.base.Throwables.propagateIfPossible;
-import java.net.SocketAddress;
-import java.util.stream.Collectors;
-import javax.net.ssl.SSLParameters;
-import org.dcache.oncrpc4j.grizzly.GrizzlyRpcTransport;
+import static org.dcache.oncrpc4j.grizzly.GrizzlyUtils.getNIOStrategy;
 import static org.dcache.oncrpc4j.grizzly.GrizzlyUtils.getSelectorPoolCfg;
 import static org.dcache.oncrpc4j.grizzly.GrizzlyUtils.rpcMessageReceiverFor;
 import static org.dcache.oncrpc4j.grizzly.GrizzlyUtils.transportFor;
@@ -149,7 +149,7 @@ public class OncRpcSvc {
             final TCPNIOTransport tcpTransport = TCPNIOTransportBuilder
                     .newInstance()
                     .setReuseAddress(true)
-                    .setIOStrategy(SameThreadIOStrategy.getInstance())
+                    .setIOStrategy(getNIOStrategy(ioStrategy))
                     .setSelectorThreadPoolConfig(selectorPoolConfig)
                     .setSelectorRunnersCount(selectorPoolConfig.getMaxPoolSize())
                     .build();
@@ -160,7 +160,7 @@ public class OncRpcSvc {
             final UDPNIOTransport udpTransport = UDPNIOTransportBuilder
                     .newInstance()
                     .setReuseAddress(true)
-                    .setIOStrategy(SameThreadIOStrategy.getInstance())
+                    .setIOStrategy(getNIOStrategy(ioStrategy))
                     .setSelectorThreadPoolConfig(selectorPoolConfig)
                     .setSelectorRunnersCount(selectorPoolConfig.getMaxPoolSize())
                     .build();