You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by "ddanielr (via GitHub)" <gi...@apache.org> on 2026/04/10 23:52:41 UTC
[PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
ddanielr opened a new pull request, #6311:
URL: https://github.com/apache/accumulo/pull/6311
This PR is part of the larger refactor work in #6291 and contains the first set of changes needed for #6291.
This PR moves the existing ThriftService Enum type to a new enum class in the rpc package.
This new RpcService enum is used instead of strings in the ThriftClientTypes constructor which links it to the ThriftProcessorTypes class.
This change now links the thrift classes to the enum as well as the ServiceLockData.
The Exec and ExecVoid interfaces are also moved from ThriftClientTypes to TServiceClient to simply that class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3082850462
##########
core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProtocol.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * https://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 org.apache.accumulo.core.rpc;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TMessageType;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolDecorator;
+
+public class AccumuloTMultiplexedProtocol extends TProtocolDecorator {
+ private final RpcService rpcService;
+
+ /**
+ * Wrap the specified protocol, allowing it to be used to communicate with a multiplexing server.
+ * The <code>rpcService</code> is required as it is written to the end of the message so that the
+ * multiplexing server can broker the function call to the proper service.
+ *
+ * @param protocol Your communication protocol of choice, e.g. <code>TBinaryProtocol</code>.
+ * @param rpcService The RpcService communicating via this protocol.
+ */
+ public AccumuloTMultiplexedProtocol(TProtocol protocol, RpcService rpcService) {
+ super(protocol);
+ this.rpcService = rpcService;
+ }
+
+ /**
+ * Appends the RpcService shortId to the message body.
+ *
+ * @param tMessage The original message.
+ * @throws TException Passed through from wrapped <code>TProtocol</code> instance.
+ */
+ @Override
+ public void writeMessageBegin(TMessage tMessage) throws TException {
+ if (tMessage.type == TMessageType.CALL || tMessage.type == TMessageType.ONEWAY) {
+ super.writeMessageBegin(new TMessage(tMessage.name, tMessage.type, tMessage.seqid));
Review Comment:
```suggestion
super.writeMessageBegin(tMessage);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3076099428
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java:
##########
@@ -21,79 +21,64 @@
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.manager.thrift.FateWorkerService;
-import org.apache.thrift.TException;
+import org.apache.accumulo.core.rpc.RpcService;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
-public abstract class ThriftClientTypes<C extends TServiceClient> {
+public class ThriftClientTypes<C extends TServiceClient> {
- public static final ClientServiceThriftClient CLIENT = new ClientServiceThriftClient("client");
+ public static final ClientServiceThriftClient CLIENT =
+ new ClientServiceThriftClient(RpcService.CLIENT);
public static final CompactorServiceThriftClient COMPACTOR =
- new CompactorServiceThriftClient("compactor");
+ new CompactorServiceThriftClient(RpcService.COMPACTOR);
public static final CompactionCoordinatorServiceThriftClient COORDINATOR =
- new CompactionCoordinatorServiceThriftClient("coordinator");
+ new CompactionCoordinatorServiceThriftClient(RpcService.COORDINATOR);
- public static final FateThriftClient FATE = new FateThriftClient("fate");
+ public static final FateThriftClient FATE = new FateThriftClient(RpcService.FATE_CLIENT);
- public static final GCMonitorServiceThriftClient GC = new GCMonitorServiceThriftClient("gc");
+ public static final GCMonitorServiceThriftClient GC =
+ new GCMonitorServiceThriftClient(RpcService.GC);
- public static final ManagerThriftClient MANAGER = new ManagerThriftClient("mgr");
+ public static final ManagerThriftClient MANAGER = new ManagerThriftClient(RpcService.MANAGER);
public static final TabletServerThriftClient TABLET_SERVER =
- new TabletServerThriftClient("tserver");
+ new TabletServerThriftClient(RpcService.TSERV);
public static final TabletScanClientServiceThriftClient TABLET_SCAN =
- new TabletScanClientServiceThriftClient("scan");
+ new TabletScanClientServiceThriftClient(RpcService.TABLET_SCAN);
public static final TabletIngestClientServiceThriftClient TABLET_INGEST =
- new TabletIngestClientServiceThriftClient("ingest");
+ new TabletIngestClientServiceThriftClient(RpcService.TABLET_INGEST);
public static final TabletManagementClientServiceThriftClient TABLET_MGMT =
- new TabletManagementClientServiceThriftClient("tablet");
+ new TabletManagementClientServiceThriftClient(RpcService.TABLET_MANAGEMENT);
Review Comment:
The RpcService enum could have a method on it that gives a short/multiplex name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3076208818
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java:
##########
@@ -41,15 +41,15 @@ public class TabletManagementClientServiceThriftClient extends ThriftClientTypes
LoggerFactory.getLogger(TabletManagementClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- public TabletManagementClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ public TabletManagementClientServiceThriftClient(RpcService service) {
+ super(service, new Client.Factory());
}
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
boolean preferCachedConnections) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.TABLET_MANAGEMENT);
+ warnedAboutTServersBeingDown, RpcService.TABLET_MANAGEMENT);
Review Comment:
Could call getService
```suggestion
warnedAboutTServersBeingDown, getService());
```
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java:
##########
@@ -36,15 +36,15 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
private static final Logger LOG = LoggerFactory.getLogger(ClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- ClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ ClientServiceThriftClient(RpcService service) {
+ super(service, new Client.Factory());
}
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
boolean preferCachedConnections) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.CLIENT);
+ warnedAboutTServersBeingDown, RpcService.CLIENT);
Review Comment:
```suggestion
warnedAboutTServersBeingDown, getService());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3096516000
##########
core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProcessor.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * https://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 org.apache.accumulo.core.rpc;
+
+import java.util.Objects;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.TMultiplexedProcessor;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TMessageType;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolDecorator;
+import org.apache.thrift.protocol.TProtocolException;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * AccumuloTMultiplexedProcessor is a {@link TMultiplexedProcessor} variant that uses the
+ * {@link AccumuloTMultiplexedProtocol} to register additional processors and then forward messages
+ * based on a single byte entry instead of having to create substrings of the original message.
+ */
+public class AccumuloTMultiplexedProcessor extends TMultiplexedProcessor {
+
+ // Support enough slots as RPC service types.
+ private final TProcessor[] PROCESSORS = new TProcessor[RpcService.values().length];
+
+ public void registerProcessor(RpcService service, TProcessor processor) {
+ int serviceID = Byte.toUnsignedInt(service.getShortId());
+ // Ensure that TProcessors are only registered once.
+ Preconditions.checkState(PROCESSORS[serviceID] == null,
+ "processor already exists with id " + serviceID);
+ PROCESSORS[serviceID] = Objects.requireNonNull(processor,
Review Comment:
This would fail if someone creates a new RpcService and does not use the next one up number. Not a show stopper, but something to be aware of. Might make sense to make this very clear in RpcService, that the maximum allowable value is the number of RpcService enum values - 1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3095527136
##########
core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProtocol.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * https://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 org.apache.accumulo.core.rpc;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TMessageType;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolDecorator;
+
+public class AccumuloTMultiplexedProtocol extends TProtocolDecorator {
+ private final RpcService rpcService;
+
+ /**
+ * Wrap the specified protocol, allowing it to be used to communicate with a multiplexing server.
+ * The <code>rpcService</code> is required as it is written to the end of the message so that the
+ * multiplexing server can broker the function call to the proper service.
+ *
+ * @param protocol Your communication protocol of choice, e.g. <code>TBinaryProtocol</code>.
+ * @param rpcService The RpcService communicating via this protocol.
+ */
+ public AccumuloTMultiplexedProtocol(TProtocol protocol, RpcService rpcService) {
+ super(protocol);
+ this.rpcService = rpcService;
+ }
+
+ /**
+ * Appends the RpcService shortId to the message body.
+ *
+ * @param tMessage The original message.
+ * @throws TException Passed through from wrapped <code>TProtocol</code> instance.
+ */
+ @Override
+ public void writeMessageBegin(TMessage tMessage) throws TException {
+ if (tMessage.type == TMessageType.CALL || tMessage.type == TMessageType.ONEWAY) {
+ super.writeMessageBegin(new TMessage(tMessage.name, tMessage.type, tMessage.seqid));
Review Comment:
good catch! Removed that in 8a419ff
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3095525398
##########
core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProcessor.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * https://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 org.apache.accumulo.core.rpc;
+
+import java.util.Objects;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.TMultiplexedProcessor;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TMessageType;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolDecorator;
+import org.apache.thrift.protocol.TProtocolException;
+
+/**
+ * AccumuloTMultiplexedProcessor is a {@link TMultiplexedProcessor} variant that uses the
+ * {@link AccumuloTMultiplexedProtocol} to register additional processors and then forward messages
+ * based on a single byte entry instead of having to create substrings of the original message.
+ */
+public class AccumuloTMultiplexedProcessor extends TMultiplexedProcessor {
+
+ // Support enough slots as RPC service types.
+ private final TProcessor[] PROCESSORS = new TProcessor[RpcService.values().length];
+
+ public void registerProcessor(RpcService service, TProcessor processor) {
+ PROCESSORS[Byte.toUnsignedInt(service.getShortId())] = Objects.requireNonNull(processor,
Review Comment:
Added that check in 8a419ff
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3072954054
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java:
##########
@@ -36,15 +36,15 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
private static final Logger LOG = LoggerFactory.getLogger(ClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- ClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ ClientServiceThriftClient(RpcService service) {
Review Comment:
See my comment in ThriftClientTypes regarding the service name and the multiplexed connection. I don't think the constructors for the ThriftClientTypes need to take an argument at this point, there is only one valid argument, RpcService.CLIENT.
The argument was a String in case we needed to change it, but things are working as expected. I think we just want to use something meaningful and short.
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java:
##########
@@ -21,79 +21,64 @@
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.manager.thrift.FateWorkerService;
-import org.apache.thrift.TException;
+import org.apache.accumulo.core.rpc.RpcService;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
-public abstract class ThriftClientTypes<C extends TServiceClient> {
+public class ThriftClientTypes<C extends TServiceClient> {
Review Comment:
```suggestion
public abstract class ThriftClientTypes<C extends TServiceClient> {
```
I think we want to leave this modifier here as it prevents direct usage and forces the user to use a subclass.
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java:
##########
@@ -21,79 +21,64 @@
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.manager.thrift.FateWorkerService;
-import org.apache.thrift.TException;
+import org.apache.accumulo.core.rpc.RpcService;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
-public abstract class ThriftClientTypes<C extends TServiceClient> {
+public class ThriftClientTypes<C extends TServiceClient> {
- public static final ClientServiceThriftClient CLIENT = new ClientServiceThriftClient("client");
+ public static final ClientServiceThriftClient CLIENT =
+ new ClientServiceThriftClient(RpcService.CLIENT);
public static final CompactorServiceThriftClient COMPACTOR =
- new CompactorServiceThriftClient("compactor");
+ new CompactorServiceThriftClient(RpcService.COMPACTOR);
public static final CompactionCoordinatorServiceThriftClient COORDINATOR =
- new CompactionCoordinatorServiceThriftClient("coordinator");
+ new CompactionCoordinatorServiceThriftClient(RpcService.COORDINATOR);
- public static final FateThriftClient FATE = new FateThriftClient("fate");
+ public static final FateThriftClient FATE = new FateThriftClient(RpcService.FATE_CLIENT);
- public static final GCMonitorServiceThriftClient GC = new GCMonitorServiceThriftClient("gc");
+ public static final GCMonitorServiceThriftClient GC =
+ new GCMonitorServiceThriftClient(RpcService.GC);
- public static final ManagerThriftClient MANAGER = new ManagerThriftClient("mgr");
+ public static final ManagerThriftClient MANAGER = new ManagerThriftClient(RpcService.MANAGER);
public static final TabletServerThriftClient TABLET_SERVER =
- new TabletServerThriftClient("tserver");
+ new TabletServerThriftClient(RpcService.TSERV);
public static final TabletScanClientServiceThriftClient TABLET_SCAN =
- new TabletScanClientServiceThriftClient("scan");
+ new TabletScanClientServiceThriftClient(RpcService.TABLET_SCAN);
public static final TabletIngestClientServiceThriftClient TABLET_INGEST =
- new TabletIngestClientServiceThriftClient("ingest");
+ new TabletIngestClientServiceThriftClient(RpcService.TABLET_INGEST);
public static final TabletManagementClientServiceThriftClient TABLET_MGMT =
- new TabletManagementClientServiceThriftClient("tablet");
+ new TabletManagementClientServiceThriftClient(RpcService.TABLET_MANAGEMENT);
Review Comment:
`getServiceName` is called from ThriftProcessortTypes when setting up the Multiplexed Thrift Server. The name is sent as part of the client request and when the server receives the request it passes the client request to the handler mapped to the name. Just providing some context for the short names ("tablet" vs "TABLET_MANAGEMENT") in the older code. My thought was to use something meaningful, but shorter rather than longer from a bytes-over-the-network perspective.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "dlmarion (via GitHub)" <gi...@apache.org>.
dlmarion commented on PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#issuecomment-4251791487
I feel like the changes in d41bfa7 might be overkill. My previous comments suggested that we keep the string short because it's passed in every message and used for directing traffic on the server side. The current values are several characters and the initial changes in this PR seemed like they could have made them longer. Going to a single byte might be the best in terms of performance, but is it much different than using a single character string? If we used a single character string, then we wouldn't need the new custom Multiplexer classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3081149410
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ThriftClientTypes.java:
##########
@@ -21,79 +21,64 @@
import static com.google.common.util.concurrent.Uninterruptibles.sleepUninterruptibly;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
-import org.apache.accumulo.core.client.AccumuloException;
-import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.clientImpl.ClientContext;
import org.apache.accumulo.core.manager.thrift.FateWorkerService;
-import org.apache.thrift.TException;
+import org.apache.accumulo.core.rpc.RpcService;
import org.apache.thrift.TServiceClient;
import org.apache.thrift.TServiceClientFactory;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
-public abstract class ThriftClientTypes<C extends TServiceClient> {
+public class ThriftClientTypes<C extends TServiceClient> {
Review Comment:
fixed in d41bfa7
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3082157433
##########
core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProcessor.java:
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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
+ *
+ * https://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 org.apache.accumulo.core.rpc;
+
+import java.util.Objects;
+
+import org.apache.thrift.TException;
+import org.apache.thrift.TMultiplexedProcessor;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TMessage;
+import org.apache.thrift.protocol.TMessageType;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.protocol.TProtocolDecorator;
+import org.apache.thrift.protocol.TProtocolException;
+
+/**
+ * AccumuloTMultiplexedProcessor is a {@link TMultiplexedProcessor} variant that uses the
+ * {@link AccumuloTMultiplexedProtocol} to register additional processors and then forward messages
+ * based on a single byte entry instead of having to create substrings of the original message.
+ */
+public class AccumuloTMultiplexedProcessor extends TMultiplexedProcessor {
+
+ // Support enough slots as RPC service types.
+ private final TProcessor[] PROCESSORS = new TProcessor[RpcService.values().length];
+
+ public void registerProcessor(RpcService service, TProcessor processor) {
+ PROCESSORS[Byte.toUnsignedInt(service.getShortId())] = Objects.requireNonNull(processor,
Review Comment:
Could also check `PROCESSORS[Byte.toUnsignedInt(service.getShortId())] == null` before setting it. Assuming that processors are registered once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3095523203
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/TabletManagementClientServiceThriftClient.java:
##########
@@ -41,15 +41,15 @@ public class TabletManagementClientServiceThriftClient extends ThriftClientTypes
LoggerFactory.getLogger(TabletManagementClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- public TabletManagementClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ public TabletManagementClientServiceThriftClient(RpcService service) {
+ super(service, new Client.Factory());
}
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
boolean preferCachedConnections) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.TABLET_MANAGEMENT);
+ warnedAboutTServersBeingDown, RpcService.TABLET_MANAGEMENT);
Review Comment:
fixed in 8a419ff
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java:
##########
@@ -36,15 +36,15 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
private static final Logger LOG = LoggerFactory.getLogger(ClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- ClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ ClientServiceThriftClient(RpcService service) {
+ super(service, new Client.Factory());
}
@Override
public Pair<String,Client> getThriftServerConnection(ClientContext context,
boolean preferCachedConnections) throws TTransportException {
return getThriftServerConnection(LOG, this, context, preferCachedConnections,
- warnedAboutTServersBeingDown, ThriftService.CLIENT);
+ warnedAboutTServersBeingDown, RpcService.CLIENT);
Review Comment:
fixed in 8a419ff
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on code in PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#discussion_r3076106929
##########
core/src/main/java/org/apache/accumulo/core/rpc/clients/ClientServiceThriftClient.java:
##########
@@ -36,15 +36,15 @@ public class ClientServiceThriftClient extends ThriftClientTypes<Client>
private static final Logger LOG = LoggerFactory.getLogger(ClientServiceThriftClient.class);
private final AtomicBoolean warnedAboutTServersBeingDown = new AtomicBoolean(false);
- ClientServiceThriftClient(String serviceName) {
- super(serviceName, new Client.Factory());
+ ClientServiceThriftClient(RpcService service) {
Review Comment:
Maybe it could change to
```java
ClientServiceThriftClient() {
super(RpcService.CLIENT, new Client.Factory());
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ctubbsii (via GitHub)" <gi...@apache.org>.
ctubbsii commented on PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#issuecomment-4930762668
It looks like this PR got a bit out of date while waiting for review. I'll look at it after it's rebased
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "keith-turner (via GitHub)" <gi...@apache.org>.
keith-turner commented on PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#issuecomment-4263021187
> Based on these benchmarks, the new processor is significantly faster. However the real benefit is the reduction of GC pressure by creating less, short-lived objects for each TMessage received.
This is neat stuff, seems like its doing significantly less work per message.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Adds rpc services and refactors ThriftClientTypes [accumulo]
Posted by "ddanielr (via GitHub)" <gi...@apache.org>.
ddanielr commented on PR #6311:
URL: https://github.com/apache/accumulo/pull/6311#issuecomment-4262913558
> I feel like the changes in [d41bfa7](https://github.com/apache/accumulo/commit/d41bfa79cc5d02b9b179a3b299d42a6215b007a1) might be overkill. My previous comments suggested that we keep the string short because it's passed in every message and used for directing traffic on the server side. The current values are several characters and the initial changes in this PR seemed like they could have made them longer. Going to a single byte might be the best in terms of performance, but is it much different than using a single character string? If we used a single character string, then we wouldn't need the new custom Multiplexer classes.
I had another set of changes that reduced the RpcService Enum name lengths down to 5 characters but when I was looking at why we were using strings I examined the multiplexer code and found that it was creating multiple string objects per message that were immediately discarded and could be putting more pressure on GC.
Thrift's Multiplexing Overview
-----
The current TMultiplexedProcessor uses a multiplexer protocol (TMultiplexedProtocol) to craft custom messages that identify which Tprocessor should receive the message.
#### TMultiplexedProtocol message creation.
The protocol takes in a string on initialization and assigns it to SERVICE_NAME.
When the protocol has to write a Tmessage, if the TMessage type is CALL or ONEWAY it prepends that SERVICE_NAME string and a separator character onto every TMessage.
```
public void writeMessageBegin(TMessage tMessage) throws TException {
if (tMessage.type == TMessageType.CALL || tMessage.type == TMessageType.ONEWAY) {
super.writeMessageBegin(
new TMessage(SERVICE_NAME + SEPARATOR + tMessage.name, tMessage.type, tMessage.seqid));
} else {
super.writeMessageBegin(tMessage);
}
}
```
The separator character is needed to allow for strings of various lengths to exist.
#### TMultiplexedProcessor message processing
The TMultiplexedProcessor takes in strings and processors and stores them in a HashMap with Strings used as keys.
For each TMessage the processor needs to process, it has to find the index of the separator character.
`int index = message.name.indexOf(TMultiplexedProtocol.SEPARATOR);`
Then a new string is created from the message and used as the lookup key in the HashMap.
` String serviceName = message.name.substring(0, index);`
` TProcessor actualProcessor = SERVICE_PROCESSOR_MAP.get(serviceName);`
Once the processor is found, a new TMessage object is created with the processor service name removed.
```
// Create a new TMessage, removing the service name
TMessage standardMessage =
new TMessage(
message.name.substring(serviceName.length() + TMultiplexedProtocol.SEPARATOR.length()),
message.type,
message.seqid);
```
And that new message is finally set to the specific processor
`actualProcessor.process(new StoredMessageProtocol(iprot, standardMessage), oprot);`
New Accumulo Modified Multiplexing
-----
The first major change is using a single byte as the processor identifier instead of String values.
By not using Strings we can use an array for registering the processors. We can also be precise with the array size.
`private final TProcessor[] PROCESSORS = new TProcessor[RpcService.values().length];`
Using bytes also removes the need for a "indexOf" operation on the processor side since thrift supports a `readByte` operation on the TProtocol.
https://github.com/apache/accumulo/blob/8a419ff87b99bec29967d37e4104291936b77850/core/src/main/java/org/apache/accumulo/core/rpc/AccumuloTMultiplexedProcessor.java#L107-L108
This also trades the String.hash call for a Byte.toUnsignedInt call.
The final change was writing the shortId after the writeMessageBegin call so the original message can be passed onto the correct processor.
This removes the need to create a new TMessage based on a substring of the original message.
This was similar to the work done in the AccumuloProtocol with the ClientHeaders and [AccumuloProtocolFactory.readAndValidateHeader()](https://github.com/apache/accumulo/blob/d41bfa79cc5d02b9b179a3b299d42a6215b007a1/core/src/main/java/org/apache/accumulo/core/rpc/AccumuloProtocolFactory.java#L120)
I tested this by writing some jmh tests to see how efficient the new processor was compared to the original thrift processor.
I also wrote test cases that only tested the routing cost to understand if the GC benefits were there.
https://gist.github.com/ddanielr/9b2c65d03ffdae4c098a2fdea41acd2d
Based on these benchmarks, the new processor is significantly faster. However the real benefit is the reduction of GC pressure by creating less, short-lived objects for each TMessage received.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org