dcache-xrootd: add ability to override default timeout for server res…

…ponse (TPC)

Motivation:

Support the changes made in #11712
(master@19457a65a78ed8a26029adbcc38e15bb413cd6e0).

Modification:

Add property for default third-party-copy response timeout from server.
Add field on pool mover xrootd transfer service.
Set this value when creating the TPC client.
Add command to change this value through the admin interface.

Result:

Ability to adjust, if necessary, the default timeout for
responses from the server to the third-party client.

Target: master
Request: 5.1
Request: 5.0
Request: 4.2
Requires-notes: yes
Requires-book: yes
Acked-by: Tigran

docs/TheBook/src/main/markdown/config-xrootd.md

@@ -309,6 +309,25 @@ as TPC source), the third-party server needs only a valid certificate issued
 by a recognized CA; anonymous read access is granted to files (even privately
 owned) on the basis of the rendezvous token submitted with the request.
 
+#### Client timeout control
+
+The Third-party embedded client has a timer which will interrupt and return
+an error if the response from the server does not arrive after a given
+amount of time.
+
+The default values for this can be controlled by the properties:
+
+    pool.mover.xrootd.tpc-server-response-timeout
+    pool.mover.xrootd.tpc-server-response-timeout.unit
+
+These are set to 2 seconds to match the aggressive behavior of the SLAC
+implementation.  However, dCache allows you to control this dynamically
+as well, using the admin command:
+
+    \s <xrootd-door> xrootd set server response timeout
+
+This could conceivably be necessary under heavier load.
+
 
 ### Signed hash verification support
 
@@ -345,6 +364,7 @@ pool configurations:
 
             pool.mover.xrootd.tpc-authn-plugins=gsi,unix
 
+
 <!--  [???]: #intouch-web
   []: http://people.web.psi.ch/feichtinger/doc/authz.pdf
   [1]: #cf-gplazma

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/pool/XrootdTransferService.java

@@ -29,6 +29,7 @@
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
+import javax.annotation.Resource;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -37,13 +38,18 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.Callable;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 
 import diskCacheV111.util.CacheException;
 
+import dmg.cells.nucleus.CellCommandListener;
 import dmg.cells.nucleus.CellPath;
+import dmg.util.command.Argument;
+import dmg.util.command.Command;
+import dmg.util.command.Option;
 
 import org.dcache.pool.movers.NettyMover;
 import org.dcache.pool.movers.NettyTransferService;
@@ -96,24 +102,58 @@
  *   third-party embedded clients.
  */
 public class XrootdTransferService extends NettyTransferService<XrootdProtocolInfo>
+                implements CellCommandListener
 {
     private static final Logger LOGGER =
             LoggerFactory.getLogger(XrootdTransferService.class);
 
-    private int                               maxFrameSize;
-    private List<ChannelHandlerFactory>       plugins;
-    private List<ChannelHandlerFactory>       accessLogPlugins;
-    private List<ChannelHandlerFactory>       tpcClientPlugins;
-    private Map<String, String>               queryConfig;
-    private NioEventLoopGroup                 thirdPartyClientGroup;
-    private ScheduledExecutorService          thirdPartyShutdownExecutor;
-    private SigningPolicy                     signingPolicy;
+    @Command(name = "xrootd set server response timeout",
+                    hint = "time in seconds a server has to reply "
+                                    + "to the third-party client",
+                    description = "Sets the timeout on the third-party client.  "
+                                    + "The default mirrors the aggressive "
+                                    + "behavior of the SLAC xrootd "
+                                    + "server; see the property ")
+    class TimeoutCommand implements Callable<String> {
+            @Argument(usage = "Timeout.")
+            Long timeout = 2L;
+
+            @Option(name = "unit",
+                    usage = "Time unit for the timeout.")
+            TimeUnit unit;
+
+            @Override
+            public String call() throws Exception {
+                tpcServerResponseTimeout = timeout;
+                if (unit != null) {
+                    tpcServerResponseTimeoutUnit = unit;
+                }
+                return "Timeout now set to " + getTpcServerResponseTimeoutInSeconds()
+                                + " seconds; this affects only future transfers, "
+                                + "not those currently running.";
+            }
+    }
+
+    private int                         maxFrameSize;
+    private List<ChannelHandlerFactory> plugins;
+    private List<ChannelHandlerFactory> accessLogPlugins;
+    private List<ChannelHandlerFactory> tpcClientPlugins;
+    private Map<String, String>         queryConfig;
+    private NioEventLoopGroup           thirdPartyClientGroup;
+    private ScheduledExecutorService    thirdPartyShutdownExecutor;
+    private SigningPolicy               signingPolicy;
+    private long                        tpcServerResponseTimeout;
+    private TimeUnit                    tpcServerResponseTimeoutUnit;
 
     public XrootdTransferService()
     {
         super("xrootd");
     }
 
+    public long getTpcServerResponseTimeoutInSeconds() {
+        return tpcServerResponseTimeoutUnit.toSeconds(tpcServerResponseTimeout);
+    }
+
     public NioEventLoopGroup getThirdPartyClientGroup()
     {
         return thirdPartyClientGroup;
@@ -157,6 +197,18 @@ public void setSigningPolicy(SigningPolicy signingPolicy)
         this.signingPolicy = signingPolicy;
     }
 
+    @Resource
+    public void setTpcServerResponseTimeout(long timeout)
+    {
+        this.tpcServerResponseTimeout = timeout;
+    }
+
+    @Resource
+    public void setTpcServerResponseTimeoutUnit(TimeUnit unit)
+    {
+        this.tpcServerResponseTimeoutUnit = unit;
+    }
+
     @Required
     public void setThirdPartyShutdownExecutor(
                     ScheduledExecutorService thirdPartyShutdownExecutor)

modules/dcache-xrootd/src/main/java/org/dcache/xrootd/tpc/TpcWriteDescriptor.java

@@ -138,6 +138,7 @@ public TpcWriteDescriptor(NettyTransferService<XrootdProtocolInfo>.NettyMoverCha
                                      info,
                                      this,
                                      service.getThirdPartyShutdownExecutor());
+        client.setResponseTimeout(service.getTpcServerResponseTimeoutInSeconds());
         group = service.getThirdPartyClientGroup();
         authPlugins = service.getTpcClientPlugins();
         isFirstSync = true;

modules/dcache/src/main/resources/org/dcache/pool/classic/pool.xml

@@ -395,7 +395,9 @@
           </bean>
       </property>
       <property name="thirdPartyShutdownExecutor" ref="workerThreadPool"/>
-    <property name="signingPolicy" ref="signing-policy"/>
+      <property name="signingPolicy" ref="signing-policy"/>
+      <property name="tpcServerResponseTimeout" value="${pool.mover.xrootd.tpc-server-response-timeout}"/>
+      <property name="tpcServerResponseTimeoutUnit" value="${pool.mover.xrootd.tpc-server-response-timeout.unit}"/>
   </bean>
 
   <bean id="http-transfer-service-parent" class="org.dcache.http.HttpTransferService"

pom.xml

@@ -67,7 +67,7 @@
         <version.smc>6.6.0</version.smc>
         <version.xerces>2.11.0</version.xerces>
         <version.jetty>9.4.12.v20180830</version.jetty>
-        <version.xrootd4j>3.4.1</version.xrootd4j>
+        <version.xrootd4j>3.4.2</version.xrootd4j>
         <version.jersey>2.26</version.jersey>
         <version.dcache-view>1.5.3</version.dcache-view>
         <version.netty>4.1.10.Final</version.netty>

skel/share/defaults/pool.properties

@@ -369,6 +369,15 @@ pool.mover.xrootd.timeout.idle = 300000
 pool.mover.xrootd.timeout.connect = 300
 (one-of?MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)pool.mover.xrootd.timeout.connect.unit = SECONDS
 
+#  ---- Xrootd tpc server response timeout
+#
+#   Timeout that the third-party client will wait for a response from a server
+#   before raising a timeout exception.  The
+#   aggressive default mirrors that of the xrootd server standard implementation.
+#
+pool.mover.xrootd.tpc-server-response-timeout = 2
+(one-of?MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)pool.mover.xrootd.tpc-server-response-timeout.unit = SECONDS
+
 #  ---- Xrootd mover port range
 pool.mover.xrootd.port.min = ${dcache.net.lan.port.min}
 pool.mover.xrootd.port.max = ${dcache.net.lan.port.max}

skel/share/services/pool.batch

@@ -33,6 +33,8 @@ check -strong pool.mover.xrootd.timeout.idle
 check -strong pool.mover.xrootd.timeout.idle.unit
 check -strong pool.mover.xrootd.timeout.connect
 check -strong pool.mover.xrootd.timeout.connect.unit
+check -strong pool.mover.xrootd.tpc-server-response-timeout
+check -strong pool.mover.xrootd.tpc-server-response-timeout.unit
 check -strong pool.mover.xrootd.frame-size
 check -strong pool.mover.xrootd.port.min
 check -strong pool.mover.xrootd.port.max